atlas-iac/testing/tests/test_hermes_cli_support.py

"""Focused tests for Agent Hermes' direct Codex and Claude Kanban lanes."""

from __future__ import annotations

import errno

import importlib.util

import hashlib

import json

import os

import signal

import stat

import sys

import threading

import time

from contextlib import nullcontext

from pathlib import Path

from types import SimpleNamespace

import pytest

import yaml

SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"

sys.path.insert(0, str(SCRIPTS))

HERMES = Path(__file__).parents[2] / "services/hermes"

KEYCLOAK = Path(__file__).parents[2] / "services/keycloak"

FLUX_HERMES = (
    Path(__file__).parents[2]
    / "clusters/atlas/flux-system/applications/hermes/kustomization.yaml"
)

def _load(name: str):
    spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module

_lane_facade = _load("cli_lane_runner")


class _LaneSurface:
    """Patch every module-local import behind the compatibility facade."""

    def __init__(self, facade):
        object.__setattr__(self, "_facade", facade)
        object.__setattr__(
            self,
            "_modules",
            tuple(
                module
                for name, module in sys.modules.items()
                if name == "cli_lane_runner" or name.startswith("cli_lane_")
            ),
        )

    def __getattr__(self, name):
        facade = object.__getattribute__(self, "_facade")
        if hasattr(facade, name):
            return getattr(facade, name)
        for module in object.__getattribute__(self, "_modules"):
            if hasattr(module, name):
                return getattr(module, name)
        raise AttributeError(name)

    def __setattr__(self, name, value):
        matched = False
        for module in object.__getattribute__(self, "_modules"):
            if hasattr(module, name):
                setattr(module, name, value)
                matched = True
        if not matched:
            setattr(object.__getattribute__(self, "_facade"), name, value)


lanes = _LaneSurface(_lane_facade)

policy = _load("claude_command_policy")

migration = _load("migrate_herdr_state")

auth_patch = _load("patch_hermes_auth")

tui_gateway_patch = _load("patch_tui_gateway")

codex_runtime_patch = _load("patch_codex_runtime")

ttyd_patch = _load("patch_ttyd_index")

client_config = _load("configure_agent_clients")

def _agent_deployment() -> dict:
    return yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())

def _services() -> dict[str, dict]:
    return {
        item["metadata"]["name"]: item
        for item in yaml.safe_load_all((HERMES / "service.yaml").read_text())
        if item
    }

def _oauth_deployment(name: str) -> dict:
    documents = [
        item
        for item in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
        if item
    ]
    return next(
        item
        for item in documents
        if item["kind"] == "Deployment" and item["metadata"]["name"] == name
    )

class _SwitchyardResponse:
    """Minimal context-managed response used by routing contract tests."""

    def __init__(self, selected: str, rationale: str = "local classifier vote"):
        self.headers = {
            "x-model-router-selected-model": selected,
            "x-model-router-rationale": rationale,
        }

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def read(self):
        return b"{}"

def _completed_result(summary: str = "done") -> dict:
    return {
        "status": "completed",
        "summary": summary,
        "changed_files": [],
        "tests_run": [],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    }

def _pending_terminal_record(board: str, task_id: str, run_id: int, summary: str) -> dict:
    structured = _completed_result(summary)
    return {
        "board": board,
        "task_id": task_id,
        "expected_run_id": run_id,
        "result": json.dumps(structured, sort_keys=True),
        "summary": summary,
        "metadata": {},
        "kanban_state": "pending",
        "recorded_at": lanes.utc_now(),
    }

def _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) -> None:
    class Connection:
        def close(self):
            return None

    def complete_task(_conn, _task_id, **kwargs):
        if task.status != "running" or task.current_run_id != kwargs["expected_run_id"]:
            return False
        task.status = "done"
        task.result = kwargs["result"]
        task.current_run_id = None
        completions.append(kwargs["result"])
        return True

    def reclaim_task(_conn, _task_id, **kwargs):
        reclaims.append(kwargs)
        return False

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=complete_task,
        reclaim_task=reclaim_task,
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

__all__ = [
    'FLUX_HERMES',
    'HERMES',
    'KEYCLOAK',
    'Path',
    'SCRIPTS',
    'SimpleNamespace',
    '_SwitchyardResponse',
    '_agent_deployment',
    '_completed_result',
    '_install_terminal_recovery_db',
    '_load',
    '_oauth_deployment',
    '_pending_terminal_record',
    '_services',
    'auth_patch',
    'client_config',
    'codex_runtime_patch',
    'errno',
    'hashlib',
    'importlib',
    'json',
    'lanes',
    'migration',
    'nullcontext',
    'os',
    'policy',
    'pytest',
    'signal',
    'stat',
    'sys',
    'threading',
    'time',
    'ttyd_patch',
    'tui_gateway_patch',
    'yaml',
]