m2-market/indexer/tests/test_reindex_flow.py

"""Inline httpx MockTransport test for the reindex flow (T015 verify step).

Registry fixture: one `published` + one `delisted` listing. Asserts the
resulting memory-api calls are exactly: drop partition, then upsert the
published listing only (the delisted one is silently omitted by a
drop-then-upsert-published rebuild — contracts/catalog-index.md "reindex").

Run directly: `python indexer/tests/test_reindex_flow.py`
Or via pytest: `pytest indexer/tests/test_reindex_flow.py`
"""

from __future__ import annotations

import base64
import json
import os
import sys
from pathlib import Path

import httpx

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from jsonschema import Draft202012Validator  # noqa: E402
from m2_market_indexer.client import MemoryClient, RegistryClient  # noqa: E402
from m2_market_indexer.reindex import run  # noqa: E402

SCHEMA_PATH = Path(__file__).resolve().parents[2] / "schemas" / "listing.schema.json"
# Pin the schema for config.listing_schema_path(): when the package under test is the
# venv-installed copy (uv workspace), its parents[3]-based resolution lands outside the
# repo and loads the wrong file. The test's fixture contract is THIS repo's schema.
os.environ["LISTING_SCHEMA_PATH"] = str(SCHEMA_PATH)

LISTING_PUBLISHED = {
    "schema_version": "m2.listing.v1",
    "listing_id": "lst_pdf-export",
    "solution_id": "sol_pdf-export",
    "solution_version": "1.2.0",
    "inventory_type": "solution",
    "name": "PDF Export",
    "summary": "Branded PDF report generation",
    "category": "reporting",
    "keywords": ["pdf", "report", "branding"],
    "price": {"amount": 50, "currency": "m2cr", "model": "one_time"},
    "seller": "op_nasr",
    "evidence_summary": "12 installs, 4.6 avg rating",
    "tenant_visibility": ["*"],
    "stats": {"installs": 12, "proposals_shown": 20, "proposals_accepted": 12},
    "status": "published",
    "install_ref": "lst_pdf-export-v1.2.0",
}

LISTING_DELISTED = {
    **LISTING_PUBLISHED,
    "listing_id": "lst_old-tool",
    "name": "Old Tool",
    "status": "delisted",
    "install_ref": "lst_old-tool-v0.9.0",
}

SOLUTION_PUBLISHED = {"intent": "generate a branded PDF report from a data export"}


def _b64_contents(payload: dict) -> dict:
    encoded = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii")
    return {"content": encoded}


def _make_handlers():
    calls: list[tuple[str, str]] = []

    def registry_handler(request: httpx.Request) -> httpx.Response:
        calls.append(("registry", f"{request.method} {request.url.path}"))
        path = request.url.path

        if path == "/api/v1/repos/m2/market-registry/contents/listings":
            return httpx.Response(
                200,
                json=[
                    {"name": "lst_pdf-export", "type": "dir"},
                    {"name": "lst_old-tool", "type": "dir"},
                ],
            )
        listings_base = "/api/v1/repos/m2/market-registry/contents/listings"
        if path == f"{listings_base}/lst_pdf-export/listing.json":
            return httpx.Response(200, json=_b64_contents(LISTING_PUBLISHED))
        if path == f"{listings_base}/lst_pdf-export/solution.json":
            return httpx.Response(200, json=_b64_contents(SOLUTION_PUBLISHED))
        if path == f"{listings_base}/lst_old-tool/listing.json":
            return httpx.Response(200, json=_b64_contents(LISTING_DELISTED))
        return httpx.Response(404, json={"message": "not found"})

    def memory_handler(request: httpx.Request) -> httpx.Response:
        calls.append(("memory", f"{request.method} {request.url.path}"))
        if request.url.path == "/memory/search":
            # exists-check: nothing indexed yet
            return httpx.Response(200, json={"results": [], "routing": {}})
        if request.url.path == "/memory/store":
            return httpx.Response(200, json={"id": "mem-1", "status": "stored"})
        return httpx.Response(404, json={"message": "not found"})

    return calls, registry_handler, memory_handler


def _exercise_reindex_flow():
    calls, registry_handler, memory_handler = _make_handlers()

    registry = RegistryClient(
        "https://git.machinemachine.ai",
        "frg_test",
        "m2/market-registry",
        "main",
        transport=httpx.MockTransport(registry_handler),
    )
    memory = MemoryClient(
        "https://memory.machinemachine.ai",
        "mem_test",
        "market:catalog",
        partition_path="/partitions/{partition}",
        document_path="/partitions/{partition}/documents/{doc_id}",
        transport=httpx.MockTransport(memory_handler),
    )
    with SCHEMA_PATH.open() as f:
        validator = Draft202012Validator(json.load(f))

    counts = run(registry, memory, validator)

    assert counts == {
        "scanned": 2,
        "invalid": 0,
        "not_published": 1,
        "upserted": 1,
    }, counts

    memory_calls = [c for kind, c in calls if kind == "memory"]
    # Real memory-api surface: drop is a no-op (store/search only); upsert is
    # exists-check (search) then store — for the one published listing only.
    assert memory_calls == [
        "POST /memory/search",
        "POST /memory/store",
    ], memory_calls

    registry_calls = [c for kind, c in calls if kind == "registry"]
    assert any("lst_pdf-export/solution.json" in c for c in registry_calls)
    assert not any("lst_old-tool/solution.json" in c for c in registry_calls)

    return counts, memory_calls


def test_reindex_flow_upserts_published_only():
    _exercise_reindex_flow()


if __name__ == "__main__":
    counts, memory_calls = _exercise_reindex_flow()
    print("OK: reindex counts ->", counts)
    print("OK: memory-api calls ->", memory_calls)