tatlock/tests/contracts/test_service_contracts.py

"""
Wire-level contract tests for external service boundaries.

Each test sends the raw request the application code sends (no client
wrappers, no mocks) and asserts on the response shape, so boundary
breakage is caught directly instead of surfacing as agent misbehavior.

Semantics:
- Service unreachable  -> skip (an outage is not a contract violation)
- Service reachable but wrong response shape -> fail

Run with: make test-contracts
"""

import json

import httpx
import pytest

from src.core.config import config

OLLAMA = str(config.OLLAMA_HOST).rstrip("/")
QDRANT = f"http://{config.QDRANT_HOST}:{config.QDRANT_PORT}"
SEARXNG = str(config.SEARXNG_HOST).rstrip("/")

CALCULATOR_TOOL = {
    "type": "function",
    "function": {
        "name": "calculator",
        "description": "Evaluate a math expression",
        "parameters": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"],
        },
    },
}


async def _get_or_skip(url: str, service: str, timeout: float = 5.0) -> httpx.Response:
    """GET a URL, skipping the test if the service is unreachable."""
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            return await client.get(url)
    except httpx.TransportError as e:
        pytest.skip(f"{service} unreachable at {url}: {e}")


async def _post_or_skip(
    url: str, service: str, payload: dict, timeout: float, headers: dict | None = None
) -> httpx.Response:
    """POST a payload, skipping the test if the service is unreachable."""
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            return await client.post(url, json=payload, headers=headers)
    except httpx.TransportError as e:
        pytest.skip(f"{service} unreachable at {url}: {e}")


@pytest.mark.contract
class TestOllamaContract:
    """Boundary: Ollama native API and its OpenAI-compat layer."""

    async def test_tags_lists_configured_model(self):
        # Mirrors check_ollama_health()
        response = await _get_or_skip(f"{OLLAMA}/api/tags", "ollama")
        assert response.status_code == 200
        names = [m["name"] for m in response.json()["models"]]
        model = config.OLLAMA_DEFAULT_MODEL
        assert (
            model in names or f"{model}:latest" in names
        ), f"{model} not pulled; available: {names}"

    async def test_generate_returns_plain_text(self):
        # Mirrors StewardAgent._call_ollama()
        response = await _post_or_skip(
            f"{OLLAMA}/api/generate",
            "ollama",
            {
                "model": config.OLLAMA_DEFAULT_MODEL,
                "prompt": "Reply with the single word: pong",
                "stream": False,
                "options": {"temperature": 0.3, "top_p": 0.9},
            },
            timeout=config.OLLAMA_TIMEOUT,
        )
        assert response.status_code == 200
        assert response.json()["response"].strip()

    async def test_openai_compat_tool_calling(self):
        # Mirrors the request PydanticAI's OpenAIChatModel sends for the
        # orchestration phase, including the extra_body tool_choice.
        response = await _post_or_skip(
            f"{OLLAMA}/v1/chat/completions",
            "ollama",
            {
                "model": config.OLLAMA_DEFAULT_MODEL,
                "messages": [{"role": "user", "content": "What is 6 * 7? Use the calculator."}],
                "tools": [CALCULATOR_TOOL],
                "tool_choice": "required",
                "stream": False,
            },
            timeout=config.OLLAMA_TIMEOUT,
        )
        assert response.status_code == 200
        message = response.json()["choices"][0]["message"]
        tool_calls = message.get("tool_calls")
        assert tool_calls, f"model answered in text instead of calling the tool: {message}"
        assert tool_calls[0]["function"]["name"] == "calculator"
        arguments = json.loads(tool_calls[0]["function"]["arguments"])
        assert "expression" in arguments


@pytest.mark.contract
class TestAnthropicContract:
    """Boundary: Anthropic Messages API (the Claude fallback backend)."""

    HEADERS_KEY = "anthropic-version"

    def _headers(self) -> dict:
        if not config.ANTHROPIC_API_KEY:
            pytest.skip("ANTHROPIC_API_KEY not configured")
        return {
            "x-api-key": config.ANTHROPIC_API_KEY,
            "anthropic-version": "2023-06-01",
        }

    async def test_minimal_message_accepted(self):
        # Mirrors check_claude_health(): tiny request, no sampling params
        response = await _post_or_skip(
            "https://api.anthropic.com/v1/messages",
            "anthropic",
            {
                "model": config.ANTHROPIC_MODEL,
                "max_tokens": 1,
                "messages": [{"role": "user", "content": "hi"}],
            },
            timeout=30.0,
            headers=self._headers(),
        )
        assert response.status_code == 200, response.text

    async def test_temperature_rejected(self):
        # Pins the Claude Sonnet 5+ contract that broke the Steward:
        # sampling parameters are rejected with a 400 (and not billed).
        response = await _post_or_skip(
            "https://api.anthropic.com/v1/messages",
            "anthropic",
            {
                "model": config.ANTHROPIC_MODEL,
                "max_tokens": 1,
                "messages": [{"role": "user", "content": "hi"}],
                "temperature": 0.3,
            },
            timeout=30.0,
            headers=self._headers(),
        )
        assert response.status_code == 400
        assert "temperature" in response.text


@pytest.mark.contract
class TestQdrantContract:
    """Boundary: Qdrant REST API (Biographer's vector memory)."""

    async def test_collections_endpoint(self):
        response = await _get_or_skip(f"{QDRANT}/collections", "qdrant")
        assert response.status_code == 200
        assert "collections" in response.json()["result"]


@pytest.mark.contract
class TestSearxngContract:
    """Boundary: SearXNG JSON search API (web search tool)."""

    async def test_json_search(self):
        response = await _get_or_skip(
            f"{SEARXNG}/search?q=test&format=json", "searxng", timeout=config.SEARXNG_TIMEOUT
        )
        assert response.status_code == 200
        assert "results" in response.json()


@pytest.mark.contract
class TestLibraryDeskContract:
    """Boundary: library-desk research API (the Librarian's backend)."""

    async def test_health(self):
        host = getattr(config, "LIBRARY_DESK_HOST", None)
        if not host:
            pytest.skip("LIBRARY_DESK_HOST not configured")
        response = await _get_or_skip(f"{str(host).rstrip('/')}/health", "library-desk")
        assert response.status_code == 200


@pytest.mark.contract
class TestRedisContract:
    """Boundary: Redis on the configured memory DB."""

    async def test_roundtrip(self):
        import redis.asyncio as redis

        client = redis.Redis(
            host=config.REDIS_HOST,
            port=config.REDIS_PORT,
            db=config.REDIS_MEMORY_DB,
            socket_connect_timeout=3,
        )
        try:
            await client.ping()
        except Exception as e:
            pytest.skip(f"redis unreachable: {e}")
        try:
            await client.set("contract-test-key", "ok", ex=30)
            assert await client.get("contract-test-key") == b"ok"
            await client.delete("contract-test-key")
        finally:
            await client.aclose()