Update files

kpops/cli/pipeline_config.py

@@ -1,11 +1,18 @@
+from __future__ import annotations
+
 from pathlib import Path
+from typing import TYPE_CHECKING, Any
 
 from pydantic import AnyHttpUrl, BaseConfig, BaseSettings, Field, parse_obj_as
-from pydantic.env_settings import SettingsSourceCallable
 
 from kpops.component_handlers.helm_wrapper.model import HelmConfig, HelmDiffConfig
 from kpops.utils.yaml_loading import load_yaml_file
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from pydantic.env_settings import SettingsSourceCallable
+
 ENV_PREFIX = "KPOPS_"
 
 
@@ -40,7 +47,7 @@ class PipelineConfig(BaseSettings):
     """Pipeline configuration unrelated to the components."""
 
     defaults_path: Path = Field(
-        default=Path("."),
+        default=Path(),
         example="defaults",
         description="The path to the folder containing the defaults.yaml file and the environment defaults files. "
         "Paths can either be absolute or relative to `config.yaml`",
@@ -116,17 +123,22 @@ def customise_sources(
             init_settings: SettingsSourceCallable,
             env_settings: SettingsSourceCallable,
             file_secret_settings: SettingsSourceCallable,
-        ):
+        ) -> tuple[
+            SettingsSourceCallable | Callable[[PipelineConfig], dict[str, Any]], ...
+        ]:
             return (
+                env_settings,
                 init_settings,
                 yaml_config_settings_source,
-                env_settings,
                 file_secret_settings,
             )
 
 
-def yaml_config_settings_source(settings: PipelineConfig) -> dict | list:
+def yaml_config_settings_source(settings: PipelineConfig) -> dict[str, Any]:
     path_to_config = settings.Config.config_path
     if path_to_config.exists():
-        return load_yaml_file(path_to_config)
+        if isinstance(source := load_yaml_file(path_to_config), dict):
+            return source
+        err_msg = f"{path_to_config} must be a mapping."
+        raise TypeError(err_msg)
     return {}

tests/pipeline/test_pipeline.py

@@ -3,6 +3,7 @@
 
 import pytest
 import yaml
+from pytest import MonkeyPatch
 from snapshottest.module import SnapshotTest
 from typer.testing import CliRunner
 
@@ -458,8 +459,34 @@ def test_default_config(self, snapshot: SnapshotTest):
 
         snapshot.assert_match(enriched_pipeline, "test-pipeline")
 
+    def test_env_vars_precedence_over_config(
+        self,
+        monkeypatch: MonkeyPatch,
+        snapshot: SnapshotTest,
+    ):
+        monkeypatch.setenv(name="KPOPS_KAFKA_BROKERS", value="env_broker")
+
+        result = runner.invoke(
+            app,
+            [
+                "generate",
+                "--pipeline-base-dir",
+                str(PIPELINE_BASE_DIR_PATH),
+                str(RESOURCE_PATH / "custom-config/pipeline.yaml"),
+                "--config",
+                str(RESOURCE_PATH / "custom-config/config.yaml"),
+            ],
+            catch_exceptions=False,
+        )
+        assert result.exit_code == 0
+        enriched_pipeline: dict = yaml.safe_load(result.stdout)
+        assert (
+            enriched_pipeline["components"][0]["app"]["streams"]["brokers"]
+            == "env_broker"
+        )
+
     def test_model_serialization(self, snapshot: SnapshotTest):
-        """Test model serialization of component containing pathlib.Path attribute"""
+        """Test model serialization of component containing pathlib.Path attribute."""
         result = runner.invoke(
             app,
             [