From 0f7a40a910df9d3ebba7620114c1f5a8b78877e1 Mon Sep 17 00:00:00 2001 From: "Tobias.Mikula" Date: Wed, 4 Nov 2026 12:01:48 +0100 Subject: [PATCH] refactor(readers): SQL-driven named query discovery via aiosql --- .github/copilot-instructions.md | 10 +- src/handlers/handler_named_query.py | 22 ++- src/readers/named_query_registry.py | 79 ---------- src/readers/query_catalog.py | 56 ++++++++ src/readers/reader_postgres.py | 77 ++++------ src/readers/sql/aggregation_queries.sql | 6 +- src/readers/sql/jobs_detail_queries.sql | 4 +- src/utils/constants.py | 4 - .../integration/test_named_query_endpoint.py | 8 -- .../unit/handlers/test_handler_named_query.py | 135 +++++++++--------- tests/unit/readers/test_aggregation_models.py | 43 ------ tests/unit/readers/test_query_catalog.py | 86 +++++++++++ tests/unit/readers/test_reader_named_query.py | 92 ++++++++---- 13 files changed, 335 insertions(+), 287 deletions(-) delete mode 100644 src/readers/named_query_registry.py create mode 100644 src/readers/query_catalog.py delete mode 100644 tests/unit/readers/test_aggregation_models.py create mode 100644 tests/unit/readers/test_query_catalog.py diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index f474ce9..be78d14 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -7,9 +7,15 @@ Structure - Main lambda (event ingestion): `src/event_gate_lambda.py` - Stats lambda (read-only queries): `src/event_stats_lambda.py` - Shared config loading: `src/utils/config_loader.py` -- Handlers: `src/handlers/` (HandlerApi, HandlerToken, HandlerTopic, HandlerHealth, HandlerStats) +- Handlers: `src/handlers/` (HandlerApi, HandlerToken, HandlerTopic, HandlerHealth, HandlerStats, HandlerNamedQuery) - Writers: `src/writers/` (inherit from `Writer` base class) - Readers: `src/readers/` (read-only database access for stats) +- Named queries: `POST /stats/{topic}/query/{query_name}`; public names are discovered from `src/readers/sql/*.sql` by `src/readers/query_catalog.py`, no Python registry + - `jobs_detail_queries.sql`: `` + `_with_cursor` (paginated) + - `aggregation_queries.sql`: `_summary` + `_timeline_hourly` + `_timeline_daily` (no `cursor`/`limit`) + - Both families accept optional `source_app` and `tenant_id` filters (exact match, `null` = no filter) + - Public names must not clash across families; `stats.sql` queries stay private + - aiosql methods are dynamic: access with `getattr(queries, "name").sql` - Config: `conf/config.json`, `conf/access.json`, `conf/topic_schemas/*.json` Python style @@ -52,6 +58,8 @@ Testing - 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 +- Keep tests minimal: only the most direct cases per behavior; no exhaustive parametrization, no tests of dataclass defaults, or duplicates of cases covered at another layer +- Do not repeat a unit-covered case in integration tests Quality gates (run after changes, fix only if below threshold) - Run all quality gates at once: `make qa` diff --git a/src/handlers/handler_named_query.py b/src/handlers/handler_named_query.py index a34cce9..682b8c7 100644 --- a/src/handlers/handler_named_query.py +++ b/src/handlers/handler_named_query.py @@ -21,7 +21,6 @@ from dataclasses import asdict, dataclass from typing import Any -from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, SUPPORTED_JOBS_DETAIL_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_aggregations_response, build_error_response, build_success_response @@ -92,13 +91,25 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: if error_response := self._validate_event_path_params(topic_name, query_name): return error_response - if query_name in SUPPORTED_AGGREGATION_QUERIES: + try: + family = self.reader_postgres.get_query_family(query_name) + except RuntimeError: + logger.exception("Internal configuration error.", extra={"query_name": query_name}) + return build_error_response( + 500, "configuration", "Internal configuration error. Please contact the service owners." + ) + + if family is None: + logger.warning("Request rejected: query is not supported.", extra={"query_name": query_name}) + return build_error_response(400, "validation", f"Query '{query_name}' is not supported.") + + if family == "aggregation": return self._handle_aggregation_query(event.get("body"), query_name, topic_name) return self._handle_jobs_detail_query(event.get("body"), query_name, topic_name) def _handle_jobs_detail_query(self, body: str | None, query_name: str, topic_name: str) -> dict[str, Any]: - """Handle a jobs-detail named query (`SUPPORTED_JOBS_DETAIL_QUERIES`). + """Handle a discovered jobs-detail named query. Args: body: The raw request body (JSON string) from the API Gateway event. query_name: The lower-cased, already-validated `query_name` path parameter. @@ -127,7 +138,7 @@ def _handle_jobs_detail_query(self, body: str | None, query_name: str, topic_nam return build_success_response(rows, pagination) def _handle_aggregation_query(self, body: str | None, query_name: str, topic_name: str) -> dict[str, Any]: - """Handle a window-wide aggregation named query (`SUPPORTED_AGGREGATION_QUERIES`). + """Handle a discovered window-wide aggregation named query. Args: body: The raw request body (JSON string) from the API Gateway event. query_name: The lower-cased, already-validated `query_name` path parameter. @@ -173,9 +184,6 @@ def _validate_event_path_params(self, topic_name: str, query_name: str) -> dict[ if not query_name: return build_error_response(400, "validation", "Missing path parameter 'query_name'.") - if query_name not in SUPPORTED_JOBS_DETAIL_QUERIES and query_name not in SUPPORTED_AGGREGATION_QUERIES: - return build_error_response(400, "validation", f"Query '{query_name}' is not supported.") - return None @staticmethod diff --git a/src/readers/named_query_registry.py b/src/readers/named_query_registry.py deleted file mode 100644 index f6291d8..0000000 --- a/src/readers/named_query_registry.py +++ /dev/null @@ -1,79 +0,0 @@ -# -# 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. - -`SUPPORTED_JOBS_DETAIL_QUERIES` maps each jobs-detail `query_name` to its aiosql query keys, with and -without a keyset-pagination cursor. `SUPPORTED_AGGREGATION_QUERIES` holds window-wide -aggregation queries instead, which have no pagination and a different response shape. -Names absent from both are rejected as a `400` at the handler layer. New named queries -are added here, not as new routes. -""" - -from dataclasses import dataclass - -from src.utils.constants import QUERY_RUNS_JOBS_DETAIL, QUERY_RUNS_JOBS_DETAIL_AGGREGATION - - -@dataclass(frozen=True) -class JobsDetailQuery: - """Definition of a single jobs-detail named query with keyset pagination. - - 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_JOBS_DETAIL_QUERIES: dict[str, JobsDetailQuery] = { - QUERY_RUNS_JOBS_DETAIL: JobsDetailQuery( - name=QUERY_RUNS_JOBS_DETAIL, - sql_key="get_runs_jobs_detail", - sql_key_with_cursor="get_runs_jobs_detail_with_cursor", - ), -} - - -@dataclass(frozen=True) -class AggregationQuery: - """Definition of a single window-wide aggregation named query. - - Attributes: - name: Stable identifier used in the request path. - summary_sql_key: aiosql query name for the single-row summary aggregates. - timeline_hourly_sql_key: aiosql query name for hourly timeline buckets. - timeline_daily_sql_key: aiosql query name for daily timeline buckets. - """ - - name: str - summary_sql_key: str - timeline_hourly_sql_key: str - timeline_daily_sql_key: str - - -SUPPORTED_AGGREGATION_QUERIES: dict[str, AggregationQuery] = { - QUERY_RUNS_JOBS_DETAIL_AGGREGATION: AggregationQuery( - name=QUERY_RUNS_JOBS_DETAIL_AGGREGATION, - summary_sql_key="get_runs_jobs_aggregations", - timeline_hourly_sql_key="get_runs_jobs_timeline_hourly", - timeline_daily_sql_key="get_runs_jobs_timeline_daily", - ), -} diff --git a/src/readers/query_catalog.py b/src/readers/query_catalog.py new file mode 100644 index 0000000..299486c --- /dev/null +++ b/src/readers/query_catalog.py @@ -0,0 +1,56 @@ +# +# 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. +# + +"""Discover public query groups from aiosql's family namespaces. + +Detail groups contain `` and `_with_cursor`; aggregation groups contain +`_summary`, `_timeline_hourly`, and `_timeline_daily`. +Only complete groups are public. Root stats methods and generated cursor helpers are private. +""" + +from typing import Literal + +from aiosql.queries import Queries + +type QueryFamily = Literal["jobs_detail", "aggregation"] + +_FAMILY_SUFFIXES: dict[QueryFamily, tuple[str, ...]] = { + "jobs_detail": ("_with_cursor", ""), + "aggregation": ("_summary", "_timeline_hourly", "_timeline_daily"), +} + + +def validate_query_groups(queries: Queries) -> None: + """Reject incomplete, unexpected, or conflicting public SQL groups.""" + public_names: set[str] = set() + for family, suffixes in _FAMILY_SUFFIXES.items(): + available = set(getattr(queries, family).available_queries) + names = {name.removesuffix(suffixes[0]) for name in available if name.endswith(suffixes[0])} + # Every authored SELECT also has an aiosql-generated `_cursor` method. + expected = {f"{name}{suffix}{helper}" for name in names for suffix in suffixes for helper in ("", "_cursor")} + if available != expected or any(not name or name.endswith("_cursor") or name != name.lower() for name in names): + raise ValueError(f"Invalid SQL query groups in {family}: {', '.join(sorted(available ^ expected))}.") + if conflicts := public_names & names: + raise ValueError(f"Conflicting public SQL query names: {', '.join(sorted(conflicts))}.") + public_names.update(names) + + +def query_family(queries: Queries, query_name: str) -> QueryFamily | None: + """Classify a public name in a validated query tree without exposing helper methods.""" + for family, suffixes in _FAMILY_SUFFIXES.items(): + if f"{query_name}{suffixes[0]}" in getattr(queries, family).available_queries: + return family + return None diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py index c08c56e..6b3e654 100644 --- a/src/readers/reader_postgres.py +++ b/src/readers/reader_postgres.py @@ -18,7 +18,6 @@ import logging import time -from dataclasses import dataclass from datetime import datetime, timezone from functools import cached_property from pathlib import Path @@ -26,10 +25,12 @@ import aiosql import psycopg2.extensions +from aiosql.queries import Queries +from aiosql.utils import SQLLoadException, SQLParseException from botocore.exceptions import BotoCoreError, ClientError from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations, Timeline, TimelineEntry -from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, SUPPORTED_JOBS_DETAIL_QUERIES +from src.readers.query_catalog import QueryFamily, query_family, validate_query_groups from src.utils.constants import ( AGGREGATION_BUCKET_THRESHOLD_MS, POSTGRES_DEFAULT_LIMIT, @@ -53,19 +54,6 @@ type AggregationQueryResult = tuple[dict[str, Any], list[dict[str, Any]]] -@dataclass(frozen=True) -class ReaderQueries: - """Typed holder for reader SQL query strings loaded via aiosql.""" - - get_stats: str - get_stats_with_cursor: str - get_runs_jobs_detail: str - get_runs_jobs_detail_with_cursor: str - get_runs_jobs_aggregations: str - get_runs_jobs_timeline_hourly: str - get_runs_jobs_timeline_daily: str - - class ReaderPostgres(PostgresBase): """Read-only Postgres accessor for run/job statistics.""" @@ -78,22 +66,20 @@ def _connect_options(self) -> str | None: return f"-c statement_timeout={POSTGRES_STATEMENT_TIMEOUT_MS} -c default_transaction_read_only=on" @cached_property - def _queries(self) -> ReaderQueries: - """Load SQL queries from the `sql/` directory via aiosql.""" - queries = aiosql.from_path(_SQL_DIR, "psycopg2") - # aiosql loads SQL files and attaches query functions as dynamic attributes at runtime. - # pylint cannot resolve these statically. - 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 - ), - get_runs_jobs_aggregations=queries.get_runs_jobs_aggregations.sql, # pylint: disable=no-member - get_runs_jobs_timeline_hourly=queries.get_runs_jobs_timeline_hourly.sql, # pylint: disable=no-member - get_runs_jobs_timeline_daily=queries.get_runs_jobs_timeline_daily.sql, # pylint: disable=no-member - ) + def _queries(self) -> Queries: + """Load native aiosql queries with separate public family namespaces.""" + try: + queries = aiosql.from_path(_SQL_DIR / "stats.sql", "psycopg2") + queries.add_child_queries("jobs_detail", aiosql.from_path(_SQL_DIR / "jobs_detail_queries.sql", "psycopg2")) + queries.add_child_queries("aggregation", aiosql.from_path(_SQL_DIR / "aggregation_queries.sql", "psycopg2")) + validate_query_groups(queries) + except (OSError, ValueError, SQLLoadException, SQLParseException) as exc: + raise RuntimeError(f"SQL query configuration error: {exc}") from exc + return queries + + def get_query_family(self, query_name: str) -> QueryFamily | None: + """Discover a public query's family without accessing database configuration.""" + return query_family(self._queries, query_name) def _require_connection_config(self) -> PostgresConfig: """Load and validate the PostgreSQL connection config, or raise. @@ -198,12 +184,12 @@ def _run_stats_query( with connection.cursor() as db_cursor: if cursor is not None: db_cursor.execute( - self._queries.get_stats_with_cursor, + getattr(self._queries, "get_stats_with_cursor").sql, {"ts_start": ts_start, "ts_end": ts_end, "cursor_id": cursor, "lim": limit + 1}, ) else: db_cursor.execute( - self._queries.get_stats, + getattr(self._queries, "get_stats").sql, {"ts_start": ts_start, "ts_end": ts_end, "lim": limit + 1}, ) if db_cursor.description is None: @@ -232,7 +218,7 @@ def read_named_query( ) -> tuple[list[QueryRow], Pagination]: """Execute a predefined named query with keyset pagination. Args: - query_name: Registered named query identifier (see `SUPPORTED_JOBS_DETAIL_QUERIES`). + query_name: Public query identifier discovered from the jobs-detail SQL file. 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). @@ -245,8 +231,7 @@ def read_named_query( Raises: RuntimeError: On unknown query name or database connectivity/query errors. """ - query = SUPPORTED_JOBS_DETAIL_QUERIES.get(query_name) - if query is None: + if self.get_query_family(query_name) != "jobs_detail": raise RuntimeError(f"Unknown named query: {query_name}.") self._require_connection_config() @@ -256,8 +241,9 @@ def read_named_query( 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) + queries = getattr(self._queries, "jobs_detail") + sql = getattr(queries, query_name).sql + sql_with_cursor = getattr(queries, f"{query_name}_with_cursor").sql try: col_names, raw_rows = self._execute_with_retry( @@ -341,7 +327,7 @@ def read_aggregations( ) -> RunsJobsAggregations: """Execute a predefined window-wide aggregation named query. Args: - query_name: Registered aggregation query identifier. + query_name: Public query identifier discovered from the aggregation SQL file. 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*). source_app: Exact source application filter, or `None` for all applications. @@ -351,8 +337,7 @@ def read_aggregations( Raises: RuntimeError: On unknown query name or database connectivity/query errors. """ - query = SUPPORTED_AGGREGATION_QUERIES.get(query_name) - if query is None: + if self.get_query_family(query_name) != "aggregation": raise RuntimeError(f"Unknown aggregation query: {query_name}.") self._require_connection_config() @@ -363,12 +348,10 @@ def read_aggregations( window_ms = ts_end - ts_start bucket = TIMELINE_BUCKET_HOUR if window_ms <= AGGREGATION_BUCKET_THRESHOLD_MS else TIMELINE_BUCKET_DAY - timeline_sql_key = ( - query.timeline_hourly_sql_key if bucket == TIMELINE_BUCKET_HOUR else (query.timeline_daily_sql_key) - ) - - summary_sql = getattr(self._queries, query.summary_sql_key) - timeline_sql = getattr(self._queries, timeline_sql_key) + timeline_suffix = "timeline_hourly" if bucket == TIMELINE_BUCKET_HOUR else "timeline_daily" + queries = getattr(self._queries, "aggregation") + summary_sql = getattr(queries, f"{query_name}_summary").sql + timeline_sql = getattr(queries, f"{query_name}_{timeline_suffix}").sql try: summary_row, timeline_rows = self._execute_with_retry( diff --git a/src/readers/sql/aggregation_queries.sql b/src/readers/sql/aggregation_queries.sql index 388f9a6..067e6c4 100644 --- a/src/readers/sql/aggregation_queries.sql +++ b/src/readers/sql/aggregation_queries.sql @@ -1,4 +1,4 @@ --- name: get_runs_jobs_aggregations(ts_start, ts_end, source_app, tenant_id) +-- name: runs_jobs_detail_aggregation_summary(ts_start, ts_end, source_app, tenant_id) -- Window-wide summary aggregates for runs_jobs_detail (job-row grain). SELECT COUNT(*) AS total_runs, COUNT(DISTINCT r.tenant_id) AS total_tenants, @@ -15,7 +15,7 @@ SELECT COUNT(*) AS total_runs, AND (:source_app IS NULL OR r.source_app = :source_app) AND (:tenant_id IS NULL OR r.tenant_id = :tenant_id); --- name: get_runs_jobs_timeline_hourly(ts_start, ts_end, source_app, tenant_id) +-- name: runs_jobs_detail_aggregation_timeline_hourly(ts_start, ts_end, source_app, tenant_id) -- Hourly timeline buckets (non-empty buckets only) for runs_jobs_detail, oldest first. SELECT (EXTRACT(EPOCH FROM date_trunc('hour', to_timestamp(r.timestamp_start / 1000.0), 'UTC')) * 1000)::bigint AS bucket_start, @@ -32,7 +32,7 @@ SELECT (EXTRACT(EPOCH FROM date_trunc('hour', to_timestamp(r.timestamp_start / 1 GROUP BY bucket_start ORDER BY bucket_start ASC; --- name: get_runs_jobs_timeline_daily(ts_start, ts_end, source_app, tenant_id) +-- name: runs_jobs_detail_aggregation_timeline_daily(ts_start, ts_end, source_app, tenant_id) -- Daily timeline buckets (non-empty buckets only) for runs_jobs_detail, oldest first. SELECT (EXTRACT(EPOCH FROM date_trunc('day', to_timestamp(r.timestamp_start / 1000.0), 'UTC')) * 1000)::bigint AS bucket_start, diff --git a/src/readers/sql/jobs_detail_queries.sql b/src/readers/sql/jobs_detail_queries.sql index c665c7c..0897772 100644 --- a/src/readers/sql/jobs_detail_queries.sql +++ b/src/readers/sql/jobs_detail_queries.sql @@ -1,4 +1,4 @@ --- name: get_runs_jobs_detail(ts_start, ts_end, lim, source_app, tenant_id) +-- name: runs_jobs_detail(ts_start, ts_end, lim, source_app, tenant_id) 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, @@ -13,7 +13,7 @@ SELECT r.event_id, r.job_ref, r.tenant_id, r.source_app, ORDER BY j.internal_id DESC LIMIT :lim; --- name: get_runs_jobs_detail_with_cursor(ts_start, ts_end, cursor_id, lim, source_app, tenant_id) +-- name: runs_jobs_detail_with_cursor(ts_start, ts_end, cursor_id, lim, source_app, tenant_id) 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, diff --git a/src/utils/constants.py b/src/utils/constants.py index ea140b6..c28fe4f 100644 --- a/src/utils/constants.py +++ b/src/utils/constants.py @@ -42,10 +42,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" -QUERY_RUNS_JOBS_DETAIL_AGGREGATION = "runs_jobs_detail_aggregation" - # Named query aggregations AGGREGATION_BUCKET_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000 # 7 days in milliseconds TIMELINE_BUCKET_HOUR = "hour" diff --git a/tests/integration/test_named_query_endpoint.py b/tests/integration/test_named_query_endpoint.py index 9ebdbff..7e4f4b5 100644 --- a/tests/integration/test_named_query_endpoint.py +++ b/tests/integration/test_named_query_endpoint.py @@ -326,14 +326,6 @@ def test_returns_populated_aggregations(self, stats_client: EventStatsTestClient assert timeline["bucket"] in ("hour", "day") assert sum(entry["total"] for entry in timeline["entries"]) == aggregations["total_runs"] - def test_cursor_or_limit_returns_400(self, stats_client: EventStatsTestClient) -> None: - """Test that cursor/limit are rejected for the aggregation query (not silently ignored).""" - response = stats_client.post_named_query("public.cps.za.runs", self._AGG_QUERY, {"limit": 10}) - - assert 400 == response["statusCode"] - body = json.loads(response["body"]) - assert "limit" in body["errors"][0]["message"] - class TestNamedQueryPagination: """Keyset pagination for runs_jobs_detail.""" diff --git a/tests/unit/handlers/test_handler_named_query.py b/tests/unit/handlers/test_handler_named_query.py index 4d01432..8f3194d 100644 --- a/tests/unit/handlers/test_handler_named_query.py +++ b/tests/unit/handlers/test_handler_named_query.py @@ -20,16 +20,20 @@ from typing import Any from unittest.mock import MagicMock +from aiosql.utils import SQLLoadException, SQLParseException import pytest +from pytest_mock import MockerFixture from src.handlers.handler_named_query import AggregationQueryParams, HandlerNamedQuery, JobsDetailQueryParams from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations +from src.readers.reader_postgres import ReaderPostgres @pytest.fixture def mock_reader() -> MagicMock: """Mock ReaderPostgres.""" - mock = MagicMock() + mock = MagicMock(spec=ReaderPostgres) + mock.get_query_family.side_effect = ReaderPostgres().get_query_family mock.read_named_query.return_value = ( [{"event_id": "ev1", "internal_id": 1, "status": "succeeded", "run_status": "succeeded"}], {"cursor": None, "has_more": False, "limit": 50}, @@ -100,23 +104,6 @@ def test_forwards_query_name_and_params_to_reader(self, handler: HandlerNamedQue class TestHandlerNamedQueryValidation: """Tests for request validation.""" - @pytest.mark.parametrize("query_name", ["runs_jobs_detail", "runs_jobs_detail_aggregation"]) - @pytest.mark.parametrize("field_name", ["source_app", "tenant_id"]) - @pytest.mark.parametrize("value", [None, "", "ordinary", "MixedCase"]) - def test_string_or_null_filters_are_forwarded( - self, - handler: HandlerNamedQuery, - mock_reader: MagicMock, - query_name: str, - field_name: str, - value: str | None, - ) -> None: - response = handler.handle_request(_make_event(query_name=query_name, body={field_name: value})) - - assert 200 == response["statusCode"] - read = mock_reader.read_aggregations if query_name.endswith("_aggregation") else mock_reader.read_named_query - assert value == read.call_args.kwargs[field_name] - @pytest.mark.parametrize("query_name", ["runs_jobs_detail", "runs_jobs_detail_aggregation"]) def test_both_filters_and_their_defaults_are_forwarded( self, handler: HandlerNamedQuery, mock_reader: MagicMock, query_name: str @@ -135,7 +122,7 @@ def test_both_filters_and_their_defaults_are_forwarded( @pytest.mark.parametrize("query_name", ["runs_jobs_detail", "runs_jobs_detail_aggregation"]) @pytest.mark.parametrize("field_name", ["source_app", "tenant_id"]) - @pytest.mark.parametrize("value", [123, True, ["app"], {"name": "app"}]) + @pytest.mark.parametrize("value", [123]) def test_invalid_filter_is_rejected_before_reading( self, handler: HandlerNamedQuery, @@ -178,31 +165,6 @@ def test_reversed_window_is_rejected_before_reading( record.getMessage() for record in caplog.records ] - @pytest.mark.parametrize("query_name", ["runs_jobs_detail", "runs_jobs_detail_aggregation"]) - @pytest.mark.parametrize( - "bounds", - [ - {"timestamp_start": 1000, "timestamp_end": 2000}, - {"timestamp_start": 1000, "timestamp_end": 1000}, - {}, - {"timestamp_start": 1000}, - {"timestamp_end": 1000}, - {"timestamp_start": None, "timestamp_end": 1000}, - {"timestamp_start": 1000, "timestamp_end": None}, - {"timestamp_start": None, "timestamp_end": None}, - ], - ) - def test_valid_or_defaulted_bounds_are_forwarded( - self, handler: HandlerNamedQuery, mock_reader: MagicMock, query_name: str, bounds: dict[str, int | None] - ) -> None: - """Keep equality, ordered bounds, and reader-side defaults unchanged.""" - response = handler.handle_request(_make_event(query_name=query_name, body=bounds)) - - assert 200 == response["statusCode"] - read = mock_reader.read_aggregations if query_name.endswith("_aggregation") else mock_reader.read_named_query - assert bounds.get("timestamp_start") == read.call_args.kwargs["timestamp_start"] - assert bounds.get("timestamp_end") == read.call_args.kwargs["timestamp_end"] - 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")) @@ -237,13 +199,27 @@ def test_missing_query_name_returns_400(self, handler: HandlerNamedQuery) -> Non body = json.loads(response["body"]) assert "query_name" in body["errors"][0]["message"] - def test_unknown_query_name_returns_400(self, handler: HandlerNamedQuery) -> None: + @pytest.mark.parametrize( + "query_name", + [ + "does_not_exist", + "get_stats", + "runs_jobs_detail_cursor", + "runs_jobs_detail_with_cursor", + "runs_jobs_detail_aggregation_summary", + "runs_jobs_detail_aggregation_timeline_hourly_cursor", + ], + ) + def test_unknown_query_name_returns_400( + self, handler: HandlerNamedQuery, query_name: str, caplog: pytest.LogCaptureFixture + ) -> None: """Test that an unknown query_name returns 400 (not 500).""" - response = handler.handle_request(_make_event(query_name="does_not_exist")) + with caplog.at_level(logging.WARNING): + response = handler.handle_request(_make_event(query_name=query_name)) assert 400 == response["statusCode"] body = json.loads(response["body"]) - assert "does_not_exist" in body["errors"][0]["message"] + assert f"Query '{query_name}' is not supported." == body["errors"][0]["message"] def test_invalid_json_body_returns_400(self, handler: HandlerNamedQuery) -> None: """Test that non-JSON body returns 400.""" @@ -302,6 +278,30 @@ def test_database_error_returns_500(self, handler: HandlerNamedQuery, mock_reade assert False is body["success"] assert "database" == body["errors"][0]["type"] + @pytest.mark.parametrize( + "error", [OSError("secret/path.sql"), SQLLoadException("secret", None), SQLParseException("secret")] + ) + def test_sql_configuration_failure_returns_sanitized_500( + self, + topics: dict[str, dict[str, Any]], + mocker: MockerFixture, + caplog: pytest.LogCaptureFixture, + error: Exception, + ) -> None: + """Test that a SQL load failure returns a sanitized 500 with one error log and no database access.""" + mocker.patch("src.readers.reader_postgres.aiosql.from_path", side_effect=error) + connection = mocker.patch.object(ReaderPostgres, "_get_connection") + handler = HandlerNamedQuery(topics=topics, reader_postgres=ReaderPostgres()) + + with caplog.at_level(logging.ERROR, logger="src.handlers.handler_named_query"): + response = handler.handle_request(_make_event()) + + assert 500 == response["statusCode"] + assert "configuration" == json.loads(response["body"])["errors"][0]["type"] + assert "secret" not in response["body"] + assert 1 == len(caplog.records) + connection.assert_not_called() + class TestHandlerAggregationQuery: """Tests for the window-wide aggregation branch (`runs_jobs_detail_aggregation`).""" @@ -350,22 +350,6 @@ def test_database_error_returns_500(self, handler: HandlerNamedQuery, mock_reade class TestValidateAggregationEventBody: """Tests for `HandlerNamedQuery._validate_aggregation_event_body`.""" - @pytest.mark.parametrize( - "body, expected_message_fragment", - [ - pytest.param("not json", "valid JSON", id="invalid_json"), - pytest.param("[1, 2, 3]", "JSON object", id="non_dict_json"), - pytest.param(json.dumps({"timestamp_start": "bad"}), "timestamp_start", id="invalid_timestamp_start"), - ], - ) - def test_returns_error_response_for_invalid_body(self, body: str, expected_message_fragment: str) -> None: - """Test that each invalid body returns a 400 with the expected message.""" - result = HandlerNamedQuery._validate_aggregation_event_body(body) - - assert isinstance(result, dict) - assert 400 == result["statusCode"] - assert expected_message_fragment in json.loads(result["body"])["errors"][0]["message"] - def test_returns_parsed_params_for_valid_body(self) -> None: """Test that a valid body with only `timestamp_start`/`timestamp_end` parses correctly.""" result = HandlerNamedQuery._validate_aggregation_event_body( @@ -395,12 +379,6 @@ class TestValidateEventPathParams: 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( @@ -421,6 +399,25 @@ def test_returns_error_response_for_invalid_params( body = json.loads(error["body"]) assert expected_message_fragment in body["errors"][0]["message"] + @pytest.mark.parametrize( + "topic, query_name, expected_status", + [ + ("", "runs_jobs_detail", 400), + ("nonexistent.topic", "runs_jobs_detail", 404), + ("public.cps.za.test", "runs_jobs_detail", 400), + ("public.cps.za.runs", "", 400), + ], + ) + def test_invalid_path_bypasses_query_discovery( + self, handler: HandlerNamedQuery, mock_reader: MagicMock, topic: str, query_name: str, expected_status: int + ) -> None: + mock_reader.get_query_family.side_effect = AssertionError("Must not load SQL for an invalid path") + + response = handler.handle_request(_make_event(topic=topic, query_name=query_name)) + + assert expected_status == response["statusCode"] + mock_reader.get_query_family.assert_not_called() + 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" diff --git a/tests/unit/readers/test_aggregation_models.py b/tests/unit/readers/test_aggregation_models.py deleted file mode 100644 index 179c32e..0000000 --- a/tests/unit/readers/test_aggregation_models.py +++ /dev/null @@ -1,43 +0,0 @@ -# -# 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. -# - - -from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations - - -def test_defaults_are_zeroed_not_missing() -> None: - aggregations = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=1000, timestamp_end=2000)) - - assert 0 == aggregations.total_runs - assert 0 == aggregations.total_tenants - assert 0 == aggregations.total_catalogs - assert 0 == aggregations.succeeded - assert 0 == aggregations.failed - assert 0 == aggregations.killed - assert 0 == aggregations.skipped - assert 0.0 == aggregations.success_rate - assert aggregations.avg_duration_ms is None - assert "hour" == aggregations.timeline.bucket - assert [] == aggregations.timeline.entries - - -def test_default_timeline_instances_are_independent() -> None: - """Each default `RunsJobsAggregations` gets its own `Timeline`/`entries` list, not a shared mutable default.""" - first = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=1000, timestamp_end=2000)) - second = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=3000, timestamp_end=4000)) - - assert first.timeline is not second.timeline - assert first.timeline.entries is not second.timeline.entries diff --git a/tests/unit/readers/test_query_catalog.py b/tests/unit/readers/test_query_catalog.py new file mode 100644 index 0000000..b2322bb --- /dev/null +++ b/tests/unit/readers/test_query_catalog.py @@ -0,0 +1,86 @@ +# +# 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 aiosql +from aiosql.queries import Queries +import pytest + +from src.readers.query_catalog import query_family, validate_query_groups + + +def _queries(detail: tuple[str, ...], aggregation: tuple[str, ...]) -> Queries: + queries = aiosql.from_str("-- name: get_stats()\nSELECT 1;", "psycopg2") + for family, names in (("jobs_detail", detail), ("aggregation", aggregation)): + sql = "\n".join(f"-- name: {name}()\nSELECT 1;" for name in names) + queries.add_child_queries(family, aiosql.from_str(sql, "psycopg2")) + return queries + + +def test_discovers_additional_groups_without_python_registration() -> None: + queries = _queries( + ("first", "first_with_cursor", "second", "second_with_cursor"), + ("totals_summary", "totals_timeline_hourly", "totals_timeline_daily"), + ) + + validate_query_groups(queries) + + assert "jobs_detail" == query_family(queries, "first") + assert "jobs_detail" == query_family(queries, "second") + assert "aggregation" == query_family(queries, "totals") + + +@pytest.mark.parametrize( + "name", + [ + "get_stats", + "detail_with_cursor", + "totals_summary", + "unknown", + ], +) +def test_private_methods_are_not_public_queries(name: str) -> None: + queries = _queries( + ("detail", "detail_with_cursor"), + ("totals_summary", "totals_timeline_hourly", "totals_timeline_daily"), + ) + validate_query_groups(queries) + + assert query_family(queries, name) is None + + +@pytest.mark.parametrize( + "detail, aggregation", + [ + (("detail",), ()), + ((), ("totals_summary", "totals_timeline_hourly")), + (("Detail", "Detail_with_cursor"), ()), + ], +) +def test_rejects_incomplete_or_unexpected_groups(detail: tuple[str, ...], aggregation: tuple[str, ...]) -> None: + queries = _queries(detail, aggregation) + + with pytest.raises(ValueError, match="Invalid SQL query groups"): + validate_query_groups(queries) + + +def test_rejects_names_shared_by_both_families() -> None: + queries = _queries( + ("shared", "shared_with_cursor"), + ("shared_summary", "shared_timeline_hourly", "shared_timeline_daily"), + ) + + with pytest.raises(ValueError, match="Conflicting public SQL query names: shared"): + validate_query_groups(queries) diff --git a/tests/unit/readers/test_reader_named_query.py b/tests/unit/readers/test_reader_named_query.py index 27ce8a1..22be8b1 100644 --- a/tests/unit/readers/test_reader_named_query.py +++ b/tests/unit/readers/test_reader_named_query.py @@ -16,11 +16,15 @@ import json +from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch +import aiosql import pytest +from pytest_mock import MockerFixture +from src.handlers.handler_named_query import HandlerNamedQuery from src.readers.reader_postgres import ReaderPostgres import src.utils.postgres_base as pb @@ -64,6 +68,70 @@ def reader(monkeypatch: pytest.MonkeyPatch) -> ReaderPostgres: return ReaderPostgres() +def test_query_discovery_is_lazy_cached_and_independent_of_database(mocker: MockerFixture) -> None: + load = mocker.spy(aiosql, "from_path") + config = mocker.patch.object(ReaderPostgres, "_require_connection_config") + connection = mocker.patch.object(ReaderPostgres, "_get_connection") + aws_session = mocker.patch("boto3.Session") + + reader = ReaderPostgres() + HandlerNamedQuery({}, reader) + load.assert_not_called() + + assert "jobs_detail" == reader.get_query_family("runs_jobs_detail") + assert "aggregation" == reader.get_query_family("runs_jobs_detail_aggregation") + assert 3 == load.call_count + config.assert_not_called() + connection.assert_not_called() + aws_session.assert_not_called() + + +@pytest.mark.parametrize( + "method, query_name", + [ + ("read_named_query", "runs_jobs_detail_aggregation"), + ("read_aggregations", "runs_jobs_detail"), + ], +) +def test_reader_rejects_private_or_wrong_family_queries_before_config( + reader: ReaderPostgres, mocker: MockerFixture, method: str, query_name: str +) -> None: + config = mocker.patch.object(reader, "_require_connection_config") + + with pytest.raises(RuntimeError, match="Unknown"): + getattr(reader, method)(query_name) + + config.assert_not_called() + + +@pytest.mark.parametrize( + "files", + [ + pytest.param({}, id="missing_sql_files"), + pytest.param( + { + "stats.sql": "-- name: get_stats()\nSELECT 1;", + "jobs_detail_queries.sql": "-- name: detail()\nSELECT 1;", + "aggregation_queries.sql": "", + }, + id="incomplete_group", + ), + ], +) +def test_sql_configuration_failure_raises_before_database_access( + reader: ReaderPostgres, tmp_path: Path, mocker: MockerFixture, files: dict[str, str] +) -> None: + for filename, sql in files.items(): + (tmp_path / filename).write_text(sql, encoding="utf-8") + mocker.patch("src.readers.reader_postgres._SQL_DIR", tmp_path) + config = mocker.patch.object(reader, "_require_connection_config") + + with pytest.raises(RuntimeError, match="SQL query configuration error"): + reader.get_query_family("detail") + + config.assert_not_called() + + 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() @@ -306,30 +374,6 @@ def test_build_aggregations_shapes_populated_rows(self) -> None: assert 1 == aggregations.timeline.entries[0].killed assert 1 == aggregations.timeline.entries[0].skipped - def test_build_aggregations_zeroes_empty_window(self) -> None: - """Test that an empty summary/timeline still produces a fully-populated, zeroed result.""" - summary_row = { - "total_runs": 0, - "total_tenants": 0, - "total_catalogs": 0, - "succeeded": 0, - "failed": 0, - "killed": 0, - "skipped": 0, - "avg_duration_ms": None, - } - - aggregations = ReaderPostgres._build_aggregations( - ts_start=1704067200000, ts_end=1704070800000, bucket="hour", summary_row=summary_row, timeline_rows=[] - ) - - assert 0 == aggregations.total_runs - assert 0 == aggregations.killed - assert 0 == aggregations.skipped - assert 0.0 == aggregations.success_rate - assert aggregations.avg_duration_ms is None - assert [] == aggregations.timeline.entries - @pytest.mark.parametrize( "window_ms, expected_bucket, expected_fragment", [