Removed unecessary of cloudevents env variable overwriting.

Author: pantierra

eoapi_notifier/outputs/cloudevents.py

@@ -5,7 +5,6 @@
 Supports standard CloudEvents environment variables and KNative SinkBinding.
 """
 
-import json
 import os
 from typing import Any
 from uuid import uuid4
@@ -40,18 +39,9 @@ def validate_endpoint(cls, v: str | None) -> str | None:
     @model_validator(mode="after")
     def apply_knative_overrides(self) -> "CloudEventsConfig":
         """Apply KNative SinkBinding environment variables as special case."""
-        # K_SINK overrides endpoint (KNative SinkBinding)
         if k_sink := os.getenv("K_SINK"):
             self.endpoint = k_sink
 
-        # K_SOURCE overrides source
-        if k_source := os.getenv("K_SOURCE"):
-            self.source = k_source
-
-        # K_TYPE overrides event_type
-        if k_type := os.getenv("K_TYPE"):
-            self.event_type = k_type
-
         return self
 
     @classmethod
@@ -207,16 +197,6 @@ def _convert_to_cloudevent(self, event: NotificationEvent) -> CloudEvent:
         source = self.config.source
         event_type_base = self.config.event_type
 
-        # Apply KNative CE overrides if present
-        ce_overrides = {}
-        if k_ce_overrides := os.getenv("K_CE_OVERRIDES"):
-            try:
-                ce_overrides = json.loads(k_ce_overrides)
-            except json.JSONDecodeError:
-                self.logger.warning(
-                    "Invalid K_CE_OVERRIDES JSON, ignoring: %s", k_ce_overrides
-                )
-
         # Map operation to event type suffix
         operation_map = {"INSERT": "created", "UPDATE": "updated", "DELETE": "deleted"}
         operation = operation_map.get(event.operation.upper(), event.operation.lower())
@@ -237,9 +217,6 @@ def _convert_to_cloudevent(self, event: NotificationEvent) -> CloudEvent:
         if event.collection:
             attributes["collection"] = event.collection
 
-        # Apply KNative CE overrides
-        attributes.update(ce_overrides)
-
         # Event data payload
         data = {
             "id": event.id,

examples/config.yaml

@@ -55,8 +55,8 @@ outputs:
       endpoint: https://example.com/webhook  # CLOUDEVENTS_ENDPOINT or K_SINK
 
       # Optional: CloudEventattributes
-      # source: "/eoapi/stac"               # CLOUDEVENTS_SOURCE or K_SOURCE
-      # event_type: "org.eoapi.stac"        # CLOUDEVENTS_EVENT_TYPE or K_TYPE
+      # source: "/eoapi/stac"               # CLOUDEVENTS_SOURCE
+      # event_type: "org.eoapi.stac"        # CLOUDEVENTS_EVENT_TYPE
 
       # Optional: HTTP settings
       # timeout: 30.0                       # CLOUDEVENTS_TIMEOUT

helm-chart/eoapi-notifier/values.yaml

@@ -78,24 +78,6 @@ secrets:
 # KNative Support:
 # The cloudevents plugin supports K_SINK variables for KNative SinkBinding:
 # - K_SINK: Overrides CLOUDEVENTS_ENDPOINT (automatically set by SinkBinding)
-# - K_SOURCE: Overrides CLOUDEVENTS_SOURCE
-# - K_TYPE: Overrides CLOUDEVENTS_EVENT_TYPE
-#
-# For KNative integration, use SinkBinding to automatically inject K_SINK:
-#   apiVersion: sources.knative.dev/v1beta1
-#   kind: SinkBinding
-#   metadata:
-#     name: eoapi-notifier-binding
-#   spec:
-#     subject:
-#       apiVersion: apps/v1
-#       kind: Deployment
-#       name: eoapi-notifier
-#     sink:
-#       ref:
-#         apiVersion: serving.knative.dev/v1
-#         kind: Service
-#         name: my-knative-service
 env: {}
   # Examples - Standard environment variables:
   # PGSTAC_HOST: postgresql-service
@@ -111,5 +93,3 @@ env: {}
   #
   # KNative examples (typically set by SinkBinding):
   # K_SINK: https://my-knative-service.default.svc.cluster.local
-  # K_SOURCE: /eoapi/stac/pgstac
-  # K_TYPE: org.eoapi.stac

tests/test_cloudevents_output.py

@@ -198,23 +198,6 @@ def test_convert_to_cloudevent(
         assert cloud_event["subject"] == "test-item"
         assert cloud_event["collection"] == "test-collection"
 
-    @patch.dict(
-        os.environ,
-        {
-            "K_SOURCE": "/custom/source",
-            "K_TYPE": "custom.type",
-        },
-    )
-    def test_convert_with_env_vars(self, sample_event: NotificationEvent) -> None:
-        """Test CloudEvent conversion with environment variables."""
-        # Create a new config and adapter after setting environment variables
-        config = CloudEventsConfig()
-        adapter = CloudEventsAdapter(config)
-        cloud_event = adapter._convert_to_cloudevent(sample_event)
-
-        assert cloud_event["source"] == "/custom/source"
-        assert cloud_event["type"] == "custom.type.created"
-
     def test_operation_mapping(self, adapter: CloudEventsAdapter) -> None:
         """Test operation to event type mapping."""
         test_cases = [