From ffa0a3b3bfefbd03595b5ac09d3caa2bca29d03c Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 18 Jun 2026 00:56:39 -0400 Subject: [PATCH] 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 --- features/acms_strategy_auto_loading.feature | 80 ++ .../steps/acms_strategy_auto_loading_steps.py | 210 +++++ .../application/services/acms_service.py | 18 +- tests/strategies/__init__.py | 0 tests/strategies/test_strategy_registry.py | 801 ------------------ 5 files changed, 300 insertions(+), 809 deletions(-) create mode 100644 features/acms_strategy_auto_loading.feature create mode 100644 features/steps/acms_strategy_auto_loading_steps.py delete mode 100644 tests/strategies/__init__.py delete mode 100644 tests/strategies/test_strategy_registry.py diff --git a/features/acms_strategy_auto_loading.feature b/features/acms_strategy_auto_loading.feature new file mode 100644 index 000000000..55af1429e --- /dev/null +++ b/features/acms_strategy_auto_loading.feature @@ -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" diff --git a/features/steps/acms_strategy_auto_loading_steps.py b/features/steps/acms_strategy_auto_loading_steps.py new file mode 100644 index 000000000..cf197be52 --- /dev/null +++ b/features/steps/acms_strategy_auto_loading_steps.py @@ -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}" + ) diff --git a/src/cleveragents/application/services/acms_service.py b/src/cleveragents/application/services/acms_service.py index 54bbfbaf7..1cc056a2b 100644 --- a/src/cleveragents/application/services/acms_service.py +++ b/src/cleveragents/application/services/acms_service.py @@ -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() diff --git a/tests/strategies/__init__.py b/tests/strategies/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/strategies/test_strategy_registry.py b/tests/strategies/test_strategy_registry.py deleted file mode 100644 index 1c9d47b6c..000000000 --- a/tests/strategies/test_strategy_registry.py +++ /dev/null @@ -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