From 45c3930d4e0b98af5252f4cf7a747406dae34e52 Mon Sep 17 00:00:00 2001 From: "Tobias.Mikula" Date: Mon, 14 Sep 2026 15:23:27 +0200 Subject: [PATCH 1/4] feat(event-stats): add named-query endpoint reproducing Qlik runs/jobs dashboard feed --- .github/copilot-instructions.md | 1 + api.yaml | 138 ++++++++ src/event_stats_lambda.py | 3 + src/handlers/handler_named_query.py | 164 ++++++++++ src/readers/named_query_registry.py | 51 +++ src/readers/reader_postgres.py | 172 +++++++++- src/readers/sql/named_queries.sql | 26 ++ src/utils/constants.py | 3 + tests/unit/conftest.py | 9 + .../unit/handlers/test_handler_named_query.py | 296 ++++++++++++++++++ tests/unit/handlers/test_handler_stats.py | 9 - tests/unit/readers/test_reader_named_query.py | 269 ++++++++++++++++ tests/unit/test_event_stats_lambda.py | 4 + 13 files changed, 1135 insertions(+), 10 deletions(-) create mode 100644 src/handlers/handler_named_query.py create mode 100644 src/readers/named_query_registry.py create mode 100644 src/readers/sql/named_queries.sql create mode 100644 tests/unit/handlers/test_handler_named_query.py create mode 100644 tests/unit/readers/test_reader_named_query.py diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 3f9ac7b..bd9cce4 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -47,6 +47,7 @@ Testing - No real API/DB calls in unit tests - Use `mocker.patch("module.dependency")` or `mocker.patch.object(Class, "method")` - Assert pattern: `assert expected == actual` +- Pylint is disabled for `tests/`. Do not add any `# pylint: disable=...` comments in test files Quality gates (run after changes, fix only if below threshold) - Run all quality gates at once: `make qa` diff --git a/api.yaml b/api.yaml index 2bcb1b3..644dc5a 100644 --- a/api.yaml +++ b/api.yaml @@ -328,6 +328,144 @@ paths: schema: $ref: '#/components/schemas/ErrorResponse' + /stats/{topic_name}/query/{query_name}: + post: + summary: Execute a predefined named query + description: > + Runs a curated, server-side named query by name and returns its + paginated result set. + security: [] + parameters: + - name: topic_name + in: path + required: true + schema: + type: string + description: Name of the topic to query + - name: query_name + in: path + required: true + schema: + type: string + enum: [runs_jobs_detail] + description: Identifier of the named query to run + requestBody: + description: Query parameters including time window, pagination cursor, and limit + required: false + content: + application/json: + schema: + type: object + properties: + timestamp_start: + type: integer + nullable: true + description: Start of time window in epoch milliseconds (default now - 7 days) + example: 1704067200000 + timestamp_end: + type: integer + nullable: true + description: End of time window in epoch milliseconds (default now) + example: 1706745600000 + cursor: + type: integer + nullable: true + description: Last internal_id from previous page (keyset pagination) + limit: + type: integer + minimum: 1 + maximum: 1000 + default: 50 + description: Maximum number of records per page + responses: + '200': + description: Paginated named-query result set with computed fields + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + example: true + statusCode: + type: integer + example: 200 + data: + type: array + items: + type: object + properties: + event_id: + type: string + job_ref: + type: string + tenant_id: + type: string + internal_id: + type: integer + catalog_id: + type: string + status: + type: string + run_date: + type: string + description: Date of the job-level timestamp_start (DD-MM-YYYY, UTC) + run_status: + type: string + description: > + Computed status bucket. Message reclassification is applied to + every row (no data received, no data produced, timeout), else the + raw status is preserved (succeeded, failed, killed, skipped). + formatted_tenant: + type: string + description: Lowercase tenant_id + elapsed_time: + type: integer + nullable: true + description: Difference in milliseconds between job end and start + start_time: + type: string + description: Formatted job start (YYYY-MM-DD HH:MM:SS UTC) + end_time: + type: string + description: Formatted job end (YYYY-MM-DD HH:MM:SS UTC) + additionalProperties: true + pagination: + type: object + properties: + cursor: + type: integer + nullable: true + has_more: + type: boolean + limit: + type: integer + '400': + description: Invalid parameters, unsupported topic, or unknown query_name + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '403': + description: User not authorized for topic + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '404': + description: Topic not found + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '500': + description: Database query error + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + /terminate: post: summary: Terminates lambda environment diff --git a/src/event_stats_lambda.py b/src/event_stats_lambda.py index f8bc438..7197b45 100644 --- a/src/event_stats_lambda.py +++ b/src/event_stats_lambda.py @@ -19,6 +19,7 @@ import os from typing import Any +from src.handlers.handler_named_query import HandlerNamedQuery from src.handlers.handler_health import HandlerHealth from src.handlers.handler_stats import HandlerStats from src.readers.reader_postgres import ReaderPostgres @@ -44,11 +45,13 @@ # Initialize EventStats handlers handler_stats = HandlerStats(topics, reader_postgres) +handler_named_query = HandlerNamedQuery(topics, reader_postgres) handler_health = HandlerHealth({"postgres_reader": reader_postgres}) # Route to handler function mapping ROUTE_MAP: dict[str, Any] = { "/stats/{topic_name}": handler_stats.handle_request, + "/stats/{topic_name}/query/{query_name}": handler_named_query.handle_request, "/health": lambda _: handler_health.get_health(), } diff --git a/src/handlers/handler_named_query.py b/src/handlers/handler_named_query.py new file mode 100644 index 0000000..ee2435c --- /dev/null +++ b/src/handlers/handler_named_query.py @@ -0,0 +1,164 @@ +# +# 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. +# + +"""Handler for the /stats/{topic_name}/query/{query_name} endpoint.""" + +import json +import logging +from dataclasses import dataclass +from typing import Any + +from src.readers.named_query_registry import SUPPORTED_QUERIES +from src.readers.reader_postgres import ReaderPostgres +from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS +from src.utils.utils import build_error_response + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class NamedQueryParams: + """Validated query parameters ready to pass to `read_named_query`. + + Attributes: + timestamp_start: Start of time window in epoch milliseconds, or `None` for the default. + timestamp_end: End of time window in epoch milliseconds, or `None` for the default. + cursor: Last `internal_id` from previous page, or `None` for the first page. + limit: Maximum number of rows per page. + """ + + timestamp_start: int | None + timestamp_end: int | None + cursor: int | None + limit: int + + +class HandlerNamedQuery: + """Handle predefined named queries for a specific topic.""" + + def __init__( + self, + topics: dict[str, dict[str, Any]], + reader_postgres: ReaderPostgres, + ) -> None: + self.topics = topics + self.reader_postgres = reader_postgres + + def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: + """Handle POST /stats/{topic_name}/query/{query_name} requests. + Args: + event: API Gateway proxy event. + Returns: + API Gateway response dict. + """ + path_params = event.get("pathParameters") or {} + topic_name = path_params.get("topic_name", "").lower() + query_name = path_params.get("query_name", "").lower() + + if error_response := self._validate_event_path_params(topic_name, query_name): + return error_response + + body_params = self._validate_event_body(event.get("body")) + if isinstance(body_params, dict): + return body_params + + try: + rows, pagination = self.reader_postgres.read_named_query( + query_name=query_name, + timestamp_start=body_params.timestamp_start, + timestamp_end=body_params.timestamp_end, + cursor=body_params.cursor, + limit=body_params.limit, + ) + except RuntimeError: + logger.exception("Named query %s failed for topic %s.", query_name, topic_name) + return build_error_response(500, "database", "Named query failed.") + + return { + "statusCode": 200, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps( + { + "success": True, + "statusCode": 200, + "data": rows, + "pagination": pagination, + }, + default=str, + ), + } + + def _validate_event_path_params(self, topic_name: str, query_name: str) -> dict[str, Any] | None: + """Validate the `topic_name`/`query_name` path parameters. + Args: + topic_name: The lower-cased `topic_name` path parameter. + query_name: The lower-cased `query_name` path parameter. + Returns: + An `error_response` dict if validation fails, or `None` if valid. + """ + if not topic_name: + return build_error_response(400, "validation", "Missing path parameter 'topic_name'.") + + if topic_name not in self.topics: + return build_error_response(404, "topic", f"Topic '{topic_name}' not found.") + + if topic_name not in SUPPORTED_STATS_TOPICS: + return build_error_response(400, "validation", f"Topic '{topic_name}' is not supported.") + + if not query_name: + return build_error_response(400, "validation", "Missing path parameter 'query_name'.") + + if query_name not in SUPPORTED_QUERIES: + return build_error_response(400, "validation", f"Query '{query_name}' is not supported. ") + + return None + + @staticmethod + def _validate_event_body(body: str | None) -> NamedQueryParams | dict[str, Any]: + """Parse and validate the request body. + Args: + body: The raw request body (JSON string) from the API Gateway event, or `None`. + Returns: + The parsed_body `NamedQueryParams`, or an `error_response` dict if validation fails. + """ + try: + parsed_body = json.loads(body or "{}") + except (json.JSONDecodeError, TypeError): + return build_error_response(400, "validation", "Request body must be valid JSON.") + + if not isinstance(parsed_body, dict): + return build_error_response(400, "validation", "Request body must be a JSON object.") + + timestamp_start = parsed_body.get("timestamp_start") + timestamp_end = parsed_body.get("timestamp_end") + cursor = parsed_body.get("cursor") + limit: int = parsed_body.get("limit", POSTGRES_DEFAULT_LIMIT) + + int_fields = ( + (timestamp_start, "timestamp_start"), + (timestamp_end, "timestamp_end"), + (cursor, "cursor"), + ) + for value, field_name in int_fields: + if value is not None and (isinstance(value, bool) or not isinstance(value, int)): + return build_error_response(400, "validation", f"Field '{field_name}' must be an integer.") + + if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: + return build_error_response(400, "validation", "Field 'limit' must be a positive integer.") + + return NamedQueryParams( + timestamp_start=timestamp_start, timestamp_end=timestamp_end, cursor=cursor, limit=limit + ) diff --git a/src/readers/named_query_registry.py b/src/readers/named_query_registry.py new file mode 100644 index 0000000..62ed004 --- /dev/null +++ b/src/readers/named_query_registry.py @@ -0,0 +1,51 @@ +# +# 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. +# + +"""Registry of predefined, server-executed named queries. + +Each named query maps a stable `query_name` to the aiosql query keys used to +run it (with and without a keyset-pagination cursor). Unknown query names are +rejected at the handler layer with a `400` response. New aggregation queries are +added here without introducing new routes. +""" + +from dataclasses import dataclass + +from src.utils.constants import QUERY_RUNS_JOBS_DETAIL + + +@dataclass(frozen=True) +class NamedQuery: + """Definition of a single named query. + + Attributes: + name: Stable identifier used in the request path. + sql_key: aiosql query name for the non-cursor variant. + sql_key_with_cursor: aiosql query name for the keyset-cursor variant. + """ + + name: str + sql_key: str + sql_key_with_cursor: str + + +SUPPORTED_QUERIES: dict[str, NamedQuery] = { + QUERY_RUNS_JOBS_DETAIL: NamedQuery( + name=QUERY_RUNS_JOBS_DETAIL, + sql_key="get_runs_jobs_detail", + sql_key_with_cursor="get_runs_jobs_detail_with_cursor", + ), +} diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py index 355f0b4..b869ce3 100644 --- a/src/readers/reader_postgres.py +++ b/src/readers/reader_postgres.py @@ -25,8 +25,10 @@ from typing import Any import aiosql +import psycopg2.extensions from botocore.exceptions import BotoCoreError, ClientError +from src.readers.named_query_registry import SUPPORTED_QUERIES from src.utils.constants import ( POSTGRES_DEFAULT_LIMIT, POSTGRES_DEFAULT_WINDOW_MS, @@ -51,6 +53,8 @@ class ReaderQueries: get_stats: str get_stats_with_cursor: str + get_runs_jobs_detail: str + get_runs_jobs_detail_with_cursor: str class ReaderPostgres(PostgresBase): @@ -73,6 +77,10 @@ def _queries(self) -> ReaderQueries: return ReaderQueries( get_stats=queries.get_stats.sql, # pylint: disable=no-member get_stats_with_cursor=queries.get_stats_with_cursor.sql, # pylint: disable=no-member + get_runs_jobs_detail=queries.get_runs_jobs_detail.sql, # pylint: disable=no-member + get_runs_jobs_detail_with_cursor=( + queries.get_runs_jobs_detail_with_cursor.sql # pylint: disable=no-member + ), ) def read_stats( @@ -144,7 +152,7 @@ def read_stats( def _run_stats_query( self, - connection: Any, + connection: psycopg2.extensions.connection, ts_start: int, ts_end: int, cursor: int | None, @@ -177,6 +185,168 @@ def _run_stats_query( self._close_connection() return col_names, raw_rows + def read_named_query( + self, + query_name: str, + timestamp_start: int | None = None, + timestamp_end: int | None = None, + cursor: int | None = None, + limit: int = POSTGRES_DEFAULT_LIMIT, + ) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Execute a predefined named query with keyset pagination. + Args: + query_name: Registered named query identifier (see `SUPPORTED_QUERIES`). + timestamp_start: Start of time window in epoch milliseconds (default to *now - 7 days*). + timestamp_end: End of time window in epoch milliseconds (default to *now*). + cursor: Last `internal_id` from previous page (keyset pagination). + limit: Maximum number of rows per page. + Returns: + Tuple of (rows, pagination), where each row is a dict with raw and + computed columns and pagination contains cursor info. + Raises: + RuntimeError: On unknown query name or database connectivity/query errors. + """ + query = SUPPORTED_QUERIES.get(query_name) + if query is None: + raise RuntimeError(f"Unknown named query: {query_name}.") + + try: + config = self._pg_config + except (BotoCoreError, ClientError, ValueError, KeyError) as exc: + raise RuntimeError(f"PostgreSQL configuration error: {exc}") from exc + + if not config.get("database"): + raise RuntimeError("PostgreSQL config missing: database.") + + missing = [field for field in REQUIRED_CONNECTION_FIELDS if not config.get(field)] + if missing: + raise RuntimeError(f"PostgreSQL config missing: {', '.join(missing)}.") + + limit = max(1, min(limit, POSTGRES_MAX_LIMIT)) + now_ms = int(time.time() * 1000) + ts_start = timestamp_start if timestamp_start is not None else (now_ms - POSTGRES_DEFAULT_WINDOW_MS) + ts_end = timestamp_end if timestamp_end is not None else now_ms + + sql = getattr(self._queries, query.sql_key) + sql_with_cursor = getattr(self._queries, query.sql_key_with_cursor) + + try: + col_names, raw_rows = self._execute_with_retry( + lambda conn: self._run_named_query(conn, sql, sql_with_cursor, ts_start, ts_end, cursor, limit) + ) + except PsycopgError as exc: + self._close_connection() + raise RuntimeError(f"Database query error: {exc}") from exc + + rows = [dict(zip(col_names, row, strict=True)) for row in raw_rows] + + has_more = len(rows) > limit + if has_more: + rows = rows[:limit] + + next_cursor: int | None = None + if has_more and rows: + next_cursor = rows[-1]["internal_id"] + + rows = [self._format_runs_jobs_detail_row(row) for row in rows] + + pagination: dict[str, Any] = { + "cursor": next_cursor, + "has_more": has_more, + "limit": limit, + } + + logger.debug("Named query %s returned %d rows.", query_name, len(rows)) + return rows, pagination + + def _run_named_query( + self, + connection: psycopg2.extensions.connection, + sql: str, + sql_with_cursor: str, + ts_start: int, + ts_end: int, + cursor: int | None, + limit: int, + ) -> RawQueryResult: + """Execute a named SQL query and return column names and raw rows.""" + try: + with connection.cursor() as db_cursor: + if cursor is not None: + db_cursor.execute( + sql_with_cursor, + {"ts_start": ts_start, "ts_end": ts_end, "cursor_id": cursor, "lim": limit + 1}, + ) + else: + db_cursor.execute( + sql, + {"ts_start": ts_start, "ts_end": ts_end, "lim": limit + 1}, + ) + if db_cursor.description is None: + raise RuntimeError("Named query returned no result description.") + col_names = [desc[0] for desc in db_cursor.description] + raw_rows = db_cursor.fetchall() + finally: + # PostgreSQL (psycopg2) wraps every statement, including SELECT, in an implicit transaction. + # Without an explicit commit or rollback, the connection stays "idle in transaction". + try: + connection.rollback() + except PsycopgError: + logger.debug("Failed to close the implicit transaction. Closing cached connection.", exc_info=True) + self._close_connection() + return col_names, raw_rows + + @staticmethod + def _format_runs_jobs_detail_row(row: dict[str, Any]) -> dict[str, Any]: + """Add computed columns to a `runs_jobs_detail` result row. + Args: + row: Raw database row dict. + Returns: + Row dict enriched with computed fields. + """ + ts_start = row.get("timestamp_start") + ts_end = row.get("timestamp_end") + + # run_date: date portion of the job-level timestamp_start (DD-MM-YYYY, UTC). + if ts_start is not None: + row["run_date"] = datetime.fromtimestamp(ts_start / 1000, tz=timezone.utc).strftime("%d-%m-%Y") + else: + row["run_date"] = None + + # run_status: message reclassification on all rows, else the raw status. + status = str(row.get("status", "")).lower() + message = str(row.get("message") or "").lower() + if "no data" in message: + row["run_status"] = "no data received" + elif "no records to send" in message: + row["run_status"] = "no data produced" + elif "timeout" in message: + row["run_status"] = "timeout" + else: + row["run_status"] = status + + # formatted_tenant + row["formatted_tenant"] = str(row.get("tenant_id", "")).lower() + + # elapsed_time: difference in milliseconds between job end and start. + if ts_start is not None and ts_end is not None: + row["elapsed_time"] = ts_end - ts_start + else: + row["elapsed_time"] = None + + # start_time / end_time: formatted job timestamps (UTC). + if ts_start is not None: + row["start_time"] = datetime.fromtimestamp(ts_start / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + else: + row["start_time"] = None + + if ts_end is not None: + row["end_time"] = datetime.fromtimestamp(ts_end / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + else: + row["end_time"] = None + + return row + @staticmethod def _format_row(row: dict[str, Any]) -> dict[str, Any]: """Add computed columns to a result row. diff --git a/src/readers/sql/named_queries.sql b/src/readers/sql/named_queries.sql new file mode 100644 index 0000000..e69c0a0 --- /dev/null +++ b/src/readers/sql/named_queries.sql @@ -0,0 +1,26 @@ +-- name: get_runs_jobs_detail(ts_start, ts_end, lim) +SELECT r.event_id, r.job_ref, r.tenant_id, r.source_app, + r.source_app_version, r.environment, + r.timestamp_start AS run_timestamp_start, + r.timestamp_end AS run_timestamp_end, + j.internal_id, j.country, j.catalog_id, j.status, + j.timestamp_start, j.timestamp_end, j.message, j.additional_info + FROM public_cps_za_runs_jobs j + INNER JOIN public_cps_za_runs r ON j.event_id = r.event_id + WHERE r.timestamp_start >= :ts_start AND r.timestamp_start <= :ts_end + ORDER BY j.internal_id DESC + LIMIT :lim; + +-- name: get_runs_jobs_detail_with_cursor(ts_start, ts_end, cursor_id, lim) +SELECT r.event_id, r.job_ref, r.tenant_id, r.source_app, + r.source_app_version, r.environment, + r.timestamp_start AS run_timestamp_start, + r.timestamp_end AS run_timestamp_end, + j.internal_id, j.country, j.catalog_id, j.status, + j.timestamp_start, j.timestamp_end, j.message, j.additional_info + FROM public_cps_za_runs_jobs j + INNER JOIN public_cps_za_runs r ON j.event_id = r.event_id + WHERE r.timestamp_start >= :ts_start AND r.timestamp_start <= :ts_end + AND j.internal_id < :cursor_id + ORDER BY j.internal_id DESC + LIMIT :lim; diff --git a/src/utils/constants.py b/src/utils/constants.py index fe75d5c..ec1ac01 100644 --- a/src/utils/constants.py +++ b/src/utils/constants.py @@ -41,3 +41,6 @@ POSTGRES_WRITE_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS, TOPIC_DLCHANGE, TOPIC_TEST, TOPIC_STATUS_CHANGE}) SUPPORTED_STATS_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS}) + +# Named query constants +QUERY_RUNS_JOBS_DETAIL = "runs_jobs_detail" diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 72f85c6..0745e40 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -149,3 +149,12 @@ def valid_payload(): "environment": "dev", "timestamp": 123, } + + +@pytest.fixture +def topics() -> dict[str, dict]: + """Minimal topics dict matching HandlerTopic.topics.""" + return { + "public.cps.za.runs": {"type": "object", "properties": {}}, + "public.cps.za.test": {"type": "object", "properties": {}}, + } diff --git a/tests/unit/handlers/test_handler_named_query.py b/tests/unit/handlers/test_handler_named_query.py new file mode 100644 index 0000000..5952404 --- /dev/null +++ b/tests/unit/handlers/test_handler_named_query.py @@ -0,0 +1,296 @@ +# +# 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 json +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from src.handlers.handler_named_query import HandlerNamedQuery, NamedQueryParams + + +@pytest.fixture +def mock_reader() -> MagicMock: + """Mock ReaderPostgres.""" + mock = MagicMock() + mock.read_named_query.return_value = ( + [{"event_id": "ev1", "internal_id": 1, "status": "succeeded", "run_status": "succeeded"}], + {"cursor": None, "has_more": False, "limit": 50}, + ) + return mock + + +@pytest.fixture +def handler( + topics: dict[str, dict[str, Any]], + mock_reader: MagicMock, +) -> HandlerNamedQuery: + """Create HandlerNamedQuery with mocked dependencies.""" + return HandlerNamedQuery( + topics=topics, + reader_postgres=mock_reader, + ) + + +def _make_event( + topic: str = "public.cps.za.runs", + query_name: str = "runs_jobs_detail", + body: Any = None, +) -> dict[str, Any]: + """Build an API Gateway-style proxy event for a named query.""" + if body is None: + body = {} + return { + "resource": "/stats/{topic_name}/query/{query_name}", + "httpMethod": "POST", + "headers": {}, + "body": json.dumps(body) if isinstance(body, dict) else body, + "pathParameters": {"topic_name": topic, "query_name": query_name}, + } + + +class TestHandlerNamedQuerySuccess: + """Tests for successful named queries.""" + + def test_returns_200_with_data(self, handler: HandlerNamedQuery) -> None: + """Test successful query returns 200 with data and pagination.""" + response = handler.handle_request(_make_event()) + + assert 200 == response["statusCode"] + body = json.loads(response["body"]) + assert True is body["success"] + assert "data" in body + assert "pagination" in body + assert "ev1" == body["data"][0]["event_id"] + + def test_forwards_query_name_and_params_to_reader(self, handler: HandlerNamedQuery, mock_reader: MagicMock) -> None: + """Test that query_name and query params are forwarded to the reader.""" + event = _make_event(body={"timestamp_start": 1000, "timestamp_end": 2000, "cursor": 42, "limit": 25}) + + handler.handle_request(event) + + call_kwargs = mock_reader.read_named_query.call_args.kwargs + assert "runs_jobs_detail" == call_kwargs["query_name"] + assert 1000 == call_kwargs["timestamp_start"] + assert 2000 == call_kwargs["timestamp_end"] + assert 42 == call_kwargs["cursor"] + assert 25 == call_kwargs["limit"] + + +class TestHandlerNamedQueryValidation: + """Tests for request validation.""" + + def test_unknown_topic_returns_404(self, handler: HandlerNamedQuery) -> None: + """Test that unknown topic returns 404.""" + response = handler.handle_request(_make_event(topic="nonexistent.topic")) + + assert 404 == response["statusCode"] + + def test_unsupported_topic_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that a known but unsupported topic returns 400.""" + response = handler.handle_request(_make_event(topic="public.cps.za.test")) + + assert 400 == response["statusCode"] + + def test_missing_topic_name_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that missing topic_name path parameter returns 400.""" + event = _make_event() + event["pathParameters"] = {"query_name": "runs_jobs_detail"} + + response = handler.handle_request(event) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "topic_name" in body["errors"][0]["message"] + + def test_missing_query_name_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that missing query_name path parameter returns 400.""" + event = _make_event() + event["pathParameters"] = {"topic_name": "public.cps.za.runs"} + + response = handler.handle_request(event) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "query_name" in body["errors"][0]["message"] + + def test_unknown_query_name_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that an unknown query_name returns 400 (not 500).""" + response = handler.handle_request(_make_event(query_name="does_not_exist")) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "does_not_exist" in body["errors"][0]["message"] + + def test_invalid_json_body_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that non-JSON body returns 400.""" + event = _make_event() + event["body"] = "not json" + + response = handler.handle_request(event) + + assert 400 == response["statusCode"] + + def test_non_dict_json_body_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that a JSON array body returns 400.""" + event = _make_event() + event["body"] = "[1, 2, 3]" + + response = handler.handle_request(event) + + assert 400 == response["statusCode"] + + def test_invalid_timestamp_start_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that noninteger timestamp_start returns 400.""" + response = handler.handle_request(_make_event(body={"timestamp_start": "bad"})) + + assert 400 == response["statusCode"] + + def test_invalid_cursor_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that noninteger cursor returns 400.""" + response = handler.handle_request(_make_event(body={"cursor": "bad"})) + + assert 400 == response["statusCode"] + + def test_invalid_limit_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that non-positive limit returns 400.""" + response = handler.handle_request(_make_event(body={"limit": 0})) + + assert 400 == response["statusCode"] + + def test_boolean_limit_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that boolean limit is rejected.""" + response = handler.handle_request(_make_event(body={"limit": True})) + + assert 400 == response["statusCode"] + + +class TestHandlerNamedQueryErrors: + """Tests for error handling.""" + + def test_database_error_returns_500(self, handler: HandlerNamedQuery, mock_reader: MagicMock) -> None: + """Test that database RuntimeError returns 500.""" + mock_reader.read_named_query.side_effect = RuntimeError("Database query failed") + + response = handler.handle_request(_make_event()) + + assert 500 == response["statusCode"] + body = json.loads(response["body"]) + assert False is body["success"] + assert "database" == body["errors"][0]["type"] + + +class TestValidateEventPathParams: + """Parametrized tests for `HandlerNamedQuery._validate_event_path_params`.""" + + @pytest.mark.parametrize( + "path_params, expected_status, expected_message_fragment", + [ + pytest.param({"query_name": "runs_jobs_detail"}, 400, "topic_name", id="missing_topic_name"), + pytest.param( + {"topic_name": "nonexistent.topic", "query_name": "runs_jobs_detail"}, + 404, + "nonexistent.topic", + id="unknown_topic", + ), + pytest.param( + {"topic_name": "public.cps.za.test", "query_name": "runs_jobs_detail"}, + 400, + "is not supported", + id="known_but_unsupported_topic", + ), + pytest.param({"topic_name": "public.cps.za.runs"}, 400, "query_name", id="missing_query_name"), + pytest.param( + {"topic_name": "public.cps.za.runs", "query_name": "does_not_exist"}, + 400, + "does_not_exist", + id="unknown_query_name", + ), + ], + ) + def test_returns_error_response_for_invalid_params( + self, + handler: HandlerNamedQuery, + path_params: dict[str, Any], + expected_status: int, + expected_message_fragment: str, + ) -> None: + """Test that each invalid path parameter combination returns the expected error.""" + topic_name = path_params.get("topic_name", "").lower() + query_name = path_params.get("query_name", "").lower() + + error = handler._validate_event_path_params(topic_name, query_name) + + assert error is not None + assert expected_status == error["statusCode"] + body = json.loads(error["body"]) + assert expected_message_fragment in body["errors"][0]["message"] + + def test_returns_none_for_valid_params(self, handler: HandlerNamedQuery) -> None: + """Test that valid path parameters return None (no error).""" + topic_name = "public.cps.za.runs" + query_name = "runs_jobs_detail" + + error = handler._validate_event_path_params(topic_name, query_name) + + assert error is None + + +class TestValidateEventBody: + """Parametrized tests for `HandlerNamedQuery._validate_event_body`.""" + + @pytest.mark.parametrize( + "body, expected_status, expected_message_fragment", + [ + pytest.param("not json", 400, "valid JSON", id="invalid_json"), + pytest.param("[1, 2, 3]", 400, "JSON object", id="non_dict_json"), + pytest.param(json.dumps({"timestamp_start": "bad"}), 400, "timestamp_start", id="invalid_timestamp_start"), + pytest.param(json.dumps({"timestamp_end": "bad"}), 400, "timestamp_end", id="invalid_timestamp_end"), + pytest.param(json.dumps({"cursor": "bad"}), 400, "cursor", id="invalid_cursor"), + pytest.param(json.dumps({"limit": 0}), 400, "limit", id="non_positive_limit"), + pytest.param(json.dumps({"limit": True}), 400, "limit", id="boolean_limit"), + ], + ) + def test_returns_error_response_for_invalid_body( + self, body: str, expected_status: int, expected_message_fragment: str + ) -> None: + """Test that each invalid body returns the expected error.""" + result = HandlerNamedQuery._validate_event_body(body) + + assert isinstance(result, dict) + assert expected_status == result["statusCode"] + parsed_body = json.loads(result["body"]) + assert expected_message_fragment in parsed_body["errors"][0]["message"] + + @pytest.mark.parametrize( + "body, expected_params", + [ + pytest.param(None, NamedQueryParams(None, None, None, 50), id="none_body_uses_defaults"), + pytest.param("{}", NamedQueryParams(None, None, None, 50), id="empty_object_uses_defaults"), + pytest.param( + json.dumps({"timestamp_start": 1000, "timestamp_end": 2000, "cursor": 42, "limit": 25}), + NamedQueryParams(1000, 2000, 42, 25), + id="full_body", + ), + ], + ) + def test_returns_parsed_params_for_valid_body(self, body: str | None, expected_params: NamedQueryParams) -> None: + """Test that valid bodies are parsed into the expected `NamedQueryParams`.""" + result = HandlerNamedQuery._validate_event_body(body) + + assert expected_params == result diff --git a/tests/unit/handlers/test_handler_stats.py b/tests/unit/handlers/test_handler_stats.py index 266dfbc..a6fd28f 100644 --- a/tests/unit/handlers/test_handler_stats.py +++ b/tests/unit/handlers/test_handler_stats.py @@ -24,15 +24,6 @@ from src.handlers.handler_stats import HandlerStats -@pytest.fixture -def topics() -> dict[str, dict[str, Any]]: - """Minimal topics dict matching HandlerTopic.topics.""" - return { - "public.cps.za.runs": {"type": "object", "properties": {}}, - "public.cps.za.test": {"type": "object", "properties": {}}, - } - - @pytest.fixture def mock_reader() -> MagicMock: """Mock ReaderPostgres.""" diff --git a/tests/unit/readers/test_reader_named_query.py b/tests/unit/readers/test_reader_named_query.py new file mode 100644 index 0000000..ddb62ed --- /dev/null +++ b/tests/unit/readers/test_reader_named_query.py @@ -0,0 +1,269 @@ +# +# 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 json +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from src.readers.reader_postgres import ReaderPostgres +import src.utils.postgres_base as pb + +_DETAIL_DESCRIPTION = [ + ("event_id",), + ("job_ref",), + ("tenant_id",), + ("source_app",), + ("source_app_version",), + ("environment",), + ("run_timestamp_start",), + ("run_timestamp_end",), + ("internal_id",), + ("country",), + ("catalog_id",), + ("status",), + ("timestamp_start",), + ("timestamp_end",), + ("message",), + ("additional_info",), +] + + +@pytest.fixture +def pg_secret() -> dict[str, Any]: + """Sample Postgres secret payload.""" + return { + "database": "eventgate", + "host": "localhost", + "port": 5432, + "user": "reader", + "password": "secret", + } + + +@pytest.fixture +def reader(monkeypatch: pytest.MonkeyPatch) -> ReaderPostgres: + """Create a ReaderPostgres instance with env vars set.""" + monkeypatch.setenv("POSTGRES_SECRET_NAME", "eventgate/postgres") + monkeypatch.setenv("POSTGRES_SECRET_REGION", "us-east-1") + return ReaderPostgres() + + +def _make_mock_connection(description: list[tuple[str, ...]], rows: list[tuple[Any, ...]]) -> MagicMock: + """Build a mock psycopg2 connection with cursor returning given rows.""" + mock_cursor = MagicMock() + mock_cursor.description = description + mock_cursor.fetchall.return_value = rows + + mock_conn = MagicMock() + mock_conn.closed = 0 + mock_conn.__enter__ = MagicMock(return_value=mock_conn) + mock_conn.__exit__ = MagicMock(return_value=False) + mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False) + return mock_conn + + +class TestFormatRunsJobsDetailRow: + """Tests for the named-query runs_jobs_detail row shaping (DIV-1, DIV-2).""" + + def test_run_date_uses_job_timestamp_start(self) -> None: + """DIV-1: run_date derives from the JOB timestamp_start, not the run one.""" + row: dict[str, Any] = { + "status": "succeeded", + "message": None, + "tenant_id": "T", + "run_timestamp_start": 1704067200000, # 2024-01-01 UTC + "timestamp_start": 1704153600000, # 2024-01-02 UTC + "timestamp_end": 1704157200000, + } + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "02-01-2024" == result["run_date"] + + def test_succeeded_status_passthrough(self) -> None: + """A succeeded job with no special message keeps status 'succeeded'.""" + row = _base_row(status="succeeded", message=None) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "succeeded" == result["run_status"] + + def test_killed_status_preserved(self) -> None: + """DIV-2: killed is preserved, not collapsed to succeed.""" + row = _base_row(status="killed", message=None) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "killed" == result["run_status"] + + def test_skipped_status_preserved(self) -> None: + """DIV-2: skipped is preserved, not collapsed to succeed.""" + row = _base_row(status="skipped", message=None) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "skipped" == result["run_status"] + + def test_failed_status_passthrough(self) -> None: + """A failed job without a special message keeps status 'failed'.""" + row = _base_row(status="failed", message="boom") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "failed" == result["run_status"] + + def test_no_data_message_on_succeeded_row(self) -> None: + """DIV-2: message reclassification applies to all rows, not only failed.""" + row = _base_row(status="succeeded", message="No Data in source") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "no data received" == result["run_status"] + + def test_no_records_to_send_message(self) -> None: + """Message 'no records to send' maps to no data produced.""" + row = _base_row(status="failed", message="There were no records to send today") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "no data produced" == result["run_status"] + + def test_timeout_message(self) -> None: + """DIV-2: timeout bucket is derived from the message.""" + row = _base_row(status="failed", message="Job Timeout after 3600s") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "timeout" == result["run_status"] + + def test_no_data_precedence_over_timeout(self) -> None: + """Precedence: 'no data' wins over 'timeout' when both are present.""" + row = _base_row(status="failed", message="no data - timeout") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "no data received" == result["run_status"] + + def test_null_message_falls_through_to_raw_status(self) -> None: + """A null message falls through to the raw status.""" + row = _base_row(status="killed", message=None) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "killed" == result["run_status"] + + def test_formatted_tenant_lowercased(self) -> None: + """formatted_tenant lowercases the tenant_id.""" + row = _base_row(status="succeeded", message=None, tenant_id="ABC") + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "abc" == result["formatted_tenant"] + + def test_elapsed_time_in_milliseconds(self) -> None: + """elapsed_time is the job duration in milliseconds.""" + row = _base_row( + status="succeeded", + message=None, + timestamp_start=1704067200000, + timestamp_end=1704070800000, + ) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert 3_600_000 == result["elapsed_time"] + + def test_start_and_end_time_are_real_utc_timestamps(self) -> None: + """start_time / end_time keep the real time-of-day in UTC.""" + row = _base_row( + status="succeeded", + message=None, + timestamp_start=1704070800000, # 2024-01-01 01:00:00 UTC + timestamp_end=1704074400000, # 2024-01-01 02:00:00 UTC + ) + result = ReaderPostgres._format_runs_jobs_detail_row(row) + + assert "2024-01-01 01:00:00" == result["start_time"] + assert "2024-01-01 02:00:00" == result["end_time"] + + +class TestReadNamedQuery: + """Tests for read_named_query execution.""" + + def test_returns_rows_and_pagination(self, reader: ReaderPostgres, pg_secret: dict[str, Any]) -> None: + """Test that read_named_query returns shaped rows and pagination.""" + rows = [ + ("ev1", "r", "T", "a", "1", "t", 0, 0, 2, "ZA", "c", "killed", 0, 0, None, None), + ("ev2", "r", "T", "a", "1", "t", 0, 0, 1, "ZA", "c", "succeeded", 0, 0, None, None), + ] + mock_conn = _make_mock_connection(_DETAIL_DESCRIPTION, rows) + + with ( + patch("boto3.Session") as mock_session, + patch.object(pb, "psycopg2") as mock_pg, + ): + mock_client = MagicMock() + mock_client.get_secret_value.return_value = {"SecretString": json.dumps(pg_secret)} + mock_session.return_value.client.return_value = mock_client + mock_pg.connect.return_value = mock_conn + + result_rows, pagination = reader.read_named_query(query_name="runs_jobs_detail", limit=50) + + assert 2 == len(result_rows) + assert "killed" == result_rows[0]["run_status"] + assert False is pagination["has_more"] + + def test_cursor_uses_cursor_variant(self, reader: ReaderPostgres, pg_secret: dict[str, Any]) -> None: + """Test that passing a cursor uses the keyset-cursor SQL variant.""" + mock_conn = _make_mock_connection(_DETAIL_DESCRIPTION, []) + + with ( + patch("boto3.Session") as mock_session, + patch.object(pb, "psycopg2") as mock_pg, + ): + mock_client = MagicMock() + mock_client.get_secret_value.return_value = {"SecretString": json.dumps(pg_secret)} + mock_session.return_value.client.return_value = mock_client + mock_pg.connect.return_value = mock_conn + + reader.read_named_query(query_name="runs_jobs_detail", cursor=100, limit=10) + + executed_sql = mock_conn.cursor.return_value.__enter__.return_value.execute.call_args[0][0] + executed_params = mock_conn.cursor.return_value.__enter__.return_value.execute.call_args[0][1] + + assert "j.internal_id <" in executed_sql + assert 100 == executed_params["cursor_id"] + + def test_unknown_query_name_raises_runtime_error(self, reader: ReaderPostgres, pg_secret: dict[str, Any]) -> None: + """Test that an unknown query_name raises RuntimeError.""" + with ( + patch("boto3.Session") as mock_session, + pytest.raises(RuntimeError, match="Unknown named query"), + ): + mock_client = MagicMock() + mock_client.get_secret_value.return_value = {"SecretString": json.dumps(pg_secret)} + mock_session.return_value.client.return_value = mock_client + reader.read_named_query(query_name="does_not_exist") + + +def _base_row( + status: str, + message: str | None, + tenant_id: str = "T", + timestamp_start: int | None = 1704067200000, + timestamp_end: int | None = 1704070800000, +) -> dict[str, Any]: + """Build a raw detail row for shaping tests.""" + return { + "status": status, + "message": message, + "tenant_id": tenant_id, + "run_timestamp_start": 1704067200000, + "timestamp_start": timestamp_start, + "timestamp_end": timestamp_end, + } diff --git a/tests/unit/test_event_stats_lambda.py b/tests/unit/test_event_stats_lambda.py index 06a19df..f427223 100644 --- a/tests/unit/test_event_stats_lambda.py +++ b/tests/unit/test_event_stats_lambda.py @@ -91,6 +91,10 @@ def test_route_map_contains_stats(self, event_stats_module) -> None: """Test that /stats/{topic_name} is in ROUTE_MAP.""" assert "/stats/{topic_name}" in event_stats_module.ROUTE_MAP + def test_route_map_contains_named_query(self, event_stats_module) -> None: + """Test that /stats/{topic_name}/query/{query_name} is in ROUTE_MAP.""" + assert "/stats/{topic_name}/query/{query_name}" in event_stats_module.ROUTE_MAP + def test_route_map_contains_health(self, event_stats_module) -> None: """Test that /health is in ROUTE_MAP.""" assert "/health" in event_stats_module.ROUTE_MAP From f2f930de40c1a5d22acb0e771294b977cb74de56 Mon Sep 17 00:00:00 2001 From: "Tobias.Mikula" Date: Tue, 15 Sep 2026 12:05:09 +0200 Subject: [PATCH 2/4] Improving the pylint findings --- .pylintrc | 4 ---- src/event_gate_lambda.py | 1 - src/event_stats_lambda.py | 1 - src/handlers/handler_named_query.py | 16 ++----------- src/handlers/handler_stats.py | 16 ++----------- src/readers/reader_postgres.py | 9 ++++---- src/utils/utils.py | 35 ++++++++++++++++++++++++++++- 7 files changed, 43 insertions(+), 39 deletions(-) diff --git a/.pylintrc b/.pylintrc index 1d4ea60..04b1906 100644 --- a/.pylintrc +++ b/.pylintrc @@ -104,10 +104,6 @@ recursive=no # source root. source-roots= -# When enabled, pylint would attempt to guess common misconfiguration and emit -# user-friendly hints instead of false-positive error messages. -suggestion-mode=yes - # Allow loading of arbitrary C extensions. Extensions are imported into the # active Python interpreter and may run arbitrary code. unsafe-load-any-extension=no diff --git a/src/event_gate_lambda.py b/src/event_gate_lambda.py index 748d213..a8e0d03 100644 --- a/src/event_gate_lambda.py +++ b/src/event_gate_lambda.py @@ -16,7 +16,6 @@ """AWS Lambda entry point for the EventGate service.""" -import logging import sys import time from typing import Any diff --git a/src/event_stats_lambda.py b/src/event_stats_lambda.py index ee3447b..d883cc9 100644 --- a/src/event_stats_lambda.py +++ b/src/event_stats_lambda.py @@ -16,7 +16,6 @@ """AWS Lambda entry point for the EventStats service.""" -import logging import time from typing import Any diff --git a/src/handlers/handler_named_query.py b/src/handlers/handler_named_query.py index ee2435c..4891fe1 100644 --- a/src/handlers/handler_named_query.py +++ b/src/handlers/handler_named_query.py @@ -24,7 +24,7 @@ from src.readers.named_query_registry import SUPPORTED_QUERIES from src.readers.reader_postgres import ReaderPostgres from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS -from src.utils.utils import build_error_response +from src.utils.utils import build_error_response, build_success_response logger = logging.getLogger(__name__) @@ -87,19 +87,7 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: logger.exception("Named query %s failed for topic %s.", query_name, topic_name) return build_error_response(500, "database", "Named query failed.") - return { - "statusCode": 200, - "headers": {"Content-Type": "application/json"}, - "body": json.dumps( - { - "success": True, - "statusCode": 200, - "data": rows, - "pagination": pagination, - }, - default=str, - ), - } + return build_success_response(rows, pagination) def _validate_event_path_params(self, topic_name: str, query_name: str) -> dict[str, Any] | None: """Validate the `topic_name`/`query_name` path parameters. diff --git a/src/handlers/handler_stats.py b/src/handlers/handler_stats.py index b5eb0c5..527f4c9 100644 --- a/src/handlers/handler_stats.py +++ b/src/handlers/handler_stats.py @@ -24,7 +24,7 @@ from src.readers.reader_postgres import ReaderPostgres from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS from src.utils.observability import append_request_context -from src.utils.utils import build_error_response, resolve_request_topic +from src.utils.utils import build_error_response, build_success_response, resolve_request_topic logger = logging.getLogger(__name__) @@ -130,16 +130,4 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: ) logger.debug("Stats query completed.") - return { - "statusCode": 200, - "headers": {"Content-Type": "application/json"}, - "body": json.dumps( - { - "success": True, - "statusCode": 200, - "data": rows, - "pagination": pagination, - }, - default=str, - ), - } + return build_success_response(rows, pagination) diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py index e213b81..9823909 100644 --- a/src/readers/reader_postgres.py +++ b/src/readers/reader_postgres.py @@ -37,6 +37,7 @@ REQUIRED_CONNECTION_FIELDS, ) from src.utils.postgres_base import PsycopgError, PostgresBase +from src.utils.utils import Pagination, QueryRow from src.writers.writer import HealthCheckError logger = logging.getLogger(__name__) @@ -89,7 +90,7 @@ def read_stats( timestamp_end: int | None = None, cursor: int | None = None, limit: int = POSTGRES_DEFAULT_LIMIT, - ) -> tuple[list[dict[str, Any]], dict[str, Any]]: + ) -> tuple[list[QueryRow], Pagination]: """Query run/job statistics with keyset pagination. Args: timestamp_start: Start of time window in epoch milliseconds. @@ -146,7 +147,7 @@ def read_stats( rows = [self._format_row(row) for row in rows] - pagination: dict[str, Any] = { + pagination: Pagination = { "cursor": next_cursor, "has_more": has_more, "limit": limit, @@ -204,7 +205,7 @@ def read_named_query( timestamp_end: int | None = None, cursor: int | None = None, limit: int = POSTGRES_DEFAULT_LIMIT, - ) -> tuple[list[dict[str, Any]], dict[str, Any]]: + ) -> tuple[list[QueryRow], Pagination]: """Execute a predefined named query with keyset pagination. Args: query_name: Registered named query identifier (see `SUPPORTED_QUERIES`). @@ -262,7 +263,7 @@ def read_named_query( rows = [self._format_runs_jobs_detail_row(row) for row in rows] - pagination: dict[str, Any] = { + pagination: Pagination = { "cursor": next_cursor, "has_more": has_more, "limit": limit, diff --git a/src/utils/utils.py b/src/utils/utils.py index 3897d29..a779015 100644 --- a/src/utils/utils.py +++ b/src/utils/utils.py @@ -20,7 +20,7 @@ import logging import time from collections.abc import Callable -from typing import Any +from typing import Any, TypedDict import boto3 @@ -28,6 +28,16 @@ logger = logging.getLogger(__name__) +type QueryRow = dict[str, Any] + + +class Pagination(TypedDict): + """Keyset pagination metadata for a paginated query result.""" + + cursor: int | None + has_more: bool + limit: int + def build_error_response(status: int, err_type: str, message: str) -> dict[str, Any]: """Build a standardized JSON error response body. @@ -51,6 +61,29 @@ def build_error_response(status: int, err_type: str, message: str) -> dict[str, } +def build_success_response(data: list[QueryRow], pagination: Pagination) -> dict[str, Any]: + """Build a standardized JSON success response body for a paginated query result. + Args: + data: The rows returned by the query. + pagination: Pagination metadata (e.g. `cursor`, `has_more`, `limit`). + Returns: + A dictionary compatible with API Gateway Lambda Proxy integration. + """ + return { + "statusCode": 200, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps( + { + "success": True, + "statusCode": 200, + "data": data, + "pagination": pagination, + }, + default=str, + ), + } + + def resolve_request_topic(event: dict[str, Any]) -> tuple[str, dict[str, Any] | None]: """Resolve the `topic_name` path parameter and bind it as a request scoped log key. Shared by every `/{...}/{topic_name}` handler so the rejection message, the status code and From 78283ffe65ebf12ea76c3f4f145cfb67fd52b330 Mon Sep 17 00:00:00 2001 From: "Tobias.Mikula" Date: Tue, 15 Sep 2026 12:59:58 +0200 Subject: [PATCH 3/4] Adding integration test for new endpoint --- DEVELOPER.md | 2 + tests/integration/conftest.py | 19 +++ .../integration/test_named_query_endpoint.py | 157 ++++++++++++++++++ 3 files changed, 178 insertions(+) create mode 100644 tests/integration/test_named_query_endpoint.py diff --git a/DEVELOPER.md b/DEVELOPER.md index 69f9241..27da002 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -27,6 +27,8 @@ EventGate ships two Lambda functions: ## Prerequisites - Python 3.13 (current required runtime) - Docker (for local integration tests using testcontainers) +- Flyway CLI 13.x, Community edition (for local integration tests). + Requires a JDK 17+ (CI uses temurin 21). See [database/README.md](database/README.md). ## Set Up Python Environment ```shell diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index c62bf4e..3dfe71a 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -535,3 +535,22 @@ def post_stats( headers=headers, path_parameters={"topic_name": topic_name}, ) + + def post_named_query( + self, + topic_name: str, + query_name: str, + body: Dict[str, Any], + token: Optional[str] = None, + ) -> Dict[str, Any]: + """Execute a predefined named query for a topic.""" + headers = {} + if token: + headers["Authorization"] = f"Bearer {token}" + return self.invoke( + "/stats/{topic_name}/query/{query_name}", + "POST", + body=body, + headers=headers, + path_parameters={"topic_name": topic_name, "query_name": query_name}, + ) diff --git a/tests/integration/test_named_query_endpoint.py b/tests/integration/test_named_query_endpoint.py new file mode 100644 index 0000000..48730de --- /dev/null +++ b/tests/integration/test_named_query_endpoint.py @@ -0,0 +1,157 @@ +# +# 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 json +import time +import uuid +from typing import Any + +import pytest + +from tests.integration.conftest import EventGateTestClient, EventStatsTestClient + +_QUERY = "runs_jobs_detail" + + +def _post_job( + client: EventGateTestClient, + token: str, + status: str, + message: str | None, + source_app: str, +) -> dict[str, Any]: + """Post a single-job run event with an explicit status and message.""" + now_ms = int(time.time() * 1000) + job: dict[str, Any] = { + "catalog_id": "db.schema.table", + "status": status, + "timestamp_start": now_ms - 60000, + "timestamp_end": now_ms, + } + if message is not None: + job["message"] = message + event = { + "event_id": str(uuid.uuid4()), + "job_ref": "spark-cq-001", + "tenant_id": "CQ_TEST", + "source_app": source_app, + "source_app_version": "2.0.0", + "environment": "test", + "timestamp_start": now_ms - 60000, + "timestamp_end": now_ms, + "jobs": [job], + } + response = client.post_event("public.cps.za.runs", event, token=token) + assert 202 == response["statusCode"], f"Seed event failed: {response}" + return event + + +def test_unknown_query_name_returns_400(stats_client: EventStatsTestClient) -> None: + """Test that an unknown query_name returns 400, not 500.""" + response = stats_client.post_named_query("public.cps.za.runs", "does_not_exist", {}) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "does_not_exist" in body["errors"][0]["message"] + + +def test_unsupported_topic_returns_400(stats_client: EventStatsTestClient) -> None: + """Test that a known but unsupported topic returns 400.""" + response = stats_client.post_named_query("public.cps.za.test", _QUERY, {}) + + assert 400 == response["statusCode"] + + +def test_nonexistent_topic_returns_404(stats_client: EventStatsTestClient) -> None: + """Test that an unknown topic returns 404.""" + response = stats_client.post_named_query("nonexistent.topic", _QUERY, {}) + + assert 404 == response["statusCode"] + + +class TestNamedQueryRunStatus: + """End-to-end run_status categorisation for runs_jobs_detail (DIV-2).""" + + @pytest.fixture(scope="class", autouse=True) + def seed_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> None: + """Seed jobs covering every run_status bucket.""" + _post_job(eventgate_client, valid_token, "succeeded", None, "cq-status-test") + _post_job(eventgate_client, valid_token, "failed", "boom", "cq-status-test") + _post_job(eventgate_client, valid_token, "killed", None, "cq-status-test") + _post_job(eventgate_client, valid_token, "skipped", None, "cq-status-test") + _post_job(eventgate_client, valid_token, "failed", "connection Timeout after 60s", "cq-status-test") + _post_job(eventgate_client, valid_token, "succeeded", "No Data found in source", "cq-status-test") + _post_job(eventgate_client, valid_token, "failed", "there were no records to send", "cq-status-test") + + def _run_statuses_for_tenant(self, stats_client: EventStatsTestClient) -> set[str]: + """Return the set of run_status values seeded by this test class.""" + response = stats_client.post_named_query("public.cps.za.runs", _QUERY, {"limit": 1000}) + assert 200 == response["statusCode"] + body = json.loads(response["body"]) + return {row["run_status"] for row in body["data"] if row["formatted_tenant"] == "cq_test"} + + def test_returns_200_with_computed_columns(self, stats_client: EventStatsTestClient) -> None: + """Test the named query returns 200 with the computed columns.""" + response = stats_client.post_named_query("public.cps.za.runs", _QUERY, {"limit": 1000}) + + assert 200 == response["statusCode"] + body = json.loads(response["body"]) + assert True is body["success"] + row = next(r for r in body["data"] if r["formatted_tenant"] == "cq_test") + for column in ("run_date", "run_status", "formatted_tenant", "elapsed_time", "start_time", "end_time"): + assert column in row + + def test_raw_statuses_preserved(self, stats_client: EventStatsTestClient) -> None: + """Test that succeeded/failed/killed/skipped are preserved, not collapsed.""" + statuses = self._run_statuses_for_tenant(stats_client) + + assert {"succeeded", "failed", "killed", "skipped"}.issubset(statuses) + + def test_message_buckets_derived(self, stats_client: EventStatsTestClient) -> None: + """Test that timeout / no data received / no data produced buckets are derived.""" + statuses = self._run_statuses_for_tenant(stats_client) + + assert {"timeout", "no data received", "no data produced"}.issubset(statuses) + + +class TestNamedQueryPagination: + """Keyset pagination for runs_jobs_detail.""" + + @pytest.fixture(scope="class", autouse=True) + def seed_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> None: + """Seed enough jobs to page through.""" + for _ in range(5): + _post_job(eventgate_client, valid_token, "succeeded", None, "cq-pagination-test") + + def test_limit_and_cursor_page_through(self, stats_client: EventStatsTestClient) -> None: + """Test that limit caps the page and the cursor fetches the next page.""" + resp1 = stats_client.post_named_query("public.cps.za.runs", _QUERY, {"limit": 2}) + body1 = json.loads(resp1["body"]) + + assert 200 == resp1["statusCode"] + assert len(body1["data"]) <= 2 + assert True is body1["pagination"]["has_more"] + cursor = body1["pagination"]["cursor"] + assert cursor is not None + + resp2 = stats_client.post_named_query("public.cps.za.runs", _QUERY, {"limit": 2, "cursor": cursor}) + body2 = json.loads(resp2["body"]) + + assert 200 == resp2["statusCode"] + first_page_ids = {row["internal_id"] for row in body1["data"]} + second_page_ids = {row["internal_id"] for row in body2["data"]} + assert first_page_ids.isdisjoint(second_page_ids) From 3437185aec2e6d030736e528188f6e8f77d308d6 Mon Sep 17 00:00:00 2001 From: "Tobias.Mikula" Date: Tue, 15 Sep 2026 13:51:21 +0200 Subject: [PATCH 4/4] Updating README.md --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 02bb563..dc4e68a 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ All responses are JSON unless otherwise noted. The POST endpoint requires a vali | GET | `/topics/{topicName}` | none | Returns JSON Schema for the topic | | POST | `/topics/{topicName}` | JWT | Validates + forwards message to configured sinks | | POST | `/stats/{topicName}` | none | Queries ingested events with filtering, sorting, and cursor pagination | +| POST | `/stats/{topicName}/query/{queryName}` | none | Executes a predefined named query with cursor pagination | | POST | `/terminate` | (internal) | Forces Lambda process exit (used to trigger cold start & config reload) | Status codes: