Save progress

bakdata · Oct 23, 2023 · 542c16a · 542c16a
2 parents 0ce9044 + 352cf9c
commit 542c16a
Show file tree

Hide file tree

Showing 11 changed files with 70 additions and 31 deletions.
diff --git a/examples/bakdata/atm-fraud-detection/config.yaml b/examples/bakdata/atm-fraud-detection/config.yaml
@@ -4,9 +4,9 @@ topic_name_config:
   default_error_topic_name: "${pipeline_name}-${component_name}-dead-letter-topic"
   default_output_topic_name: "${pipeline_name}-${component_name}-topic"
 
-brokers: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
+brokers: "http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092"
 
-schema_registry_url: "http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081"
+schema_registry_url: "http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081"
 
 kafka_rest_host: "http://localhost:8082"
 

diff --git a/examples/bakdata/atm-fraud-detection/pipeline.yaml b/examples/bakdata/atm-fraud-detection/pipeline.yaml
@@ -84,14 +84,14 @@
     connector.class: io.confluent.connect.jdbc.JdbcSinkConnector
     tasks.max: 1
     topics: ${pipeline_name}-account-linker-topic
-    connection.url: jdbc:postgresql://postgresql-dev.kpops.svc.cluster.local:5432/app_db
+    connection.url: jdbc:postgresql://postgresql-dev.${NAMESPACE}.svc.cluster.local:5432/app_db
     connection.user: app1
     connection.password: AppPassword
     connection.ds.pool.size: 5
     insert.mode: insert
     insert.mode.databaselevel: true
     value.converter: io.confluent.connect.avro.AvroConverter
-    value.converter.schema.registry.url: http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081
+    value.converter.schema.registry.url: http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081
     key.converter: org.apache.kafka.connect.storage.StringConverter
     transforms: flatten
     transforms.flatten.type: org.apache.kafka.connect.transforms.Flatten$Value

diff --git a/kpops/cli/pipeline_config.py b/kpops/cli/pipeline_config.py
@@ -1,18 +1,15 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from pydantic import BaseConfig, BaseSettings, Field
+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_"
 
 

diff --git a/kpops/component_handlers/helm_wrapper/helm.py b/kpops/component_handlers/helm_wrapper/helm.py
@@ -212,7 +212,7 @@ def __execute(self, command: list[str]) -> str:
         log.debug(f"Executing {' '.join(command)}")
         process = subprocess.run(
             command,
-            check=True,
+            check=False,
             capture_output=True,
             text=True,
         )

diff --git a/kpops/components/base_components/kafka_connector.py b/kpops/components/base_components/kafka_connector.py
@@ -3,7 +3,7 @@
 import logging
 from abc import ABC
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, NoReturn
+from typing import Any, NoReturn
 
 from pydantic import Field, validator
 from typing_extensions import override
@@ -25,13 +25,11 @@
     KafkaConnectResetterValues,
 )
 from kpops.components.base_components.base_defaults_component import deduplicate
+from kpops.components.base_components.models.from_section import FromTopic
 from kpops.components.base_components.pipeline_component import PipelineComponent
 from kpops.utils.colorify import magentaify
 from kpops.utils.docstring import describe_attr
 
-if TYPE_CHECKING:
-    from kpops.components.base_components.models.from_section import FromTopic
-
 log = logging.getLogger("KafkaConnector")
 
 

diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -47,7 +47,7 @@ pytest-mock = "^3.10.0"
 pytest-timeout = "^2.1.0"
 snapshottest = "^0.6.0"
 pre-commit = "^2.19.0"
-ruff = "^0.0.292"
+ruff = "^0.1.1"
 black = "^23.7.0"
 typer-cli = "^0.0.13"
 pyright = "^1.1.314"
@@ -140,7 +140,6 @@ select = [
   "RET", # flake8-return
   "SLOT", # flake8-slots
   "SIM", # flake8-simplify
-  "TCH", # flake8-type-checking, configure correctly and add
   "PTH", # flake8-use-pathlib
   "PGH", # pygrep-hooks
   "PL", # Pylint

diff --git a/tests/conftest.py b/tests/conftest.py
@@ -0,0 +1,16 @@
+import os
+from collections.abc import Iterator
+from unittest import mock
+
+import pytest
+
+
+@pytest.fixture()
+def mock_env() -> Iterator[os._Environ[str]]:
+    """Clear ``os.environ``.
+
+    :yield: ``os.environ``. Prevents the function and the mock
+        context from exiting.
+    """
+    with mock.patch.dict(os.environ, clear=True):
+        yield os.environ
diff --git a/tests/pipeline/snapshots/snap_test_example.py b/tests/pipeline/snapshots/snap_test_example.py
@@ -23,12 +23,12 @@
                 'replicaCount': 1,
                 'schedule': '0 12 * * *',
                 'streams': {
-                    'brokers': 'http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092',
+                    'brokers': 'http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092',
                     'extraOutputTopics': {
                     },
                     'optimizeLeaveGroupBehavior': False,
                     'outputTopic': 'bakdata-atm-fraud-detection-account-producer-topic',
-                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
                 },
                 'suspend': True
             },
