"""Robot Framework helper for ACMS backend protocol smoke tests. Provides a CLI-style interface for Robot to invoke backend creation, protocol compliance, and DI resolution. Exit code 0 = success, 1 = failure. Usage: python robot/helper_acms_backends.py """ from __future__ import annotations import sys from pathlib import Path # Ensure the src directory is on the import path. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from cleveragents.application.container import ( # noqa: E402 get_container, reset_container, ) from cleveragents.domain.models.acms.backends import ( # noqa: E402 GraphBackend, GraphResult, TextBackend, TextResult, VectorBackend, VectorResult, ) from cleveragents.domain.models.acms.stubs import ( # noqa: E402 InMemoryGraphBackend, InMemoryTextBackend, InMemoryVectorBackend, ) def main() -> int: """Entry point called by Robot Framework ``Run Process``.""" if len(sys.argv) < 2: print("Usage: helper_acms_backends.py ") return 1 command: str = sys.argv[1] if command == "text-backend": try: backend = InMemoryTextBackend() assert isinstance(backend, TextBackend) results = backend.search("test", scope=frozenset({"RES01"})) assert results == [] print("acms-text-backend-ok") return 0 except Exception as exc: print(f"acms-text-backend-fail: {exc}") return 1 if command == "vector-backend": try: backend = InMemoryVectorBackend() assert isinstance(backend, VectorBackend) results = backend.similarity_search([0.1, 0.2], scope=frozenset({"RES01"})) assert results == [] print("acms-vector-backend-ok") return 0 except Exception as exc: print(f"acms-vector-backend-fail: {exc}") return 1 if command == "graph-backend": try: backend = InMemoryGraphBackend() assert isinstance(backend, GraphBackend) result = backend.sparql_query( "SELECT ?s WHERE { ?s a uko:Container }", scope=frozenset({"RES01"}), ) assert result.triples == [] result2 = backend.get_triples("uko:subject") assert result2.triples == [] result3 = backend.traverse("uko:start", depth=2) assert result3.triples == [] print("acms-graph-backend-ok") return 0 except Exception as exc: print(f"acms-graph-backend-fail: {exc}") return 1 if command == "result-types": try: tr = TextResult(uko_uri="uko:test", content="hello", score=0.5) assert tr.uko_uri == "uko:test" vr = VectorResult(uko_uri="uko:vec", content="embed", score=0.8) assert vr.uko_uri == "uko:vec" gr = GraphResult(triples=[("a", "b", "c")]) assert len(gr.triples) == 1 print("acms-result-types-ok") return 0 except Exception as exc: print(f"acms-result-types-fail: {exc}") return 1 if command == "di-resolution": try: reset_container() container = get_container() tb = container.text_backend() assert isinstance(tb, TextBackend) vb = container.vector_backend() assert isinstance(vb, VectorBackend) gb = container.graph_backend() assert isinstance(gb, GraphBackend) print("acms-di-resolution-ok") return 0 except Exception as exc: print(f"acms-di-resolution-fail: {exc}") return 1 if command == "validation": try: # Empty query should raise backend = InMemoryTextBackend() raised = False try: backend.search("", scope=frozenset({"RES01"})) except ValueError: raised = True assert raised, "Expected ValueError for empty query" # Empty embedding should raise vb = InMemoryVectorBackend() raised = False try: vb.similarity_search([], scope=frozenset({"RES01"})) except ValueError: raised = True assert raised, "Expected ValueError for empty embedding" # Invalid score should raise raised = False try: TextResult(uko_uri="uko:x", content="x", score=2.0) except ValueError: raised = True assert raised, "Expected ValueError for score > 1.0" print("acms-validation-ok") return 0 except Exception as exc: print(f"acms-validation-fail: {exc}") return 1 print(f"Unknown command: {command}") return 1 if __name__ == "__main__": sys.exit(main())