diff --git a/.forgejo/workflows/master.yml b/.forgejo/workflows/master.yml index bb14f9ee0..862103661 100644 --- a/.forgejo/workflows/master.yml +++ b/.forgejo/workflows/master.yml @@ -3,8 +3,6 @@ name: CI on: push: branches: [master, develop] - pull_request: - branches: [master, develop] vars: docker_prefix: "http://harbor.cleverthis.com/docker/" diff --git a/CHANGELOG.md b/CHANGELOG.md index a376dc67c..ff1844bee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Added + +- **ReactiveEventBus close() lifecycle and context manager protocol** (#10916): + Added `_closed` flag and `is_closed` public property to `ReactiveEventBus`. + `emit()` now raises `RuntimeError` when called after `close()`. Added + `__enter__`/`__exit__` context manager protocol for automatic cleanup. + Added BDD scenarios covering: close marks bus as closed and clears subscriptions, + emit raises RuntimeError after close, and context manager protocol. Fixed + single-quote style inconsistency in step decorators and removed accidentally + committed `add_close_steps.py` patch script. + ### Changed - **CI coverage job now waits for unit_tests** (#10714): Added `unit_tests` to the diff --git a/features/event_bus.feature b/features/event_bus.feature index 2875b4f58..d9a3cc455 100644 --- a/features/event_bus.feature +++ b/features/event_bus.feature @@ -119,6 +119,37 @@ Feature: EventBus protocol and domain event emission Then the EventBus Protocol stubs should be callable # --------------------------------------------------------------------------- + + # --------------------------------------------------------------------------- + # ReactiveEventBus close() - RxPY subject lifecycle + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus close marks bus as closed and clears subscriptions + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I emit a "plan.created" DomainEvent + And I close the ReactiveEventBus + Then the subscribed handler should have received 1 event + And the subscriptions should be cleared + And the bus should be closed + + Scenario: ReactiveEventBus close clears the audit log + Given a ReactiveEventBus + When I emit a "plan.created" DomainEvent + And I emit a "decision.created" DomainEvent + And I close the ReactiveEventBus + Then the audit log should be empty + + Scenario: ReactiveEventBus emit raises RuntimeError after close + Given a ReactiveEventBus + When I close the ReactiveEventBus + And I emit a "plan.created" DomainEvent after close + Then a RuntimeError should be raised + + Scenario: ReactiveEventBus supports context manager protocol + When I use the ReactiveEventBus as a context manager + Then the bus should be closed after the context manager exits + # DecisionService event emission # --------------------------------------------------------------------------- diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index 35c1abd93..c6771ddf7 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -256,6 +256,79 @@ def step_subscribe_non_callable(ctx: Context) -> None: assert raised, "Expected TypeError for non-callable handler" + +# --------------------------------------------------------------------------- +# ReactiveEventBus close() steps +# --------------------------------------------------------------------------- + + +@when("I close the ReactiveEventBus") +def step_close_reactive_bus(ctx: Context) -> None: + ctx.bus.close() + + +@then("the subscribed handler should have received {n:d} event") +def step_subscribed_handler_received(ctx: Context, n: int) -> None: + assert ctx.collectors, "No collectors registered" + count = len(ctx.collectors[0].received) + assert count == n, f"Expected {n} events, got {count}" + + +@then("the bus should be closed") +def step_bus_is_closed(ctx: Context) -> None: + assert ctx.bus.is_closed, "Expected bus.is_closed to be True after close()" + + +@then("the subscriptions should be cleared") +def step_subscriptions_cleared(ctx: Context) -> None: + # Verify the bus is closed (which guarantees subscriptions are cleared) + assert ctx.bus.is_closed, "Expected bus.is_closed to be True after close()" + # Also verify via audit_log that no further events can be dispatched + assert len(ctx.bus.audit_log) == 0, ( + "Expected empty audit log after close (subscriptions and audit log cleared)" + ) + + +@then("the audit log should be empty") +def step_audit_log_empty(ctx: Context) -> None: + assert len(ctx.bus.audit_log) == 0, ( + f"Expected empty audit log after close, found {len(ctx.bus.audit_log)} events" + ) + + +@when('I emit a "{et}" DomainEvent after close') +def step_emit_after_close(ctx: Context, et: str) -> None: + ctx.exception = None + try: + ctx.bus.emit(_make_event(et)) + except RuntimeError as exc: + ctx.exception = exc + + +@then("a RuntimeError should be raised") +def step_runtime_error_raised(ctx: Context) -> None: + assert isinstance(ctx.exception, RuntimeError), ( + f"Expected RuntimeError, got {type(ctx.exception).__name__!r}: {ctx.exception}" + ) + + +@when("I use the ReactiveEventBus as a context manager") +def step_use_as_context_manager(ctx: Context) -> None: + ctx.context_manager_bus: ReactiveEventBus | None = None + with ReactiveEventBus() as bus: + ctx.context_manager_bus = bus + bus.subscribe(EventType.PLAN_CREATED, lambda e: None) + bus.emit(_make_event("plan.created")) + ctx.closed_bus = bus + + +@then("the bus should be closed after the context manager exits") +def step_bus_closed_after_context_manager(ctx: Context) -> None: + assert ctx.closed_bus.is_closed, ( + "Expected bus.is_closed to be True after context manager exit" + ) + + @then("the EventBus Protocol stubs should be callable") def step_protocol_stubs_callable(ctx: Context) -> None: """Exercise Protocol method stubs directly to ensure 100% coverage.""" diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index d55a2eb50..ca2f45303 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -60,6 +60,7 @@ class ReactiveEventBus: self._stream: Observable = self._subject.pipe(ops.map(_identity)) self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size) + self._closed: bool = False # ------------------------------------------------------------------ # Public API (satisfies EventBus protocol) @@ -93,6 +94,11 @@ class ReactiveEventBus: """Clear all retained in-memory audit events.""" self._audit_log.clear() + @property + def is_closed(self) -> bool: + """Return ``True`` if :meth:`close` has been called on this bus.""" + return self._closed + def emit(self, event: DomainEvent) -> None: """Publish *event* to the reactive stream and all type handlers. @@ -107,7 +113,12 @@ class ReactiveEventBus: Raises: TypeError: If *event* is not a :class:`DomainEvent`. + RuntimeError: If :meth:`close` has already been called. """ + if self._closed: + raise RuntimeError( + "Cannot emit events on a closed ReactiveEventBus" + ) if not isinstance(event, DomainEvent): raise TypeError( f"event must be a DomainEvent, got {type(event).__name__!r}" @@ -179,11 +190,35 @@ class ReactiveEventBus: Call this when the bus is no longer needed (e.g. in test teardown) to release any RxPY Subject resources and prevent subscription leaks between test scenarios. + + After calling ``close()``, any subsequent call to :meth:`emit` will + raise :exc:`RuntimeError`. """ + self._closed = True with contextlib.suppress(Exception): self._subject.on_completed() self._subscriptions.clear() self._audit_log.clear() + def __enter__(self) -> ReactiveEventBus: + """Support use as a context manager. + + Returns: + This :class:`ReactiveEventBus` instance. + """ + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: object, + ) -> None: + """Close the bus on context manager exit. + + Calls :meth:`close` regardless of whether an exception occurred. + """ + self.close() + __all__ = ["ReactiveEventBus"] diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 09aea2413..785acd4cb 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -1164,6 +1164,9 @@ _build_analyzer_registry # noqa: B018, F821 # ReactiveEventBus.audit_log — public property for event audit trail (#587) audit_log # noqa: B018, F821 +# ReactiveEventBus.is_closed — public property for closed state inspection (#10916) +is_closed # noqa: B018, F821 + # Test doubles — public API for BDD/Robot test infrastructure TrackingEventBus # noqa: B018, F821