forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""Step definitions for application_container_coverage_boost.feature.
|
|
|
|
Targets the remaining uncovered lines in
|
|
``cleveragents.application.container``:
|
|
|
|
- Lines 187-195: ``_build_checkpoint_service`` function body
|
|
- Lines 204-211: ``_build_trace_service`` function body
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.container import (
|
|
_build_checkpoint_service,
|
|
_build_session_factory,
|
|
_build_skill_service,
|
|
_build_trace_service,
|
|
reset_container,
|
|
)
|
|
from cleveragents.application.services.checkpoint_service import CheckpointService
|
|
from cleveragents.application.services.skill_service import SkillService
|
|
from cleveragents.application.services.trace_service import TraceService
|
|
from cleveragents.infrastructure.database.llm_trace_repository import (
|
|
LLMTraceRepository,
|
|
)
|
|
from cleveragents.infrastructure.database.repositories import CheckpointRepository
|
|
|
|
_IN_MEMORY_URL = "sqlite:///:memory:"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Background
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a clean container state for coverage boost tests")
|
|
def step_clean_container_state(context):
|
|
"""Reset the global container to avoid cross-test interference."""
|
|
reset_container()
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_checkpoint_service scenarios
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a mock plan lifecycle service")
|
|
def step_mock_plan_lifecycle_service(context):
|
|
"""Create a mock PlanLifecycleService for injection."""
|
|
context.boost_mock_lifecycle = MagicMock()
|
|
context.boost_mock_lifecycle.__class__.__name__ = "PlanLifecycleService"
|
|
|
|
|
|
@when("I build a checkpoint service with an in-memory database URL")
|
|
def step_build_checkpoint_service_default(context):
|
|
"""Call _build_checkpoint_service with only a database URL (no lifecycle)."""
|
|
context.boost_checkpoint_svc = _build_checkpoint_service(_IN_MEMORY_URL)
|
|
|
|
|
|
@when("I build a checkpoint service with the mock plan lifecycle service")
|
|
def step_build_checkpoint_service_with_lifecycle(context):
|
|
"""Call _build_checkpoint_service with a database URL and lifecycle service."""
|
|
context.boost_checkpoint_svc = _build_checkpoint_service(
|
|
_IN_MEMORY_URL,
|
|
plan_lifecycle_service=context.boost_mock_lifecycle,
|
|
)
|
|
|
|
|
|
@then("acbs the result should be a CheckpointService instance")
|
|
def step_verify_checkpoint_service_type(context):
|
|
"""Assert the returned object is a CheckpointService."""
|
|
assert isinstance(context.boost_checkpoint_svc, CheckpointService), (
|
|
f"Expected CheckpointService, got {type(context.boost_checkpoint_svc).__name__}"
|
|
)
|
|
|
|
|
|
@then("the checkpoint service should have a repository")
|
|
def step_verify_checkpoint_has_repository(context):
|
|
"""Assert the service was initialised with a CheckpointRepository."""
|
|
repo = context.boost_checkpoint_svc._repository
|
|
assert repo is not None, "Expected a repository, got None"
|
|
assert isinstance(repo, CheckpointRepository), (
|
|
f"Expected CheckpointRepository, got {type(repo).__name__}"
|
|
)
|
|
|
|
|
|
@then("the checkpoint service should have no plan lifecycle service")
|
|
def step_verify_checkpoint_no_lifecycle(context):
|
|
"""Assert the service has no plan_lifecycle_service (default None)."""
|
|
assert context.boost_checkpoint_svc._plan_lifecycle_service is None
|
|
|
|
|
|
@then("the checkpoint service should reference the mock plan lifecycle service")
|
|
def step_verify_checkpoint_has_lifecycle(context):
|
|
"""Assert the service references the mock lifecycle we injected."""
|
|
assert (
|
|
context.boost_checkpoint_svc._plan_lifecycle_service
|
|
is context.boost_mock_lifecycle
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_trace_service scenarios
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("explicit application settings for trace service")
|
|
def step_explicit_settings_for_trace(context):
|
|
"""Create a real Settings instance to pass explicitly."""
|
|
from cleveragents.config.settings import get_settings
|
|
|
|
context.boost_explicit_settings = get_settings()
|
|
|
|
|
|
@when("I build a trace service with an in-memory database URL and no explicit settings")
|
|
def step_build_trace_service_default(context):
|
|
"""Call _build_trace_service with only a database URL."""
|
|
context.boost_trace_svc = _build_trace_service(_IN_MEMORY_URL)
|
|
|
|
|
|
@when("I build a trace service with the explicit settings")
|
|
def step_build_trace_service_with_settings(context):
|
|
"""Call _build_trace_service with a database URL and explicit settings."""
|
|
context.boost_trace_svc = _build_trace_service(
|
|
_IN_MEMORY_URL,
|
|
settings=context.boost_explicit_settings,
|
|
)
|
|
|
|
|
|
@then("acbs the result should be a TraceService instance")
|
|
def step_verify_trace_service_type(context):
|
|
"""Assert the returned object is a TraceService."""
|
|
assert isinstance(context.boost_trace_svc, TraceService), (
|
|
f"Expected TraceService, got {type(context.boost_trace_svc).__name__}"
|
|
)
|
|
|
|
|
|
@then("the trace service should have a repository")
|
|
def step_verify_trace_has_repository(context):
|
|
"""Assert the service was initialised with an LLMTraceRepository."""
|
|
repo = context.boost_trace_svc._repository
|
|
assert repo is not None, "Expected a repository, got None"
|
|
assert isinstance(repo, LLMTraceRepository), (
|
|
f"Expected LLMTraceRepository, got {type(repo).__name__}"
|
|
)
|
|
|
|
|
|
@then("the trace service should have resolved settings from defaults")
|
|
def step_verify_trace_default_settings(context):
|
|
"""Assert the service has settings resolved via get_settings() fallback."""
|
|
from cleveragents.config.settings import Settings
|
|
|
|
svc_settings = context.boost_trace_svc._settings
|
|
assert svc_settings is not None, "Expected settings, got None"
|
|
assert isinstance(svc_settings, Settings), (
|
|
f"Expected Settings, got {type(svc_settings).__name__}"
|
|
)
|
|
|
|
|
|
@then("the trace service should use the explicitly provided settings")
|
|
def step_verify_trace_explicit_settings(context):
|
|
"""Assert the service uses the exact Settings object we injected."""
|
|
assert context.boost_trace_svc._settings is context.boost_explicit_settings, (
|
|
"Expected the explicitly provided settings instance"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_session_factory scenarios
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when("I build a session factory with an in-memory database URL")
|
|
def step_build_session_factory(context: Any) -> None:
|
|
"""Call _build_session_factory with an in-memory database URL."""
|
|
context.boost_session_factory = _build_session_factory(_IN_MEMORY_URL)
|
|
|
|
|
|
@then("the result should be a callable sessionmaker")
|
|
def step_verify_sessionmaker_callable(context: Any) -> None:
|
|
"""Assert the returned object is a callable sessionmaker."""
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
assert isinstance(context.boost_session_factory, sessionmaker), (
|
|
f"Expected sessionmaker, got {type(context.boost_session_factory).__name__}"
|
|
)
|
|
assert callable(context.boost_session_factory)
|
|
|
|
|
|
@then("calling the session factory should produce a Session")
|
|
def step_verify_session_factory_produces_session(context: Any) -> None:
|
|
"""Assert that calling the factory returns a Session instance."""
|
|
from sqlalchemy.orm import Session
|
|
|
|
session = context.boost_session_factory()
|
|
try:
|
|
assert isinstance(session, Session), (
|
|
f"Expected Session, got {type(session).__name__}"
|
|
)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@when("I build a session factory with an invalid database URL")
|
|
def step_build_session_factory_invalid(context: Any) -> None:
|
|
"""Call _build_session_factory with an invalid URL — error deferred to usage."""
|
|
# SQLAlchemy lazily connects; the factory itself succeeds, but
|
|
# calling it to create a session and executing SQL raises.
|
|
context.boost_session_factory = _build_session_factory(
|
|
"sqlite:///nonexistent/path/to/db.sqlite"
|
|
)
|
|
|
|
|
|
@then("a database error should be raised when the factory is called")
|
|
def step_verify_session_factory_error(context: Any) -> None:
|
|
"""Assert that using the factory with an invalid URL raises an error."""
|
|
from sqlalchemy import text as sa_text
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
session = context.boost_session_factory()
|
|
try:
|
|
# Force a connection attempt by executing a trivial query.
|
|
session.execute(sa_text("SELECT 1"))
|
|
raise AssertionError("Expected OperationalError, but no error was raised")
|
|
except OperationalError:
|
|
pass # Expected
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@when("I resolve session_factory from the container with an in-memory database URL")
|
|
def step_resolve_container_session_factory(context: Any) -> None:
|
|
"""Resolve session_factory from the DI container using an in-memory URL."""
|
|
from unittest.mock import patch
|
|
|
|
with patch.dict("os.environ", {"CLEVERAGENTS_DATABASE_URL": _IN_MEMORY_URL}):
|
|
from cleveragents.application.container import get_container
|
|
|
|
reset_container()
|
|
try:
|
|
container = get_container()
|
|
context.boost_resolved_factory = container.session_factory()
|
|
finally:
|
|
reset_container()
|
|
|
|
|
|
@then("the resolved session factory should be callable")
|
|
def step_verify_resolved_factory_callable(context: Any) -> None:
|
|
"""Assert the container-resolved factory is callable."""
|
|
assert callable(context.boost_resolved_factory), (
|
|
"Expected a callable session factory from the container"
|
|
)
|
|
|
|
|
|
@then("calling the resolved session factory should produce a Session")
|
|
def step_verify_resolved_factory_produces_session(context: Any) -> None:
|
|
"""Assert that calling the resolved factory returns a Session instance."""
|
|
from sqlalchemy.orm import Session
|
|
|
|
session = context.boost_resolved_factory()
|
|
try:
|
|
assert isinstance(session, Session), (
|
|
f"Expected Session, got {type(session).__name__}"
|
|
)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# _build_skill_service scenarios
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when("I build a skill service with an in-memory database URL")
|
|
def step_build_skill_service_default(context):
|
|
"""Call _build_skill_service with a valid in-memory database URL."""
|
|
context.boost_skill_svc = _build_skill_service(_IN_MEMORY_URL)
|
|
|
|
|
|
@when("I build a skill service with an invalid database URL")
|
|
def step_build_skill_service_invalid(context):
|
|
"""Mock create_engine to raise OperationalError, triggering the fallback."""
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
with patch(
|
|
"sqlalchemy.create_engine",
|
|
side_effect=OperationalError("mock", {}, Exception("DB unavailable")),
|
|
):
|
|
context.boost_skill_svc = _build_skill_service(
|
|
"sqlite:////nonexistent/path/db.sqlite"
|
|
)
|
|
|
|
|
|
@then("the result should be a SkillService instance")
|
|
def step_verify_skill_service_type(context):
|
|
"""Assert the returned object is a SkillService."""
|
|
assert isinstance(context.boost_skill_svc, SkillService), (
|
|
f"Expected SkillService, got {type(context.boost_skill_svc).__name__}"
|
|
)
|
|
|
|
|
|
@then("the skill service should have a repository")
|
|
def step_verify_skill_has_repository(context):
|
|
"""Assert the service was initialised with a SkillRepository."""
|
|
assert context.boost_skill_svc._skill_repo is not None, (
|
|
"Expected a SkillRepository, got None"
|
|
)
|
|
|
|
|
|
@then("the skill service should have no repository")
|
|
def step_verify_skill_no_repository(context):
|
|
"""Assert the fallback service has no DB-backed repository."""
|
|
assert context.boost_skill_svc._skill_repo is None, (
|
|
"Expected no SkillRepository for in-memory fallback"
|
|
)
|