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
316 lines
11 KiB
Python
316 lines
11 KiB
Python
"""Step definitions for sandbox_strategy_registry_coverage.feature.
|
|
|
|
These steps target specific uncovered lines in strategy_registry.py:
|
|
- Lines 65-68: CustomStrategyConfig.__init__ assignment block
|
|
- Line 167: register_from_config delegation
|
|
- Lines 196-200: register_all_from_config skip on missing module/class
|
|
- Lines 205-209: register_all_from_config error handling on load failure
|
|
"""
|
|
|
|
import logging
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.infrastructure.plugins.exceptions import (
|
|
PluginLoadError,
|
|
)
|
|
from cleveragents.infrastructure.sandbox.strategy_registry import (
|
|
CustomStrategyConfig,
|
|
SandboxStrategyRegistry,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: a class that satisfies _validate_protocol (has all 9 methods)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _CompliantStrategy:
|
|
"""Dummy class with all 9 required SandboxStrategyProtocol methods."""
|
|
|
|
def create(self): ...
|
|
def read(self): ...
|
|
def write(self): ...
|
|
def diff(self): ...
|
|
def commit(self): ...
|
|
def rollback(self): ...
|
|
def checkpoint(self): ...
|
|
def restore_checkpoint(self): ...
|
|
def cleanup(self): ...
|
|
|
|
|
|
class _NonCompliantStrategy:
|
|
"""Dummy class missing required methods."""
|
|
|
|
def create(self): ...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the sandbox strategy registry module is imported")
|
|
def step_module_imported(context):
|
|
"""Verify the module is importable."""
|
|
assert CustomStrategyConfig is not None
|
|
assert SandboxStrategyRegistry is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CustomStrategyConfig success (lines 65-68)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I create a CustomStrategyConfig with name "{name}" module "{module}" and class "{cls}"'
|
|
)
|
|
def step_create_config(context, name, module, cls):
|
|
"""Create a CustomStrategyConfig with the given values."""
|
|
context.strategy_config = CustomStrategyConfig(
|
|
name=name,
|
|
module=module,
|
|
class_name=cls,
|
|
)
|
|
|
|
|
|
@then('the config should have name "{name}" module "{module}" and class_name "{cls}"')
|
|
def step_verify_config_fields(context, name, module, cls):
|
|
"""Verify the config fields were assigned correctly."""
|
|
cfg = context.strategy_config
|
|
assert cfg.name == name, f"Expected name={name!r}, got {cfg.name!r}"
|
|
assert cfg.module == module, f"Expected module={module!r}, got {cfg.module!r}"
|
|
assert cfg.class_name == cls, f"Expected class_name={cls!r}, got {cfg.class_name!r}"
|
|
|
|
|
|
@then("the config options should default to an empty dict")
|
|
def step_verify_default_options(context):
|
|
"""Verify options defaults to empty dict when not provided."""
|
|
assert context.strategy_config.options == {}, (
|
|
f"Expected empty dict, got {context.strategy_config.options!r}"
|
|
)
|
|
|
|
|
|
@when(
|
|
'I create a CustomStrategyConfig with name "{name}" module "{module}" class "{cls}" and options'
|
|
)
|
|
def step_create_config_with_options(context, name, module, cls):
|
|
"""Create a CustomStrategyConfig with explicit options."""
|
|
context.strategy_config = CustomStrategyConfig(
|
|
name=name,
|
|
module=module,
|
|
class_name=cls,
|
|
options={"timeout": 30, "retries": 5},
|
|
)
|
|
|
|
|
|
@then("the config options should contain the provided values")
|
|
def step_verify_explicit_options(context):
|
|
"""Verify options stores the provided dict."""
|
|
opts = context.strategy_config.options
|
|
assert opts == {"timeout": 30, "retries": 5}, f"Unexpected options: {opts!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry with mock loader helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a registry with a mock loader that returns a compliant class")
|
|
def step_registry_with_mock_loader(context):
|
|
"""Create a SandboxStrategyRegistry with a mocked PluginLoader."""
|
|
mock_loader = MagicMock()
|
|
mock_loader.load_class.return_value = _CompliantStrategy
|
|
context.registry = SandboxStrategyRegistry(loader=mock_loader)
|
|
context.mock_loader = mock_loader
|
|
|
|
|
|
@given('a CustomStrategyConfig named "{name}" with module "{module}" and class "{cls}"')
|
|
def step_create_named_config(context, name, module, cls):
|
|
"""Create and store a CustomStrategyConfig for later use."""
|
|
context.strategy_config = CustomStrategyConfig(
|
|
name=name,
|
|
module=module,
|
|
class_name=cls,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# register_from_config (line 167)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call register_from_config with that config")
|
|
def step_call_register_from_config(context):
|
|
"""Call register_from_config using the stored config."""
|
|
context.returned_cls = context.registry.register_from_config(
|
|
context.strategy_config,
|
|
)
|
|
|
|
|
|
@then('the strategy "{name}" should be registered in the registry')
|
|
def step_verify_strategy_registered(context, name):
|
|
"""Verify the strategy was successfully registered."""
|
|
assert context.registry.has(name), (
|
|
f"Strategy {name!r} not found; registered: {context.registry.list_strategies()}"
|
|
)
|
|
assert context.returned_cls is _CompliantStrategy
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# register_all_from_config: skip missing module (lines 196-200)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call register_all_from_config with an entry that has an empty module")
|
|
def step_register_all_missing_module(context):
|
|
"""Call register_all_from_config with a config missing the module key."""
|
|
context.log_handler = _CapturingLogHandler()
|
|
logger = logging.getLogger("cleveragents.infrastructure.sandbox.strategy_registry")
|
|
# Re-enable the logger in case Alembic's fileConfig() disabled it
|
|
logger.disabled = False
|
|
logger.addHandler(context.log_handler)
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
def cleanup():
|
|
logger.removeHandler(context.log_handler)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
configs = {
|
|
"bad_no_module": {
|
|
"module": "",
|
|
"class": "SomeClass",
|
|
},
|
|
}
|
|
context.registered_names = context.registry.register_all_from_config(configs)
|
|
context.skipped_entry_name = "bad_no_module"
|
|
|
|
|
|
@then("the returned list should not contain the skipped entry name")
|
|
def step_verify_skipped_not_in_list(context):
|
|
"""Verify the skipped entry is not in the returned list."""
|
|
assert context.skipped_entry_name not in context.registered_names, (
|
|
f"Expected {context.skipped_entry_name!r} to be absent, "
|
|
f"got {context.registered_names}"
|
|
)
|
|
|
|
|
|
@then("a warning should be logged about the skipped entry")
|
|
def step_verify_warning_logged(context):
|
|
"""Verify a warning was logged for the skipped entry."""
|
|
warnings = [r for r in context.log_handler.records if r.levelno == logging.WARNING]
|
|
assert any("Skipping custom strategy" in r.getMessage() for r in warnings), (
|
|
f"Expected a 'Skipping custom strategy' warning, "
|
|
f"got: {[r.getMessage() for r in warnings]}"
|
|
)
|
|
|
|
|
|
@when("I call register_all_from_config with an entry that has an empty class")
|
|
def step_register_all_missing_class(context):
|
|
"""Call register_all_from_config with a config missing the class key."""
|
|
configs = {
|
|
"bad_no_class": {
|
|
"module": "some.module",
|
|
"class": "",
|
|
},
|
|
}
|
|
context.registered_names = context.registry.register_all_from_config(configs)
|
|
context.skipped_entry_name = "bad_no_class"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# register_all_from_config: PluginLoadError (lines 205-209)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a registry whose loader raises PluginLoadError")
|
|
def step_registry_loader_raises_load_error(context):
|
|
"""Create a registry whose loader always raises PluginLoadError."""
|
|
mock_loader = MagicMock()
|
|
mock_loader.load_class.side_effect = PluginLoadError("boom")
|
|
context.registry = SandboxStrategyRegistry(loader=mock_loader)
|
|
|
|
# Attach a log handler to capture error logs
|
|
context.log_handler = _CapturingLogHandler()
|
|
logger = logging.getLogger("cleveragents.infrastructure.sandbox.strategy_registry")
|
|
# Re-enable the logger in case Alembic's fileConfig() disabled it
|
|
logger.disabled = False
|
|
logger.addHandler(context.log_handler)
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
def cleanup():
|
|
logger.removeHandler(context.log_handler)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I call register_all_from_config with an entry that will fail to load")
|
|
def step_register_all_fail_load(context):
|
|
"""Call register_all_from_config with a valid-looking config that will fail."""
|
|
configs = {
|
|
"failing_strategy": {
|
|
"module": "nonexistent.module",
|
|
"class": "NoSuchClass",
|
|
},
|
|
}
|
|
context.registered_names = context.registry.register_all_from_config(configs)
|
|
|
|
|
|
@then("the returned list should be empty")
|
|
def step_verify_empty_result(context):
|
|
"""Verify no strategies were registered."""
|
|
assert context.registered_names == [], (
|
|
f"Expected empty list, got {context.registered_names!r}"
|
|
)
|
|
|
|
|
|
@then("an error should be logged about the failed registration")
|
|
def step_verify_error_logged(context):
|
|
"""Verify an error was logged for the failed registration."""
|
|
errors = [r for r in context.log_handler.records if r.levelno == logging.ERROR]
|
|
assert any(
|
|
"Failed to register custom strategy" in r.getMessage() for r in errors
|
|
), f"Expected a 'Failed to register' error, got: {[r.getMessage() for r in errors]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# register_all_from_config: ProtocolMismatchError (lines 205-209)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a registry whose loader returns a non-compliant class")
|
|
def step_registry_loader_returns_bad_class(context):
|
|
"""Create a registry whose loader returns a class that fails protocol check."""
|
|
mock_loader = MagicMock()
|
|
mock_loader.load_class.return_value = _NonCompliantStrategy
|
|
context.registry = SandboxStrategyRegistry(loader=mock_loader)
|
|
|
|
|
|
@when("I call register_all_from_config with an entry that will fail protocol check")
|
|
def step_register_all_fail_protocol(context):
|
|
"""Call register_all_from_config; the protocol validation will fail."""
|
|
configs = {
|
|
"bad_protocol": {
|
|
"module": "some.module",
|
|
"class": "BadClass",
|
|
},
|
|
}
|
|
context.registered_names = context.registry.register_all_from_config(configs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helper: capturing log handler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _CapturingLogHandler(logging.Handler):
|
|
"""A logging handler that captures LogRecords for assertion."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.records: list[logging.LogRecord] = []
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
self.records.append(record)
|