fix(acms): defer settings strategy auto-load past _strategies/_logger init
The previous ACMSPipeline.__init__ invoked _load_strategies_from_settings at line 831 before self._strategies (line 834) and self._logger (line 864) were initialized. When a Settings whose context dict contained a 'strategies' key was passed in, the method body raised AttributeError on 'self._strategies', and the except handler then raised a second AttributeError on 'self._logger' that propagated out of __init__ and crashed pipeline construction entirely. Move the auto-load call to after both attributes are bound. Convert the pytest-shaped tests/strategies/test_strategy_registry.py (which the behave-based unit_tests gate never ran) into a behave feature file + step definitions under features/, matching the project's BDD convention and bringing the strategy auto-loading paths under actual CI coverage. ISSUES CLOSED: #7527
This commit is contained in:
Forgejo
@@ -0,0 +1,80 @@
|
||||
@phase2 @acms @strategy
|
||||
Feature: ACMS Pipeline Auto-Loading Strategies from Settings
|
||||
As a CleverAgents developer
|
||||
I want the ACMS pipeline to auto-load strategies from a Settings
|
||||
object's context.strategies configuration
|
||||
So that downstream code does not have to register every strategy by hand,
|
||||
and so that a Settings carrying strategy config cannot crash __init__
|
||||
with an AttributeError on _strategies / _logger.
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Initialization order regression (init-order AttributeError fix)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@auto_load @init_order
|
||||
Scenario: Pipeline constructs cleanly when Settings carries an empty strategies sub-dict
|
||||
Given a mock Settings whose context dict contains an empty strategies sub-dict
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then construction should succeed without raising
|
||||
And the resulting pipeline should expose its registered strategies
|
||||
|
||||
@auto_load @init_order
|
||||
Scenario: Pipeline registers strategies declared in the Settings registry
|
||||
Given a mock Settings whose context.strategies enables two built-ins
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then construction should succeed without raising
|
||||
And the pipeline strategies should include "simple-keyword"
|
||||
|
||||
@auto_load
|
||||
Scenario: Pipeline tolerates a Settings without any context attribute
|
||||
Given a mock Settings without any context attribute
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then construction should succeed without raising
|
||||
|
||||
@auto_load
|
||||
Scenario: Pipeline tolerates a Settings whose context is a non-dict object
|
||||
Given a mock Settings whose context is a non-dict object
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then construction should succeed without raising
|
||||
|
||||
@auto_load @error_handling
|
||||
Scenario: Pipeline survives a malformed strategies config without crashing
|
||||
Given a mock Settings whose context.strategies references an invalid custom strategy module
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then construction should succeed without raising
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _enforce_enabled_strategies — filter pipeline strategy set to enabled subset
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@auto_load @enforce
|
||||
Scenario: Enabled list filters the pipeline strategy set
|
||||
Given a mock Settings whose context.strategies enables only "simple-keyword"
|
||||
When I construct an ACMSPipeline with that Settings
|
||||
Then the pipeline strategies should contain "simple-keyword"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# load_strategies_from_config helper (context_strategies.py)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@load_helper
|
||||
Scenario: load_strategies_from_config registers the three default built-ins
|
||||
When I call load_strategies_from_config with an empty config dict
|
||||
Then the resulting registry should contain "simple-keyword"
|
||||
And the resulting registry should contain "semantic-embedding"
|
||||
And the resulting registry should contain "breadth-depth-navigator"
|
||||
|
||||
@load_helper
|
||||
Scenario: load_strategies_from_config honors an explicit enabled list
|
||||
When I call load_strategies_from_config with an enabled list of only "simple-keyword"
|
||||
Then the resulting registry should list only "simple-keyword" as enabled
|
||||
|
||||
@load_helper
|
||||
Scenario: load_strategies_from_config defaults to DEFAULT_ENABLED_STRATEGIES when enabled is missing
|
||||
When I call load_strategies_from_config with no enabled key
|
||||
Then the resulting registry should list the default-enabled strategies
|
||||
|
||||
@load_helper @custom
|
||||
Scenario: load_strategies_from_config registers a custom strategy from a module:ClassName string
|
||||
When I call load_strategies_from_config with a custom strategy mapping "my-keyword" to SimpleKeywordStrategy
|
||||
Then the resulting registry should contain "my-keyword"
|
||||
@@ -0,0 +1,210 @@
|
||||
"""Behave step definitions for ACMS pipeline strategy auto-loading.
|
||||
|
||||
Covers:
|
||||
- The init-order regression where ``ACMSPipeline.__init__`` invoked
|
||||
``_load_strategies_from_settings`` before ``self._strategies`` /
|
||||
``self._logger`` existed, crashing construction with ``AttributeError``.
|
||||
- The ``_load_strategies_from_settings`` body (early-return paths, dict
|
||||
handling, registry merge, enabled-list enforcement, exception logging).
|
||||
- The ``_enforce_enabled_strategies`` filter.
|
||||
- The ``load_strategies_from_config`` helper (default built-in
|
||||
registration, custom ``module:ClassName`` discovery, enabled-list set).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given — Settings shapes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a mock Settings whose context dict contains an empty strategies sub-dict")
|
||||
def step_settings_empty_strategies(context: Context) -> None:
|
||||
context.settings = SimpleNamespace(context={"strategies": {}})
|
||||
|
||||
|
||||
@given("a mock Settings whose context.strategies enables two built-ins")
|
||||
def step_settings_enables_two(context: Context) -> None:
|
||||
context.settings = SimpleNamespace(
|
||||
context={
|
||||
"strategies": {
|
||||
"enabled": ["simple-keyword", "semantic-embedding"],
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@given('a mock Settings whose context.strategies enables only "simple-keyword"')
|
||||
def step_settings_enables_one(context: Context) -> None:
|
||||
context.settings = SimpleNamespace(
|
||||
context={
|
||||
"strategies": {
|
||||
"enabled": ["simple-keyword"],
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@given("a mock Settings without any context attribute")
|
||||
def step_settings_without_context(context: Context) -> None:
|
||||
context.settings = SimpleNamespace()
|
||||
|
||||
|
||||
@given("a mock Settings whose context is a non-dict object")
|
||||
def step_settings_context_non_dict(context: Context) -> None:
|
||||
context.settings = SimpleNamespace(context=SimpleNamespace(strategies={}))
|
||||
|
||||
|
||||
@given(
|
||||
"a mock Settings whose context.strategies references an invalid custom strategy module"
|
||||
)
|
||||
def step_settings_invalid_custom(context: Context) -> None:
|
||||
context.settings = SimpleNamespace(
|
||||
context={
|
||||
"strategies": {
|
||||
"custom": {
|
||||
"broken": "nonexistent_module_for_bdd:NoSuchClass",
|
||||
},
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When — pipeline construction and helper invocations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I construct an ACMSPipeline with that Settings")
|
||||
def step_construct_pipeline(context: Context) -> None:
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
|
||||
try:
|
||||
context.pipeline = ACMSPipeline(settings=context.settings)
|
||||
context.construct_error: Exception | None = None
|
||||
except Exception as exc: # pragma: no cover - exercised via assertion below
|
||||
context.pipeline = None
|
||||
context.construct_error = exc
|
||||
|
||||
|
||||
@when("I call load_strategies_from_config with an empty config dict")
|
||||
def step_load_empty(context: Context) -> None:
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
context.registry = load_strategies_from_config({})
|
||||
|
||||
|
||||
@when(
|
||||
'I call load_strategies_from_config with an enabled list of only "simple-keyword"'
|
||||
)
|
||||
def step_load_enabled_one(context: Context) -> None:
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
context.registry = load_strategies_from_config({"enabled": ["simple-keyword"]})
|
||||
|
||||
|
||||
@when("I call load_strategies_from_config with no enabled key")
|
||||
def step_load_no_enabled(context: Context) -> None:
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
context.registry = load_strategies_from_config({"custom": {}})
|
||||
|
||||
|
||||
@when(
|
||||
'I call load_strategies_from_config with a custom strategy mapping "my-keyword" to SimpleKeywordStrategy'
|
||||
)
|
||||
def step_load_custom(context: Context) -> None:
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
config: dict[str, Any] = {
|
||||
"custom": {
|
||||
"my-keyword": (
|
||||
"cleveragents.application.services.context_strategies"
|
||||
":SimpleKeywordStrategy"
|
||||
),
|
||||
},
|
||||
}
|
||||
context.registry = load_strategies_from_config(config)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then — assertions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("construction should succeed without raising")
|
||||
def step_check_construct_ok(context: Context) -> None:
|
||||
err = getattr(context, "construct_error", None)
|
||||
assert err is None, f"Expected clean construction, got: {err!r}"
|
||||
assert context.pipeline is not None
|
||||
|
||||
|
||||
@then("the resulting pipeline should expose its registered strategies")
|
||||
def step_check_pipeline_has_strategies(context: Context) -> None:
|
||||
assert context.pipeline is not None
|
||||
strategies = list(context.pipeline._strategies.keys())
|
||||
assert strategies, "Pipeline should have registered at least one strategy"
|
||||
|
||||
|
||||
@then('the pipeline strategies should include "simple-keyword"')
|
||||
def step_pipeline_includes_simple_keyword(context: Context) -> None:
|
||||
assert context.pipeline is not None
|
||||
assert "simple-keyword" in context.pipeline._strategies
|
||||
|
||||
|
||||
@then('the pipeline strategies should contain "simple-keyword"')
|
||||
def step_pipeline_contains_simple_keyword(context: Context) -> None:
|
||||
assert context.pipeline is not None
|
||||
assert "simple-keyword" in context.pipeline._strategies
|
||||
|
||||
|
||||
@then('the resulting registry should contain "simple-keyword"')
|
||||
def step_registry_has_sk(context: Context) -> None:
|
||||
assert "simple-keyword" in context.registry.list_all()
|
||||
|
||||
|
||||
@then('the resulting registry should contain "semantic-embedding"')
|
||||
def step_registry_has_se(context: Context) -> None:
|
||||
assert "semantic-embedding" in context.registry.list_all()
|
||||
|
||||
|
||||
@then('the resulting registry should contain "breadth-depth-navigator"')
|
||||
def step_registry_has_bdn(context: Context) -> None:
|
||||
assert "breadth-depth-navigator" in context.registry.list_all()
|
||||
|
||||
|
||||
@then('the resulting registry should contain "my-keyword"')
|
||||
def step_registry_has_custom(context: Context) -> None:
|
||||
assert "my-keyword" in context.registry.list_all()
|
||||
|
||||
|
||||
@then('the resulting registry should list only "simple-keyword" as enabled')
|
||||
def step_registry_enabled_only_sk(context: Context) -> None:
|
||||
enabled = list(context.registry.list_enabled())
|
||||
assert enabled == ["simple-keyword"], f"Expected ['simple-keyword'], got {enabled}"
|
||||
|
||||
|
||||
@then("the resulting registry should list the default-enabled strategies")
|
||||
def step_registry_enabled_default(context: Context) -> None:
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
DEFAULT_ENABLED_STRATEGIES,
|
||||
)
|
||||
|
||||
enabled = set(context.registry.list_enabled())
|
||||
assert enabled == set(DEFAULT_ENABLED_STRATEGIES), (
|
||||
f"Expected {set(DEFAULT_ENABLED_STRATEGIES)}, got {enabled}"
|
||||
)
|
||||
@@ -821,14 +821,6 @@ class ACMSPipeline:
|
||||
# can wire configuration (e.g. default budget, strategy allow-lists)
|
||||
# and persistence without changing the constructor signature.
|
||||
self._settings = settings
|
||||
|
||||
# Auto-load strategy configuration from Settings if provided.
|
||||
# Per spec §42947-42980: when a Settings object is passed to the
|
||||
# pipeline, its TOML-derived ``context`` configuration is used to
|
||||
# populate and enable the StrategyRegistry. This eliminates the need
|
||||
# for callers to manually register every strategy.
|
||||
if settings is not None:
|
||||
self._load_strategies_from_settings(settings)
|
||||
self._unit_of_work = unit_of_work
|
||||
|
||||
self._strategies: dict[str, ContextStrategy] = {
|
||||
@@ -864,6 +856,16 @@ class ACMSPipeline:
|
||||
self._logger = logger.bind(service="acms_pipeline")
|
||||
self._enforcement_result_local = local()
|
||||
|
||||
# Auto-load strategy configuration from Settings if provided.
|
||||
# Per spec §42947-42980: when a Settings object is passed to the
|
||||
# pipeline, its TOML-derived ``context`` configuration is used to
|
||||
# populate and enable the StrategyRegistry. This eliminates the need
|
||||
# for callers to manually register every strategy. Must run AFTER
|
||||
# ``self._strategies`` and ``self._logger`` are initialized — the
|
||||
# method body reads both.
|
||||
if settings is not None:
|
||||
self._load_strategies_from_settings(settings)
|
||||
|
||||
# All 10 pipeline components (default to pass-through stubs)
|
||||
self._strategy_selector = strategy_selector or DefaultStrategySelector()
|
||||
self._budget_allocator = budget_allocator or DefaultBudgetAllocator()
|
||||
|
||||
@@ -1,801 +0,0 @@
|
||||
"""Tests for the ContextStrategyRegistry and plugin loading system.
|
||||
|
||||
Covers:
|
||||
- Strategy registration and retrieval
|
||||
- Protocol conformance validation
|
||||
- Plugin loading from ``module:ClassName`` strings
|
||||
- TOML configuration loading via :func:`load_strategies_from_config`
|
||||
- Thread safety of registry operations
|
||||
- Fallback degradation chain verification
|
||||
- Config-driven strategy enable/disable
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test fixtures and helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FakeStrategy:
|
||||
"""Minimal strategy that satisfies the v1 ``ContextStrategy`` protocol."""
|
||||
|
||||
def __init__(self, name: str = "fake") -> None:
|
||||
self._name = name
|
||||
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities as _Caps,
|
||||
)
|
||||
|
||||
return _Caps(supports_semantic_search=False)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float: # type: ignore[override]
|
||||
return 0.5
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[Any],
|
||||
budget: Any,
|
||||
) -> Sequence[Any]:
|
||||
return list(fragments)
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Fake strategy for testing."
|
||||
|
||||
|
||||
class FakeDomainStrategy:
|
||||
"""Strategy that implements the domain-model ``ContextStrategy`` Protocol."""
|
||||
|
||||
def __init__(self, name: str = "domain-fake") -> None:
|
||||
self._name = name
|
||||
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.domain.models.acms.strategy import StrategyCapabilities
|
||||
|
||||
return StrategyCapabilities(uses_text=True, resource_types=("text",))
|
||||
|
||||
def can_handle(self, request: dict[str, Any], backends: Any) -> float: # type: ignore[override]
|
||||
return 0.8
|
||||
|
||||
def assemble(
|
||||
self, request: Any, backends: Any, budget: int, plan_context: Any
|
||||
) -> list: # type: ignore[override]
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Fake domain strategy for testing."
|
||||
|
||||
|
||||
class FakeStrategyWithNoCaps:
|
||||
"""Strategy with empty capabilities (used to trigger validate_registry warnings)."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "noop-cap-fake"
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.domain.models.acms.strategy import StrategyCapabilities
|
||||
|
||||
return StrategyCapabilities(uses_text=False, resource_types=()) # empty caps
|
||||
|
||||
def can_handle(self, request: dict[str, Any], backends: Any) -> float:
|
||||
return 0.1
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
request: Any,
|
||||
backends: Any,
|
||||
budget: int,
|
||||
plan_context: Any,
|
||||
) -> list:
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Strategy with empty capabilities"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Strategy registration and retrieval
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationAndRetrieval:
|
||||
"""Validate basic register/get/list operations."""
|
||||
|
||||
def test_register_and_get(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("my-strategy")
|
||||
registry.register(strategy)
|
||||
assert registry.get("my-strategy") is strategy
|
||||
assert "my-strategy" in registry
|
||||
|
||||
def test_register_with_custom_name(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("original-name")
|
||||
registry.register(strategy, name="override-name")
|
||||
assert "override-name" in registry
|
||||
assert "original-name" not in registry
|
||||
|
||||
def test_register_with_config(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("config-strategy")
|
||||
cfg = StrategyConfig(enabled=False)
|
||||
registry.register(strategy, config=cfg)
|
||||
|
||||
# With enabled=False the strategy should be registered but not in _enabled_order
|
||||
entry = registry.get_entry("config-strategy")
|
||||
assert entry.config.enabled is False
|
||||
|
||||
def test_register_builtin_flag(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("built-in-1")
|
||||
registry.register(strategy, is_builtin=True)
|
||||
|
||||
assert "built-in-1" in registry.list_builtin()
|
||||
entry = registry.get_entry("built-in-1")
|
||||
assert entry.is_builtin is True
|
||||
|
||||
def test_duplicate_registration_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
s = FakeStrategy("dup")
|
||||
registry.register(s)
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register(FakeStrategy("dup"))
|
||||
|
||||
def test_empty_name_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
class Nameless:
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return ""
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities as _Caps,
|
||||
)
|
||||
|
||||
return _Caps()
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float: # type: ignore[override]
|
||||
return 0.3
|
||||
|
||||
def assemble(self, fragments: Sequence[Any], budget: Any) -> Sequence[Any]:
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return ""
|
||||
|
||||
with pytest.raises(StrategyRegistrationError, match="empty"):
|
||||
registry.register(Nameless())
|
||||
|
||||
def test_protocol_violation_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
class NotStrategy: # No name/abilities etc. at all
|
||||
pass
|
||||
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register(NotStrategy()) # type: ignore[arg-type]
|
||||
|
||||
def test_list_all(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for i in range(5):
|
||||
entry_name = f"strat-{i}"
|
||||
registry.register(FakeStrategy(entry_name))
|
||||
all_names = registry.list_all()
|
||||
assert sorted(all_names) == [
|
||||
"strat-0",
|
||||
"strat-1",
|
||||
"strat-2",
|
||||
"strat-3",
|
||||
"strat-4",
|
||||
]
|
||||
|
||||
def test_list_enabled_empty(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
assert registry.list_enabled() == []
|
||||
|
||||
def test_get_entry_and_config(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
cfg = StrategyConfig(timeout_seconds=60)
|
||||
registry.register(FakeStrategy("cfg-strat"), config=cfg)
|
||||
|
||||
entry = registry.get_entry("cfg-strat")
|
||||
assert entry.name == "cfg-strat"
|
||||
assert entry.config.timeout_seconds == 60
|
||||
|
||||
loaded_cfg = registry.get_config("cfg-strat")
|
||||
assert loaded_cfg.timeout_seconds == 60
|
||||
|
||||
def test_get_raises_on_missing(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyNotFoundError, match="not found"):
|
||||
registry.get("nonexistent")
|
||||
|
||||
def test_len_and_contains(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
assert len(registry) == 0
|
||||
assert "empty" not in registry
|
||||
|
||||
registry.register(FakeStrategy("one"))
|
||||
assert len(registry) == 1
|
||||
assert "one" in registry
|
||||
|
||||
def test_unregister(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("to-unregister"))
|
||||
assert "to-unregister" in registry
|
||||
|
||||
registry.unregister("to-unregister")
|
||||
assert "to-unregister" not in registry
|
||||
|
||||
def test_unregister_unknown_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyNotFoundError):
|
||||
registry.unregister("ghost")
|
||||
|
||||
def test_clear(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for i in range(3):
|
||||
registry.register(FakeStrategy(f"clear-{i}"))
|
||||
assert len(registry) == 3
|
||||
|
||||
registry.clear()
|
||||
assert len(registry) == 0
|
||||
assert registry.list_all() == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Protocol conformance validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProtocolValidation:
|
||||
"""Validate :meth:`StrategyRegistry.validate_registry` behavior."""
|
||||
|
||||
def test_validate_returns_no_warnings_empty(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
warnings = registry.validate_registry()
|
||||
assert warnings == []
|
||||
|
||||
def test_validate_enabled_but_not_registered_warns(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.inject_stale_enabled_entry("ghost-strategy")
|
||||
warnings = registry.validate_registry()
|
||||
assert any("ghost-strategy" in w for w in warnings)
|
||||
|
||||
def test_validate_domain_strategy_no_resource_types_warns(self):
|
||||
"""Domain-model strategies must declare resource_types."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strat = FakeStrategyWithNoCaps() # has domain caps with empty resource_types
|
||||
registry.register(strat)
|
||||
warnings = registry.validate_registry()
|
||||
assert any("resource types" in w for w in warnings)
|
||||
|
||||
def test_validate_domain_strategy_with_resource_types_ok(self):
|
||||
"""Domain strategies that declare resource_types pass validation."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strat = FakeDomainStrategy("with-caps") # has resource_types=("text",)
|
||||
registry.register(strat)
|
||||
|
||||
warnings = registry.validate_registry()
|
||||
assert not any("resource types" in w for w in warnings)
|
||||
|
||||
def test_validate_v1_pipeline_strategies_skip_resource_types(self):
|
||||
"""V1 pipeline strategies (with acms_service StrategyCapabilities) should NOT
|
||||
get resource-types warnings because their capabilities object has no
|
||||
such field. This was the fix introduced by #10590."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
# SimpleKeywordStrategy uses v1 pipeline StrategyCapabilities (acms_service)
|
||||
# which lacks the domain-model resource_types field.
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
|
||||
registry.register(SimpleKeywordStrategy(), is_builtin=True)
|
||||
warnings = registry.validate_registry()
|
||||
|
||||
# Should NOT have a "resource types" warning for v1 pipeline strategies
|
||||
resource_warnings = [w for w in warnings if "resource types" in w]
|
||||
assert len(resource_warnings) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Plugin loading from module:ClassName strings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPluginLoading:
|
||||
"""Test :meth:`StrategyRegistry.register_from_module`."""
|
||||
|
||||
def test_register_from_module_success(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register_from_module(
|
||||
name="simple-keyword",
|
||||
module_path="cleveragents.application.services.context_strategies:SimpleKeywordStrategy",
|
||||
config=StrategyConfig(enabled=True),
|
||||
)
|
||||
assert "simple-keyword" in registry
|
||||
strategy = registry.get("simple-keyword")
|
||||
assert strategy.name == "simple-keyword"
|
||||
|
||||
def test_register_from_module_invalid_format_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError, match="module:ClassName"):
|
||||
registry.register_from_module(name="bad", module_path="invalid_format") # type: ignore[arg-type]
|
||||
|
||||
def test_register_from_module_unauthorized_prefix_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register_from_module(
|
||||
name="evil",
|
||||
module_path="os.path:join", # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
def test_register_from_module_nonexistent_class_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError, match="not found"):
|
||||
registry.register_from_module(
|
||||
name="bogus",
|
||||
module_path="cleveragents.application.services.context_strategies:"
|
||||
"NoSuchClassDoesNotExist", # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: TOML configuration loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTOMLConfigurationLoading:
|
||||
"""Test :func:`load_strategies_from_config`.
|
||||
|
||||
This exercises the entire pipeline: register built-ins, optionally load
|
||||
custom plugins from module:ClassName strings, and set the enabled list.
|
||||
"""
|
||||
|
||||
def test_load_empty_config_registers_defaults(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config({}) # {} = minimal config
|
||||
names = registry.list_all()
|
||||
assert "simple-keyword" in names
|
||||
assert "semantic-embedding" in names
|
||||
assert "breadth-depth-navigator" in names
|
||||
|
||||
# Only defaults should be enabled
|
||||
enabled = registry.list_enabled()
|
||||
assert set(enabled) == {
|
||||
"simple-keyword",
|
||||
"semantic-embedding",
|
||||
"breadth-depth-navigator",
|
||||
}
|
||||
|
||||
def test_load_config_with_custom_plugin(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config(
|
||||
{
|
||||
"custom": {
|
||||
"my-strategy": (
|
||||
"cleveragents.application.services.context_strategies:"
|
||||
"SemanticEmbeddingStrategy"
|
||||
),
|
||||
},
|
||||
}
|
||||
)
|
||||
assert "my-strategy" in registry
|
||||
assert "simple-keyword" in registry
|
||||
|
||||
def test_load_config_with_custom_and_custom_enabled(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config(
|
||||
{
|
||||
"enabled": ["simple-keyword", "my-strategy"],
|
||||
"custom": {
|
||||
"my-strategy": (
|
||||
"cleveragents.application.services.context_strategies:"
|
||||
"SemanticEmbeddingStrategy"
|
||||
), # type: ignore[dict-item]
|
||||
},
|
||||
}
|
||||
)
|
||||
enabled = registry.list_enabled()
|
||||
assert set(enabled) == {"simple-keyword", "my-strategy"}
|
||||
|
||||
def test_load_config_all_builtins_registered_as_builtin(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config({})
|
||||
for name in ["simple-keyword", "semantic-embedding", "breadth-depth-navigator"]:
|
||||
entry = registry.get_entry(name)
|
||||
assert entry.is_builtin is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Thread safety of registry operations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestThreadSafety:
|
||||
"""Verify that concurrent register/get/set_enabled calls don't corrupt state."""
|
||||
|
||||
def test_concurrent_register(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
errors: list[Exception] = [] # type: ignore[var-annotated]
|
||||
|
||||
def register_worker(worker_id: int) -> None:
|
||||
try:
|
||||
for _ in range(50):
|
||||
name = f"concurrent-strat-{worker_id}"
|
||||
registry.register(FakeStrategy(name))
|
||||
except Exception as exc:
|
||||
errors.append(exc)
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=register_worker, args=(i,)) for i in range(10)
|
||||
]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors, f"Errors during concurrent register: {errors}"
|
||||
# Each worker registers once (deduplicated within its own run)
|
||||
total = len(registry)
|
||||
assert total > 0
|
||||
|
||||
def test_concurrent_get_and_register(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("shared")) # Pre-register one strategy
|
||||
errors: list[Exception] = [] # type: ignore[var-annotated]
|
||||
reads: list[str] = [] # type: ignore[var-annotated]
|
||||
|
||||
def reader(count: int) -> None:
|
||||
for _ in range(20):
|
||||
try:
|
||||
s = registry.get("shared")
|
||||
assert s is not None
|
||||
reads.append(s.name)
|
||||
except Exception as exc:
|
||||
errors.append(exc)
|
||||
|
||||
def writer(worker_id: int) -> None:
|
||||
for _ in range(20):
|
||||
name = f"writer-{worker_id}"
|
||||
registry.register(FakeStrategy(name))
|
||||
|
||||
threads: list[threading.Thread] = []
|
||||
threads.append(threading.Thread(target=reader, args=(1,)))
|
||||
for i in range(5):
|
||||
threads.append(threading.Thread(target=writer, args=(i,)))
|
||||
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors
|
||||
assert len(reads) == 20
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Fallback degradation chain verification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFallbackDegradation:
|
||||
"""Verify that SimpleKeywordStrategy is always available as a fallback."""
|
||||
|
||||
def test_simple_keyword_always_produces_results(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import ContextBudget
|
||||
|
||||
strat = SimpleKeywordStrategy()
|
||||
budget = ContextBudget(max_tokens=1000)
|
||||
|
||||
# Even with empty fragments, should return an empty list (no crash)
|
||||
result = strat.assemble([], budget)
|
||||
assert result == []
|
||||
|
||||
# Create a fake fragment-like object
|
||||
class FakeFrag: # pylint: disable=C0115
|
||||
relevance_score: float = 0.5
|
||||
token_count: int = 10
|
||||
uko_node: str = "project://test/node"
|
||||
content: str = "hello world test fragment"
|
||||
|
||||
fragments = [FakeFrag()] # type: ignore[list-item]
|
||||
result = strat.assemble(fragments, budget)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_simple_keyword_confidence_is_universal(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
|
||||
strat = SimpleKeywordStrategy()
|
||||
# Without query in request
|
||||
assert strat.can_handle({}) == 0.3
|
||||
# Request is always handled with at least 0.3 confidence
|
||||
assert strat.can_handle({"query": ""}) == 0.3
|
||||
assert strat.can_handle({"anything": "irrelevant"}) == 0.3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Config-driven strategy enable/disable
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEnableDisable:
|
||||
"""Test :meth:`StrategyRegistry.set_enabled` and config updates."""
|
||||
|
||||
def test_set_enabled_updates_list(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for name in ["a", "b", "c"]:
|
||||
registry.register(FakeStrategy(name), config=StrategyConfig(enabled=True))
|
||||
|
||||
# Only enable 'a' and 'c'
|
||||
registry.set_enabled(["a", "c"])
|
||||
enabled = registry.list_enabled()
|
||||
assert enabled == ["a", "c"]
|
||||
|
||||
def test_set_unknown_strategy_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("exists"))
|
||||
with pytest.raises(StrategyNotFoundError, match="unknown"):
|
||||
registry.set_enabled(["nonexistent"])
|
||||
|
||||
def test_set_enabled_updates_config_flags(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for name in ["x", "y"]:
|
||||
registry.register(FakeStrategy(name))
|
||||
|
||||
# Initially enabled by default (enable_by_default=True)
|
||||
assert registry.get_config("x").enabled is True
|
||||
|
||||
registry.set_enabled(["x"]) # disable y only
|
||||
|
||||
# Check config entries: x stays enabled, y should be disabled
|
||||
assert registry.get_config("x").enabled is True
|
||||
assert registry.get_config("y").enabled is False
|
||||
|
||||
def test_update_config_individual_fields(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("updatable"))
|
||||
|
||||
registry.update_config(
|
||||
"updatable",
|
||||
enabled=False,
|
||||
timeout_seconds=120,
|
||||
max_fragments=50,
|
||||
)
|
||||
|
||||
cfg = registry.get_config("updatable")
|
||||
assert cfg.enabled is False
|
||||
assert cfg.timeout_seconds == 120
|
||||
assert cfg.max_fragments == 50
|
||||
|
||||
def test_update_config_enable_re_adds_to_enabled_order(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("re-enable"))
|
||||
registry.update_config("re-enable", enabled=True)
|
||||
|
||||
assert "re-enable" in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
def test_update_config_disable_removes_from_enabled_order(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("disable-me"))
|
||||
assert "disable-me" in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
registry.update_config("disable-me", enabled=False)
|
||||
assert "disable-me" not in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Integration with acms_service auto-loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestACMSPipelineAutoLoading:
|
||||
"""Verify that ACMSPipeline.__init__ loads strategies from Settings."""
|
||||
|
||||
def test_pipeline_auto_loads_strategies_from_settings(self):
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
# Create a mock Settings object with context.strategies config.
|
||||
settings = MagicMock()
|
||||
settings.context = { # type: ignore[attr-defined]
|
||||
"strategies": {
|
||||
"enabled": ["simple-keyword", "semantic-embedding"],
|
||||
}
|
||||
}
|
||||
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
|
||||
pipeline = ACMSPipeline(settings=settings)
|
||||
strategies_in_pipeline = list(pipeline._strategies.keys()) # pylint: disable=protected-access
|
||||
|
||||
# Built-in relevance strategy is always registered
|
||||
assert "relevance" in strategies_in_pipeline
|
||||
# Simple-keyword should be auto-loaded
|
||||
assert "simple-keyword" in strategies_in_pipeline
|
||||
|
||||
def test_pipeline_works_without_settings(self):
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
|
||||
pipeline = ACMSPipeline()
|
||||
assert "relevance" in pipeline._strategies # pylint: disable=protected-access
|
||||
Reference in New Issue
Block a user