forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
381 lines
14 KiB
Python
381 lines
14 KiB
Python
"""Step definitions for application_container_coverage_r3.feature.
|
|
|
|
Targets remaining uncovered lines in
|
|
``cleveragents.application.container``:
|
|
|
|
- Lines 117, 119: ``_build_stream_router`` function body
|
|
- Lines 124, 126: ``_build_langgraph_bridge`` function body
|
|
- Lines 131, 133: ``_build_route_bridge`` function body
|
|
- Lines 138, 142: ``_lazy_register_registry_agents`` function body
|
|
- Lines 386-391, 393: ``_build_skill_service`` ImportError fallback
|
|
- Lines 851-857: ``get_container`` audit_event_subscriber exception
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import builtins
|
|
import sys
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.container import (
|
|
_build_langgraph_bridge,
|
|
_build_route_bridge,
|
|
_build_skill_service,
|
|
_build_stream_router,
|
|
_lazy_register_registry_agents,
|
|
get_container,
|
|
reset_container,
|
|
)
|
|
|
|
_IN_MEMORY_URL = "sqlite:///:memory:"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Background
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("acncov3 a clean container state")
|
|
def step_acncov3_clean_container(context: Any) -> None:
|
|
"""Reset the global container to avoid cross-test interference."""
|
|
reset_container()
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_stream_router (lines 117, 119)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when("acncov3 I call _build_stream_router")
|
|
def step_acncov3_call_build_stream_router(context: Any) -> None:
|
|
"""Call _build_stream_router which lazily imports ReactiveStreamRouter."""
|
|
try:
|
|
context.acncov3_stream_router_result = _build_stream_router()
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_stream_router_result = None
|
|
context.acncov3_error = exc
|
|
|
|
|
|
@then("acncov3 the result should be a ReactiveStreamRouter instance")
|
|
def step_acncov3_verify_stream_router(context: Any) -> None:
|
|
"""Assert the returned object is a ReactiveStreamRouter."""
|
|
from cleveragents.reactive.stream_router import ReactiveStreamRouter
|
|
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error, got: {context.acncov3_error}"
|
|
)
|
|
result = context.acncov3_stream_router_result
|
|
assert isinstance(result, ReactiveStreamRouter), (
|
|
f"Expected ReactiveStreamRouter, got {type(result).__name__}"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_langgraph_bridge (lines 124, 126)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("acncov3 a mock stream router")
|
|
def step_acncov3_mock_stream_router(context: Any) -> None:
|
|
"""Create a mock stream router for injection."""
|
|
context.acncov3_mock_stream_router = MagicMock(
|
|
name="MockReactiveStreamRouter",
|
|
)
|
|
|
|
|
|
@when("acncov3 I call _build_langgraph_bridge with the mock stream router")
|
|
def step_acncov3_call_build_langgraph_bridge(context: Any) -> None:
|
|
"""Call _build_langgraph_bridge with the mock stream router."""
|
|
try:
|
|
context.acncov3_bridge_result = _build_langgraph_bridge(
|
|
context.acncov3_mock_stream_router,
|
|
)
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_bridge_result = None
|
|
context.acncov3_error = exc
|
|
|
|
|
|
@then("acncov3 the result should be a RxPyLangGraphBridge instance")
|
|
def step_acncov3_verify_langgraph_bridge(context: Any) -> None:
|
|
"""Assert the returned object is a RxPyLangGraphBridge."""
|
|
from cleveragents.langgraph.bridge import RxPyLangGraphBridge
|
|
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error, got: {context.acncov3_error}"
|
|
)
|
|
result = context.acncov3_bridge_result
|
|
assert isinstance(result, RxPyLangGraphBridge), (
|
|
f"Expected RxPyLangGraphBridge, got {type(result).__name__}"
|
|
)
|
|
|
|
|
|
@then("acncov3 the bridge should reference the mock stream router")
|
|
def step_acncov3_verify_bridge_router_ref(context: Any) -> None:
|
|
"""Assert the bridge stores the stream router we injected."""
|
|
bridge = context.acncov3_bridge_result
|
|
assert bridge.stream_router is context.acncov3_mock_stream_router, (
|
|
"Expected the bridge to reference the mock stream router"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_route_bridge (lines 131, 133)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("acncov3 a mock agents registry")
|
|
def step_acncov3_mock_agents(context: Any) -> None:
|
|
"""Create a mock agents / actor registry for injection."""
|
|
context.acncov3_mock_agents = MagicMock(name="MockActorRegistry")
|
|
|
|
|
|
@when("acncov3 I call _build_route_bridge with the mock stream router and agents")
|
|
def step_acncov3_call_build_route_bridge(context: Any) -> None:
|
|
"""Call _build_route_bridge with mock stream_router and agents."""
|
|
try:
|
|
context.acncov3_route_bridge_result = _build_route_bridge(
|
|
context.acncov3_mock_stream_router,
|
|
context.acncov3_mock_agents,
|
|
)
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_route_bridge_result = None
|
|
context.acncov3_error = exc
|
|
|
|
|
|
@then("acncov3 the result should be a RouteBridge instance")
|
|
def step_acncov3_verify_route_bridge(context: Any) -> None:
|
|
"""Assert the returned object is a RouteBridge."""
|
|
from cleveragents.reactive.route_bridge import RouteBridge
|
|
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error, got: {context.acncov3_error}"
|
|
)
|
|
result = context.acncov3_route_bridge_result
|
|
assert isinstance(result, RouteBridge), (
|
|
f"Expected RouteBridge, got {type(result).__name__}"
|
|
)
|
|
|
|
|
|
@then("acncov3 the route bridge should reference the mock stream router")
|
|
def step_acncov3_verify_route_bridge_router_ref(context: Any) -> None:
|
|
"""Assert the route bridge stores the stream router we injected."""
|
|
bridge = context.acncov3_route_bridge_result
|
|
assert bridge.stream_router is context.acncov3_mock_stream_router, (
|
|
"Expected the route bridge to reference the mock stream router"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _lazy_register_registry_agents (lines 138, 142)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("acncov3 a mock stream router for registration")
|
|
def step_acncov3_mock_stream_router_for_reg(context: Any) -> None:
|
|
"""Create a mock stream router for the registration call."""
|
|
context.acncov3_reg_stream_router = MagicMock(
|
|
name="MockStreamRouterForReg",
|
|
)
|
|
|
|
|
|
@given("acncov3 a mock route bridge for registration")
|
|
def step_acncov3_mock_route_bridge_for_reg(context: Any) -> None:
|
|
"""Create a mock route bridge for the registration call."""
|
|
context.acncov3_reg_route_bridge = MagicMock(name="MockRouteBridgeForReg")
|
|
|
|
|
|
@given("acncov3 a mock actor registry for registration")
|
|
def step_acncov3_mock_actor_registry_for_reg(context: Any) -> None:
|
|
"""Create a mock actor registry that returns an empty list."""
|
|
mock_registry = MagicMock(name="MockActorRegistryForReg")
|
|
mock_registry.list_actors.return_value = []
|
|
context.acncov3_reg_actor_registry = mock_registry
|
|
|
|
|
|
@when("acncov3 I call _lazy_register_registry_agents")
|
|
def step_acncov3_call_lazy_register(context: Any) -> None:
|
|
"""Call _lazy_register_registry_agents with mock dependencies.
|
|
|
|
We patch ``register_registry_agents`` in the adapter module so we
|
|
can verify it was called without side-effects, while still
|
|
exercising the lazy import path inside the container function.
|
|
"""
|
|
patcher = patch(
|
|
"cleveragents.application.reactive_registry_adapter.register_registry_agents",
|
|
)
|
|
mock_register = patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
context.acncov3_mock_register_fn = mock_register
|
|
|
|
try:
|
|
context.acncov3_reg_result = _lazy_register_registry_agents(
|
|
context.acncov3_reg_stream_router,
|
|
context.acncov3_reg_route_bridge,
|
|
context.acncov3_reg_actor_registry,
|
|
)
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_reg_result = None
|
|
context.acncov3_error = exc
|
|
|
|
|
|
@then("acncov3 register_registry_agents should have been invoked")
|
|
def step_acncov3_verify_register_called(context: Any) -> None:
|
|
"""Assert that register_registry_agents was called with our mocks."""
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error, got: {context.acncov3_error}"
|
|
)
|
|
context.acncov3_mock_register_fn.assert_called_once_with(
|
|
context.acncov3_reg_stream_router,
|
|
context.acncov3_reg_route_bridge,
|
|
context.acncov3_reg_actor_registry,
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_skill_service ImportError (lines 386-391, 393)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when("acncov3 I call _build_skill_service with SQLAlchemy import blocked")
|
|
def step_acncov3_skill_service_import_blocked(context: Any) -> None:
|
|
"""Force an ImportError for sqlalchemy inside _build_skill_service.
|
|
|
|
We temporarily remove sqlalchemy entries from ``sys.modules`` and
|
|
install a custom ``__import__`` that raises ``ImportError`` for
|
|
sqlalchemy. This exercises the except-ImportError fallback path
|
|
(lines 386-393).
|
|
"""
|
|
# 1. Save and remove all sqlalchemy entries from sys.modules
|
|
saved_modules: dict[str, Any] = {}
|
|
for key in list(sys.modules.keys()):
|
|
if key == "sqlalchemy" or key.startswith("sqlalchemy."):
|
|
saved_modules[key] = sys.modules.pop(key)
|
|
|
|
# 2. Install a blocking __import__
|
|
original_import = builtins.__import__
|
|
|
|
def blocking_import(name: str, *args: Any, **kwargs: Any) -> Any:
|
|
if name == "sqlalchemy" or name.startswith("sqlalchemy."):
|
|
raise ImportError(f"acncov3-mock: No module named '{name}'")
|
|
return original_import(name, *args, **kwargs)
|
|
|
|
builtins.__import__ = blocking_import # type: ignore[assignment]
|
|
|
|
try:
|
|
context.acncov3_skill_result = _build_skill_service(_IN_MEMORY_URL)
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_skill_result = None
|
|
context.acncov3_error = exc
|
|
finally:
|
|
# 3. Restore original __import__ and sys.modules
|
|
builtins.__import__ = original_import # type: ignore[assignment]
|
|
sys.modules.update(saved_modules)
|
|
|
|
|
|
@then("acncov3 the result should be an in-memory SkillService")
|
|
def step_acncov3_verify_inmemory_skill(context: Any) -> None:
|
|
"""Assert the returned object is a SkillService (in-memory fallback)."""
|
|
from cleveragents.application.services.skill_service import SkillService
|
|
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error from _build_skill_service, got: {context.acncov3_error}"
|
|
)
|
|
result = context.acncov3_skill_result
|
|
assert isinstance(result, SkillService), (
|
|
f"Expected SkillService, got {type(result).__name__}"
|
|
)
|
|
|
|
|
|
@then("acncov3 the in-memory SkillService should have no repository")
|
|
def step_acncov3_verify_skill_no_repo(context: Any) -> None:
|
|
"""Assert the fallback SkillService has no DB-backed repository."""
|
|
assert context.acncov3_skill_result._skill_repo is None, (
|
|
"Expected no SkillRepository for the ImportError fallback"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# get_container audit_event_subscriber exception (lines 851-857)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("acncov3 the container is reset")
|
|
def step_acncov3_container_reset(context: Any) -> None:
|
|
"""Ensure the global container is None so get_container creates fresh."""
|
|
reset_container()
|
|
|
|
|
|
@given("acncov3 audit_event_subscriber is patched to raise an error")
|
|
def step_acncov3_patch_audit_subscriber(context: Any) -> None:
|
|
"""Patch AuditEventSubscriber.__init__ to raise RuntimeError.
|
|
|
|
When get_container() calls ``_container.audit_event_subscriber()``
|
|
the Singleton provider tries to construct AuditEventSubscriber.
|
|
Making its ``__init__`` raise exercises the except branch (lines 851-857).
|
|
"""
|
|
patcher = patch(
|
|
"cleveragents.application.services.audit_event_subscriber"
|
|
".AuditEventSubscriber.__init__",
|
|
side_effect=RuntimeError("acncov3-mock: DB not initialised"),
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
|
|
@when("acncov3 I call get_container")
|
|
def step_acncov3_call_get_container(context: Any) -> None:
|
|
"""Call get_container which should catch the audit subscriber error."""
|
|
try:
|
|
context.acncov3_container = get_container()
|
|
context.acncov3_error = None
|
|
except Exception as exc:
|
|
context.acncov3_container = None
|
|
context.acncov3_error = exc
|
|
|
|
|
|
@then("acncov3 a valid container should be returned despite the error")
|
|
def step_acncov3_verify_container_returned(context: Any) -> None:
|
|
"""Assert get_container returned a Container even though the subscriber failed.
|
|
|
|
dependency-injector's ``DeclarativeContainer()`` returns a
|
|
``DynamicContainer`` at runtime, so we check against the base
|
|
``containers.Container`` type.
|
|
"""
|
|
from dependency_injector import containers as di_containers
|
|
|
|
assert context.acncov3_error is None, (
|
|
f"Expected no error from get_container, got: {context.acncov3_error}"
|
|
)
|
|
assert context.acncov3_container is not None, (
|
|
"Expected a container instance, got None"
|
|
)
|
|
assert isinstance(context.acncov3_container, di_containers.Container), (
|
|
f"Expected a DI Container, got {type(context.acncov3_container).__name__}"
|
|
)
|
|
# Verify it has the expected providers
|
|
assert hasattr(context.acncov3_container, "settings"), (
|
|
"Container should have a 'settings' provider"
|
|
)
|
|
assert hasattr(context.acncov3_container, "audit_event_subscriber"), (
|
|
"Container should have an 'audit_event_subscriber' provider"
|
|
)
|
|
|
|
|
|
@then("acncov3 the container should be cached as the global singleton")
|
|
def step_acncov3_verify_container_cached(context: Any) -> None:
|
|
"""Assert the returned container is cached (second call returns same)."""
|
|
second_call = get_container()
|
|
assert second_call is context.acncov3_container, (
|
|
"Expected the same container instance on second call (singleton)"
|
|
)
|
|
# Clean up: reset for other tests
|
|
reset_container()
|