From d98d282f5fa4de75b1bd41fcec903a229730bcef Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Thu, 5 Feb 2026 09:56:08 -0500 Subject: [PATCH] tests: fixed unit tests so they are now working --- features/steps/actor_cli_steps.py | 302 +++++++++++---------- src/cleveragents/cli/commands/actor_run.py | 2 +- src/cleveragents/reactive/stream_router.py | 77 +++++- 3 files changed, 229 insertions(+), 152 deletions(-) diff --git a/features/steps/actor_cli_steps.py b/features/steps/actor_cli_steps.py index 5d7258d4..1326533d 100644 --- a/features/steps/actor_cli_steps.py +++ b/features/steps/actor_cli_steps.py @@ -195,8 +195,19 @@ def step_impl(context): @when("I run actor add with that config") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() + mock_actor_service = MagicMock() + + # Mock ensure_built_in_actors to return empty list + mock_actor_registry.ensure_built_in_actors.return_value = [] + + # Set up the actor service to handle the built-in actor calls + mock_actor_service.upsert_actor.return_value = _make_actor(name="openai/gpt-4o") + mock_actor_service.get_default_actor.return_value = _make_actor( + name="openai/gpt-4o", is_default=True + ) + if isinstance(context.actor_config_data, dict): mock_actor = _make_actor(config=context.actor_config_data) mock_actor_registry.upsert_actor.return_value = mock_actor @@ -204,8 +215,9 @@ def step_impl(context): mock_actor_registry.upsert_actor.side_effect = ValidationError( "Config must be a JSON/YAML object" ) - mock_container.return_value.actor_service.return_value = MagicMock() - mock_container.return_value.actor_registry.return_value = mock_actor_registry + + # Return the mocked services + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -245,12 +257,24 @@ def step_impl(context): formatted = json.dumps(value) if not isinstance(value, str) else value option_args.extend(["--option", f"{key}={formatted}"]) - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() + mock_actor_service = MagicMock() + + # Mock ensure_built_in_actors to return empty list + mock_actor_registry.ensure_built_in_actors.return_value = [] + + # Set up the actor service to handle the built-in actor calls + mock_actor_service.upsert_actor.return_value = _make_actor(name="openai/gpt-4o") + mock_actor_service.get_default_actor.return_value = _make_actor( + name="openai/gpt-4o", is_default=True + ) + mock_actor = _make_actor(config=context.actor_config_data) mock_actor_registry.upsert_actor.return_value = mock_actor - mock_container.return_value.actor_service.return_value = MagicMock() - mock_container.return_value.actor_registry.return_value = mock_actor_registry + + # Return the mocked services + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -275,12 +299,24 @@ def step_impl(context): @when("I run actor add with boolean option override") def step_impl(context): option_args = ["--option", "enabled=false"] - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() + mock_actor_service = MagicMock() + + # Mock ensure_built_in_actors to return empty list + mock_actor_registry.ensure_built_in_actors.return_value = [] + + # Set up the actor service to handle the built-in actor calls + mock_actor_service.upsert_actor.return_value = _make_actor(name="openai/gpt-4o") + mock_actor_service.get_default_actor.return_value = _make_actor( + name="openai/gpt-4o", is_default=True + ) + mock_actor = _make_actor(config=context.actor_config_data) mock_actor_registry.upsert_actor.return_value = mock_actor - mock_container.return_value.actor_service.return_value = MagicMock() - mock_container.return_value.actor_registry.return_value = mock_actor_registry + + # Return the mocked services + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -342,8 +378,19 @@ def step_impl(context): @when("I run actor add with that config and unsafe flag") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() + mock_actor_service = MagicMock() + + # Mock ensure_built_in_actors to return empty list + mock_actor_registry.ensure_built_in_actors.return_value = [] + + # Set up the actor service to handle the built-in actor calls + mock_actor_service.upsert_actor.return_value = _make_actor(name="openai/gpt-4o") + mock_actor_service.get_default_actor.return_value = _make_actor( + name="openai/gpt-4o", is_default=True + ) + if isinstance(context.actor_config_data, dict): mock_actor = _make_actor(config=context.actor_config_data, unsafe=True) mock_actor_registry.upsert_actor.return_value = mock_actor @@ -351,8 +398,9 @@ def step_impl(context): mock_actor_registry.upsert_actor.side_effect = ValidationError( "Config must be a JSON/YAML object" ) - mock_container.return_value.actor_service.return_value = MagicMock() - mock_container.return_value.actor_registry.return_value = mock_actor_registry + + # Return the mocked services + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -383,25 +431,18 @@ def step_impl(context): @when("I run actor add via service with that config") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: + actor_service = MagicMock() + graph_descriptor = context.actor_config_data.get("graph") + actor_service.upsert_actor.return_value = _make_actor( + name="service/graph-actor", + provider=context.actor_config_data.get("provider", "openai"), + model=context.actor_config_data.get("model", "gpt-4o"), + config=context.actor_config_data, + graph_descriptor=graph_descriptor, + ) + mock_get_services.return_value = (actor_service, None) - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - graph_descriptor = context.actor_config_data.get("graph") - actor_service.upsert_actor.return_value = _make_actor( - name="service/graph-actor", - provider=context.actor_config_data.get("provider", "openai"), - model=context.actor_config_data.get("model", "gpt-4o"), - config=context.actor_config_data, - graph_descriptor=graph_descriptor, - ) - - with patch("cleveragents.application.container.get_container") as mock_container: - mock_container.return_value = ServiceOnlyContainer(actor_service) context.result = context.runner.invoke( actor_app, [ @@ -466,16 +507,10 @@ def step_impl(context): @when("I run actor add via service without unsafe flag") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: + actor_service = MagicMock() + mock_get_services.return_value = (actor_service, None) - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - with patch("cleveragents.application.container.get_container") as mock_container: - mock_container.return_value = ServiceOnlyContainer(actor_service) context.result = context.runner.invoke( actor_app, [ @@ -491,35 +526,30 @@ def step_impl(context): @when("I run actor add via service with unsafe canonical blob") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service - - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - resolved = MagicMock( - provider="openai", - model="gpt-4o", - graph_descriptor=None, - unsafe=True, - options=None, - ) - canonical_blob = { - "provider": resolved.provider, - "model": resolved.model, - "unsafe": True, - } - with ( patch( "cleveragents.cli.commands.actor._canonicalize_actor_config", - return_value=(resolved, canonical_blob, False), + return_value=( + MagicMock( + provider="openai", + model="gpt-4o", + graph_descriptor=None, + unsafe=True, + options=None, + ), + { + "provider": "openai", + "model": "gpt-4o", + "unsafe": True, + }, + False, + ), ), - patch("cleveragents.application.container.get_container") as mock_container, + patch("cleveragents.cli.commands.actor._get_services") as mock_get_services, ): - mock_container.return_value = ServiceOnlyContainer(actor_service) + actor_service = MagicMock() + mock_get_services.return_value = (actor_service, None) + context.result = context.runner.invoke( actor_app, [ @@ -536,7 +566,7 @@ def step_impl(context): @when("I run actor update with safe flag and yaml config") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() mock_actor_service = MagicMock() current_actor = _make_actor( @@ -555,8 +585,7 @@ def step_impl(context): unsafe=False, ) mock_actor_registry.upsert_actor.return_value = updated_actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -574,7 +603,7 @@ def step_impl(context): @when("I run actor update with unsafe flag") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_registry = MagicMock() mock_actor_service = MagicMock() current_actor = _make_actor( @@ -593,8 +622,7 @@ def step_impl(context): unsafe=True, ) mock_actor_registry.upsert_actor.return_value = updated_actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -632,14 +660,13 @@ def step_impl(context): @when("I run actor update for missing actor") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() mock_actor_registry.get_actor.side_effect = NotFoundError( resource_type="actor", resource_id="local/missing" ) - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -669,7 +696,7 @@ def step_impl(context): @when("I run actor update requiring unsafe confirmation") def step_impl(context): with ( - patch("cleveragents.application.container.get_container") as mock_container, + patch("cleveragents.cli.commands.actor._get_services") as mock_get_services, patch( "cleveragents.cli.commands.actor._canonicalize_actor_config" ) as mock_canonicalize, @@ -678,8 +705,7 @@ def step_impl(context): mock_actor_registry = MagicMock() current_actor = _make_actor(name="local/needs-confirm") mock_actor_registry.get_actor.return_value = current_actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) mock_resolved = MagicMock(unsafe=True) mock_canonicalize.return_value = ( @@ -696,24 +722,17 @@ def step_impl(context): @when("I run actor update via service without unsafe flag") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: + actor_service = MagicMock() + current_actor = _make_actor( + name="local/service-update", + provider="service-provider", + model="service-model", + config={"provider": "service-provider", "model": "service-model"}, + ) + actor_service.get_actor.return_value = current_actor + mock_get_services.return_value = (actor_service, None) - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - current_actor = _make_actor( - name="local/service-update", - provider="service-provider", - model="service-model", - config={"provider": "service-provider", "model": "service-model"}, - ) - actor_service.get_actor.return_value = current_actor - - with patch("cleveragents.application.container.get_container") as mock_container: - mock_container.return_value = ServiceOnlyContainer(actor_service) context.result = context.runner.invoke( actor_app, [ @@ -730,7 +749,7 @@ def step_impl(context): @when("I run actor update with option overrides") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() current_actor = _make_actor( @@ -751,8 +770,7 @@ def step_impl(context): config=current_actor.config_blob, ) mock_actor_registry.upsert_actor.return_value = updated_actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke( actor_app, @@ -773,42 +791,37 @@ def step_impl(context): @when("I run actor update via service with unsafe canonical blob") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service - - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - current_actor = _make_actor( - name="local/service-canonical", - provider="service-provider", - model="service-model", - config={"provider": "service-provider", "model": "service-model"}, - ) - actor_service.get_actor.return_value = current_actor - resolved = MagicMock( - provider=current_actor.provider, - model=current_actor.model, - graph_descriptor=current_actor.graph_descriptor, - unsafe=True, - options=None, - ) - canonical_blob = { - "provider": resolved.provider, - "model": resolved.model, - "unsafe": True, - } - with ( patch( "cleveragents.cli.commands.actor._canonicalize_actor_config", - return_value=(resolved, canonical_blob, False), + return_value=( + MagicMock( + provider="service-provider", + model="service-model", + graph_descriptor=None, + unsafe=True, + options=None, + ), + { + "provider": "service-provider", + "model": "service-model", + "unsafe": True, + }, + False, + ), ), - patch("cleveragents.application.container.get_container") as mock_container, + patch("cleveragents.cli.commands.actor._get_services") as mock_get_services, ): - mock_container.return_value = ServiceOnlyContainer(actor_service) + actor_service = MagicMock() + current_actor = _make_actor( + name="local/service-canonical", + provider="service-provider", + model="service-model", + config={"provider": "service-provider", "model": "service-model"}, + ) + actor_service.get_actor.return_value = current_actor + mock_get_services.return_value = (actor_service, None) + context.result = context.runner.invoke( actor_app, [ @@ -825,11 +838,10 @@ def step_impl(context): @when("I run actor remove successfully") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke(actor_app, ["remove", "local/removable"]) context.mock_actor_registry = mock_actor_registry @@ -851,16 +863,10 @@ def step_impl(context): @when("I run actor remove via service path") def step_impl(context): - class ServiceOnlyContainer: - def __init__(self, actor_service: MagicMock): - self._actor_service = actor_service + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: + actor_service = MagicMock() + mock_get_services.return_value = (actor_service, None) - def actor_service(self): - return self._actor_service - - actor_service = MagicMock() - with patch("cleveragents.application.container.get_container") as mock_container: - mock_container.return_value = ServiceOnlyContainer(actor_service) context.result = context.runner.invoke(actor_app, ["remove", "local/removable"]) context.actor_service = actor_service @@ -868,12 +874,11 @@ def step_impl(context): @when("I run actor list with no actors") def step_impl(context): - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() mock_actor_registry.list_actors.return_value = [] - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke(actor_app, ["list"]) @@ -884,12 +889,11 @@ def step_impl(context): _make_actor(name="local/first", provider="p1", model="m1"), _make_actor(name="local/second", provider="p2", model="m2", unsafe=True), ] - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() mock_actor_registry.list_actors.return_value = actors - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke(actor_app, ["list"]) context.actors = actors @@ -898,12 +902,11 @@ def step_impl(context): @when("I run actor show successfully") def step_impl(context): actor = _make_actor(name="local/show", provider="provider", model="model") - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() mock_actor_registry.get_actor.return_value = actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke(actor_app, ["show", actor.name]) context.actor = actor @@ -925,12 +928,11 @@ def step_impl(context): @when("I run set-default actor successfully") def step_impl(context): actor = _make_actor(name="provider/model", provider="provider", model="model") - with patch("cleveragents.application.container.get_container") as mock_container: + with patch("cleveragents.cli.commands.actor._get_services") as mock_get_services: mock_actor_service = MagicMock() mock_actor_registry = MagicMock() mock_actor_registry.set_default_actor.return_value = actor - mock_container.return_value.actor_service.return_value = mock_actor_service - mock_container.return_value.actor_registry.return_value = mock_actor_registry + mock_get_services.return_value = (mock_actor_service, mock_actor_registry) context.result = context.runner.invoke(actor_app, ["set-default", actor.name]) context.actor = actor diff --git a/src/cleveragents/cli/commands/actor_run.py b/src/cleveragents/cli/commands/actor_run.py index 63c7f696..71bb2b70 100644 --- a/src/cleveragents/cli/commands/actor_run.py +++ b/src/cleveragents/cli/commands/actor_run.py @@ -73,7 +73,7 @@ def run( help="Allow RxPy stream routes in run mode (bypass validation)", ), ] = False, -): +) -> None: """Run the reactive network once with actor-first configs.""" app_exec = ReactiveCleverAgentsApp( config_files=config, diff --git a/src/cleveragents/reactive/stream_router.py b/src/cleveragents/reactive/stream_router.py index e336fff2..599ea1eb 100644 --- a/src/cleveragents/reactive/stream_router.py +++ b/src/cleveragents/reactive/stream_router.py @@ -420,5 +420,80 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes if stream_name not in self.streams: raise StreamRoutingError(f"Stream '{stream_name}' not found") meta = metadata or {} - msg = StreamMessage(content=message, metadata=meta) + import time + + msg = StreamMessage(content=message, metadata=meta, timestamp=time.time()) self.streams[stream_name].on_next(msg) + + def split_stream( + self, source_stream: str, conditions: list[dict[str, Any]] + ) -> None: + """Split a stream based on conditions.""" + if source_stream not in self.streams: + raise StreamRoutingError(f"Stream '{source_stream}' not found") + + source = self.streams[source_stream] + + def splitter(msg: Any) -> None: + for cond in conditions: + target = cond.get("target") + condition = cond.get("condition", {}) + + if ( + target + and target in self.streams + and self._evaluate_condition(msg, condition) + ): + self.streams[target].on_next(msg) + + subscription = source.subscribe(splitter) + self.subscriptions.append(subscription) + + def merge_streams(self, source_streams: list[str], target_stream: str) -> None: + """Merge multiple streams into a single target stream.""" + # Create target stream if it doesn't exist + if target_stream not in self.streams: + self.create_stream(StreamConfig(name=target_stream)) + + target = self.streams[target_stream] + + for source_name in source_streams: + if source_name in self.streams: + subscription = self.streams[source_name].subscribe(target) + self.subscriptions.append(subscription) + + def subscribe_to_output(self, observer: Callable[[Any], None]) -> Any: + """Subscribe to the output stream.""" + return self.streams["__output__"].subscribe(observer) + + def subscribe_to_error(self, observer: Callable[[Any], None]) -> Any: + """Subscribe to the error stream.""" + return self.streams["__error__"].subscribe(observer) + + def dispose(self) -> None: + """Dispose of all streams and subscriptions.""" + # Dispose subscriptions first + for subscription in self.subscriptions[:]: + try: + if hasattr(subscription, "dispose"): + subscription.dispose() + except Exception: # pylint: disable=broad-except + pass # Ignore disposal errors + + # Clear subscriptions + self.subscriptions.clear() + + # Dispose streams + for stream_name in list(self.streams.keys()): + stream = self.streams[stream_name] + try: + if hasattr(stream, "dispose"): + stream.dispose() + except Exception: # pylint: disable=broad-except + pass # Ignore disposal errors + + # Clear all dictionaries + self.streams.clear() + self.observables.clear() + self.stream_configs.clear() + self.agents.clear()