diff --git a/DEVELOPER.md b/DEVELOPER.md index 0eeff70..312c885 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -164,6 +164,12 @@ EventGate uses a **direct invocation approach** for integration testing: - Docker running (Docker Desktop on macOS/Windows, or Docker Engine on Linux) - Python 3.13 with dependencies installed +When using colima on macOS, set the following environment variables before running tests: +```shell +export DOCKER_HOST=unix://$HOME/.colima/default/docker.sock +export TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock +``` + ### Run Integration Tests Containers start and stop automatically: diff --git a/adr/001-status-change-db/001-status-change-db.md b/adr/001-status-change-db/001-status-change-db.md new file mode 100644 index 0000000..42e1f8d --- /dev/null +++ b/adr/001-status-change-db/001-status-change-db.md @@ -0,0 +1,185 @@ +# ADR 001. Event Bus -- Status change database + +## Decision +Aggregate events and insert into newly created database table. + +## Background +The status change topic receives job status events, i.e. it is an event log. +For monitoring, a database table with the current status is needed, i.e. events need to be aggregated into the latest state +So, the table stores the latest merged state per `job_id`, not the full event log. + +This ADR shall describe the database schema as well as the aggregation logic and some of the queries that the database should support. + +## Database schema + +```sql +CREATE TABLE job ( + job_id UUID PRIMARY KEY, + job_group_id UUID, + parent_job_id UUID, + initial_job_id UUID, + + -- Identity / business attributes + job_ref TEXT, + job_name TEXT, + definition_id TEXT, + definition_version TEXT, + tenant_id TEXT, + country TEXT, + source_app TEXT, + source_app_version TEXT, + environment TEXT, + + -- Execution context + platform TEXT, + platform_metadata JSONB, + input_arguments JSONB, + additional_context JSONB, + attempt_number INTEGER NOT NULL CHECK (attempt_number > 0), + + -- Current lifecycle status + status_type TEXT CHECK (status_type IN ('WAITING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'KILLED')), + status_subtype TEXT, + status_detail TEXT, + + -- Lifecycle timestamps derived from events + created_at TIMESTAMPTZ, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + last_updated_at TIMESTAMPTZ NOT NULL, + + CONSTRAINT fk_job_parent + FOREIGN KEY (parent_job_id) + REFERENCES job (job_id) + ON DELETE RESTRICT + + CONSTRAINT fk_job_initial + FOREIGN KEY (initial_job_id) + REFERENCES job (job_id) + ON DELETE RESTRICT +); +``` + +## Event insertion and aggregation + +### Assumptions +- Most events are received in order per `job_id`, but some events may be received out-of-order (during failures scenarios). The order of events is determined by the field `timestamp_event`. If an event is out-of-order, it means that an event with a higher `timestamp_event` has already been processed for the database insert / update. +- Duplicate events can occur and must be handled idempotently, i.e. duplicate events must not lead to two rows in the database table. +- The `job` table stores the latest snapshot of the job's status, not historical versions. + +### Field merge strategy +The field merge strategy is defined in the SQL upsert query. Therefore the field names of the database table rather than the Kafka event are used below. + +The field merge strategy takes into account out-of-order updates and the idempotency of duplicates. The true event order is determined by the field `last_updated_at` (same as `timestamp_event` in Kafka record). There are three merge strategies: +- Take latest non-null: If either the incoming or the current field value is non-null, then the non-null value is accepted. If both incoming and current value are non-null, then the newer value (determined by `last_updated_at`) is accepted. This means that a value that has been set, will never be set to null again. This merge strategy applies to most fields. + +- Take latest: The newer value is accepted, even if it is null. This applies to the `status_subtype` and `status_detail` fields, as they are coupled to the `status_type` field, which is mandatory for all events. If an initial status like `RUNNING` has a `status_detail` (for whatever reason), then it should not be kept when the status changes to `FINISHED`, but instead set to null. However, usually only terminal statuses should have `status_detail` and `status_subtype` + +- Take latest non-default: For fields with default values, default values are treated like null values in terms of the merge. This means that non-default values are preferred, even if they are not the latest value. Once a value different to the default has been set, it is not possible to set it back to the default value. This applies to the field `attempt_number`. + +- Cumulative merge: For `additional_context`, the merge strategy is a cumulative merge, i.e. fields from both the incoming and current record are retained. Note that this is not the case for `input_arguments` and `platform_metadata` + +### Timestamp conversion +Timestamps in Kafka events are epoch milliseconds, which should be converted to `TIMESTAMPTZ` in Postgres. Using a dedicated timestamp type greatly simplifies querying and reading from the table, especially in BI-tools. + +## Sample query patterns +**Get jobs by job name** +```sql +SELECT * +FROM job j +WHERE job_name = :my_job_name +ORDER BY last_updated_at; +``` + +**Get all jobs grouped by job group id (including retries)** +```sql +SELECT + j.*, + MIN(started_at) OVER (PARTITION BY job_group_id) AS group_started_at +FROM job j +ORDER BY + group_started_at, + job_group_id, + started_at; +``` + +**Get job hierarchy roots, including only latest attempt** +```sql +WITH latest_root_attempts AS ( + SELECT * + FROM ( + SELECT + j.*, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(initial_job_id, job_id) + ORDER BY attempt_number DESC, started_at DESC + ) AS rn + FROM job j + WHERE job_name = 'Aqueduct Ingestion' + ) t + WHERE rn = 1 +) +SELECT + * +FROM latest_root_attempts +ORDER BY + job_group_id, + started_at; +``` + +**Get all jobs in a job hierarchy given a specific job group id, including only latest attempts** +```sql +SELECT * +FROM ( + SELECT + j.*, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(initial_job_id, job_id) + ORDER BY attempt_number DESC, started_at DESC + ) AS rn + FROM job j + WHERE job_group_id = '11111111-1111-4111-8111-111111111111'::uuid + ) t +WHERE rn = 1; +``` + +**Get all jobs in all job hierarchies, including only latest attempts** +```sql +WITH RECURSIVE latest_attempts AS ( + SELECT * + FROM ( + SELECT + j.*, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(initial_job_id, job_id) + ORDER BY attempt_number DESC, started_at DESC + ) AS rn + FROM job j + ) t + WHERE rn = 1 +), +latest_root AS ( + SELECT job_id + FROM latest_attempts + WHERE parent_job_id IS NULL + ORDER BY attempt_number DESC, started_at DESC + LIMIT 1 +), +tree AS ( + -- start from latest root + SELECT la.* + FROM latest_attempts la + JOIN latest_root r ON la.job_id = r.job_id + + UNION ALL + + -- traverse hierarchy + SELECT child.* + FROM latest_attempts child + JOIN tree parent + ON child.parent_job_id = parent.job_id +) +SELECT * +FROM tree +ORDER BY started_at; +``` diff --git a/conf/access.json b/conf/access.json index 9d80b59..bf36283 100644 --- a/conf/access.json +++ b/conf/access.json @@ -10,5 +10,5 @@ }, "public.cps.za.dlchange": ["FooUser", "BarUser"], "public.cps.za.test": ["TestUser"], - "public.cps.za.status_change": ["TestUser"] + "public.cps.za.status_change": ["TestUser", "IntegrationTestUser"] } diff --git a/src/utils/constants.py b/src/utils/constants.py index 7ef26a2..fe75d5c 100644 --- a/src/utils/constants.py +++ b/src/utils/constants.py @@ -39,5 +39,5 @@ TOPIC_TEST = "public.cps.za.test" TOPIC_STATUS_CHANGE = "public.cps.za.status_change" -POSTGRES_WRITE_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS, TOPIC_DLCHANGE, TOPIC_TEST}) +POSTGRES_WRITE_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS, TOPIC_DLCHANGE, TOPIC_TEST, TOPIC_STATUS_CHANGE}) SUPPORTED_STATS_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS}) diff --git a/src/writers/sql/inserts.sql b/src/writers/sql/inserts.sql index f6db68d..ae8e7b9 100644 --- a/src/writers/sql/inserts.sql +++ b/src/writers/sql/inserts.sql @@ -35,3 +35,141 @@ INSERT INTO public_cps_za_test VALUES (:event_id, :tenant_id, :source_app, :environment, :timestamp_event, :additional_info); + +-- name: upsert_status_change(job_id, job_group_id, parent_job_id, initial_job_id, job_ref, job_name, definition_id, definition_version, tenant_id, country, source_app, source_app_version, environment, platform, platform_metadata, input_arguments, additional_context, attempt_number, status_type, status_subtype, status_detail, created_at, started_at, finished_at, last_updated_at)! +-- Upsert a status_change event into the aggregated job table using merge logic per ADR 001. +INSERT INTO public_cps_za_status_change_aggregated_job AS t ( + job_id, job_group_id, parent_job_id, initial_job_id, + job_ref, job_name, definition_id, definition_version, + tenant_id, country, source_app, source_app_version, environment, + platform, platform_metadata, input_arguments, additional_context, + attempt_number, status_type, status_subtype, status_detail, + created_at, started_at, finished_at, last_updated_at +) +VALUES ( + :job_id, :job_group_id, :parent_job_id, :initial_job_id, + :job_ref, :job_name, :definition_id, :definition_version, + :tenant_id, :country, :source_app, :source_app_version, :environment, + :platform, :platform_metadata, :input_arguments, :additional_context, + :attempt_number, :status_type, :status_subtype, :status_detail, + :created_at, :started_at, :finished_at, :last_updated_at +) +ON CONFLICT (job_id) DO UPDATE SET + job_group_id = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.job_group_id, t.job_group_id) + ELSE COALESCE(t.job_group_id, EXCLUDED.job_group_id) + END, + parent_job_id = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.parent_job_id, t.parent_job_id) + ELSE COALESCE(t.parent_job_id, EXCLUDED.parent_job_id) + END, + initial_job_id = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.initial_job_id, t.initial_job_id) + ELSE COALESCE(t.initial_job_id, EXCLUDED.initial_job_id) + END, + job_ref = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.job_ref, t.job_ref) + ELSE COALESCE(t.job_ref, EXCLUDED.job_ref) + END, + job_name = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.job_name, t.job_name) + ELSE COALESCE(t.job_name, EXCLUDED.job_name) + END, + definition_id = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.definition_id, t.definition_id) + ELSE COALESCE(t.definition_id, EXCLUDED.definition_id) + END, + definition_version = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.definition_version, t.definition_version) + ELSE COALESCE(t.definition_version, EXCLUDED.definition_version) + END, + tenant_id = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.tenant_id, t.tenant_id) + ELSE COALESCE(t.tenant_id, EXCLUDED.tenant_id) + END, + country = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.country, t.country) + ELSE COALESCE(t.country, EXCLUDED.country) + END, + source_app = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.source_app, t.source_app) + ELSE COALESCE(t.source_app, EXCLUDED.source_app) + END, + source_app_version = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.source_app_version, t.source_app_version) + ELSE COALESCE(t.source_app_version, EXCLUDED.source_app_version) + END, + environment = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.environment, t.environment) + ELSE COALESCE(t.environment, EXCLUDED.environment) + END, + platform = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.platform, t.platform) + ELSE COALESCE(t.platform, EXCLUDED.platform) + END, + platform_metadata = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.platform_metadata, t.platform_metadata) + ELSE COALESCE(t.platform_metadata, EXCLUDED.platform_metadata) + END, + input_arguments = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.input_arguments, t.input_arguments) + ELSE COALESCE(t.input_arguments, EXCLUDED.input_arguments) + END, + additional_context = CASE + WHEN EXCLUDED.additional_context IS NULL THEN t.additional_context + WHEN t.additional_context IS NULL THEN EXCLUDED.additional_context + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN t.additional_context || EXCLUDED.additional_context + ELSE EXCLUDED.additional_context || t.additional_context + END, + attempt_number = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(NULLIF(EXCLUDED.attempt_number, 1), t.attempt_number, 1) + ELSE COALESCE(NULLIF(t.attempt_number, 1), EXCLUDED.attempt_number, 1) + END, + status_type = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN EXCLUDED.status_type + ELSE t.status_type + END, + status_subtype = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN EXCLUDED.status_subtype + ELSE t.status_subtype + END, + status_detail = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN EXCLUDED.status_detail + ELSE t.status_detail + END, + created_at = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.created_at, t.created_at) + ELSE COALESCE(t.created_at, EXCLUDED.created_at) + END, + started_at = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.started_at, t.started_at) + ELSE COALESCE(t.started_at, EXCLUDED.started_at) + END, + finished_at = CASE + WHEN EXCLUDED.last_updated_at >= t.last_updated_at + THEN COALESCE(EXCLUDED.finished_at, t.finished_at) + ELSE t.finished_at + END, + last_updated_at = GREATEST(EXCLUDED.last_updated_at, t.last_updated_at); diff --git a/src/writers/writer_postgres.py b/src/writers/writer_postgres.py index 06df7c1..f3ef6e8 100644 --- a/src/writers/writer_postgres.py +++ b/src/writers/writer_postgres.py @@ -19,6 +19,7 @@ import json import logging from dataclasses import dataclass +from datetime import datetime, timezone from functools import cached_property from pathlib import Path from typing import Any @@ -30,6 +31,7 @@ REQUIRED_CONNECTION_FIELDS, TOPIC_DLCHANGE, TOPIC_RUNS, + TOPIC_STATUS_CHANGE, TOPIC_TEST, POSTGRES_WRITE_TOPICS, ) @@ -51,6 +53,7 @@ class WriterQueries: insert_run: str insert_run_job: str insert_test: str + upsert_status_change: str class WriterPostgres(Writer, PostgresBase): @@ -72,6 +75,7 @@ def _queries(self) -> WriterQueries: insert_run=queries.insert_run.sql, # pylint: disable=no-member insert_run_job=queries.insert_run_job.sql, # pylint: disable=no-member insert_test=queries.insert_test.sql, # pylint: disable=no-member + upsert_status_change=queries.upsert_status_change.sql, # pylint: disable=no-member ) def _insert_dlchange(self, cursor: Any, message: dict[str, Any]) -> None: @@ -158,6 +162,67 @@ def _insert_test(self, cursor: Any, message: dict[str, Any]) -> None: }, ) + def _upsert_status_change(self, cursor: Any, message: dict[str, Any]) -> None: + """Upsert a status_change event into the aggregated job table. + Args: + cursor: Database cursor. + message: Event payload. + """ + logger.debug("Sending to Postgres - status_change.") + ts = datetime.fromtimestamp(message["timestamp_event"] / 1000.0, tz=timezone.utc) + event_type = message["event_type"] + + created_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + + if event_type == "JobCreatedEvent": + created_at = ts + elif event_type == "JobCreatedAndStartedEvent": + created_at = ts + started_at = ts + elif event_type == "JobStartedEvent": + started_at = ts + elif event_type == "JobFinishedEvent": + finished_at = ts + + cursor.execute( + self._queries.upsert_status_change, + { + "job_id": message["job_id"], + "job_group_id": message.get("job_group_id"), + "parent_job_id": message.get("parent_job_id"), + "initial_job_id": message.get("initial_job_id"), + "job_ref": message.get("job_ref"), + "job_name": message.get("job_name"), + "definition_id": message.get("definition_id"), + "definition_version": message.get("definition_version"), + "tenant_id": message.get("tenant_id"), + "country": message.get("country"), + "source_app": message.get("source_app"), + "source_app_version": message.get("source_app_version"), + "environment": message.get("environment"), + "platform": message.get("platform"), + "platform_metadata": ( + json.dumps(message["platform_metadata"]) if message.get("platform_metadata") is not None else None + ), + "input_arguments": ( + json.dumps(message["input_arguments"]) if message.get("input_arguments") is not None else None + ), + "additional_context": ( + json.dumps(message["additional_context"]) if message.get("additional_context") is not None else None + ), + "attempt_number": message.get("attempt_number") or 1, + "status_type": message.get("status_type"), + "status_subtype": message.get("status_subtype"), + "status_detail": message.get("status_detail"), + "created_at": created_at, + "started_at": started_at, + "finished_at": finished_at, + "last_updated_at": ts, + }, + ) + def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") -> None: """Dispatch insertion for a topic into the correct Postgres table(s). Args: @@ -211,6 +276,8 @@ def _write_topic(self, connection: Any, topic_name: str, message: dict[str, Any] self._insert_run(cursor, message) elif topic_name == TOPIC_TEST: self._insert_test(cursor, message) + elif topic_name == TOPIC_STATUS_CHANGE: + self._upsert_status_change(cursor, message) connection.commit() @staticmethod diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index debd6dc..8f76a95 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -175,7 +175,23 @@ def postgres_container() -> Generator[str, None, None]: dsn = _convert_dsn(container.get_connection_url()) logger.debug("PostgreSQL started, initializing schema.") - conn = psycopg2.connect(dsn) + # Attempt up to 5 times to connect to Postgres (2s interval). + # This is a workaround for colima on macOS. The built-in wait strategy for Postgres is only tested against Docker Desktop, see https://github.com/testcontainers/testcontainers-java/issues/5986 + conn = None + last_exc: Optional[Exception] = None + for attempt in range(1, 6): + try: + conn = psycopg2.connect(dsn) + break + except psycopg2.OperationalError as exc: + last_exc = exc + logger.debug("Postgres not ready yet (attempt %d/5): %s. Retrying in 2s.", attempt, exc) + if attempt < 5: + time.sleep(2) + + if conn is None: + raise TimeoutError(f"Timed out waiting for Postgres to become available after 5 attempts: {last_exc}") + conn.autocommit = True with conn.cursor() as cursor: cursor.execute(SCHEMA_SQL) diff --git a/tests/integration/schemas/postgres_schema.py b/tests/integration/schemas/postgres_schema.py index de46006..1a286e8 100644 --- a/tests/integration/schemas/postgres_schema.py +++ b/tests/integration/schemas/postgres_schema.py @@ -68,4 +68,33 @@ timestamp_event BIGINT, additional_info JSONB ); + +-- Table for test_status_change_writer +CREATE TABLE IF NOT EXISTS public_cps_za_status_change_aggregated_job ( + job_id UUID PRIMARY KEY, + job_group_id UUID, + parent_job_id UUID, + initial_job_id UUID, + job_ref TEXT, + job_name TEXT, + definition_id TEXT, + definition_version TEXT, + tenant_id TEXT, + country TEXT, + source_app TEXT, + source_app_version TEXT, + environment TEXT, + platform TEXT, + platform_metadata JSONB, + input_arguments JSONB, + additional_context JSONB, + attempt_number INTEGER NOT NULL DEFAULT 1 CHECK (attempt_number > 0), + status_type TEXT CHECK (status_type IN ('WAITING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'KILLED')), + status_subtype TEXT, + status_detail TEXT, + created_at TIMESTAMPTZ, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + last_updated_at TIMESTAMPTZ NOT NULL +); """ diff --git a/tests/integration/test_status_change_writer.py b/tests/integration/test_status_change_writer.py new file mode 100644 index 0000000..708c87f --- /dev/null +++ b/tests/integration/test_status_change_writer.py @@ -0,0 +1,436 @@ +# +# Copyright 2026 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict + +import psycopg2 +import pytest + +from tests.integration.conftest import EventGateTestClient + +_TOPIC = "public.cps.za.status_change" + + +def _fetch_job_row(conn: Any, job_id: str) -> dict[str, Any] | None: + """Fetch a single row from the job table by job_id as a column-keyed dict.""" + with conn.cursor() as cursor: + cursor.execute("SELECT * FROM public_cps_za_status_change_aggregated_job WHERE job_id = %s", (job_id,)) + columns = [desc[0] for desc in cursor.description] + row = cursor.fetchone() + if row is None: + return None + return dict(zip(columns, row)) + + +class TestStatusChangeJobCreatedAndStarted: + """Verify a JobCreatedAndStartedEvent is upserted into the job table.""" + + @pytest.fixture(scope="class") + def created_event(self, eventgate_client: EventGateTestClient, valid_token: str) -> Dict[str, Any]: + """Post a JobCreatedAndStartedEvent and return the payload.""" + job_id = str(uuid.uuid4()) + event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "job_ref": "myJobRef", + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646400000, + "country": "za", + "job_id": job_id, + "parent_job_id": str(uuid.uuid4()), + "initial_job_id": str(uuid.uuid4()), + "job_group_id": job_id, + "job_name": "IngestApp Ingestion", + "platform": "aws.stepfunctions", + "platform_metadata": {"platform_key": "platform_value"}, + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19", "triggerType": "SCHEDULE"}, + "definition_id": "1234", + "definition_version": "v1.0", + "status_type": "RUNNING", + "status_subtype": "myStatusSubtype", + "status_detail": "myStatusDetail", + "additional_context": {"context_key": "context_value"}, + } + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + return event + + def test_all_fields_populated(self, created_event: Dict[str, Any], postgres_container: str) -> None: + """source_app, platform, tenant_id and environment must be stored from the event.""" + conn = psycopg2.connect(postgres_container) + try: + row = _fetch_job_row(conn, created_event["job_id"]) + assert row is not None + assert created_event["job_id"] == str(row["job_id"]) + assert created_event["job_group_id"] == str(row["job_group_id"]) + assert created_event["parent_job_id"] == str(row["parent_job_id"]) + assert created_event["initial_job_id"] == str(row["initial_job_id"]) + assert created_event["job_ref"] == row["job_ref"] + assert created_event["job_name"] == row["job_name"] + assert created_event["definition_id"] == row["definition_id"] + assert created_event["definition_version"] == row["definition_version"] + assert created_event["tenant_id"] == row["tenant_id"] + assert created_event["country"] == row["country"] + assert created_event["source_app"] == row["source_app"] + assert created_event["source_app_version"] == row["source_app_version"] + assert created_event["environment"] == row["environment"] + assert created_event["platform"] == row["platform"] + assert created_event["platform_metadata"] == row["platform_metadata"] + assert created_event["input_arguments"] == row["input_arguments"] + assert created_event["additional_context"] == row["additional_context"] + assert 1 == row["attempt_number"] + assert created_event["status_type"] == row["status_type"] + assert created_event["status_subtype"] == row["status_subtype"] + assert created_event["status_detail"] == row["status_detail"] + ts = datetime.fromtimestamp(created_event["timestamp_event"] / 1000, tz=timezone.utc) + assert ts == row["started_at"] + assert ts == row["created_at"] + assert None is row["finished_at"] + assert ts == row["last_updated_at"] + + finally: + conn.close() + + +class TestStatusChangeJobFinished: + """Verify a JobFinishedEvent overwrites terminal status and timestamps.""" + + def create_event(self, job_id) -> Dict[str, Any]: + return { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "job_ref": "myJobRef", + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1000000000000, + "country": "za", + "job_id": job_id, + "parent_job_id": str(uuid.uuid4()), + "initial_job_id": str(uuid.uuid4()), + "job_group_id": job_id, + "job_name": "IngestApp Ingestion", + "platform": "aws.stepfunctions", + "platform_metadata": {"create_platform_key": "create_platform_value"}, + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19", "triggerType": "SCHEDULE"}, + "definition_id": "1234", + "definition_version": "v1.0", + "status_type": "RUNNING", + "status_subtype": "myStatusSubtype", + "status_detail": "myStatusDetail", + "additional_context": {"create_key": "create_value", "common_key": "create_value"}, + } + + def finish_event(self, job_id) -> Dict[str, Any]: + return { + "event_type": "JobFinishedEvent", + "event_id": str(uuid.uuid4()), + "timestamp_event": 2000000000000, + "job_id": job_id, + "status_type": "SUCCEEDED", + "additional_context": {"finish_key": "finish_value", "common_key": "finish_value"}, + "platform_metadata": {"finish_platform_key": "finish_platform_value"}, + } + + def update_event(self, job_id) -> Dict[str, Any]: + return { + "event_type": "JobUpdatedEvent", + "event_id": str(uuid.uuid4()), + "timestamp_event": 1500000000000, + "job_id": job_id, + "attempt_number": 2, + "status_type": "RUNNING", + } + + @pytest.fixture(scope="class") + def in_order_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> Dict[str, Any]: + """Post a JobCreatedAndStartedEvent followed by a SUCCEEDED JobFinishedEvent.""" + job_id = str(uuid.uuid4()) + create_event = self.create_event(job_id) + finish_event = self.finish_event(job_id) + update_event = self.update_event(job_id) + for event in (create_event, update_event, finish_event): + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + return {"create_event": create_event} + + @pytest.fixture(scope="class") + def out_of_order_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> Dict[str, Any]: + """First post a JobFinishedEvent followed by a JobCreatedAndStartedEvent.""" + job_id = str(uuid.uuid4()) + create_event = self.create_event(job_id) + finish_event = self.finish_event(job_id) + late_update_event = self.update_event(job_id) + for event in (finish_event, create_event, late_update_event): + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + return {"create_event": create_event} + + def _test_all_fields(self, create_event: Dict[str, Any], postgres_container: str) -> None: + conn = psycopg2.connect(postgres_container) + try: + row = _fetch_job_row(conn, create_event["job_id"]) + + assert create_event["job_id"] == str(row["job_id"]) + assert create_event["job_group_id"] == str(row["job_group_id"]) + assert create_event["parent_job_id"] == str(row["parent_job_id"]) + assert create_event["initial_job_id"] == str(row["initial_job_id"]) + assert create_event["job_ref"] == row["job_ref"] + assert create_event["job_name"] == row["job_name"] + assert create_event["definition_id"] == row["definition_id"] + assert create_event["definition_version"] == row["definition_version"] + assert create_event["tenant_id"] == row["tenant_id"] + assert create_event["country"] == row["country"] + assert create_event["source_app"] == row["source_app"] + assert create_event["source_app_version"] == row["source_app_version"] + assert create_event["environment"] == row["environment"] + assert create_event["platform"] == row["platform"] + assert {"finish_platform_key": "finish_platform_value"} == row["platform_metadata"] + assert create_event["input_arguments"] == row["input_arguments"] + # Cumulative merge of additional_context + assert {"create_key": "create_value", "finish_key": "finish_value", "common_key": "finish_value"} == row[ + "additional_context" + ] + # Take non-default value even if it's not from latest event + assert 2 == row["attempt_number"] + # Take latest with nulls for status fields + assert "SUCCEEDED" == row["status_type"] + assert None is row["status_subtype"] + assert None is row["status_detail"] + ts_create = datetime.fromtimestamp(1000000000000 / 1000, tz=timezone.utc) + ts_finish = datetime.fromtimestamp(2000000000000 / 1000, tz=timezone.utc) + assert ts_create == row["started_at"] + assert ts_create == row["created_at"] + assert ts_finish == row["finished_at"] + assert ts_finish == row["last_updated_at"] + finally: + conn.close() + + def test_all_fields_with_in_order_events(self, in_order_events: Dict[str, Any], postgres_container: str) -> None: + """Status fields must use latest values, even if null, and timestamps must be updated to the latest event.""" + self._test_all_fields(in_order_events["create_event"], postgres_container) + + def test_all_fields_with_out_of_order_events( + self, out_of_order_events: Dict[str, Any], postgres_container: str + ) -> None: + """Out of order events must lead to same outcome as in order events""" + self._test_all_fields(out_of_order_events["create_event"], postgres_container) + + def test_single_row_after_create_and_finish(self, in_order_events: Dict[str, Any], postgres_container: str) -> None: + """Two events for the same job_id must produce exactly one row.""" + conn = psycopg2.connect(postgres_container) + try: + with conn.cursor() as cursor: + cursor.execute( + "SELECT COUNT(*) FROM public_cps_za_status_change_aggregated_job WHERE job_id = %s", + (in_order_events["create_event"]["job_id"],), + ) + count = cursor.fetchone()[0] + assert 1 == count + finally: + conn.close() + + +class TestNestedJobsScenario: + + @pytest.fixture(scope="class") + def nested_jobs_posted(self, eventgate_client: EventGateTestClient, valid_token: str) -> Dict[str, Any]: + """Post a JobCreatedAndStartedEvent followed by a nested event.""" + job_id = str(uuid.uuid4()) + parent_event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646400000, + "country": "za", + "job_id": job_id, + "job_group_id": job_id, + "job_name": "IngestApp Ingestion", + "platform": "aws.stepfunctions", + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19", "triggerType": "SCHEDULE"}, + "definition_id": "1234", + "status_type": "RUNNING", + } + + nested_event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646410000, + "country": "za", + "job_id": str(uuid.uuid4()), + "parent_job_id": job_id, + "job_group_id": job_id, + "job_name": "Create Pramen Config", + "platform": "aws.lambda", + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19"}, + "definition_id": "1234", + "status_type": "RUNNING", + } + + for event in (parent_event, nested_event): + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + return {"job_id": job_id, "parent_event": parent_event, "nested_event": nested_event} + + def test_parent_job_row_created(self, nested_jobs_posted: Dict[str, Any], postgres_container: str) -> None: + """The parent job must have a row in the job table.""" + conn = psycopg2.connect(postgres_container) + try: + parent_row = _fetch_job_row(conn, nested_jobs_posted["parent_event"]["job_id"]) + assert parent_row is not None + assert "RUNNING" == parent_row["status_type"] + + finally: + conn.close() + + def test_nested_job_row_created(self, nested_jobs_posted: Dict[str, Any], postgres_container: str) -> None: + """The nested job must have a row in the job table and the parent job id must be set.""" + conn = psycopg2.connect(postgres_container) + try: + nested_row = _fetch_job_row(conn, nested_jobs_posted["nested_event"]["job_id"]) + assert nested_row is not None + assert "RUNNING" == nested_row["status_type"] + assert nested_jobs_posted["parent_event"]["job_id"] == str(nested_row["parent_job_id"]) + finally: + conn.close() + + +class TestJobRetryScenario: + @pytest.fixture(scope="class") + def retry_jobs_posted(self, eventgate_client: EventGateTestClient, valid_token: str) -> Dict[str, Any]: + """Post a JobCreatedAndStartedEvent followed by a FAILED JobFinishedEvent.""" + initial_job_id = str(uuid.uuid4()) + initial_event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646400000, + "country": "za", + "job_id": initial_job_id, + "job_group_id": initial_job_id, + "job_name": "IngestApp Ingestion", + "platform": "aws.stepfunctions", + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19", "triggerType": "SCHEDULE"}, + "definition_id": "1234", + "status_type": "RUNNING", + } + + retry_event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646800000, + "job_id": str(uuid.uuid4()), + "initial_job_id": initial_job_id, + "job_name": "IngestApp Ingestion", + "attempt_number": 2, + "platform": "aws.glue", + "input_arguments": {"pipelineId": 1234, "currentDate": "2026-05-19"}, + "definition_id": "1234", + "status_type": "RUNNING", + } + + for event in (initial_event, retry_event): + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + return {"job_id": initial_job_id, "initial_event": initial_event, "retry_event": retry_event} + + def test_initial_job_row_created(self, retry_jobs_posted: Dict[str, Any], postgres_container: str) -> None: + """The initial job must have a row in the job table.""" + conn = psycopg2.connect(postgres_container) + try: + initial_row = _fetch_job_row(conn, retry_jobs_posted["initial_event"]["job_id"]) + assert initial_row is not None + assert "RUNNING" == initial_row["status_type"] + + finally: + conn.close() + + def test_retry_job_row_created(self, retry_jobs_posted: Dict[str, Any], postgres_container: str) -> None: + """The retry job must have a row in the job table and the initial job id must be set.""" + conn = psycopg2.connect(postgres_container) + try: + retry_row = _fetch_job_row(conn, retry_jobs_posted["retry_event"]["job_id"]) + assert retry_row is not None + assert "RUNNING" == retry_row["status_type"] + assert retry_jobs_posted["initial_event"]["job_id"] == str(retry_row["initial_job_id"]) + assert 2 == retry_row["attempt_number"] + finally: + conn.close() + + +class TestStatusChangeIdempotency: + """Verify that posting the same event twice produces exactly one job row.""" + + def test_duplicate_event_produces_one_row( + self, + eventgate_client: EventGateTestClient, + valid_token: str, + postgres_container: str, + ) -> None: + """Posting the identical event twice must not create duplicate rows.""" + job_id = str(uuid.uuid4()) + event: Dict[str, Any] = { + "event_type": "JobCreatedAndStartedEvent", + "event_id": str(uuid.uuid4()), + "tenant_id": "abcd", + "source_app": "ingestapp", + "source_app_version": "2.14.0", + "environment": "dev", + "timestamp_event": 1747646400000, + "job_id": job_id, + "job_group_id": job_id, + "job_name": "Idempotency Test Job", + "platform": "aws.stepfunctions", + "definition_id": "1234", + "status_type": "RUNNING", + "additional_context": {"test_key": "test_value"}, + } + for _ in range(2): + response = eventgate_client.post_event(_TOPIC, event, token=valid_token) + assert 202 == response["statusCode"] + conn = psycopg2.connect(postgres_container) + try: + with conn.cursor() as cursor: + cursor.execute( + "SELECT COUNT(*) FROM public_cps_za_status_change_aggregated_job WHERE job_id = %s", (job_id,) + ) + count = cursor.fetchone()[0] + row = _fetch_job_row(conn, job_id) + + assert 1 == count + assert {"test_key": "test_value"} == row["additional_context"] + finally: + conn.close()