From 4b3cd6a7c2ab88baea31bcba6aa57cecf47833b8 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Fri, 24 Apr 2026 13:06:40 +0000 Subject: [PATCH] fix(audit): forward DomainEvent.timestamp to AuditService.record() - Replace 8 # type: ignore[arg-type] suppressions in _row_to_entry() with cast() calls per project type-safety standards - Add argument validation for timestamp parameter (TypeError for non-datetime, ValueError for naive datetime) - Restore container_wiring integration test to use _ensure_session() + Base.metadata.create_all() + _async_mode=False pattern (broken by prior rebase) - Restore user_identity_* integration test functions and robot tests that were incorrectly removed during prior rebase - Fix sys.path insertion in helper_audit_wiring.py to ensure repo src takes precedence over system-level /app/src installation - Add preserve_event_timestamp integration test to robot suite ISSUES CLOSED: #719 --- robot/audit_service_wiring.robot | 30 +++++ robot/helper_audit_wiring.py | 127 +++++++++++++++++- .../application/services/audit_service.py | 42 ++++-- 3 files changed, 185 insertions(+), 14 deletions(-) diff --git a/robot/audit_service_wiring.robot b/robot/audit_service_wiring.robot index 48cce2717..85c6aa81e 100644 --- a/robot/audit_service_wiring.robot +++ b/robot/audit_service_wiring.robot @@ -58,6 +58,36 @@ E2E Plan Lifecycle Through Audit Pipeline Should Be Equal As Integers ${result.rc} 0 Should Contain ${result.stdout} E2E_OK +User Identity Field Propagates Through Audit Pipeline + [Documentation] DomainEvent.user_identity propagates via AuditEventSubscriber to AuditService DB entry + [Tags] observability audit identity + ${result}= Run Process ${PYTHON} ${HELPER} user_identity_field + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} IDENTITY_FIELD_OK + +User Identity Fallback From Details + [Documentation] Backward compat: user_identity extracted from details when field is None + [Tags] observability audit identity + ${result}= Run Process ${PYTHON} ${HELPER} user_identity_details_fallback + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} IDENTITY_FALLBACK_OK + +User Identity Field Precedence Over Details + [Documentation] DomainEvent.user_identity field takes precedence over details dict + [Tags] observability audit identity + ${result}= Run Process ${PYTHON} ${HELPER} user_identity_field_precedence + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} IDENTITY_PRECEDENCE_OK + Audit Subscriber Preserves Event Timestamp [Documentation] Verify DomainEvent.timestamp is forwarded to audit entry created_at (#719) [Tags] observability audit timestamp diff --git a/robot/helper_audit_wiring.py b/robot/helper_audit_wiring.py index a98ba6eae..fa8863cb7 100644 --- a/robot/helper_audit_wiring.py +++ b/robot/helper_audit_wiring.py @@ -9,9 +9,14 @@ import sys from datetime import UTC, datetime from pathlib import Path -# Ensure local source tree is importable +# Ensure local source tree is importable and takes precedence over any +# system-level installation (e.g. /app/src from the dev container). _SRC = str(Path(__file__).resolve().parents[1] / "src") -if _SRC not in sys.path: +if sys.path and sys.path[0] != _SRC: + # Remove any existing entry for _SRC to avoid duplicates, then + # re-insert at position 0 so the repo's source takes precedence. + if _SRC in sys.path: + sys.path.remove(_SRC) sys.path.insert(0, _SRC) from sqlalchemy import create_engine # noqa: E402 @@ -151,9 +156,21 @@ def container_wiring() -> None: sys.exit(1) # Functional verification: emit through container's EventBus + # and verify the event reaches the AuditService. + audit_svc = container.audit_service() + + # The container's in-memory DB has no tables yet — create them + # so the audit service can persist entries. + db_session = audit_svc._ensure_session() + audit_engine = db_session.get_bind() + Base.metadata.create_all(audit_engine) + + # Disable async mode so writes happen synchronously and we can + # verify the count immediately without race conditions. + audit_svc._async_mode = False + bus = container.event_bus() bus.emit(DomainEvent(event_type=EventType.PLAN_APPLIED, plan_id="WIRE-001")) - audit_svc = container.audit_service() count = audit_svc.count() if count < 1: print( @@ -212,11 +229,110 @@ def e2e_plan_lifecycle() -> None: print("E2E_OK") +def user_identity_field() -> None: + """Verify DomainEvent.user_identity propagates to audit entry.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id="SES-FIELD-INT", + user_identity="integration-user@example.com", + details={"mode": "interactive"}, + ) + ) + + entries = svc.list_entries(event_type="session_created") + if not entries: + print("FAIL: no session_created entry found", file=sys.stderr) + sys.exit(1) + if entries[0].user_identity != "integration-user@example.com": + print( + f"FAIL: user_identity={entries[0].user_identity!r}, " + f"expected 'integration-user@example.com'", + file=sys.stderr, + ) + sys.exit(1) + print("IDENTITY_FIELD_OK") + + +def user_identity_details_fallback() -> None: + """Verify backward compat: user_identity from details when field is None.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id="SES-FALLBACK-INT", + details={ + "mode": "interactive", + "user_identity": "fallback-user@example.com", + }, + ) + ) + + entries = svc.list_entries(event_type="session_created") + if not entries: + print("FAIL: no session_created entry found", file=sys.stderr) + sys.exit(1) + if entries[0].user_identity != "fallback-user@example.com": + print( + f"FAIL: user_identity={entries[0].user_identity!r}, " + f"expected 'fallback-user@example.com'", + file=sys.stderr, + ) + sys.exit(1) + print("IDENTITY_FALLBACK_OK") + + +def user_identity_field_precedence() -> None: + """Verify field takes precedence over details for user_identity.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id="SES-PREC-INT", + user_identity="field-wins@example.com", + details={ + "mode": "interactive", + "user_identity": "details-loses@example.com", + }, + ) + ) + + entries = svc.list_entries(event_type="session_created") + if not entries: + print("FAIL: no session_created entry found", file=sys.stderr) + sys.exit(1) + if entries[0].user_identity != "field-wins@example.com": + print( + f"FAIL: user_identity={entries[0].user_identity!r}, " + f"expected 'field-wins@example.com'", + file=sys.stderr, + ) + sys.exit(1) + # Also verify the details dict does NOT contain user_identity + # (it should have been stripped to avoid duplication). + if "user_identity" in entries[0].details: + print( + "FAIL: user_identity still present in details after field precedence", + file=sys.stderr, + ) + sys.exit(1) + print("IDENTITY_PRECEDENCE_OK") + + # --------------------------------------------------------------------------- # Dispatch # --------------------------------------------------------------------------- - def preserve_event_timestamp() -> None: """Verify DomainEvent.timestamp is preserved in audit entry created_at (#719). @@ -260,6 +376,9 @@ _COMMANDS = { "ignore_non_security": ignore_non_security, "container_wiring": container_wiring, "e2e_plan_lifecycle": e2e_plan_lifecycle, + "user_identity_field": user_identity_field, + "user_identity_details_fallback": user_identity_details_fallback, + "user_identity_field_precedence": user_identity_field_precedence, "preserve_event_timestamp": preserve_event_timestamp, } diff --git a/src/cleveragents/application/services/audit_service.py b/src/cleveragents/application/services/audit_service.py index 86ebebe1c..b456c4a3d 100644 --- a/src/cleveragents/application/services/audit_service.py +++ b/src/cleveragents/application/services/audit_service.py @@ -44,7 +44,7 @@ import queue import threading from dataclasses import dataclass from datetime import UTC, datetime, timedelta -from typing import Any, Self +from typing import Any, Self, cast import structlog from sqlalchemy import create_engine @@ -360,6 +360,9 @@ class AuditService: Raises: ValueError: If *event_type* is not one of the spec-defined event types in :data:`VALID_EVENT_TYPES`. + TypeError: If *timestamp* is not a ``datetime`` instance or + ``None``. + ValueError: If *timestamp* is a naive datetime (no ``tzinfo``). Returns: The created :class:`AuditLogEntry`. In async mode the ``id`` @@ -370,6 +373,17 @@ class AuditService: f"Unknown event_type {event_type!r}; " f"expected one of {sorted(VALID_EVENT_TYPES)}" ) + if timestamp is not None: + if not isinstance(timestamp, datetime): + raise TypeError( + f"timestamp must be a datetime instance or None, " + f"got {type(timestamp).__name__!r}" + ) + if timestamp.tzinfo is None: + raise ValueError( + "timestamp must be timezone-aware (UTC); " + "got a naive datetime with no tzinfo" + ) created_at = ( timestamp.strftime(_TIMESTAMP_FMT) if timestamp is not None @@ -512,18 +526,26 @@ class AuditService: @staticmethod def _row_to_entry(row: AuditLogModel) -> AuditLogEntry: - """Convert a SQLAlchemy model row to a domain data class.""" + """Convert a SQLAlchemy model row to a domain data class. + + ``AuditLogModel`` uses legacy ``Column()`` declarations with + ``__allow_unmapped__ = True``, so Pyright cannot infer the + concrete types of the ORM accessor attributes. ``cast()`` is + used here to assert the expected types explicitly, which is the + project-standard alternative to ``# type: ignore`` suppressions. + """ try: - details = json.loads(row.details) if row.details else {} # type: ignore[arg-type] + details_raw = cast("str | None", row.details) + details = json.loads(details_raw) if details_raw else {} except (json.JSONDecodeError, TypeError): details = {} return AuditLogEntry( - id=row.id, # type: ignore[arg-type] - event_type=row.event_type, # type: ignore[arg-type] - plan_id=row.plan_id, # type: ignore[arg-type] - project_name=row.project_name, # type: ignore[arg-type] - actor_name=row.actor_name, # type: ignore[arg-type] - user_identity=row.user_identity, # type: ignore[arg-type] + id=cast(int, row.id), + event_type=cast(str, row.event_type), + plan_id=cast("str | None", row.plan_id), + project_name=cast("str | None", row.project_name), + actor_name=cast("str | None", row.actor_name), + user_identity=cast("str | None", row.user_identity), details=details, - created_at=row.created_at, # type: ignore[arg-type] + created_at=cast(str, row.created_at), )