cleveragents-core/robot/helper_component_resolver.py

"""Helper script for component_resolver.robot end-to-end tests.

Each subcommand is a self-contained check that prints a sentinel on success.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any, Protocol, runtime_checkable

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.services.component_resolver import (  # noqa: E402
    ComponentNotFoundError,
    ComponentRegistrationError,
    ComponentResolver,
    ScopeLevel,
)

# ---------------------------------------------------------------------------
# Test protocol and implementations
# ---------------------------------------------------------------------------


@runtime_checkable
class _Greeter(Protocol):
    def greet(self) -> str: ...


class _DefaultGreeter:
    def greet(self) -> str:
        return "hello-global"


class _ProjectGreeter:
    def greet(self) -> str:
        return "hello-project"


class _PlanGreeter:
    def greet(self) -> str:
        return "hello-plan"


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def global_default_resolution() -> None:
    """Verify global default resolution."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    result = resolver.resolve(_Greeter)
    if result.scope == ScopeLevel.GLOBAL and result.component.greet() == "hello-global":
        print("component-resolver-global-ok")
    else:
        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
        sys.exit(1)


def project_override() -> None:
    """Verify project override takes precedence over global."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    resolver.register_project("proj-1", _Greeter, _ProjectGreeter())
    result = resolver.resolve(_Greeter, project_id="proj-1")
    if (
        result.scope == ScopeLevel.PROJECT
        and result.component.greet() == "hello-project"
    ):
        print("component-resolver-project-override-ok")
    else:
        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
        sys.exit(1)


def plan_override() -> None:
    """Verify plan override takes highest precedence."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    resolver.register_project("proj-1", _Greeter, _ProjectGreeter())
    resolver.register_plan("plan-1", _Greeter, _PlanGreeter())
    result = resolver.resolve(_Greeter, plan_id="plan-1", project_id="proj-1")
    if result.scope == ScopeLevel.PLAN and result.component.greet() == "hello-plan":
        print("component-resolver-plan-override-ok")
    else:
        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
        sys.exit(1)


def fallthrough_to_global() -> None:
    """Verify fallthrough when no plan or project override exists."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    result = resolver.resolve(_Greeter, plan_id="no-plan", project_id="no-project")
    if result.scope == ScopeLevel.GLOBAL and result.component.greet() == "hello-global":
        print("component-resolver-fallthrough-ok")
    else:
        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
        sys.exit(1)


def missing_component_error() -> None:
    """Verify ComponentNotFoundError for unregistered components."""
    resolver = ComponentResolver()
    try:
        resolver.resolve(_Greeter)
        print("FAIL: expected ComponentNotFoundError", file=sys.stderr)
        sys.exit(1)
    except ComponentNotFoundError:
        print("component-resolver-not-found-ok")


def caching_behavior() -> None:
    """Verify resolution caching."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    r1 = resolver.resolve(_Greeter)
    r2 = resolver.resolve(_Greeter)
    if r1 is r2 and resolver.cache_size() == 1:
        print("component-resolver-caching-ok")
    else:
        print(f"FAIL: same={r1 is r2}, cache={resolver.cache_size()}", file=sys.stderr)
        sys.exit(1)


def security_module_prefix() -> None:
    """Verify module prefix restriction."""
    resolver = ComponentResolver()
    try:
        resolver._import_component("evil_module:Hacker")
        print("FAIL: expected error", file=sys.stderr)
        sys.exit(1)
    except ComponentRegistrationError:
        print("component-resolver-security-ok")


def introspection_apis() -> None:
    """Verify introspection methods."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    resolver.register_project("p1", _Greeter, _ProjectGreeter())
    resolver.register_plan("pl1", _Greeter, _PlanGreeter())
    g = resolver.list_global_types()
    p = resolver.list_project_types("p1")
    pl = resolver.list_plan_types("pl1")
    if "_Greeter" in g and "_Greeter" in p and "_Greeter" in pl:
        print("component-resolver-introspection-ok")
    else:
        print(f"FAIL: global={g}, project={p}, plan={pl}", file=sys.stderr)
        sys.exit(1)


def clear_all() -> None:
    """Verify clear removes all registrations."""
    resolver = ComponentResolver()
    resolver.register_global(_Greeter, _DefaultGreeter())
    resolver.register_project("p1", _Greeter, _ProjectGreeter())
    resolver.register_plan("pl1", _Greeter, _PlanGreeter())
    resolver.clear()
    ok = (
        not resolver.has_global(_Greeter)
        and not resolver.has_project("p1", _Greeter)
        and not resolver.has_plan("pl1", _Greeter)
    )
    if ok:
        print("component-resolver-clear-ok")
    else:
        print("FAIL: clear did not remove all", file=sys.stderr)
        sys.exit(1)


# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Any] = {
    "global-default": global_default_resolution,
    "project-override": project_override,
    "plan-override": plan_override,
    "fallthrough": fallthrough_to_global,
    "not-found": missing_component_error,
    "caching": caching_behavior,
    "security": security_module_prefix,
    "introspection": introspection_apis,
    "clear": clear_all,
}

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        print(f"Commands: {', '.join(sorted(_COMMANDS))}", file=sys.stderr)
        sys.exit(1)
    cmd = sys.argv[1]
    fn = _COMMANDS.get(cmd)
    if fn is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)
    fn()