Enable transparent OS environment lookups from internal environment (#…

…368)

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()

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"

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,
+    }