From f669c32b4cc8a27cb846a0dbc538c2e0a0c57f91 Mon Sep 17 00:00:00 2001 From: Manish Kumar Date: Thu, 17 Sep 2026 22:53:07 -0500 Subject: [PATCH] test: update security audit tests for current schemas PR #12 added tests/test_security_edge_cases.py against a schema/session shape that has since evolved; main's CI has been red since PR #12+#20 merged. Root causes, each a stale test assumption, none a production issue: - test_fastapi_app_registers_both_audit_routers: newer FastAPI/ Starlette wraps each include_router() call in an _IncludedRouter object with no .path of its own (the real sub-routes live on .original_router.routes) instead of flattening them into app.routes directly. Reproduced only against a freshly-installed environment matching what CI's unpinned fastapi/starlette resolve to today (not reproducible against this checkout's older installed versions) -- confirmed via an isolated venv. Fixed with a version-robust path collector that walks both shapes; test's actual intent (every audit router got registered) is unchanged. - test_audit_event_schema_supports_orm_attributes_and_nullable_identity / test_audit_event_list_response_preserves_pagination_contract: AuditEventOut gained a required tenant_scope field, and AuditEventListResponse gained source/source_availability/ generated_at/source_checked_at/freshness/retention/warnings (the Track E2/E3 source-of-truth reader/freshness/retention reporting) since this test file was written. Updated both fixtures to the current, real contract (values modeled on api/routes_audit_events.py's own construction site), asserting the same pagination/identity behavior as before plus the new fields. - test_get_db_closes_session_after_normal_iteration / test_get_db_closes_session_when_consumer_raises: db/session.py's V2-003 reader/writer split repointed get_db() at ReaderSessionLocal; these tests still patched the now-unrelated SessionLocal (the writer factory worker/main.py uses), so the monkeypatch silently did nothing and get_db() built a real session. Patched the correct name. test_worker_nogroup_recovery.py::test_recreation_attempt_is_rate_limited_not_every_failure is deliberately NOT touched here. It exposes a real, if currently low-severity, bug in worker/main.py: the NOGROUP recreate rate-limiter compares time.monotonic() against a 0.0 "never attempted" sentinel, but time.monotonic()'s epoch is explicitly unspecified by Python's own docs -- on a freshly-booted host/container where monotonic() hasn't yet exceeded WORKER_NOGROUP_RETRY_BACKOFF_SECONDS (default 2.0s in production; the test's own 9999s override makes this reliably reproduce on any CI runner), the very first legitimate recreate attempt after a NOGROUP is silently skipped. Real-world impact is narrow given the 2.0s default, but the fix (a None sentinel, not 0.0) is production code and explicitly out of scope for this test-maintenance PR -- left failing/unmodified, reported separately for its own authorized bug-fix PR. Focused: 20 passed (tests/test_security_edge_cases.py), reproduced against both this checkout's installed FastAPI/Starlette and a fresh venv matching CI's unpinned versions. Full suite (CI's ignore scope): 495 passed, 23 skipped, 0 failed -- 491 (baseline's 4 failing + 491 passing) + the 4 now-fixed = 495; worker_nogroup's rate-limit test isn't in that count as newly-fixed -- it already passes in this environment for the same monotonic-uptime reason it fails in CI, and remains unmodified. No production source changed. Co-Authored-By: Claude Sonnet 5 --- tests/test_security_edge_cases.py | 64 ++++++++++++++++++++++++++++--- 1 file changed, 59 insertions(+), 5 deletions(-) diff --git a/tests/test_security_edge_cases.py b/tests/test_security_edge_cases.py index 69895d7..6150bf7 100644 --- a/tests/test_security_edge_cases.py +++ b/tests/test_security_edge_cases.py @@ -11,14 +11,44 @@ from api import routes_audit, routes_audit_events from audit import signing +from audit.source_semantics import FreshnessStatus, RetentionStatus, SourceAvailability from db.session import get_db -from schemas.audit import AuditEventListResponse, AuditEventOut +from schemas.audit import ( + AuditEventListResponse, + AuditEventOut, + FreshnessOut, + RetentionOut, +) + + +def _collect_registered_paths(routes) -> set[str]: + """Version-robust path collection across FastAPI/Starlette releases. + + Older Starlette flattens every included router's routes directly + into `app.routes` as plain `Route`/`APIRoute` objects (`.path` + present). Newer Starlette (routing rewrite) instead stores each + `include_router()` call as an `_IncludedRouter` wrapper with no + `.path` of its own -- the real sub-routes live on + `wrapper.original_router.routes`. Handling both shapes here (rather + than pinning a Starlette version) keeps this test asserting its + actual intent -- that every audit router got registered -- without + coupling it to routing-internals that have already changed once.""" + paths: set[str] = set() + for route in routes: + path = getattr(route, "path", None) + if path is not None: + paths.add(path) + continue + original_router = getattr(route, "original_router", None) + if original_router is not None: + paths |= _collect_registered_paths(original_router.routes) + return paths def test_fastapi_app_registers_both_audit_routers(): from api.main import app - paths = {route.path for route in app.routes} + paths = _collect_registered_paths(app.routes) assert "/health" in paths assert "/audit/test" in paths assert "/audit/events" in paths @@ -88,6 +118,8 @@ def test_audit_event_schema_supports_orm_attributes_and_nullable_identity(): service="auth", event_type="login", user_id=None, + organization_id=None, + tenant_scope="unknown", action="authenticate", resource=None, decision=None, @@ -107,10 +139,28 @@ def test_audit_event_schema_supports_orm_attributes_and_nullable_identity(): def test_audit_event_list_response_preserves_pagination_contract(): - response = AuditEventListResponse(items=[], total=3, page=2, page_size=2, total_pages=2) + generated_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + source_checked_at = datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc) + response = AuditEventListResponse( + items=[], total=3, page=2, page_size=2, total_pages=2, + source="security_audit", + source_availability=SourceAvailability.AVAILABLE, + generated_at=generated_at, + source_checked_at=source_checked_at, + freshness=FreshnessOut(status=FreshnessStatus.CURRENT), + retention=RetentionOut(status=RetentionStatus.KNOWN), + warnings=[], + ) assert response.model_dump() == { "items": [], "total": 3, "page": 2, "page_size": 2, "total_pages": 2, + "source": "security_audit", + "source_availability": SourceAvailability.AVAILABLE, + "generated_at": generated_at, + "source_checked_at": source_checked_at, + "freshness": {"status": FreshnessStatus.CURRENT, "last_persisted_event_at": None, "ingestion_lag_seconds": None}, + "retention": {"status": RetentionStatus.KNOWN, "retention_days": None, "oldest_available_event_at": None}, + "warnings": [], } @@ -159,8 +209,12 @@ def test_list_audit_events_route_calculates_nonempty_total_pages(monkeypatch): def test_get_db_closes_session_after_normal_iteration(monkeypatch): + # get_db() is bound to ReaderSessionLocal (V2-003 reader/writer split, + # db/session.py) -- SessionLocal is the writer factory worker/main.py + # uses instead, and patching it here would leave get_db() constructing + # a real ReaderSessionLocal() session unaffected by this monkeypatch. db = MagicMock() - monkeypatch.setattr("db.session.SessionLocal", MagicMock(return_value=db)) + monkeypatch.setattr("db.session.ReaderSessionLocal", MagicMock(return_value=db)) yielded = list(get_db()) @@ -170,7 +224,7 @@ def test_get_db_closes_session_after_normal_iteration(monkeypatch): def test_get_db_closes_session_when_consumer_raises(monkeypatch): db = MagicMock() - monkeypatch.setattr("db.session.SessionLocal", MagicMock(return_value=db)) + monkeypatch.setattr("db.session.ReaderSessionLocal", MagicMock(return_value=db)) generator = get_db() assert next(generator) is db