cleveragents-core/robot/helper_server_stubs.py

"""Helper script for server_stubs.robot smoke tests.

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

from __future__ import annotations

import os
import sys
from pathlib import Path

# 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.a2a.clients import (  # noqa: E402
    AuthClient,
    RemoteExecutionClient,
    ServerClient,
    StubAuthClient,
    StubRemoteExecutionClient,
    StubServerClient,
)
from cleveragents.a2a.server_config import ServerConnectionConfig  # noqa: E402
from cleveragents.application.services.config_service import _REGISTRY  # noqa: E402

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


def stub_clients() -> None:
    """Verify all stub clients raise NotImplementedError."""
    server = StubServerClient()
    try:
        server.health_check()
        print("FAIL: should have raised NotImplementedError", file=sys.stderr)
        sys.exit(1)
    except NotImplementedError:
        pass

    try:
        server.get_version()
        print("FAIL: should have raised NotImplementedError", file=sys.stderr)
        sys.exit(1)
    except NotImplementedError:
        pass

    exec_client = StubRemoteExecutionClient()
    try:
        exec_client.execute_plan("PLAN001")
        print("FAIL: should have raised NotImplementedError", file=sys.stderr)
        sys.exit(1)
    except NotImplementedError:
        pass

    auth_client = StubAuthClient()
    try:
        auth_client.authenticate("tok_test")
        print("FAIL: should have raised NotImplementedError", file=sys.stderr)
        sys.exit(1)
    except NotImplementedError:
        pass

    # Verify protocol conformance
    if not isinstance(server, ServerClient):
        print("FAIL: StubServerClient not ServerClient", file=sys.stderr)
        sys.exit(1)
    if not isinstance(exec_client, RemoteExecutionClient):
        print(
            "FAIL: StubRemoteExecutionClient not RemoteExecutionClient", file=sys.stderr
        )
        sys.exit(1)
    if not isinstance(auth_client, AuthClient):
        print("FAIL: StubAuthClient not AuthClient", file=sys.stderr)
        sys.exit(1)

    print("server-stub-clients-ok")


def config_validation() -> None:
    """Verify ServerConnectionConfig validation."""
    # Valid config
    cfg = ServerConnectionConfig(server_url="https://agents.example.com")
    if cfg.server_url != "https://agents.example.com":
        print("FAIL: wrong url", file=sys.stderr)
        sys.exit(1)
    if cfg.namespace != "default":
        print("FAIL: wrong namespace", file=sys.stderr)
        sys.exit(1)
    if cfg.tls_verify is not True:
        print("FAIL: wrong tls_verify", file=sys.stderr)
        sys.exit(1)

    # Invalid: empty URL
    try:
        ServerConnectionConfig(server_url="")
        print("FAIL: should have rejected empty URL", file=sys.stderr)
        sys.exit(1)
    except Exception:
        pass

    # Invalid: non-HTTP URL
    try:
        ServerConnectionConfig(server_url="ftp://bad.example.com")
        print("FAIL: should have rejected ftp URL", file=sys.stderr)
        sys.exit(1)
    except Exception:
        pass

    print("server-config-validation-ok")


def config_keys() -> None:
    """Verify server config keys are registered."""
    required_keys = [
        "server.url",
        "server.token",
        "server.tls-verify",
        "server.namespace",
    ]
    for key in required_keys:
        if key not in _REGISTRY:
            print(f"FAIL: '{key}' not in registry", file=sys.stderr)
            sys.exit(1)

    # Check defaults
    if _REGISTRY["server.tls-verify"].default is not True:
        print("FAIL: server.tls-verify default not True", file=sys.stderr)
        sys.exit(1)
    if _REGISTRY["server.namespace"].default != "default":
        print("FAIL: server.namespace default not 'default'", file=sys.stderr)
        sys.exit(1)

    print("server-config-keys-ok")


def server_mode() -> None:
    """Verify resolve_server_mode returns disabled when no URL is configured."""
    import tempfile

    os.environ.pop("CLEVERAGENTS_SERVER_URL", None)

    # Point HOME to a clean temporary directory so resolve_server_mode
    # cannot pick up a server.url written by a previous test run.
    with tempfile.TemporaryDirectory() as tmpdir:
        old_home = os.environ.get("HOME")
        os.environ["HOME"] = tmpdir
        try:
            from cleveragents.cli.commands.server import resolve_server_mode

            mode = resolve_server_mode()
        finally:
            if old_home is not None:
                os.environ["HOME"] = old_home
            else:
                os.environ.pop("HOME", None)

    if mode != "disabled":
        print(f"FAIL: expected 'disabled', got '{mode}'", file=sys.stderr)
        sys.exit(1)

    print("server-mode-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

_COMMANDS = {
    "stub-clients": stub_clients,
    "config-validation": config_validation,
    "config-keys": config_keys,
    "server-mode": server_mode,
}


def main() -> None:
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(2)
    _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    main()