HAL9000
c4ffc7facc
- Add features/scope_chain_resolver.feature with 15 scenarios covering ScopeResolutionContext construction, ScopeResolverRegistry register/ unregister/resolve/introspection, and all _discover_resolvers paths (load failure, outer exception, dict-style fallback) - Add features/steps/scope_chain_resolver_steps.py with mock-based helpers to exercise the entry-point discovery branches ISSUES CLOSED: #8867
256 lines
9.1 KiB
Python
256 lines
9.1 KiB
Python
"""Step definitions for features/scope_chain_resolver.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.metadata as _importlib_metadata
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.contexts.scope_chain_resolver import (
|
|
ScopeResolutionContext,
|
|
ScopeResolverRegistry,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _StaticResolver:
|
|
"""Test resolver that returns a fixed fragment list for one matching scope."""
|
|
|
|
def __init__(self, target_scope: str, fragments: list[str]) -> None:
|
|
self._target = target_scope
|
|
self._fragments = fragments
|
|
|
|
def resolve(self, scope: str, context: ScopeResolutionContext) -> list[str]:
|
|
if scope == self._target:
|
|
return list(self._fragments)
|
|
return []
|
|
|
|
|
|
class _NullResolver:
|
|
"""Test resolver that always returns an empty list."""
|
|
|
|
def resolve(self, scope: str, context: ScopeResolutionContext) -> list[str]:
|
|
return []
|
|
|
|
|
|
def _make_resolution_ctx(scope: str = "test:0") -> ScopeResolutionContext:
|
|
return ScopeResolutionContext(scope=scope)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh ScopeResolverRegistry")
|
|
def step_fresh_registry(context: Context) -> None:
|
|
context.registry = ScopeResolverRegistry()
|
|
|
|
|
|
@given("a ScopeResolverRegistry with a failing entry point")
|
|
def step_registry_failing_ep(context: Context) -> None:
|
|
bad_ep = MagicMock()
|
|
bad_ep.name = "bad"
|
|
bad_ep.load.side_effect = RuntimeError("simulated load failure")
|
|
|
|
mock_eps = MagicMock()
|
|
mock_eps.select.return_value = [bad_ep]
|
|
|
|
with patch.object(_importlib_metadata, "entry_points", return_value=mock_eps):
|
|
context.registry = ScopeResolverRegistry()
|
|
|
|
|
|
@given("a ScopeResolverRegistry where entry_points raises an exception")
|
|
def step_registry_ep_raises(context: Context) -> None:
|
|
with patch.object(
|
|
_importlib_metadata,
|
|
"entry_points",
|
|
side_effect=RuntimeError("simulated entry_points failure"),
|
|
):
|
|
context.registry = ScopeResolverRegistry()
|
|
|
|
|
|
@given("a ScopeResolverRegistry with dict-style entry points returning no scope resolvers")
|
|
def step_registry_dict_eps(context: Context) -> None:
|
|
# Simulate older importlib_metadata API: entry_points() returns a plain dict
|
|
# without a 'select' method, triggering the else-branch fallback.
|
|
mock_dict: dict[str, list[Any]] = {"other.group": []}
|
|
|
|
with patch.object(_importlib_metadata, "entry_points", return_value=mock_dict):
|
|
context.registry = ScopeResolverRegistry()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a ScopeResolutionContext with scope "{scope}"')
|
|
def step_create_ctx_minimal(context: Context, scope: str) -> None:
|
|
context.resolution_ctx = ScopeResolutionContext(scope=scope)
|
|
|
|
|
|
@when(
|
|
'I create a full ScopeResolutionContext with scope "{scope}"'
|
|
' and metadata key "{key}" value "{value}"'
|
|
' and resolved_fragment "{frag}"'
|
|
)
|
|
def step_create_ctx_full(
|
|
context: Context, scope: str, key: str, value: str, frag: str
|
|
) -> None:
|
|
context.resolution_ctx = ScopeResolutionContext(
|
|
scope=scope,
|
|
metadata={key: value},
|
|
resolved_fragments=[frag],
|
|
)
|
|
|
|
|
|
@when(
|
|
'I register a resolver named "{name}" with priority {priority:d}'
|
|
' that returns "{fragment}" for scope "{target_scope}"'
|
|
)
|
|
def step_register_static_resolver(
|
|
context: Context, name: str, priority: int, fragment: str, target_scope: str
|
|
) -> None:
|
|
resolver = _StaticResolver(target_scope, [fragment])
|
|
context.registry.register(name, resolver, priority)
|
|
|
|
|
|
@when('I register a null resolver named "{name}" with priority {priority:d}')
|
|
def step_register_null_resolver(context: Context, name: str, priority: int) -> None:
|
|
context.registry.register(name, _NullResolver(), priority)
|
|
|
|
|
|
@when('I resolve scope "{scope}" with the registry')
|
|
def step_resolve_scope(context: Context, scope: str) -> None:
|
|
ctx = _make_resolution_ctx(scope)
|
|
context.resolution_result = context.registry.resolve(scope, ctx)
|
|
|
|
|
|
@when('I unregister the resolver named "{name}"')
|
|
def step_unregister(context: Context, name: str) -> None:
|
|
context.registry.unregister(name)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the ScopeResolutionContext scope should be "{expected}"')
|
|
def step_check_ctx_scope(context: Context, expected: str) -> None:
|
|
assert context.resolution_ctx.scope == expected, (
|
|
f"Expected scope {expected!r}, got {context.resolution_ctx.scope!r}"
|
|
)
|
|
|
|
|
|
@then("the ScopeResolutionContext metadata should be empty")
|
|
def step_check_ctx_metadata_empty(context: Context) -> None:
|
|
assert context.resolution_ctx.metadata == {}
|
|
|
|
|
|
@then("the ScopeResolutionContext resolved_fragments should be empty")
|
|
def step_check_ctx_frags_empty(context: Context) -> None:
|
|
assert context.resolution_ctx.resolved_fragments == []
|
|
|
|
|
|
@then('the ScopeResolutionContext metadata key "{key}" should be "{expected}"')
|
|
def step_check_ctx_metadata_key(context: Context, key: str, expected: str) -> None:
|
|
assert context.resolution_ctx.metadata.get(key) == expected, (
|
|
f"Expected metadata[{key!r}]={expected!r}, got {context.resolution_ctx.metadata}"
|
|
)
|
|
|
|
|
|
@then('the ScopeResolutionContext resolved_fragments should contain "{frag}"')
|
|
def step_check_ctx_frags_contain(context: Context, frag: str) -> None:
|
|
assert frag in context.resolution_ctx.resolved_fragments, (
|
|
f"Expected {frag!r} in resolved_fragments {context.resolution_ctx.resolved_fragments}"
|
|
)
|
|
|
|
|
|
@then("the resolution result should be empty")
|
|
def step_result_empty(context: Context) -> None:
|
|
assert context.resolution_result == [], (
|
|
f"Expected empty result, got {context.resolution_result}"
|
|
)
|
|
|
|
|
|
@then('the resolution result should contain "{frag}"')
|
|
def step_result_contains(context: Context, frag: str) -> None:
|
|
assert frag in context.resolution_result, (
|
|
f"Expected {frag!r} in result {context.resolution_result}"
|
|
)
|
|
|
|
|
|
@then('the resolution result should not contain "{frag}"')
|
|
def step_result_not_contains(context: Context, frag: str) -> None:
|
|
assert frag not in context.resolution_result, (
|
|
f"Did not expect {frag!r} in result {context.resolution_result}"
|
|
)
|
|
|
|
|
|
@then('the registry should contain a resolver named "{name}"')
|
|
def step_registry_has_resolver(context: Context, name: str) -> None:
|
|
assert name in context.registry.get_resolvers(), (
|
|
f"Expected resolver {name!r} in registry, got {list(context.registry.get_resolvers())}"
|
|
)
|
|
|
|
|
|
@then('the registry should not contain a resolver named "{name}"')
|
|
def step_registry_no_resolver(context: Context, name: str) -> None:
|
|
assert name not in context.registry.get_resolvers(), (
|
|
f"Did not expect resolver {name!r} in registry"
|
|
)
|
|
|
|
|
|
@then('the resolver named "{name}" should have priority {expected:d}')
|
|
def step_resolver_priority(context: Context, name: str, expected: int) -> None:
|
|
resolvers = context.registry.get_resolvers()
|
|
assert name in resolvers, f"Resolver {name!r} not found"
|
|
_, priority = resolvers[name]
|
|
assert priority == expected, f"Expected priority {expected}, got {priority}"
|
|
|
|
|
|
@then("get_resolvers should return {count:d} entries")
|
|
def step_get_resolvers_count(context: Context, count: int) -> None:
|
|
actual = len(context.registry.get_resolvers())
|
|
assert actual == count, f"Expected {count} resolvers, got {actual}"
|
|
|
|
|
|
@then('get_resolvers should include key "{key}"')
|
|
def step_get_resolvers_has_key(context: Context, key: str) -> None:
|
|
assert key in context.registry.get_resolvers(), (
|
|
f"Expected key {key!r} in get_resolvers()"
|
|
)
|
|
|
|
|
|
@then("list_resolvers should return {count:d} entries in priority order")
|
|
def step_list_resolvers_count_and_order(context: Context, count: int) -> None:
|
|
result = context.registry.list_resolvers()
|
|
assert len(result) == count, f"Expected {count} entries, got {len(result)}: {result}"
|
|
if context.table:
|
|
for i, row in enumerate(context.table):
|
|
expected_name = row["name"]
|
|
expected_priority = int(row["priority"])
|
|
actual_name, actual_priority = result[i]
|
|
assert actual_name == expected_name, (
|
|
f"Index {i}: expected name {expected_name!r}, got {actual_name!r}"
|
|
)
|
|
assert actual_priority == expected_priority, (
|
|
f"Index {i}: expected priority {expected_priority}, got {actual_priority}"
|
|
)
|
|
|
|
|
|
@then("the scope resolver registry should be empty")
|
|
def step_registry_empty(context: Context) -> None:
|
|
assert context.registry.get_resolvers() == {}, (
|
|
f"Expected empty registry, got {list(context.registry.get_resolvers())}"
|
|
)
|