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:8090 (self-signed cert, verify=False)
- Token server  -> http://127.0.0.1:8091 (host network mode)
- LLM API       -> GEMMA_BASE_URL (default http://192.168.86.2:8023/v1)
- In-container  -> docker exec voice ...
"""

import json
import os
import subprocess

import httpx
import pytest

CONTAINER = "voice"
WEB_BASE_URL = os.environ.get("VOICE_WEB_URL", "https://localhost:8090")
TOKEN_URL = os.environ.get("VOICE_TOKEN_URL", "http://127.0.0.1:8091/token")
GEMMA_BASE_URL = os.environ.get("GEMMA_BASE_URL", "http://192.168.86.2:8023/v1").rstrip("/")
GEMMA_MODEL = os.environ.get("GEMMA_MODEL", "gemma-4-e4b")
COGNEE_MCP_URL = os.environ.get("COGNEE_MCP_URL", "http://192.168.86.2:8003/mcp")

# 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 Gemma LLM API (same one Hope uses)."""
    with httpx.Client(base_url=GEMMA_BASE_URL, 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 cognee_client():
    """Factory that calls tools on the remote Cognee MCP server (Streamable HTTP).

    Returns (result_text, is_error). Uses the mcp client library from the
    agent venv inside the container.
    """

    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({COGNEE_MCP_URL!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"Cognee 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