Added K_CE_OVERRIDE to cloudevents plugin.

Author: pantierra

eoapi_notifier/outputs/cloudevents.py

@@ -5,6 +5,7 @@
 Supports standard CloudEvents environment variables and KNative SinkBinding.
 """
 
+import json
 import os
 from typing import Any
 from uuid import uuid4
@@ -28,6 +29,7 @@ class CloudEventsConfig(BasePluginConfig):
     timeout: float = 30.0
     max_retries: int = 3
     retry_backoff: float = 1.0
+    overrides: dict[str, str] = {}
 
     @field_validator("endpoint")
     @classmethod
@@ -42,6 +44,10 @@ def apply_knative_overrides(self) -> "CloudEventsConfig":
         if k_sink := os.getenv("K_SINK"):
             self.endpoint = k_sink
 
+        if k_ce_overrides := os.getenv("K_CE_OVERRIDES"):
+            overrides_data = json.loads(k_ce_overrides)
+            self.overrides = overrides_data.get("extensions", {})
+
         return self
 
     @classmethod
@@ -197,6 +203,17 @@ 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_extensions = {}
+        if k_ce_overrides := os.getenv("K_CE_OVERRIDES"):
+            try:
+                overrides_data = json.loads(k_ce_overrides)
+                ce_extensions = overrides_data.get("extensions", {})
+            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())
@@ -217,6 +234,10 @@ def _convert_to_cloudevent(self, event: NotificationEvent) -> CloudEvent:
         if event.collection:
             attributes["collection"] = event.collection
 
+        # Apply KNative CE extension overrides
+        for key, value in ce_extensions.items():
+            attributes[key] = str(value)
+
         # Event data payload
         data = {
             "id": event.id,

helm-chart/eoapi-notifier/values.yaml

@@ -78,6 +78,7 @@ secrets:
 # KNative Support:
 # The cloudevents plugin supports K_SINK variables for KNative SinkBinding:
 # - K_SINK: Overrides CLOUDEVENTS_ENDPOINT (automatically set by SinkBinding)
+# - K_CE_OVERRIDE: A JSON object that specifies overrides to the outbound event.
 env: {}
   # Examples - Standard environment variables:
   # PGSTAC_HOST: postgresql-service

tests/test_cloudevents_output.py

@@ -53,6 +53,29 @@ def test_connection_info(self) -> None:
         config = CloudEventsConfig(endpoint="https://example.com/webhook")
         assert "POST https://example.com/webhook" in config.get_connection_info()
 
+    @patch.dict(
+        os.environ,
+        {"K_CE_OVERRIDES": '{"extensions": {"extra": "test", "num": 42}}'},
+    )
+    def test_k_ce_overrides_valid(self) -> None:
+        """Test K_CE_OVERRIDES with valid extensions."""
+        config = CloudEventsConfig()
+        assert config.overrides == {"extra": "test", "num": 42}
+
+    @patch.dict(os.environ, {"K_CE_OVERRIDES": '{"other": "field"}'})
+    def test_k_ce_overrides_no_extensions(self) -> None:
+        """Test K_CE_OVERRIDES without extensions field."""
+        config = CloudEventsConfig()
+        assert config.overrides == {}
+
+    @patch.dict(os.environ, {"K_CE_OVERRIDES": "invalid-json"})
+    def test_k_ce_overrides_invalid_json(self) -> None:
+        """Test K_CE_OVERRIDES with invalid JSON."""
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            CloudEventsConfig()
+
 
 class TestCloudEventsAdapter:
     """Test CloudEvents output adapter."""
@@ -229,3 +252,52 @@ async def test_health_check(self, adapter: CloudEventsAdapter) -> None:
         # Running with client
         adapter._client = MagicMock()
         assert await adapter.health_check() is True
+
+    @patch.dict(
+        os.environ,
+        {
+            "K_CE_OVERRIDES": (
+                '{"extensions": {"extra": "test-value", "priority": "high"}}'
+            )
+        },
+    )
+    def test_convert_to_cloudevent_with_overrides(
+        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+    ) -> None:
+        """Test CloudEvent conversion with K_CE_OVERRIDES."""
+        cloud_event = adapter._convert_to_cloudevent(sample_event)
+
+        assert isinstance(cloud_event, CloudEvent)
+        assert cloud_event["extra"] == "test-value"
+        assert cloud_event["priority"] == "high"
+
+    @patch.dict(os.environ, {"K_CE_OVERRIDES": '{"extensions": {"number": 123}}'})
+    def test_convert_to_cloudevent_with_number_override(
+        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+    ) -> None:
+        """Test CloudEvent conversion with number in K_CE_OVERRIDES."""
+        cloud_event = adapter._convert_to_cloudevent(sample_event)
+
+        assert cloud_event["number"] == "123"  # Should be converted to string
+
+    @patch.dict(os.environ, {"K_CE_OVERRIDES": "invalid-json"})
+    def test_convert_to_cloudevent_invalid_overrides(
+        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+    ) -> None:
+        """Test CloudEvent conversion with invalid K_CE_OVERRIDES JSON."""
+        cloud_event = adapter._convert_to_cloudevent(sample_event)
+
+        # Should work normally without overrides
+        assert isinstance(cloud_event, CloudEvent)
+        assert cloud_event["source"] == "/eoapi/stac"
+
+    @patch.dict(os.environ, {"K_CE_OVERRIDES": '{"other": "field"}'})
+    def test_convert_to_cloudevent_no_extensions(
+        self, adapter: CloudEventsAdapter, sample_event: NotificationEvent
+    ) -> None:
+        """Test CloudEvent conversion with K_CE_OVERRIDES but no extensions field."""
+        cloud_event = adapter._convert_to_cloudevent(sample_event)
+
+        # Should work normally without extensions
+        assert isinstance(cloud_event, CloudEvent)
+        assert cloud_event["source"] == "/eoapi/stac"