@@ -74,12 +74,12 @@
                 'replicaCount': 1,
                 'schedule': '0 12 * * *',
                 'streams': {
-                    'brokers': 'http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092',
+                    'brokers': 'http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092',
                     'extraOutputTopics': {
                     },
                     'optimizeLeaveGroupBehavior': False,
                     'outputTopic': 'bakdata-atm-fraud-detection-transaction-avro-producer-topic',
-                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
                 },
                 'suspend': True
             },
@@ -129,14 +129,14 @@
                 },
                 'replicaCount': 1,
                 'streams': {
-                    'brokers': 'http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092',
+                    'brokers': 'http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092',
                     'errorTopic': 'bakdata-atm-fraud-detection-transaction-joiner-dead-letter-topic',
                     'inputTopics': [
                         'bakdata-atm-fraud-detection-transaction-avro-producer-topic'
                     ],
                     'optimizeLeaveGroupBehavior': False,
                     'outputTopic': 'bakdata-atm-fraud-detection-transaction-joiner-topic',
-                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
                 }
             },
             'name': 'transaction-joiner',
@@ -191,14 +191,14 @@
                 },
                 'replicaCount': 1,
                 'streams': {
-                    'brokers': 'http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092',
+                    'brokers': 'http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092',
                     'errorTopic': 'bakdata-atm-fraud-detection-fraud-detector-dead-letter-topic',
                     'inputTopics': [
                         'bakdata-atm-fraud-detection-transaction-joiner-topic'
                     ],
                     'optimizeLeaveGroupBehavior': False,
                     'outputTopic': 'bakdata-atm-fraud-detection-fraud-detector-topic',
-                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
                 }
             },
             'name': 'fraud-detector',
@@ -253,7 +253,7 @@
                 },
                 'replicaCount': 1,
                 'streams': {
-                    'brokers': 'http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092',
+                    'brokers': 'http://k8kafka-cp-kafka-headless.${NAMESPACE}.svc.cluster.local:9092',
                     'errorTopic': 'bakdata-atm-fraud-detection-account-linker-dead-letter-topic',
                     'extraInputTopics': {
                         'accounts': [
@@ -265,7 +265,7 @@
                     ],
                     'optimizeLeaveGroupBehavior': False,
                     'outputTopic': 'bakdata-atm-fraud-detection-account-linker-topic',
-                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                    'schemaRegistryUrl': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
                 }
             },
             'from': {
@@ -315,7 +315,7 @@
                 'auto.create': True,
                 'connection.ds.pool.size': 5,
                 'connection.password': 'AppPassword',
-                'connection.url': 'jdbc:postgresql://postgresql-dev.kpops.svc.cluster.local:5432/app_db',
+                'connection.url': 'jdbc:postgresql://postgresql-dev.${NAMESPACE}.svc.cluster.local:5432/app_db',
                 'connection.user': 'app1',
                 'connector.class': 'io.confluent.connect.jdbc.JdbcSinkConnector',
                 'errors.deadletterqueue.context.headers.enable': True,
@@ -333,7 +333,7 @@
                 'transforms': 'flatten',
                 'transforms.flatten.type': 'org.apache.kafka.connect.transforms.Flatten$Value',
                 'value.converter': 'io.confluent.connect.avro.AvroConverter',
-                'value.converter.schema.registry.url': 'http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081'
+                'value.converter.schema.registry.url': 'http://k8kafka-cp-schema-registry.${NAMESPACE}.svc.cluster.local:8081'
             },
             'name': 'postgresql-connector',
             'namespace': '${NAMESPACE}',

diff --git a/tests/pipeline/test_example.py b/tests/pipeline/test_example.py
@@ -1,3 +1,4 @@
+import pytest
 import yaml
 from snapshottest.module import SnapshotTest
 from typer.testing import CliRunner
@@ -7,6 +8,7 @@
 runner = CliRunner()
 
 
+@pytest.mark.usefixtures("mock_env")
 class TestExample:
     def test_atm_fraud(self, snapshot: SnapshotTest):
         result = runner.invoke(

diff --git a/tests/pipeline/test_pipeline.py b/tests/pipeline/test_pipeline.py
@@ -16,6 +16,7 @@
 PIPELINE_BASE_DIR_PATH = RESOURCE_PATH.parent
 
 
+@pytest.mark.usefixtures("mock_env")
 class TestPipeline:
     def test_python_api(self):
         pipeline = kpops.generate(