"""Robot Framework helper for ScoredFragment and ContextFragment integration tests. Provides a CLI-style interface for Robot to invoke pipeline model operations and verify results. Exit code 0 = success, 1 = failure. Usage: python robot/helper_context_fragment_models.py scored-create python robot/helper_context_fragment_models.py scored-breakdown python robot/helper_context_fragment_models.py scored-equality python robot/helper_context_fragment_models.py scored-dedup python robot/helper_context_fragment_models.py scored-from-relevance python robot/helper_context_fragment_models.py scored-frozen python robot/helper_context_fragment_models.py strategy-source """ from __future__ import annotations import sys from collections.abc import Callable from pathlib import Path from pydantic import ValidationError _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from cleveragents.domain.contexts.fragment import ScoredFragment # noqa: E402 from cleveragents.domain.models.core.context_fragment import ( # noqa: E402 ContextFragment, FragmentProvenance, ) _DEFAULT_PROV = FragmentProvenance(resource_uri="test://robot") def _make_fragment(**kwargs: object) -> ContextFragment: """Create a ContextFragment with sensible test defaults.""" defaults: dict[str, object] = { "uko_node": "test://default", "content": "test content", "token_count": 10, "provenance": _DEFAULT_PROV, } defaults.update(kwargs) return ContextFragment(**defaults) # type: ignore[arg-type] def _cmd_scored_create() -> int: """Create a ScoredFragment and verify defaults.""" frag = _make_fragment(uko_node="project://app/main.py", content="hello") scored = ScoredFragment(fragment=frag, composite_score=0.85) if scored.composite_score != 0.85: print(f"fail: expected composite_score=0.85, got {scored.composite_score}") return 1 if scored.rank != 0: print(f"fail: expected rank=0, got {scored.rank}") return 1 if scored.score_breakdown != {}: print(f"fail: expected empty breakdown, got {scored.score_breakdown}") return 1 if scored.uko_node != "project://app/main.py": print(f"fail: expected uko_node=project://app/main.py, got {scored.uko_node}") return 1 print("scored-create-ok") return 0 def _cmd_scored_breakdown() -> int: """Create a ScoredFragment with full breakdown.""" frag = _make_fragment(uko_node="project://app/io.py", content="async IO") breakdown = {"relevance": 0.9, "hierarchy": 0.8, "recency": 0.95} scored = ScoredFragment( fragment=frag, composite_score=0.92, score_breakdown=breakdown, rank=1, ) if scored.composite_score != 0.92: print(f"fail: expected composite_score=0.92, got {scored.composite_score}") return 1 if scored.rank != 1: print(f"fail: expected rank=1, got {scored.rank}") return 1 if scored.score_breakdown.get("relevance") != 0.9: print(f"fail: expected relevance=0.9, got {scored.score_breakdown}") return 1 print("scored-breakdown-ok") return 0 def _cmd_scored_equality() -> int: """Verify equality based on uko_node + detail_depth.""" frag_a = _make_fragment( uko_node="project://app/main.py", content="A", detail_depth=3, ) frag_b = _make_fragment( uko_node="project://app/main.py", content="B", detail_depth=3, ) sa = ScoredFragment.from_fragment(frag_a, composite_score=0.5) sb = ScoredFragment.from_fragment(frag_b, composite_score=0.9) if sa != sb: print("fail: expected equal scored fragments") return 1 if hash(sa) != hash(sb): print("fail: expected same hash") return 1 # Different uko_node should be not equal frag_c = _make_fragment( uko_node="project://app/other.py", content="C", detail_depth=3, ) sc = ScoredFragment.from_fragment(frag_c, composite_score=0.5) if sa == sc: print("fail: expected different scored fragments") return 1 print("scored-equality-ok") return 0 def _cmd_scored_dedup() -> int: """Verify set-based deduplication.""" frag_a = _make_fragment( uko_node="project://app/main.py", content="A", detail_depth=3, ) frag_b = _make_fragment( uko_node="project://app/main.py", content="B", detail_depth=3, ) frag_c = _make_fragment( uko_node="project://app/other.py", content="C", detail_depth=3, ) scored_set = { ScoredFragment.from_fragment(frag_a, composite_score=0.5), ScoredFragment.from_fragment(frag_b, composite_score=0.9), ScoredFragment.from_fragment(frag_c, composite_score=0.7), } if len(scored_set) != 2: print(f"fail: expected 2 unique, got {len(scored_set)}") return 1 print("scored-dedup-ok") return 0 def _cmd_scored_from_relevance() -> int: """Create ScoredFragment using from_relevance factory.""" frag = _make_fragment( uko_node="project://app/util.py", content="utils", relevance_score=0.75, ) scored = ScoredFragment.from_relevance(frag) if scored.composite_score != 0.75: print(f"fail: expected composite_score=0.75, got {scored.composite_score}") return 1 if scored.score_breakdown.get("relevance") != 0.75: print(f"fail: expected relevance=0.75, got {scored.score_breakdown}") return 1 print("scored-from-relevance-ok") return 0 def _cmd_scored_frozen() -> int: """Verify ScoredFragment is frozen.""" frag = _make_fragment() scored = ScoredFragment(fragment=frag, composite_score=0.5) try: scored.composite_score = 0.99 # type: ignore[misc] print("fail: expected error when modifying frozen model") return 1 except ValidationError: pass print("scored-frozen-ok") return 0 def _cmd_strategy_source() -> int: """Verify strategy_source field on ContextFragment.""" # Default should be empty string frag = _make_fragment() if frag.strategy_source != "": print(f"fail: expected empty strategy_source, got '{frag.strategy_source}'") return 1 # Explicit value frag2 = _make_fragment(strategy_source="keyword-search") if frag2.strategy_source != "keyword-search": print( f"fail: expected strategy_source='keyword-search', " f"got '{frag2.strategy_source}'" ) return 1 print("strategy-source-ok") return 0 _COMMANDS: dict[str, Callable[[], int]] = { "scored-create": _cmd_scored_create, "scored-breakdown": _cmd_scored_breakdown, "scored-equality": _cmd_scored_equality, "scored-dedup": _cmd_scored_dedup, "scored-from-relevance": _cmd_scored_from_relevance, "scored-frozen": _cmd_scored_frozen, "strategy-source": _cmd_strategy_source, } def main() -> int: """Dispatch to the requested sub-command.""" if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: valid = ", ".join(sorted(_COMMANDS)) print(f"Usage: {sys.argv[0]} <{valid}>", file=sys.stderr) return 1 return _COMMANDS[sys.argv[1]]() if __name__ == "__main__": raise SystemExit(main())