cleveragents-core/robot/helper_acms_backends.py

"""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 <command>
"""

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 <command>")
        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())