hope-voice-api/tests/conftest.py

"""Shared fixtures for the Hope voice-assistant test suite.

Tests run on the HOST machine against the live "voice" container:
- HTTPS web UI  -> https://localhost:9178 (self-signed cert, verify=False)
- Token server  -> http://127.0.0.1:8091 (host network mode)
- LLM API       -> LLM_BASE_URL (default http://192.168.86.2:3005/api)
- In-container  -> docker exec voice ...
"""

import json
import os
import subprocess
from pathlib import Path

import httpx
import pytest

CONTAINER = "voice"
WEB_BASE_URL = os.environ.get("VOICE_WEB_URL", "https://localhost:9178")
TOKEN_URL = os.environ.get("VOICE_TOKEN_URL", "http://127.0.0.1:8091/token")
LLM_BASE_URL = (
    os.environ.get("LLM_BASE_URL")
    or os.environ.get("GEMMA_BASE_URL")
    or "http://192.168.86.2:3005/api"
).rstrip("/")
LLM_MODEL = (
    os.environ.get("LLM_MODEL")
    or os.environ.get("GEMMA_MODEL")
    or "gemma-4-e4b"
)
LLM_API_KEY = (
    os.environ.get("LLM_API_KEY")
    or os.environ.get("GEMMA_API_KEY")
    or "not-needed"
)
_users = json.loads((Path(__file__).parents[1] / "config" / "users.json").read_text())
_shane_memory = _users["users"]["shane"]["memory"]
MEMORY_MCP_URL = os.environ.get("MEMORY_MCP_URL", _shane_memory["url"])
MEMORY_MCP_TOKEN = os.environ.get("MEMORY_MCP_TOKEN", _shane_memory["api_key"])

# Path to the agent's venv python inside the container (has mcp, livekit.agents)
CONTAINER_PYTHON = "/opt/voice-agent/.venv/bin/python"
AGENT_DIR_IN_CONTAINER = "/opt/voice-agent"


def pytest_configure(config):
    config.addinivalue_line("markers", "slow: marks tests as slow (LLM / network calls)")


# ── docker helpers ───────────────────────────────────────────────────────────


def docker_exec(*args, timeout=60, env=None, check=True) -> subprocess.CompletedProcess:
    """Run a command inside the voice container. Returns CompletedProcess."""
    cmd = ["docker", "exec"]
    if env:
        for k, v in env.items():
            cmd += ["-e", f"{k}={v}"]
    cmd += [CONTAINER, *args]
    return subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=check,
    )


def docker_exec_python(
    code: str, *, timeout=120, cwd=None, env=None, check=True
) -> subprocess.CompletedProcess:
    """Run a python snippet inside the container with the agent venv."""
    return docker_exec(CONTAINER_PYTHON, "-c", code, timeout=timeout, env=env, check=check)


# ── fixtures ─────────────────────────────────────────────────────────────────


@pytest.fixture(scope="session")
def base_url() -> str:
    """Base URL of the web UI (HTTPS, self-signed)."""
    return WEB_BASE_URL


@pytest.fixture(scope="session")
def client(base_url):
    """httpx client for the HTTPS web UI with TLS verification disabled."""
    with httpx.Client(base_url=base_url, verify=False, timeout=15) as c:
        yield c


@pytest.fixture(scope="session")
def token() -> dict:
    """Fetch a fresh LiveKit token from the token endpoint (direct, no proxy)."""
    resp = httpx.post(TOKEN_URL, json={}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    assert "token" in data, f"token endpoint did not return a token: {data.keys()}"
    return data


@pytest.fixture(scope="session")
def llm_client():
    """httpx client pointed at the LLM API (same one Hope uses)."""
    headers = {}
    if LLM_API_KEY and LLM_API_KEY != "not-needed":
        headers["Authorization"] = f"Bearer {LLM_API_KEY}"
    with httpx.Client(base_url=LLM_BASE_URL, headers=headers, timeout=60) as c:
        yield c


@pytest.fixture(scope="session")
def mcp_client_factory():
    """Factory that runs an MCP stdio tool call inside the container.

    Returns (result_text, is_error). Spawns the MCP server script via the
    agent venv python and drives it with a raw JSON-RPC session using the
    mcp client library (available in the container, not on the host).
    """

    def _call(script: str, tool_name: str, args: dict | None = None, env: dict | None = None) -> tuple[str, bool]:
        import os as host_os

        merged_env = {**host_os.environ, **(env or {})}
        code = f"""
import asyncio, json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    params = StdioServerParameters(
        command={CONTAINER_PYTHON!r},
        args=[{AGENT_DIR_IN_CONTAINER + "/" + script!r}],
        env={json.dumps(merged_env)},
    )
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool({tool_name!r}, {json.dumps(args or {})})
            text = ""
            for block in result.content:
                if getattr(block, "type", None) == "text":
                    text += block.text
            print(json.dumps({{"text": text, "isError": bool(result.isError)}}))

asyncio.run(main())
"""
        proc = docker_exec_python(code, timeout=120, check=False)
        if proc.returncode != 0:
            raise AssertionError(
                f"MCP call {tool_name} failed (rc={proc.returncode}): "
                f"stderr={proc.stderr[-800:]}"
            )
        data = json.loads(proc.stdout.strip().splitlines()[-1])
        return data["text"], data["isError"]

    return _call


@pytest.fixture(scope="session")
def memory_client():
    """Call Shane's authenticated remote memory MCP server."""

    def _call(tool_name: str, args: dict | None = None) -> tuple[str, bool]:
        code = f"""
import asyncio, json
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    async with streamablehttp_client(
        {MEMORY_MCP_URL!r},
        headers={{"Authorization": "Bearer " + {MEMORY_MCP_TOKEN!r}}},
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool({tool_name!r}, {json.dumps(args or {})})
            text = ""
            for block in result.content:
                if getattr(block, "type", None) == "text":
                    text += block.text
            print(json.dumps({{"text": text, "isError": bool(result.isError)}}))

asyncio.run(main())
"""
        proc = docker_exec_python(code, timeout=120, check=False)
        if proc.returncode != 0:
            raise AssertionError(
                f"Memory MCP call {tool_name} failed (rc={proc.returncode}): "
                f"stderr={proc.stderr[-800:]}"
            )
        data = json.loads(proc.stdout.strip().splitlines()[-1])
        return data["text"], data["isError"]

    return _call