ai-resume/corpus/tests/support.py

"""Shared fixtures for the corpus tests: a throwaway content tree and a fake Qdrant."""

from __future__ import annotations

import contextlib
import os
import shutil
import tempfile
from unittest import mock

from corpus.config import Config
from corpus.tests.fake_qdrant import FakeQdrant

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

DOC = """---
id: {id}
area: {area}
level: {level}
title: "{title}"
claim: "{claim}"
confidence: {confidence}
period: "2023–"
tech: [Go, SQLite]
metrics: [{{label: endpoints, value: 8}}]
artefacts: []
public: true
last_verified: {last_verified}
tags: [{tags}]
---

# {title}

{body}
"""

FENCE = """A paragraph before the code block explains the constraint.

```go
package main

func main() {{
	println("chunk boundary must never land inside this fence")
}}
```

A paragraph after the code block.
"""

TABLE = """Measured results, one row per endpoint:

| endpoint | latency | loss |
|---|---|---|
| alpha | 12ms | 0.0% |
| beta | 14ms | 0.1% |
| gamma | 9ms | 0.0% |

And a closing paragraph.
"""

PROSE = "\n\n".join(
    f"Paragraph {index} describes a delivery constraint in corporate language, "
    "with enough words that the chunker has to think about where to break, and "
    "never inside a fenced block, a table or a front-matter block."
    for index in range(1, 40)
)

DEFAULT_DOCS = {
    "ai/ai-overview.md": dict(
        id="ai-overview", area="ai", level=1, title="Agent Systems",
        claim="Harnesses, memory and routing.", confidence="running",
        last_verified="2026-10-01", tags="agents, llm", body=PROSE,
    ),
    "edge/l3/edge-zigbee.md": dict(
        id="edge-zigbee", area="edge", level=3, title="Low Power Mesh Sensing",
        claim="A constrained mesh that survives power loss.", confidence="delivered",
        last_verified="2026-09-15", tags="zigbee, mesh", body=FENCE,
    ),
    "platform/l3/platform-containers.md": dict(
        id="platform-containers", area="platform", level=3, title="Container Estate",
        claim="Fifty containers with measured resource limits.", confidence="in-progress",
        last_verified="2026-08-01", tags="docker", body=TABLE,
    ),
}


def temp_repo(docs: dict = None) -> str:
    """Create a throwaway repository root with tools/ and content/."""
    root = tempfile.mkdtemp(prefix="corpus-test-")
    os.makedirs(os.path.join(root, "tools"), exist_ok=True)
    shutil.copy(
        os.path.join(REPO_ROOT, "tools", "validate_content.py"),
        os.path.join(root, "tools", "validate_content.py"),
    )
    write_docs(root, docs or DEFAULT_DOCS)
    return root


def write_docs(root: str, docs: dict) -> None:
    for rel, fields in docs.items():
        path = os.path.join(root, "content", rel)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="utf-8") as handle:
            handle.write(DOC.format(**fields))


def write_doc(root: str, rel: str, **fields) -> None:
    path = os.path.join(root, "content", rel)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as handle:
        handle.write(DOC.format(**fields))
    # A fresh interpreter's import cache is fine; the loader re-reads the file.


def make_config(root: str, **overrides) -> Config:
    values = dict(
        provider="local-hashing",
        model="local-hashing-v1",
        dim=32,
        qdrant_url="fake://qdrant",
        collection="test_corpus",
    )
    values.update(overrides)
    return Config.from_env(root=root, **values)


@contextlib.contextmanager
def fake_qdrant(cfg, fake: FakeQdrant = None):
    """Point every module that builds a Qdrant client at one in-memory fake."""
    fake = fake or FakeQdrant(cfg.dim)
    with mock.patch("corpus.pipeline.Qdrant", return_value=fake), mock.patch(
        "corpus.freshness.Qdrant", return_value=fake
    ):
        yield fake


class StubServer:
    """A throwaway HTTP server on a loopback ephemeral port, for provider tests.

    `responder(method, path, body)` returns `(status, payload)`; every request is
    recorded with its parsed body so a test can assert the wire shape without a
    real provider or a real Qdrant.
    """

    def __init__(self, responder):
        import json
        import threading
        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

        self.requests = []
        self._responder = responder
        outer = self

        class Handler(BaseHTTPRequestHandler):
            def _handle(self):
                length = int(self.headers.get("Content-Length") or 0)
                raw = self.rfile.read(length).decode("utf-8") if length else ""
                body = json.loads(raw) if raw.strip() else None
                outer.requests.append(
                    {"method": self.command, "path": self.path, "body": body,
                     "headers": dict(self.headers)}
                )
                status, payload = outer._responder(self.command, self.path, body)
                data = payload.encode("utf-8") if isinstance(payload, str) else json.dumps(payload).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            do_GET = do_POST = do_PUT = do_DELETE = _handle

            def log_message(self, *_args):
                pass

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.url = f"http://127.0.0.1:{self.server.server_address[1]}"
        self._thread = threading.Thread(target=self.server.serve_forever, daemon=True)

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *_exc):
        self.server.shutdown()
        self.server.server_close()