diff --git a/kpops/utils/environment.py b/kpops/utils/environment.py index 0ed7ae920..fdabf1703 100644 --- a/kpops/utils/environment.py +++ b/kpops/utils/environment.py @@ -1,36 +1,42 @@ import os -import platform from collections import UserDict -from collections.abc import Callable +from collections.abc import ItemsView, KeysView, MutableMapping, ValuesView -class Environment(UserDict): - def __init__(self, mapping=None, /, **kwargs) -> None: - transformation = Environment.__get_transformation() - if mapping is not None: - mapping = {transformation(key): value for key, value in mapping.items()} - else: +class Environment(UserDict[str, str]): + """Internal environment wrapping OS environment.""" + + def __init__( + self, mapping: MutableMapping[str, str] | None = None, /, **kwargs: str + ) -> None: + self._global = os.environ + if mapping is None: mapping = {} if kwargs: - mapping.update( - {transformation(key): value for key, value in kwargs.items()} - ) + mapping.update(**kwargs) super().__init__(mapping) - @staticmethod - def __key_camel_case_transform(key: str) -> str: - return key.lower() + def __getitem__(self, key: str) -> str: + try: + return self.data[key] + except KeyError: + return self._global[key] + + def __contains__(self, key: object) -> bool: + return super().__contains__(key) or self._global.__contains__(key) + + @property + def _dict(self) -> dict[str, str]: + return {**self._global, **self.data} + + def keys(self) -> KeysView[str]: + return KeysView(self._dict) - @staticmethod - def __key_identity_transform(key: str) -> str: - return key + def values(self) -> ValuesView[str]: + return ValuesView(self._dict) - @staticmethod - def __get_transformation() -> Callable[[str], str]: - if platform.system() == "Windows": - return Environment.__key_camel_case_transform - else: - return Environment.__key_identity_transform + def items(self) -> ItemsView[str, str]: + return ItemsView(self._dict) -ENV = Environment(os.environ) +ENV = Environment() diff --git a/tests/pipeline/resources/pipeline-with-env-defaults/defaults.yaml b/tests/pipeline/resources/pipeline-with-env-defaults/defaults.yaml index 639c176e5..8faa86ad5 100644 --- a/tests/pipeline/resources/pipeline-with-env-defaults/defaults.yaml +++ b/tests/pipeline/resources/pipeline-with-env-defaults/defaults.yaml @@ -34,6 +34,5 @@ kafka-connector: key.ignore: "false" linger.ms: "5000" max.buffered.records: "20000" - name: "sink-connector" read.timeout.ms: "120000" tasks.max: "1" diff --git a/tests/utils/test_environment.py b/tests/utils/test_environment.py index 8fc02c826..d1dcb8f5e 100644 --- a/tests/utils/test_environment.py +++ b/tests/utils/test_environment.py @@ -1,117 +1,81 @@ -from unittest.mock import patch +import os +from collections.abc import ItemsView, KeysView, ValuesView +from unittest.mock import ANY import pytest from kpops.utils.environment import Environment -@pytest.fixture() -def fake_environment_windows(): - return {"MY": "fake", "ENVIRONMENT": "here"} +@pytest.fixture(autouse=True) +def environment(monkeypatch: pytest.MonkeyPatch) -> Environment: + for key in os.environ: + monkeypatch.delenv(key) + monkeypatch.setenv("MY", "fake") + monkeypatch.setenv("ENVIRONMENT", "here") + return Environment() -@pytest.fixture() -def fake_environment_linux(): - return {"my": "fake", "environment": "here"} +def test_get_item(environment: Environment): + assert environment["MY"] == "fake" + assert environment["ENVIRONMENT"] == "here" -@patch("platform.system") -def test_normal_behaviour_get_item(system, fake_environment_linux): - system.return_value = "Linux" - environment = Environment(fake_environment_linux) - - assert environment["my"] == "fake" - assert environment["environment"] == "here" - - -@patch("platform.system") -def test_normal_behaviour_get_item_as_kwargs(system, fake_environment_linux): - system.return_value = "Linux" - environment = Environment(**fake_environment_linux) - - assert environment["my"] == "fake" - assert environment["environment"] == "here" - - -@patch("platform.system") -def test_normal_behaviour_keys_transformation(system, fake_environment_linux): - system.return_value = "Linux" - environment = Environment(fake_environment_linux) - keys = set(environment.keys()) - - assert "my" in keys - assert "environment" in keys - - -@patch("platform.system") -def test_normal_behaviour_set_key(system, fake_environment_linux): - system.return_value = "Linux" - environment = Environment(fake_environment_linux) +def test_set_item(environment: Environment): environment["extra"] = "key" keys = set(environment.keys()) - assert "my" in keys - assert "environment" in keys + assert "MY" in keys + assert "ENVIRONMENT" in keys assert "extra" in keys assert environment["extra"] == "key" -@patch("platform.system") -def test_windows_behaviour_set_key(system, fake_environment_windows): - system.return_value = "Windows" - environment = Environment(fake_environment_windows) - environment["extra"] = "key" - - keys = set(environment.keys()) - assert "my" in keys - assert "environment" in keys - assert "extra" in keys - assert environment["extra"] == "key" - - -@patch("platform.system") -def test_normal_behaviour_keys_transformation_kwargs(system, fake_environment_linux): - system.return_value = "Linux" - environment = Environment(**fake_environment_linux) - - keys = set(environment.keys()) - assert "my" in keys - assert "environment" in keys - - -@patch("platform.system") -def test_windows_behaviour_keys_transformation(system, fake_environment_windows): - system.return_value = "Windows" - environment = Environment(fake_environment_windows) - - keys = set(environment.keys()) - assert "my" in keys - assert "environment" in keys - - -@patch("platform.system") -def test_windows_behaviour_keys_transformation_as_kwargs( - system, fake_environment_windows -): - system.return_value = "Windows" - environment = Environment(**fake_environment_windows) - keys = set(environment.keys()) - assert "my" in keys - assert "environment" in keys - - -@patch("platform.system") -def test_windows_behaviour_get_item(system, fake_environment_windows): - system.return_value = "Windows" - - environment = Environment(fake_environment_windows) - assert environment["my"] == "fake" - assert environment["environment"] == "here" - - -@patch("platform.system") -def test_windows_behaviour_get_item_as_kwargs(system, fake_environment_windows): - system.return_value = "Windows" - environment = Environment(**fake_environment_windows) - assert environment["my"] == "fake" - assert environment["environment"] == "here" +def test_update_os_environ(environment: Environment): + with pytest.raises(KeyError): + environment["TEST"] + os.environ["TEST"] = "test" + assert "TEST" in environment + assert environment["TEST"] == "test" + keys = environment.keys() + assert isinstance(keys, KeysView) + assert "TEST" in keys + values = environment.values() + assert isinstance(values, ValuesView) + assert "test" in values + items = environment.items() + assert isinstance(items, ItemsView) + d = dict(items) + assert d["TEST"] == "test" + + +def test_mapping(): + environment = Environment({"kwarg1": "value1", "kwarg2": "value2"}) + assert environment["MY"] == "fake" + assert environment["ENVIRONMENT"] == "here" + assert environment["kwarg1"] == "value1" + assert environment["kwarg2"] == "value2" + + +def test_kwargs(): + environment = Environment(kwarg1="value1", kwarg2="value2") + assert environment["MY"] == "fake" + assert environment["ENVIRONMENT"] == "here" + assert environment["kwarg1"] == "value1" + assert environment["kwarg2"] == "value2" + + +def test_dict(environment: Environment): + assert environment._dict == { + "MY": "fake", + "ENVIRONMENT": "here", + "PYTEST_CURRENT_TEST": ANY, + } + + +def test_dict_unpacking(environment: Environment): + assert {**environment} == { + "MY": "fake", + "ENVIRONMENT": "here", + "PYTEST_CURRENT_TEST": ANY, + }