"""Step definitions for AuditService EventBus wiring (Forgejo #581). Tests that the :class:`AuditEventSubscriber` correctly bridges :class:`EventBus` events to :class:`AuditService.record()` with secret masking, event type mapping, and error resilience. """ from __future__ import annotations from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from cleveragents.application.services.audit_event_subscriber import ( AuditEventSubscriber, ) from cleveragents.application.services.audit_service import AuditService from cleveragents.config.settings import Settings from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.reactive import ReactiveEventBus from cleveragents.infrastructure.events.types import EventType __all__: list[str] = [] # ── Helpers ─────────────────────────────────────────────────────── def _make_settings() -> Settings: """Create a fresh Settings instance for testing.""" Settings._instance = None return Settings(database_url="sqlite:///:memory:", audit_async=False) def _fresh_audit_service() -> AuditService: """Create an AuditService backed by an in-memory SQLite database.""" settings = _make_settings() engine = create_engine("sqlite:///:memory:", echo=False) Base.metadata.create_all(engine) session = sessionmaker(bind=engine)() return AuditService(settings=settings, session=session) # ── Given ───────────────────────────────────────────────────────── @given("an event bus with an audit subscriber") def step_event_bus_with_audit_subscriber(context: Context) -> None: context.audit_service = _fresh_audit_service() context.event_bus = ReactiveEventBus() context.subscriber = AuditEventSubscriber( audit_service=context.audit_service, event_bus=context.event_bus, ) @given("an event bus with a failing audit service") def step_event_bus_with_failing_audit_service(context: Context) -> None: context.event_bus = ReactiveEventBus() mock_service = MagicMock(spec=AuditService) mock_service.record.side_effect = RuntimeError("DB unavailable") context.subscriber = AuditEventSubscriber( audit_service=mock_service, event_bus=context.event_bus, ) # ── When ────────────────────────────────────────────────────────── @when("a plan_applied event is emitted") def step_emit_plan_applied(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.PLAN_APPLIED, plan_id="PLAN-001", details={"action_name": "local/deploy"}, ) ) @when("a plan_cancelled event is emitted") def step_emit_plan_cancelled(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.PLAN_CANCELLED, plan_id="PLAN-002", details={"reason": "user requested"}, ) ) @when("a config_changed event is emitted") def step_emit_config_changed(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.CONFIG_CHANGED, details={"key": "log_level", "value": "DEBUG"}, ) ) @when("an entity_deleted event is emitted") def step_emit_entity_deleted(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.ENTITY_DELETED, details={"entity": "resource/old-file"}, ) ) @when("a session_created event is emitted") def step_emit_session_created(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.SESSION_CREATED, session_id="SES-001", details={"mode": "interactive"}, ) ) @when("a correction_applied event is emitted") def step_emit_correction_applied(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.CORRECTION_APPLIED, plan_id="PLAN-003", details={"correction": "fix import"}, ) ) @when("a resource_modified event is emitted") def step_emit_resource_modified(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.RESOURCE_MODIFIED, details={"path": "src/main.py", "action": "write"}, ) ) @when("an auth_success event is emitted") def step_emit_auth_success(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.AUTH_SUCCESS, details={"user": "admin", "method": "token"}, ) ) @when("an auth_failure event is emitted") def step_emit_auth_failure(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.AUTH_FAILURE, details={"user": "unknown", "reason": "invalid credentials"}, ) ) @when("an event with sensitive details is emitted") def step_emit_event_with_sensitive_details(context: Context) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.CONFIG_CHANGED, details={ "key": "openai_api_key", "api_key": "sk-proj-ABCDEFGHIJ1234567890", "normal_field": "visible", }, ) ) @when("multiple security events are emitted") def step_emit_multiple_events(context: Context) -> None: context.event_bus.emit(DomainEvent(event_type=EventType.PLAN_APPLIED, plan_id="P1")) context.event_bus.emit( DomainEvent(event_type=EventType.PLAN_CANCELLED, plan_id="P2") ) context.event_bus.emit( DomainEvent(event_type=EventType.SESSION_CREATED, session_id="S1") ) @when("a non-security event is emitted") def step_emit_non_security_event(context: Context) -> None: context.event_bus.emit( DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="P-nonsec") ) @when('a plan_applied event with plan_id "{plan_id}" is emitted') def step_emit_plan_applied_with_plan_id(context: Context, plan_id: str) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.PLAN_APPLIED, plan_id=plan_id, details={"action_name": "local/test"}, ) ) @when('a config_changed event with actor_name "{actor_name}" is emitted') def step_emit_config_changed_with_actor(context: Context, actor_name: str) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.CONFIG_CHANGED, actor_name=actor_name, details={"key": "debug_enabled", "value": "true"}, ) ) @when('a session_created event with session_id "{session_id}" is emitted') def step_emit_session_created_with_session_id( context: Context, session_id: str ) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.SESSION_CREATED, session_id=session_id, details={"mode": "interactive"}, ) ) @when('a plan_applied event with correlation_id "{correlation_id}" is emitted') def step_emit_plan_applied_with_correlation_id( context: Context, correlation_id: str ) -> None: context.event_bus.emit( DomainEvent( event_type=EventType.PLAN_APPLIED, plan_id="PLAN-CORR", correlation_id=correlation_id, details={"action_name": "local/test"}, ) ) @when("a plan_applied event is emitted on the failing bus") def step_emit_on_failing_bus(context: Context) -> None: context.exception_raised = False try: context.event_bus.emit( DomainEvent( event_type=EventType.PLAN_APPLIED, plan_id="PLAN-FAIL", ) ) except Exception: context.exception_raised = True # ── Then ────────────────────────────────────────────────────────── @then("the audit log should contain a plan_applied entry") def step_audit_has_plan_applied(context: Context) -> None: entries = context.audit_service.list_entries(event_type="plan_applied") assert len(entries) == 1, f"Expected plan_applied entry, got {len(entries)}" @then("the audit log should contain a plan_cancelled entry") def step_audit_has_plan_cancelled(context: Context) -> None: entries = context.audit_service.list_entries(event_type="plan_cancelled") assert len(entries) == 1, f"Expected plan_cancelled entry, got {len(entries)}" @then("the audit log should contain a config_changed entry") def step_audit_has_config_changed(context: Context) -> None: entries = context.audit_service.list_entries(event_type="config_changed") assert len(entries) == 1, f"Expected config_changed entry, got {len(entries)}" @then("the audit log should contain an entity_deleted entry") def step_audit_has_entity_deleted(context: Context) -> None: entries = context.audit_service.list_entries(event_type="entity_deleted") assert len(entries) == 1, f"Expected entity_deleted entry, got {len(entries)}" @then("the audit log should contain a session_created entry") def step_audit_has_session_created(context: Context) -> None: entries = context.audit_service.list_entries(event_type="session_created") assert len(entries) == 1, f"Expected session_created entry, got {len(entries)}" @then("the audit log should contain a correction_applied entry") def step_audit_has_correction_applied(context: Context) -> None: entries = context.audit_service.list_entries(event_type="correction_applied") assert len(entries) == 1, f"Expected correction_applied entry, got {len(entries)}" @then("the audit log should contain a resource_modified entry") def step_audit_has_resource_modified(context: Context) -> None: entries = context.audit_service.list_entries(event_type="resource_modified") assert len(entries) == 1, f"Expected resource_modified entry, got {len(entries)}" @then("the audit log should contain an auth_success entry") def step_audit_has_auth_success(context: Context) -> None: entries = context.audit_service.list_entries(event_type="auth_success") assert len(entries) == 1, f"Expected auth_success entry, got {len(entries)}" @then("the audit log should contain an auth_failure entry") def step_audit_has_auth_failure(context: Context) -> None: entries = context.audit_service.list_entries(event_type="auth_failure") assert len(entries) == 1, f"Expected auth_failure entry, got {len(entries)}" @then("the audit log entry should have redacted values") def step_audit_entry_has_redacted(context: Context) -> None: entries = context.audit_service.list_entries(event_type="config_changed") assert len(entries) >= 1, "Expected at least one config_changed entry" details = entries[0].details assert details.get("api_key") == "***REDACTED***", ( f"Expected api_key to be redacted, got {details.get('api_key')!r}" ) assert details.get("normal_field") == "visible", ( f"Expected normal_field to be 'visible', got {details.get('normal_field')!r}" ) @then("all events should appear in the audit log") def step_all_events_in_audit_log(context: Context) -> None: total = context.audit_service.count() assert total == 3, f"Expected 3 audit entries, got {total}" @then("the audit log should not contain the event") def step_audit_log_no_event(context: Context) -> None: total = context.audit_service.count() assert total == 0, f"Expected 0 audit entries, got {total}" @then('the audit entry event_type should be the string "{expected_type}"') def step_audit_entry_type_string(context: Context, expected_type: str) -> None: entries = context.audit_service.list_entries(event_type=expected_type) assert len(entries) >= 1, f"No entries with event_type={expected_type!r}" assert entries[0].event_type == expected_type @then('the audit entry should have plan_id "{expected_plan_id}"') def step_audit_entry_plan_id(context: Context, expected_plan_id: str) -> None: entries = context.audit_service.list_entries(event_type="plan_applied") assert len(entries) >= 1 assert entries[0].plan_id == expected_plan_id, ( f"Expected plan_id={expected_plan_id!r}, got {entries[0].plan_id!r}" ) @then('the audit entry should have actor_name "{expected_actor}"') def step_audit_entry_actor_name(context: Context, expected_actor: str) -> None: entries = context.audit_service.list_entries(event_type="config_changed") assert len(entries) >= 1 assert entries[0].actor_name == expected_actor, ( f"Expected actor_name={expected_actor!r}, got {entries[0].actor_name!r}" ) @then('the audit entry details should contain session_id "{expected_sid}"') def step_audit_entry_details_session_id(context: Context, expected_sid: str) -> None: entries = context.audit_service.list_entries(event_type="session_created") assert len(entries) >= 1, "No session_created entries found" details = entries[0].details assert details.get("session_id") == expected_sid, ( f"Expected session_id={expected_sid!r}, got {details.get('session_id')!r}" ) @then('the audit entry details should contain correlation_id "{expected_cid}"') def step_audit_entry_details_correlation_id( context: Context, expected_cid: str ) -> None: entries = context.audit_service.list_entries(event_type="plan_applied") assert len(entries) >= 1, "No plan_applied entries found" details = entries[0].details assert details.get("correlation_id") == expected_cid, ( f"Expected correlation_id={expected_cid!r}, " f"got {details.get('correlation_id')!r}" ) @then("no audit subscriber exception should propagate") def step_no_audit_subscriber_exception(context: Context) -> None: assert not getattr(context, "exception_raised", True), ( "An exception propagated from the audit subscriber" )