"""Step definitions for event_system_taxonomy.feature. Tests the complete Event System Domain Event Taxonomy: EventType enum completeness, DomainEvent model fields and serialisation, EventBus Protocol, ReactiveEventBus emit/subscribe/stream/audit_log behaviour, and log correlation. """ from __future__ import annotations import re from datetime import UTC from typing import Any from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context # type: ignore[import-untyped] from pydantic import ValidationError from rx.core.observable.observable import Observable from ulid import ULID from cleveragents.infrastructure.events import ( DomainEvent, EventBus, EventType, LoggingEventBus, ReactiveEventBus, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _DOT_PATTERN = re.compile(r"^[a-z_]+\.[a-z_]+$") def _make_event(event_type_str: str, **kwargs: Any) -> DomainEvent: return DomainEvent(event_type=EventType(event_type_str), **kwargs) class _TaxonomyCollector: """In-test stub that collects emitted events.""" def __init__(self) -> None: self.received: list[DomainEvent] = [] def __call__(self, event: DomainEvent) -> None: self.received.append(event) # --------------------------------------------------------------------------- # EventType enum completeness # --------------------------------------------------------------------------- @then("the EventType enum should have at least {n:d} members") def step_event_type_min_members(ctx: Context, n: int) -> None: count = len(EventType) assert count >= n, f"EventType has {count} members, expected >= {n}" @then('EventType should contain "{value}"') def step_event_type_contains_value(ctx: Context, value: str) -> None: values = {e.value for e in EventType} assert value in values, f"{value!r} not in EventType values: {sorted(values)}" @then('every EventType value should match the pattern "."') def step_event_type_dot_pattern(ctx: Context) -> None: for member in EventType: assert _DOT_PATTERN.match(member.value), ( f"EventType.{member.name} value {member.value!r} does not match " f"the . pattern" ) # --------------------------------------------------------------------------- # DomainEvent model # --------------------------------------------------------------------------- @when('I create a DomainEvent for taxonomy with event_type "{et}"') def step_create_taxonomy_event(ctx: Context, et: str) -> None: ctx.taxonomy_event = _make_event(et) @then('the taxonomy event should have field "{field}"') def step_taxonomy_event_has_field(ctx: Context, field: str) -> None: assert field in ctx.taxonomy_event.model_fields, ( f"DomainEvent missing field {field!r}" ) @then("the taxonomy event correlation_id should be {n:d} characters long") def step_taxonomy_correlation_id_length(ctx: Context, n: int) -> None: cid = ctx.taxonomy_event.correlation_id assert len(cid) == n, f"correlation_id length {len(cid)}, expected {n}" @then("the taxonomy event timestamp should be UTC") def step_taxonomy_timestamp_utc(ctx: Context) -> None: ts = ctx.taxonomy_event.timestamp assert ts.tzinfo is not None, "timestamp has no timezone" assert ts.tzinfo == UTC, f"timestamp timezone is {ts.tzinfo!r}, expected UTC" @then("mutating the taxonomy event should raise a validation error") def step_taxonomy_event_frozen(ctx: Context) -> None: raised = False try: ctx.taxonomy_event.event_type = EventType.PLAN_CANCELLED # type: ignore[misc] except ValidationError: raised = True assert raised, "Expected ValidationError when mutating frozen DomainEvent" @when("I create a DomainEvent for taxonomy with all fields populated") def step_create_taxonomy_full_event(ctx: Context) -> None: ctx.taxonomy_event = DomainEvent( event_type=EventType.PLAN_CREATED, plan_id=str(ULID()), root_plan_id=str(ULID()), session_id=str(ULID()), actor_name="local/strategist", project_name="my-project", details={"phase": "strategize", "actor": "strategists/planner"}, ) @then("the taxonomy event should serialize to JSON and deserialize back identically") def step_taxonomy_event_json_roundtrip(ctx: Context) -> None: json_str = ctx.taxonomy_event.model_dump_json() restored = DomainEvent.model_validate_json(json_str) assert restored == ctx.taxonomy_event, ( f"Round-trip mismatch:\n original={ctx.taxonomy_event}\n restored={restored}" ) @when("I create a DomainEvent for taxonomy with correlation fields") def step_create_taxonomy_event_correlation(ctx: Context) -> None: ctx.taxonomy_event = DomainEvent( event_type=EventType.PLAN_CREATED, plan_id=str(ULID()), root_plan_id=str(ULID()), session_id=str(ULID()), ) @then("the taxonomy event plan_id should be set") def step_taxonomy_plan_id_set(ctx: Context) -> None: assert ctx.taxonomy_event.plan_id is not None @then("the taxonomy event root_plan_id should be set") def step_taxonomy_root_plan_id_set(ctx: Context) -> None: assert ctx.taxonomy_event.root_plan_id is not None @then("the taxonomy event session_id should be set") def step_taxonomy_session_id_set(ctx: Context) -> None: assert ctx.taxonomy_event.session_id is not None @then("the taxonomy event user_identity should be None") def step_taxonomy_event_user_identity_none(ctx: Context) -> None: assert ctx.taxonomy_event.user_identity is None, ( f"Expected user_identity=None, got {ctx.taxonomy_event.user_identity!r}" ) @when('I create a DomainEvent for taxonomy with user_identity "{uid}"') def step_create_taxonomy_event_with_user_identity(ctx: Context, uid: str) -> None: ctx.taxonomy_event = DomainEvent( event_type=EventType.PLAN_CREATED, user_identity=uid, ) @then('the taxonomy event user_identity should be "{expected}"') def step_taxonomy_event_user_identity_value(ctx: Context, expected: str) -> None: assert ctx.taxonomy_event.user_identity == expected, ( f"Expected user_identity={expected!r}, got {ctx.taxonomy_event.user_identity!r}" ) @when("I create a DomainEvent for taxonomy with typed details") def step_create_taxonomy_event_with_details(ctx: Context) -> None: ctx.taxonomy_event = DomainEvent( event_type=EventType.PLAN_PHASE_CHANGED, details={"old_phase": "strategize", "new_phase": "execute"}, ) @then("the taxonomy event details should contain the expected keys") def step_taxonomy_event_details_keys(ctx: Context) -> None: details = ctx.taxonomy_event.details assert "old_phase" in details, "Missing 'old_phase' in details" assert "new_phase" in details, "Missing 'new_phase' in details" # --------------------------------------------------------------------------- # EventBus Protocol # --------------------------------------------------------------------------- @then('the EventBus Protocol should define method "{method}"') def step_protocol_has_method(ctx: Context, method: str) -> None: assert hasattr(EventBus, method), f"EventBus Protocol missing method {method!r}" @then("a ReactiveEventBus instance should satisfy the EventBus Protocol") def step_reactive_satisfies_protocol(ctx: Context) -> None: bus = ReactiveEventBus() assert isinstance(bus, EventBus), "ReactiveEventBus does not satisfy EventBus" @then("a LoggingEventBus instance should satisfy the EventBus Protocol") def step_logging_satisfies_protocol(ctx: Context) -> None: bus = LoggingEventBus() assert isinstance(bus, EventBus), "LoggingEventBus does not satisfy EventBus" # --------------------------------------------------------------------------- # ReactiveEventBus — emit / subscribe / stream # --------------------------------------------------------------------------- @given("a fresh ReactiveEventBus for taxonomy tests") def step_given_fresh_bus(ctx: Context) -> None: ctx.taxonomy_bus = ReactiveEventBus() ctx.taxonomy_collectors = [] @given("a ReactiveEventBus for taxonomy tests with max_audit_log_size {n:d}") def step_given_capped_bus(ctx: Context, n: int) -> None: ctx.taxonomy_bus = ReactiveEventBus(max_audit_log_size=n) ctx.taxonomy_collectors = [] @then( "creating a ReactiveEventBus with max_audit_log_size {n:d} should raise ValueError" ) def step_reactive_bus_invalid_audit_log_size(ctx: Context, n: int) -> None: raised = False try: ReactiveEventBus(max_audit_log_size=n) except ValueError: raised = True assert raised, f"Expected ValueError for max_audit_log_size={n}" @when('I taxonomy-subscribe to "{et}" events') def step_taxonomy_subscribe(ctx: Context, et: str) -> None: collector = _TaxonomyCollector() ctx.taxonomy_collectors.append(collector) ctx.taxonomy_bus.subscribe(EventType(et), collector) @when('I taxonomy-emit a "{et}" DomainEvent') def step_taxonomy_emit(ctx: Context, et: str) -> None: ctx.taxonomy_bus.emit(_make_event(et)) @then("the taxonomy handler should have received {n:d} event(s)") def step_taxonomy_handler_received(ctx: Context, n: int) -> None: assert ctx.taxonomy_collectors, "No taxonomy collectors registered" count = len(ctx.taxonomy_collectors[0].received) assert count == n, f"Expected {n} event(s), got {count}" @when('I taxonomy-subscribe two handlers to "{et}" events') def step_taxonomy_subscribe_two(ctx: Context, et: str) -> None: for _ in range(2): collector = _TaxonomyCollector() ctx.taxonomy_collectors.append(collector) ctx.taxonomy_bus.subscribe(EventType(et), collector) @then("each taxonomy handler should have received {n:d} event") def step_each_taxonomy_handler_received(ctx: Context, n: int) -> None: for i, col in enumerate(ctx.taxonomy_collectors): assert len(col.received) == n, ( f"Handler {i} received {len(col.received)}, expected {n}" ) @then("the taxonomy bus stream should be an Observable") def step_taxonomy_bus_stream_observable(ctx: Context) -> None: stream = ctx.taxonomy_bus.stream assert isinstance(stream, Observable), ( f"Expected Observable, got {type(stream).__name__}" ) @then('the taxonomy bus stream should not expose method "{method}"') def step_taxonomy_bus_stream_not_expose_method(ctx: Context, method: str) -> None: stream = ctx.taxonomy_bus.stream assert not hasattr(stream, method), ( f"Expected stream to hide {method!r}, but it is present" ) @when('I taxonomy-subscribe a failing handler and a normal handler to "{et}" events') def step_taxonomy_subscribe_failing_and_normal(ctx: Context, et: str) -> None: normal_collector = _TaxonomyCollector() ctx.taxonomy_collectors.append(normal_collector) def _failing_handler(event: DomainEvent) -> None: raise RuntimeError(f"failing taxonomy handler for {event.event_type.value}") ctx.taxonomy_bus.subscribe(EventType(et), _failing_handler) ctx.taxonomy_bus.subscribe(EventType(et), normal_collector) @then("the taxonomy normal handler should have received {n:d} event(s)") def step_taxonomy_normal_handler_received(ctx: Context, n: int) -> None: assert ctx.taxonomy_collectors, "No taxonomy normal collector registered" count = len(ctx.taxonomy_collectors[-1].received) assert count == n, f"Expected normal handler to receive {n} event(s), got {count}" @when('I observe the taxonomy bus stream and emit a "{et}" event') def step_taxonomy_observe_and_emit(ctx: Context, et: str) -> None: ctx.stream_events = [] ctx.taxonomy_bus.stream.subscribe(on_next=ctx.stream_events.append) ctx.taxonomy_bus.emit(_make_event(et)) @when('I observe taxonomy stream with a failing observer and emit a "{et}" event') def step_taxonomy_stream_failing_observer(ctx: Context, et: str) -> None: def _failing_observer(event: DomainEvent) -> None: raise RuntimeError(f"failing stream observer for {event.event_type.value}") ctx.taxonomy_bus.stream.subscribe(on_next=_failing_observer) ctx.taxonomy_bus.emit(_make_event(et)) @then("the stream observer should have captured the event") def step_stream_observer_captured(ctx: Context) -> None: assert len(ctx.stream_events) == 1, ( f"Expected 1 stream event, got {len(ctx.stream_events)}" ) @then("taxonomy-emitting a non-DomainEvent should raise TypeError") def step_taxonomy_emit_bad_type(ctx: Context) -> None: raised = False try: ctx.taxonomy_bus.emit("not-an-event") # type: ignore[arg-type] except TypeError: raised = True assert raised, "Expected TypeError for non-DomainEvent" @then("taxonomy-subscribing with a non-EventType should raise TypeError") def step_taxonomy_subscribe_bad_type(ctx: Context) -> None: raised = False try: ctx.taxonomy_bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type] except TypeError: raised = True assert raised, "Expected TypeError for non-EventType" @then("taxonomy-subscribing with a non-callable should raise TypeError") def step_taxonomy_subscribe_non_callable(ctx: Context) -> None: raised = False try: ctx.taxonomy_bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] except TypeError: raised = True assert raised, "Expected TypeError for non-callable handler" # --------------------------------------------------------------------------- # In-memory event trail (audit_log) # --------------------------------------------------------------------------- @then("the audit_log should contain {n:d} events") def step_audit_log_count(ctx: Context, n: int) -> None: log = ctx.taxonomy_bus.audit_log assert len(log) == n, f"audit_log has {len(log)} events, expected {n}" @then('the audit_log should include event types "{event_types}"') def step_audit_log_contains_types(ctx: Context, event_types: str) -> None: expected = [value.strip() for value in event_types.split(",")] actual = [event.event_type.value for event in ctx.taxonomy_bus.audit_log] assert actual == expected, f"audit_log types {actual} != expected {expected}" @then('the audit_log events should be in order "{order}"') def step_audit_log_order(ctx: Context, order: str) -> None: expected = [value.strip() for value in order.split(",")] log = ctx.taxonomy_bus.audit_log actual = [e.event_type.value for e in log] assert actual == expected, f"audit_log order {actual} != expected {expected}" @then("mutating the returned audit_log should not affect the bus") def step_audit_log_defensive_copy(ctx: Context) -> None: log1 = ctx.taxonomy_bus.audit_log original_len = len(log1) log1.clear() # mutate the returned list log2 = ctx.taxonomy_bus.audit_log assert len(log2) == original_len, ( f"audit_log was mutated: expected {original_len}, got {len(log2)}" ) @when("I clear the taxonomy bus audit_log") def step_clear_taxonomy_audit_log(ctx: Context) -> None: ctx.taxonomy_bus.clear_audit_log() # --------------------------------------------------------------------------- # Log correlation # --------------------------------------------------------------------------- @when("I emit two taxonomy events with the same correlation_id") def step_emit_correlated_events(ctx: Context) -> None: shared_cid = str(ULID()) ctx.taxonomy_bus.emit( DomainEvent( event_type=EventType.PLAN_CREATED, correlation_id=shared_cid, ) ) ctx.taxonomy_bus.emit( DomainEvent( event_type=EventType.PLAN_PHASE_CHANGED, correlation_id=shared_cid, ) ) ctx.shared_correlation_id = shared_cid @then("both events in the audit_log should share the same correlation_id") def step_correlated_events_in_log(ctx: Context) -> None: log = ctx.taxonomy_bus.audit_log assert len(log) == 2, f"Expected 2 events, got {len(log)}" assert log[0].correlation_id == ctx.shared_correlation_id assert log[1].correlation_id == ctx.shared_correlation_id @when("I create two independent DomainEvents for taxonomy") def step_create_two_independent_events(ctx: Context) -> None: ctx.taxonomy_event_one = DomainEvent(event_type=EventType.PLAN_CREATED) ctx.taxonomy_event_two = DomainEvent(event_type=EventType.DECISION_CREATED) @then("the two taxonomy events should have different correlation_ids") def step_taxonomy_events_have_different_correlation_ids(ctx: Context) -> None: cid_one = ctx.taxonomy_event_one.correlation_id cid_two = ctx.taxonomy_event_two.correlation_id assert cid_one != cid_two, ( "Expected independently created DomainEvents to have unique " f"correlation_ids, got {cid_one!r} and {cid_two!r}" ) # --------------------------------------------------------------------------- # LoggingEventBus — emit / subscribe # --------------------------------------------------------------------------- @given("a fresh LoggingEventBus for taxonomy tests") def step_given_fresh_logging_bus(ctx: Context) -> None: ctx.logging_taxonomy_bus = LoggingEventBus() ctx.logging_taxonomy_collectors = [] @when('I logging-taxonomy-subscribe to "{et}" events') def step_logging_taxonomy_subscribe(ctx: Context, et: str) -> None: collector = _TaxonomyCollector() ctx.logging_taxonomy_collectors.append(collector) ctx.logging_taxonomy_bus.subscribe(EventType(et), collector) @when('I logging-taxonomy-emit a "{et}" DomainEvent') def step_logging_taxonomy_emit(ctx: Context, et: str) -> None: ctx.logging_taxonomy_bus.emit(_make_event(et)) @then("the logging taxonomy handler should have received {n:d} event(s)") def step_logging_taxonomy_handler_received(ctx: Context, n: int) -> None: assert ctx.logging_taxonomy_collectors, "No logging taxonomy collectors registered" count = len(ctx.logging_taxonomy_collectors[0].received) assert count == n, f"Expected {n} event(s), got {count}" @when( 'I logging-taxonomy-subscribe a failing handler and a normal handler to "{et}" events' ) def step_logging_taxonomy_subscribe_failing_and_normal(ctx: Context, et: str) -> None: normal_collector = _TaxonomyCollector() ctx.logging_taxonomy_collectors.append(normal_collector) def _failing_handler(event: DomainEvent) -> None: raise RuntimeError(f"failing logging handler for {event.event_type.value}") ctx.logging_taxonomy_bus.subscribe(EventType(et), _failing_handler) ctx.logging_taxonomy_bus.subscribe(EventType(et), normal_collector) @then("the logging taxonomy normal handler should have received {n:d} event(s)") def step_logging_taxonomy_normal_handler_received(ctx: Context, n: int) -> None: assert ctx.logging_taxonomy_collectors, ( "No logging taxonomy normal collector registered" ) count = len(ctx.logging_taxonomy_collectors[-1].received) assert count == n, f"Expected normal handler to receive {n} event(s), got {count}" @then("logging-taxonomy-emitting a non-DomainEvent should raise TypeError") def step_logging_taxonomy_emit_bad_type(ctx: Context) -> None: raised = False try: ctx.logging_taxonomy_bus.emit("not-an-event") # type: ignore[arg-type] except TypeError: raised = True assert raised, "Expected TypeError for non-DomainEvent" @then("logging-taxonomy-subscribing with a non-EventType should raise TypeError") def step_logging_taxonomy_subscribe_bad_type(ctx: Context) -> None: raised = False try: ctx.logging_taxonomy_bus.subscribe( "plan.created", lambda e: None, # type: ignore[arg-type] ) except TypeError: raised = True assert raised, "Expected TypeError for non-EventType" @then("logging-taxonomy-subscribing with a non-callable should raise TypeError") def step_logging_taxonomy_subscribe_non_callable(ctx: Context) -> None: raised = False try: ctx.logging_taxonomy_bus.subscribe( EventType.PLAN_CREATED, "not-callable", # type: ignore[arg-type] ) except TypeError: raised = True assert raised, "Expected TypeError for non-callable handler"