cleveragents-core/robot/helper_audit_wiring.py

"""Helper script for audit_service_wiring.robot integration tests.

Each subcommand is a self-contained check that prints a sentinel on success.
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

from cleveragents.application.services.audit_event_subscriber import (  # noqa: E402
    SECURITY_EVENT_MAP,
    AuditEventSubscriber,
)
from cleveragents.application.services.audit_service import (  # noqa: E402
    AuditService,
)
from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.events.models import DomainEvent  # noqa: E402
from cleveragents.infrastructure.events.reactive import (  # noqa: E402
    ReactiveEventBus,
)
from cleveragents.infrastructure.events.types import EventType  # noqa: E402


def _make_audit_service() -> AuditService:
    """Create an in-memory AuditService for testing."""
    Settings._instance = None
    settings = Settings(database_url="sqlite:///:memory:")
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    return AuditService(settings=settings, session=session)


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def record_events() -> None:
    """Verify subscriber records security events through AuditService."""
    svc = _make_audit_service()
    bus = ReactiveEventBus()
    AuditEventSubscriber(audit_service=svc, event_bus=bus)

    # Emit all 9 security-relevant event types
    for event_type in SECURITY_EVENT_MAP:
        bus.emit(DomainEvent(event_type=event_type, plan_id="TEST-001"))

    count = svc.count()
    if count != len(SECURITY_EVENT_MAP):
        print(
            f"FAIL: expected {len(SECURITY_EVENT_MAP)} entries, got {count}",
            file=sys.stderr,
        )
        sys.exit(1)
    print("AUDIT_OK")


def redact_secrets() -> None:
    """Verify sensitive data in event details is redacted."""
    svc = _make_audit_service()
    bus = ReactiveEventBus()
    AuditEventSubscriber(audit_service=svc, event_bus=bus)

    bus.emit(
        DomainEvent(
            event_type=EventType.CONFIG_CHANGED,
            details={
                "api_key": "sk-proj-ABCDEFGHIJ1234567890",
                "normal": "visible",
            },
        )
    )

    entries = svc.list_entries(event_type="config_changed")
    if not entries:
        print("FAIL: no config_changed entry found", file=sys.stderr)
        sys.exit(1)

    details = entries[0].details
    if details.get("api_key") != "***REDACTED***":
        print(
            f"FAIL: api_key not redacted: {details.get('api_key')!r}",
            file=sys.stderr,
        )
        sys.exit(1)
    if details.get("normal") != "visible":
        print(
            f"FAIL: normal field changed: {details.get('normal')!r}",
            file=sys.stderr,
        )
        sys.exit(1)
    print("REDACT_OK")


def ignore_non_security() -> None:
    """Verify non-security events are not recorded."""
    svc = _make_audit_service()
    bus = ReactiveEventBus()
    AuditEventSubscriber(audit_service=svc, event_bus=bus)

    bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="P-nonsec"))

    count = svc.count()
    if count != 0:
        print(f"FAIL: expected 0 entries, got {count}", file=sys.stderr)
        sys.exit(1)
    print("FILTER_OK")


def container_wiring() -> None:
    """Verify DI container registers a functional AuditEventSubscriber.

    Goes beyond type checking: emits an event through the container's
    EventBus and verifies it reaches the AuditService via the subscriber.
    """
    from dependency_injector import providers

    from cleveragents.application.container import Container

    # Reset Settings singleton to avoid stale state from earlier tests.
    Settings._instance = None

    container = Container()
    # Override database_url to use an in-memory SQLite database so the
    # test does not depend on the filesystem (the default file-based URL
    # requires the .cleveragents/ directory to exist).
    container.database_url.override(providers.Object("sqlite:///:memory:"))

    subscriber = container.audit_event_subscriber()
    if not isinstance(subscriber, AuditEventSubscriber):
        print(
            f"FAIL: expected AuditEventSubscriber, got {type(subscriber).__name__}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Functional verification: emit through container's EventBus
    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(
            f"FAIL: expected >=1 audit entries after emit, got {count}",
            file=sys.stderr,
        )
        sys.exit(1)
    print("CONTAINER_OK")


def e2e_plan_lifecycle() -> None:
    """E2E: PlanLifecycleService -> EventBus -> AuditEventSubscriber -> DB.

    Exercises the full domain-to-audit pipeline:
    1. Create PlanLifecycleService with an EventBus (in-memory mode).
    2. Wire AuditEventSubscriber on the same bus.
    3. Create an action, use it to create a plan.
    4. Cancel the plan (triggers PLAN_CANCELLED event).
    5. Verify a plan_cancelled audit entry exists in the DB.
    """
    svc = _make_audit_service()
    bus = ReactiveEventBus()
    AuditEventSubscriber(audit_service=svc, event_bus=bus)

    settings = Settings._instance or Settings(database_url="sqlite:///:memory:")
    plc = PlanLifecycleService(settings=settings, event_bus=bus)

    # Create and register an action
    plc.create_action(
        name="local/e2e-audit-test",
        description="E2E test action",
        definition_of_done="verify audit wiring",
        strategy_actor="test-strategy",
        execution_actor="test-execution",
    )

    # Use the action to create a plan
    plan = plc.use_action("local/e2e-audit-test")

    # Cancel the plan -- this emits PLAN_CANCELLED via the EventBus
    plan_id = plan.identity.plan_id
    plc.cancel_plan(plan_id, reason="e2e audit verification")

    entries = svc.list_entries(event_type="plan_cancelled")
    if not entries:
        print(
            "FAIL: no plan_cancelled audit entry after cancel_plan()", file=sys.stderr
        )
        sys.exit(1)
    if entries[0].plan_id != plan_id:
        print(
            f"FAIL: plan_id mismatch: {entries[0].plan_id!r} != {plan_id!r}",
            file=sys.stderr,
        )
        sys.exit(1)
    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
# ---------------------------------------------------------------------------

_COMMANDS = {
    "record_events": record_events,
    "redact_secrets": redact_secrets,
    "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,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        print(f"Commands: {list(_COMMANDS)}", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()