cleveragents-core/robot/helper_event_bus.py

"""Helper script for event_bus.robot smoke 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 cleveragents.infrastructure.events import (  # noqa: E402
    DomainEvent,
    EventBus,
    EventType,
    LoggingEventBus,
    ReactiveEventBus,
)

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


def event_type_values() -> None:
    """Verify key EventType values are present."""
    required = {
        "plan.created",
        "decision.created",
        "tool.invoked",
        "actor.completed",
        "budget.exceeded",
    }
    actual = {e.value for e in EventType}
    missing = required - actual
    if missing:
        print(f"FAIL: missing EventType values: {missing}", file=sys.stderr)
        sys.exit(1)
    print("event-type-values-ok")


def domain_event_creation() -> None:
    """Verify DomainEvent constructs with defaults."""
    event = DomainEvent(event_type=EventType.PLAN_CREATED)
    if event.plan_id is not None:
        print("FAIL: expected plan_id=None", file=sys.stderr)
        sys.exit(1)
    if not event.correlation_id:
        print("FAIL: expected non-empty correlation_id", file=sys.stderr)
        sys.exit(1)
    print("domain-event-creation-ok")


def protocol_conformance() -> None:
    """Verify both bus implementations satisfy EventBus protocol."""
    for cls in (ReactiveEventBus, LoggingEventBus):
        bus = cls()
        if not isinstance(bus, EventBus):
            print(f"FAIL: {cls.__name__} does not satisfy EventBus", file=sys.stderr)
            sys.exit(1)
    print("protocol-conformance-ok")


def reactive_emit_subscribe() -> None:
    """Verify ReactiveEventBus dispatches events to handlers."""
    received: list[DomainEvent] = []
    bus = ReactiveEventBus()
    bus.subscribe(EventType.PLAN_CREATED, received.append)
    bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED))
    if len(received) != 1:
        print(f"FAIL: expected 1 event, got {len(received)}", file=sys.stderr)
        sys.exit(1)
    if str(received[0].event_type) != "plan.created":
        print(f"FAIL: wrong event type: {received[0].event_type}", file=sys.stderr)
        sys.exit(1)
    print("reactive-emit-subscribe-ok")


def logging_emit_subscribe() -> None:
    """Verify LoggingEventBus dispatches events to handlers."""
    received: list[DomainEvent] = []
    bus = LoggingEventBus()
    bus.subscribe(EventType.DECISION_CREATED, received.append)
    bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED))
    if len(received) != 1:
        print(f"FAIL: expected 1 event, got {len(received)}", file=sys.stderr)
        sys.exit(1)
    print("logging-emit-subscribe-ok")


def reactive_type_filtering() -> None:
    """Verify ReactiveEventBus only dispatches to matching event type."""
    received: list[DomainEvent] = []
    bus = ReactiveEventBus()
    bus.subscribe(EventType.PLAN_CREATED, received.append)
    bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED))
    if received:
        print("FAIL: handler should not have been called", file=sys.stderr)
        sys.exit(1)
    print("reactive-type-filtering-ok")


def decision_service_emits_event() -> None:
    """Verify DecisionService emits DECISION_CREATED via injected bus."""
    import sys
    from unittest.mock import MagicMock, create_autospec

    from ulid import ULID

    from cleveragents.application.services.decision_service import DecisionService
    from cleveragents.config.settings import Settings
    from cleveragents.domain.models.core.decision import DecisionType
    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork

    received: list[DomainEvent] = []
    bus = ReactiveEventBus()
    bus.subscribe(EventType.DECISION_CREATED, received.append)

    mock_uow = create_autospec(UnitOfWork, instance=True)
    mock_tx_ctx = MagicMock()
    mock_tx_ctx.decisions = MagicMock()
    mock_uow.transaction.return_value.__enter__.return_value = mock_tx_ctx
    mock_uow.transaction.return_value.__exit__.return_value = False

    svc = DecisionService(
        settings=create_autospec(Settings, instance=True),
        unit_of_work=mock_uow,
        event_bus=bus,
    )
    plan_id = str(ULID())
    svc.record_decision(
        plan_id=plan_id,
        decision_type=DecisionType.STRATEGY_CHOICE,
        question="Q?",
        chosen_option="A",
        rationale="because",
    )
    if not received:
        print("FAIL: no DECISION_CREATED event emitted", file=sys.stderr)
        sys.exit(1)
    print("decision-service-emits-event-ok")


def plan_lifecycle_emits_event() -> None:
    """Verify PlanLifecycleService emits PLAN_CREATED via injected bus."""
    import sys
    from unittest.mock import create_autospec

    from cleveragents.application.services.plan_lifecycle_service import (
        PlanLifecycleService,
    )
    from cleveragents.config.settings import Settings

    received: list[DomainEvent] = []
    bus = ReactiveEventBus()
    bus.subscribe(EventType.PLAN_CREATED, received.append)

    svc = PlanLifecycleService(
        settings=create_autospec(Settings, instance=True), event_bus=bus
    )
    svc.create_action(
        name="test/action",
        description="test action",
        definition_of_done="Done when implemented",
        strategy_actor="local/strategist",
        execution_actor="local/executor",
    )
    svc.use_action(action_name="test/action")
    if not received:
        print("FAIL: no PLAN_CREATED event emitted", file=sys.stderr)
        sys.exit(1)
    print("plan-lifecycle-emits-event-ok")


def container_event_bus_singleton() -> None:
    """Verify DI container provides ReactiveEventBus as Singleton."""
    from cleveragents.application.container import Container

    container = Container()
    bus_a = container.event_bus()
    bus_b = container.event_bus()
    if bus_a is not bus_b:
        print("FAIL: event_bus not a singleton", file=sys.stderr)
        sys.exit(1)
    print("container-event-bus-singleton-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

_COMMANDS = {
    "event_type_values": event_type_values,
    "domain_event_creation": domain_event_creation,
    "protocol_conformance": protocol_conformance,
    "reactive_emit_subscribe": reactive_emit_subscribe,
    "logging_emit_subscribe": logging_emit_subscribe,
    "reactive_type_filtering": reactive_type_filtering,
    "decision_service_emits_event": decision_service_emits_event,
    "plan_lifecycle_emits_event": plan_lifecycle_emits_event,
    "container_event_bus_singleton": container_event_bus_singleton,
}

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]]()