forked from cleveragents/cleveragents-core
freemo
1d15eca866
Add the Depth/Breadth Projection System and Skeleton Context Propagation as specified in docs/specification.md §25265-25340 and §43057-43128: - ProjectionSpec: frozen Pydantic model capturing a projection request (focus, breadth, depth, gradient, domain) - ProjectedNode: frozen model for materialized graph nodes with resolved depth and distance - DepthBreadthProjector: stateless BFS projector over UKO graph adjacency with depth gradient (linear reduction by distance) - PlanContextInheritance: service computing child plan context from parent assembled context with skeleton injection - ChildContextResult: frozen result model with request and skeleton - InheritanceConfig: frozen config for skeleton_ratio (default 0.2) - Built-in DetailLevelMap presets for code, docs, and database Includes 27 Behave BDD scenarios, 9 Robot Framework integration tests, and ASV benchmarks for all components. ISSUES CLOSED: #544
649 lines
22 KiB
Python
649 lines
22 KiB
Python
"""Step definitions for features/depth_breadth_projection.feature.
|
|
|
|
Tests the Depth/Breadth Projection System and Skeleton Context Propagation:
|
|
ProjectionSpec, ProjectedNode, DepthBreadthProjector, PlanContextInheritance,
|
|
ChildContextResult, InheritanceConfig, and built-in DetailLevelMap presets.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.application.services.depth_breadth_projection import (
|
|
ChildContextResult,
|
|
DepthBreadthProjector,
|
|
InheritanceConfig,
|
|
PlanContextInheritance,
|
|
ProjectedNode,
|
|
ProjectionSpec,
|
|
code_detail_map,
|
|
database_detail_map,
|
|
docs_detail_map,
|
|
)
|
|
from cleveragents.domain.models.acms.crp import (
|
|
ContextFragment,
|
|
ContextRequest,
|
|
FragmentProvenance,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PROV = FragmentProvenance(resource_uri="test://projection")
|
|
|
|
|
|
def _make_frag(
|
|
uko_node: str = "test://node",
|
|
content: str = "test",
|
|
token_count: int = 100,
|
|
relevance_score: float = 0.8,
|
|
detail_depth: int = 3,
|
|
) -> ContextFragment:
|
|
return ContextFragment(
|
|
uko_node=uko_node,
|
|
content=content,
|
|
token_count=token_count,
|
|
relevance_score=relevance_score,
|
|
detail_depth=detail_depth,
|
|
provenance=_PROV,
|
|
)
|
|
|
|
|
|
class _FakeAssembledContext:
|
|
"""Minimal stand-in for AssembledContext."""
|
|
|
|
def __init__(
|
|
self,
|
|
fragments: list[ContextFragment],
|
|
context_hash: str = "abc123",
|
|
) -> None:
|
|
self.fragments = fragments
|
|
self.context_hash = context_hash
|
|
|
|
|
|
class _MockSkeletonCompressor:
|
|
"""Compressor that returns a fixed number of fragments."""
|
|
|
|
def __init__(self, n_fragments: int = 2) -> None:
|
|
self._n = n_fragments
|
|
|
|
def compress(
|
|
self,
|
|
fragments: tuple[ContextFragment, ...],
|
|
skeleton_budget: int,
|
|
) -> tuple[ContextFragment, ...]:
|
|
return tuple(fragments[: self._n])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the depth/breadth projection modules are available")
|
|
def step_modules_available(context: Context) -> None:
|
|
pass
|
|
|
|
|
|
@given("a UKO graph with nodes A B C and edges A->B B->C")
|
|
def step_graph_abc(context: Context) -> None:
|
|
context.adjacency = {
|
|
"A": ["B"],
|
|
"B": ["C"],
|
|
}
|
|
|
|
|
|
@given("a UKO graph with nodes A B and edges A->B")
|
|
def step_graph_ab(context: Context) -> None:
|
|
context.adjacency = {
|
|
"A": ["B"],
|
|
}
|
|
|
|
|
|
@given("a UKO graph with nodes A and no edges")
|
|
def step_graph_a_only(context: Context) -> None:
|
|
adj: dict[str, list[str]] = {}
|
|
context.adjacency = adj
|
|
|
|
|
|
@given("a UKO graph with nodes A B C D and edges A->B C->D")
|
|
def step_graph_abcd(context: Context) -> None:
|
|
context.adjacency = {
|
|
"A": ["B"],
|
|
"C": ["D"],
|
|
}
|
|
|
|
|
|
@given('a code DetailLevelMap is registered for domain "{domain}"')
|
|
def step_register_code_map(context: Context, domain: str) -> None:
|
|
projector = DepthBreadthProjector()
|
|
projector.register_detail_map(domain, code_detail_map())
|
|
context.projector = projector
|
|
|
|
|
|
@given("a parent assembled context with {n:d} fragments at detail_depth {depth:d}")
|
|
def step_parent_context(context: Context, n: int, depth: int) -> None:
|
|
frags = [
|
|
_make_frag(
|
|
uko_node=f"module://mod_{i}",
|
|
content=f"module_{i}",
|
|
token_count=100,
|
|
detail_depth=depth,
|
|
)
|
|
for i in range(n)
|
|
]
|
|
context.parent_context = _FakeAssembledContext(frags, context_hash="parenthash123")
|
|
|
|
|
|
@given("a PlanContextInheritance service with default config")
|
|
def step_inheritance_default(context: Context) -> None:
|
|
context.inheritance_service = PlanContextInheritance()
|
|
|
|
|
|
@given("a mock skeleton compressor that returns {n:d} fragments")
|
|
def step_mock_compressor(context: Context, n: int) -> None:
|
|
context.mock_compressor = _MockSkeletonCompressor(n_fragments=n)
|
|
|
|
|
|
@given("a PlanContextInheritance service with the mock compressor")
|
|
def step_inheritance_with_compressor(context: Context) -> None:
|
|
context.inheritance_service = PlanContextInheritance(
|
|
skeleton_compressor=context.mock_compressor,
|
|
)
|
|
|
|
|
|
@given(
|
|
"a PlanContextInheritance service with the mock compressor and min_skeleton_tokens {min_tok:d}"
|
|
)
|
|
def step_inheritance_with_min_tokens(context: Context, min_tok: int) -> None:
|
|
config = InheritanceConfig(min_skeleton_tokens=min_tok)
|
|
context.inheritance_service = PlanContextInheritance(
|
|
config=config,
|
|
skeleton_compressor=context.mock_compressor,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — ProjectionSpec
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I create a ProjectionSpec with focus "{focus}" and breadth {breadth:d} and depth {depth:d}'
|
|
)
|
|
def step_create_spec(context: Context, focus: str, breadth: int, depth: int) -> None:
|
|
context.proj_spec = ProjectionSpec(
|
|
focus=(focus,),
|
|
breadth=breadth,
|
|
depth=depth,
|
|
)
|
|
|
|
|
|
@when("I create a ProjectionSpec with empty focus")
|
|
def step_create_spec_empty_focus(context: Context) -> None:
|
|
context.proj_error = None
|
|
try:
|
|
context.proj_spec = ProjectionSpec(focus=(), breadth=2, depth=3)
|
|
except ValidationError as exc:
|
|
context.proj_error = exc
|
|
|
|
|
|
@when("I create a ProjectionSpec with negative depth")
|
|
def step_create_spec_negative_depth(context: Context) -> None:
|
|
context.proj_error = None
|
|
try:
|
|
context.proj_spec = ProjectionSpec(focus=("A",), breadth=2, depth=-1)
|
|
except (ValueError, ValidationError) as exc:
|
|
context.proj_error = exc
|
|
|
|
|
|
@when('I create a ProjectionSpec with focus "{focus}" and named depth "{depth}"')
|
|
def step_create_spec_named(context: Context, focus: str, depth: str) -> None:
|
|
context.proj_spec = ProjectionSpec(
|
|
focus=(focus,),
|
|
depth=depth,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — ProjectedNode
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I create a ProjectedNode with uri "{uri}" and distance {dist:d} and resolved_depth {depth:d}'
|
|
)
|
|
def step_create_projected_node(
|
|
context: Context, uri: str, dist: int, depth: int
|
|
) -> None:
|
|
context.proj_node = ProjectedNode(uri=uri, distance=dist, resolved_depth=depth)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — DepthBreadthProjector
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I project with focus "{focus}" and breadth {breadth:d} and depth {depth:d}')
|
|
def step_project(context: Context, focus: str, breadth: int, depth: int) -> None:
|
|
projector = getattr(context, "projector", None) or DepthBreadthProjector()
|
|
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
|
|
context.proj_result = projector.project(spec, context.adjacency)
|
|
|
|
|
|
@when(
|
|
'I project with focus "{focus}" and breadth {breadth:d} and depth {depth:d} and gradient {gradient}'
|
|
)
|
|
def step_project_gradient(
|
|
context: Context, focus: str, breadth: int, depth: int, gradient: str
|
|
) -> None:
|
|
grad = gradient.strip().lower() == "true"
|
|
projector = DepthBreadthProjector()
|
|
spec = ProjectionSpec(
|
|
focus=(focus,), breadth=breadth, depth=depth, depth_gradient=grad
|
|
)
|
|
context.proj_result = projector.project(spec, context.adjacency)
|
|
|
|
|
|
@when(
|
|
'I project with focus "{focus}" and breadth {breadth:d} and named depth "{depth}" in domain "{domain}"'
|
|
)
|
|
def step_project_named(
|
|
context: Context, focus: str, breadth: int, depth: str, domain: str
|
|
) -> None:
|
|
projector = getattr(context, "projector", None) or DepthBreadthProjector()
|
|
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth, domain=domain)
|
|
context.proj_error = None
|
|
try:
|
|
context.proj_result = projector.project(spec, context.adjacency)
|
|
except ValueError as exc:
|
|
context.proj_error = exc
|
|
|
|
|
|
@when(
|
|
'I project with multi-focus "{f1}" and "{f2}" and breadth {breadth:d} and depth {depth:d}'
|
|
)
|
|
def step_project_multi_focus(
|
|
context: Context, f1: str, f2: str, breadth: int, depth: int
|
|
) -> None:
|
|
projector = DepthBreadthProjector()
|
|
spec = ProjectionSpec(focus=(f1, f2), breadth=breadth, depth=depth)
|
|
context.proj_result = projector.project(spec, context.adjacency)
|
|
|
|
|
|
@when(
|
|
'I project to fragments with focus "{focus}" and breadth {breadth:d} and depth {depth:d}'
|
|
)
|
|
def step_project_fragments(
|
|
context: Context, focus: str, breadth: int, depth: int
|
|
) -> None:
|
|
projector = DepthBreadthProjector()
|
|
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
|
|
context.proj_fragments = projector.project_to_fragments(spec, context.adjacency)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — DetailLevelMap presets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create the code detail level map preset")
|
|
def step_code_map(context: Context) -> None:
|
|
context.code_map = code_detail_map()
|
|
|
|
|
|
@when("I create the docs detail level map preset")
|
|
def step_docs_map(context: Context) -> None:
|
|
context.docs_map = docs_detail_map()
|
|
|
|
|
|
@when("I create the database detail level map preset")
|
|
def step_db_map(context: Context) -> None:
|
|
context.db_map = database_detail_map()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — PlanContextInheritance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I compute child context with focus "{focus}" and budget {budget:d}')
|
|
def step_compute_child(context: Context, focus: str, budget: int) -> None:
|
|
context.child_error = None
|
|
try:
|
|
context.child_result = context.inheritance_service.compute_child_context(
|
|
parent_context=context.parent_context,
|
|
child_focus=[focus],
|
|
child_token_budget=budget,
|
|
)
|
|
except ValueError as exc:
|
|
context.child_error = exc
|
|
|
|
|
|
@when(
|
|
'I compute child context with focus "{focus}" and budget {budget:d} at depth_in_tree {pd:d} to {cd:d}'
|
|
)
|
|
def step_compute_child_depths(
|
|
context: Context, focus: str, budget: int, pd: int, cd: int
|
|
) -> None:
|
|
context.child_result = context.inheritance_service.compute_child_context(
|
|
parent_context=context.parent_context,
|
|
child_focus=[focus],
|
|
child_token_budget=budget,
|
|
parent_depth_in_tree=pd,
|
|
child_depth_in_tree=cd,
|
|
)
|
|
|
|
|
|
@when("I compute child context with parent_depth {pd:d} and child_depth {cd:d}")
|
|
def step_compute_child_invalid_depth(context: Context, pd: int, cd: int) -> None:
|
|
context.child_error = None
|
|
try:
|
|
context.child_result = context.inheritance_service.compute_child_context(
|
|
parent_context=context.parent_context,
|
|
child_focus=["class://X"],
|
|
child_token_budget=1000,
|
|
parent_depth_in_tree=pd,
|
|
child_depth_in_tree=cd,
|
|
)
|
|
except ValueError as exc:
|
|
context.child_error = exc
|
|
|
|
|
|
@when("I create an InheritanceConfig with defaults")
|
|
def step_default_config(context: Context) -> None:
|
|
context.inheritance_config = InheritanceConfig()
|
|
|
|
|
|
@when('I extract child focus from decisions "{d1}" and "{d2}"')
|
|
def step_extract_focus(context: Context, d1: str, d2: str) -> None:
|
|
context.child_focus = PlanContextInheritance.extract_child_focus([d1, d2])
|
|
|
|
|
|
@when('I extract child focus from empty decisions with fallback "{fallback}"')
|
|
def step_extract_focus_fallback(context: Context, fallback: str) -> None:
|
|
context.child_focus = PlanContextInheritance.extract_child_focus(
|
|
[], fallback_focus=[fallback]
|
|
)
|
|
|
|
|
|
@when("I create a ChildContextResult with a request and skeleton data")
|
|
def step_create_child_result(context: Context) -> None:
|
|
req = ContextRequest(focus=["class://Test"], breadth=1, depth=5)
|
|
frag = _make_frag(uko_node="test://skel", content="skeleton", token_count=50)
|
|
context.child_ctx_result = ChildContextResult(
|
|
request=req,
|
|
skeleton_fragments=(frag,),
|
|
parent_context_hash="hash123",
|
|
skeleton_budget=200,
|
|
depth_delta=1,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — ProjectionSpec
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the projection spec focus should contain "{uri}"')
|
|
def step_spec_focus(context: Context, uri: str) -> None:
|
|
assert uri in context.proj_spec.focus, (
|
|
f"Expected {uri!r} in focus {context.proj_spec.focus}"
|
|
)
|
|
|
|
|
|
@then("the projection spec breadth should be {expected:d}")
|
|
def step_spec_breadth(context: Context, expected: int) -> None:
|
|
assert context.proj_spec.breadth == expected
|
|
|
|
|
|
@then("the projection spec depth should be {expected:d}")
|
|
def step_spec_depth_int(context: Context, expected: int) -> None:
|
|
assert context.proj_spec.depth == expected
|
|
|
|
|
|
@then('the projection spec depth should be "{expected}"')
|
|
def step_spec_depth_str(context: Context, expected: str) -> None:
|
|
assert context.proj_spec.depth == expected
|
|
|
|
|
|
@then("the projection spec depth_gradient should be True")
|
|
def step_spec_gradient_true(context: Context) -> None:
|
|
assert context.proj_spec.depth_gradient is True
|
|
|
|
|
|
@then('the projection spec domain should be "{expected}"')
|
|
def step_spec_domain(context: Context, expected: str) -> None:
|
|
assert context.proj_spec.domain == expected
|
|
|
|
|
|
@then("the projection spec should be immutable")
|
|
def step_spec_immutable(context: Context) -> None:
|
|
try:
|
|
context.proj_spec.breadth = 99 # type: ignore[misc]
|
|
assert False, "Should have raised" # noqa: B011
|
|
except (ValidationError, TypeError, AttributeError):
|
|
pass
|
|
|
|
|
|
@then("a projection ValidationError should be raised")
|
|
def step_proj_validation_error(context: Context) -> None:
|
|
assert context.proj_error is not None, "Expected ValidationError but none raised"
|
|
assert isinstance(context.proj_error, ValidationError)
|
|
|
|
|
|
@then('a projection ValueError should be raised mentioning "{keyword}"')
|
|
def step_proj_value_error(context: Context, keyword: str) -> None:
|
|
error = getattr(context, "proj_error", None) or getattr(
|
|
context, "child_error", None
|
|
)
|
|
assert error is not None, f"Expected ValueError mentioning {keyword!r}"
|
|
assert keyword.lower() in str(error).lower(), f"Expected '{keyword}' in: {error}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — ProjectedNode
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the projected node uri should be "{expected}"')
|
|
def step_node_uri(context: Context, expected: str) -> None:
|
|
assert context.proj_node.uri == expected
|
|
|
|
|
|
@then("the projected node distance should be {expected:d}")
|
|
def step_node_distance(context: Context, expected: int) -> None:
|
|
assert context.proj_node.distance == expected
|
|
|
|
|
|
@then("the projected node resolved_depth should be {expected:d}")
|
|
def step_node_resolved_depth(context: Context, expected: int) -> None:
|
|
assert context.proj_node.resolved_depth == expected
|
|
|
|
|
|
@then("the projected node should be immutable")
|
|
def step_node_immutable(context: Context) -> None:
|
|
try:
|
|
context.proj_node.distance = 99 # type: ignore[misc]
|
|
assert False, "Should have raised" # noqa: B011
|
|
except (ValidationError, TypeError, AttributeError):
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — DepthBreadthProjector
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the projection should contain exactly {count:d} node")
|
|
def step_proj_count_singular(context: Context, count: int) -> None:
|
|
assert len(context.proj_result) == count, (
|
|
f"Expected {count} nodes, got {len(context.proj_result)}"
|
|
)
|
|
|
|
|
|
@then("the projection should contain exactly {count:d} nodes")
|
|
def step_proj_count(context: Context, count: int) -> None:
|
|
assert len(context.proj_result) == count, (
|
|
f"Expected {count} nodes, got {len(context.proj_result)}"
|
|
)
|
|
|
|
|
|
@then('the projected nodes should include "{uri}" at distance {dist:d}')
|
|
def step_proj_node_at_distance(context: Context, uri: str, dist: int) -> None:
|
|
found = [n for n in context.proj_result if n.uri == uri]
|
|
assert found, f"Node {uri!r} not found in projection"
|
|
assert found[0].distance == dist, (
|
|
f"Expected {uri!r} at distance {dist}, got {found[0].distance}"
|
|
)
|
|
|
|
|
|
@then('the projected node "{uri}" should have resolved_depth {depth:d}')
|
|
def step_proj_node_depth(context: Context, uri: str, depth: int) -> None:
|
|
found = [n for n in context.proj_result if n.uri == uri]
|
|
assert found, f"Node {uri!r} not found"
|
|
assert found[0].resolved_depth == depth, (
|
|
f"Expected {uri!r} depth={depth}, got {found[0].resolved_depth}"
|
|
)
|
|
|
|
|
|
@then('the projected node "{uri}" should have resolved_depth greater than {depth:d}')
|
|
def step_proj_node_depth_gt(context: Context, uri: str, depth: int) -> None:
|
|
found = [n for n in context.proj_result if n.uri == uri]
|
|
assert found, f"Node {uri!r} not found"
|
|
assert found[0].resolved_depth > depth, (
|
|
f"Expected {uri!r} depth > {depth}, got {found[0].resolved_depth}"
|
|
)
|
|
|
|
|
|
@then('the projected node "{uri}" should have resolved_depth less than {depth:d}')
|
|
def step_proj_node_depth_lt(context: Context, uri: str, depth: int) -> None:
|
|
found = [n for n in context.proj_result if n.uri == uri]
|
|
assert found, f"Node {uri!r} not found"
|
|
assert found[0].resolved_depth < depth, (
|
|
f"Expected {uri!r} depth < {depth}, got {found[0].resolved_depth}"
|
|
)
|
|
|
|
|
|
@then("the result should contain {count:d} ContextFragments")
|
|
def step_fragment_count(context: Context, count: int) -> None:
|
|
assert len(context.proj_fragments) == count
|
|
|
|
|
|
@then('the first fragment uko_node should be "{expected}"')
|
|
def step_first_frag_uko(context: Context, expected: str) -> None:
|
|
assert context.proj_fragments[0].uko_node == expected
|
|
|
|
|
|
@then('all fragments should have metadata key "{key}" set to True')
|
|
def step_fragments_metadata(context: Context, key: str) -> None:
|
|
for frag in context.proj_fragments:
|
|
assert frag.metadata.get(key) is True, (
|
|
f"Fragment {frag.uko_node} missing metadata key {key!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — DetailLevelMap presets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the code map should resolve "{level}" to {expected:d}')
|
|
def step_code_resolve(context: Context, level: str, expected: int) -> None:
|
|
assert context.code_map.resolve(level) == expected
|
|
|
|
|
|
@then('the docs map should resolve "{level}" to {expected:d}')
|
|
def step_docs_resolve(context: Context, level: str, expected: int) -> None:
|
|
assert context.docs_map.resolve(level) == expected
|
|
|
|
|
|
@then('the database map should resolve "{level}" to {expected:d}')
|
|
def step_db_resolve(context: Context, level: str, expected: int) -> None:
|
|
assert context.db_map.resolve(level) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — PlanContextInheritance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the child result should contain a ContextRequest")
|
|
def step_child_has_request(context: Context) -> None:
|
|
assert hasattr(context.child_result, "request"), "No request in result"
|
|
assert isinstance(context.child_result.request, ContextRequest)
|
|
|
|
|
|
@then('the child request focus should contain "{uri}"')
|
|
def step_child_focus_contains(context: Context, uri: str) -> None:
|
|
assert uri in context.child_result.request.focus
|
|
|
|
|
|
@then("the child request depth_gradient should be True")
|
|
def step_child_gradient(context: Context) -> None:
|
|
assert context.child_result.request.depth_gradient is True
|
|
|
|
|
|
@then("the child result depth_delta should be {expected:d}")
|
|
def step_child_depth_delta(context: Context, expected: int) -> None:
|
|
assert context.child_result.depth_delta == expected
|
|
|
|
|
|
@then("the child request depth should be greater than {depth:d}")
|
|
def step_child_depth_gt(context: Context, depth: int) -> None:
|
|
assert context.child_result.request.depth > depth, (
|
|
f"Expected depth > {depth}, got {context.child_result.request.depth}"
|
|
)
|
|
|
|
|
|
@then("the child request breadth should be at most {max_breadth:d}")
|
|
def step_child_breadth_max(context: Context, max_breadth: int) -> None:
|
|
assert context.child_result.request.breadth <= max_breadth, (
|
|
f"Expected breadth <= {max_breadth}, got {context.child_result.request.breadth}"
|
|
)
|
|
|
|
|
|
@then("the child result skeleton_fragments should have {count:d} entries")
|
|
def step_child_skeleton_count(context: Context, count: int) -> None:
|
|
assert len(context.child_result.skeleton_fragments) == count, (
|
|
f"Expected {count} skeleton fragments, got {len(context.child_result.skeleton_fragments)}"
|
|
)
|
|
|
|
|
|
@then("the inheritance config skeleton_ratio should be {expected:g}")
|
|
def step_config_ratio(context: Context, expected: float) -> None:
|
|
assert context.inheritance_config.skeleton_ratio == expected
|
|
|
|
|
|
@then('the child focus should contain "{uri}"')
|
|
def step_focus_contains_single(context: Context, uri: str) -> None:
|
|
assert uri in context.child_focus
|
|
|
|
|
|
@then('the child focus should include both "{uri1}" and "{uri2}"')
|
|
def step_focus_contains_both(context: Context, uri1: str, uri2: str) -> None:
|
|
assert uri1 in context.child_focus
|
|
assert uri2 in context.child_focus
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — ChildContextResult
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the child context result should be immutable")
|
|
def step_result_immutable(context: Context) -> None:
|
|
try:
|
|
context.child_ctx_result.depth_delta = 99 # type: ignore[misc]
|
|
assert False, "Should have raised" # noqa: B011
|
|
except (ValidationError, TypeError, AttributeError):
|
|
pass
|
|
|
|
|
|
@then("the child context result should contain the request")
|
|
def step_result_has_request(context: Context) -> None:
|
|
assert isinstance(context.child_ctx_result.request, ContextRequest)
|