tests/test_archive_api.py

"""Synthetic contract and security tests; no model, token, corpus or DB needed."""

import asyncio
import importlib.util
import sys
import threading
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient

from api.archive_security import ArchiveSecurityMiddleware, safe_validation_errors
from api.v1.archive import router
from core.config import Settings
from core.inference_runner import BoundedInferenceRunner, InferenceBusy, InferenceTimedOut
from models.translator import TextTranslator
from services.archive_inference import ArchiveInferenceService

SYNTHETIC_KEY = "test-only-key-not-an-operator-secret-0000"


class FakeEmbedder:
    model_name = "synthetic-model"

    def encode_documents(self, texts, batch_size, max_length):
        return {"dense_vecs": [[0.5, 0.5] for _ in texts],
                "lexical_weights": [{"123": 0.25, "567": 0.75} for _ in texts]}


class ImmediateRunner:
    async def run(self, operation):
        return operation()


def client_for(key=SYNTHETIC_KEY, embedder=None, translator=None):
    config = Settings(_env_file=None, ARCHIVE_INFERENCE_API_KEY=key,
                      TRANSLATION_PROVIDER="disabled")
    app = FastAPI()
    app.add_middleware(ArchiveSecurityMiddleware, configuration=config)
    app.add_exception_handler(RequestValidationError, safe_validation_errors)
    app.state.inference_runner = ImmediateRunner()
    app.state.archive_inference_service = ArchiveInferenceService(
        embedder or FakeEmbedder(), translator or TextTranslator(config))
    app.include_router(router, prefix="/api/v1")
    app.get("/api/v1/health/ping")(lambda: {"message": "pong"})
    return TestClient(app)


def machine_post(client, route, payload, key=SYNTHETIC_KEY):
    return client.post("/api/v1/archive/" + route, json=payload,
                       headers={"Authorization": "Bearer " + key})


def test_public_health_remains_public_and_archive_fails_closed():
    client = client_for("")
    assert client.get("/api/v1/health/ping").status_code == 200
    assert machine_post(client, "embeddings", {"texts": ["Synthetic text"]}).status_code == 503


@pytest.mark.parametrize("key", ["", "wrong", "x" * 600])
def test_invalid_machine_authentication_is_rejected(key):
    response = machine_post(client_for(), "embeddings", {"texts": ["Synthetic text"]}, key)
    assert response.status_code == 401
    assert response.headers["cache-control"] == "no-store"


def test_dense_and_sparse_contract_uses_original_token_ids():
    response = machine_post(client_for(), "embeddings", {"texts": ["Synthetic text", "Other text"]})
    assert response.status_code == 200
    body = response.json()
    assert body["model"] == "synthetic-model" and body["dimension"] == 2
    assert len(body["embeddings"]) == 2
    assert body["embeddings"][0]["sparse"] == {"123": 0.25, "567": 0.75}
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.parametrize("payload", [
    {"texts": []}, {"texts": [" "]}, {"texts": ["x" * 8001]},
    {"texts": ["x"] * 65}, {"texts": ["한" * 8000] * 3},
    {"texts": ["Synthetic text"], "unknown": "synthetic-private-input"},
    {"texts": ["Synthetic text"], "model": "arbitrary-model-to-download"},
])
def test_request_limits_and_model_allowlist(payload):
    response = machine_post(client_for(), "embeddings", payload)
    assert response.status_code == 422
    assert "synthetic-private-input" not in response.text
    assert "arbitrary-model-to-download" not in response.text


def test_body_cap_enforced_before_parsing():
    response = client_for().post("/api/v1/archive/embeddings", content="x" * 512001,
                                headers={"Authorization": "Bearer " + SYNTHETIC_KEY})
    assert response.status_code == 413


def test_disabled_translation_does_not_download_or_use_external_api():
    response = machine_post(client_for(), "translate", {
        "texts": ["Synthetic translation text"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 503


def test_same_locale_is_explicit_identity():
    response = machine_post(client_for(), "translate", {
        "texts": ["Synthetic text"], "source_locale": "en", "target_locale": "en"})
    assert response.json() == {"model": "identity", "translations": ["Synthetic text"]}


def test_translation_contract_preserves_batch_order():
    translator = SimpleNamespace(translate=lambda texts, source, target:
                                 ("synthetic-translator", ["합성 " + text for text in texts]))
    response = machine_post(client_for(translator=translator), "translate", {
        "texts": ["A", "B"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 200
    assert response.json() == {"model": "synthetic-translator", "translations": ["합성 A", "합성 B"]}


def test_provider_exception_does_not_leak_source_or_credential():
    def fail(*_):
        raise RuntimeError("synthetic-private-input synthetic-private-key")
    translator = SimpleNamespace(translate=fail)
    response = machine_post(client_for(translator=translator), "translate", {
        "texts": ["synthetic-private-input"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 502
    assert "synthetic-private" not in response.text


def test_invalid_provider_response_rejected():
    translator = SimpleNamespace(translate=lambda *_: ("synthetic", []))
    response = machine_post(client_for(translator=translator), "translate", {
        "texts": ["Synthetic text"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 502


def test_marian_license_gate_and_revision_do_not_load_model():
    config = Settings(_env_file=None, TRANSLATION_PROVIDER="marian",
                      TRANSLATION_LICENSE_ACKNOWLEDGED=False)
    response = machine_post(client_for(translator=TextTranslator(config)), "translate", {
        "texts": ["Synthetic text"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 503
    config.TRANSLATION_LICENSE_ACKNOWLEDGED = True
    config.TRANSLATION_EN_KO_REVISION = "main"
    response = machine_post(client_for(translator=TextTranslator(config)), "translate", {
        "texts": ["Synthetic text"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 503


def test_openai_rejects_insecure_endpoint_before_transport():
    config = Settings(_env_file=None, TRANSLATION_PROVIDER="openai",
                      TRANSLATION_OPENAI_BASE_URL="http://localhost:9999/v1",
                      TRANSLATION_OPENAI_API_KEY="synthetic", TRANSLATION_OPENAI_MODEL="synthetic")
    response = machine_post(client_for(translator=TextTranslator(config)), "translate", {
        "texts": ["Synthetic text"], "source_locale": "en", "target_locale": "ko"})
    assert response.status_code == 503


def test_timeout_retains_capacity_until_cpu_work_actually_finishes():
    async def scenario():
        runner = BoundedInferenceRunner(outstanding=1, timeout_seconds=0.02)
        finished = threading.Event()
        try:
            with pytest.raises(InferenceTimedOut):
                await runner.run(lambda: finished.wait(2))
            with pytest.raises(InferenceBusy):
                await runner.run(lambda: "would queue hidden work")
            finished.set()
            await asyncio.sleep(0.04)
            assert await runner.run(lambda: "ready") == "ready"
        finally:
            finished.set()
            runner.close()
    asyncio.run(scenario())


def test_long_input_model_rejection_is_sanitized():
    class RejectingEmbedder(FakeEmbedder):
        def encode_documents(self, *_args, **_kwargs):
            raise ValueError("synthetic-private-input exceeds token limit")
    response = machine_post(client_for(embedder=RejectingEmbedder()), "embeddings", {
        "texts": ["synthetic-private-input"]})
    assert response.status_code == 422
    assert "synthetic-private-input" not in response.text


def test_original_query_embedder_does_not_retokenize_numeric_ids(monkeypatch):
    monkeypatch.setitem(sys.modules, "FlagEmbedding", SimpleNamespace(BGEM3FlagModel=object))
    monkeypatch.setitem(sys.modules, "torch", SimpleNamespace())
    path = Path(__file__).resolve().parents[1] / "models" / "embedder.py"
    specification = importlib.util.spec_from_file_location("synthetic_embedder_test", path)
    module = importlib.util.module_from_spec(specification)
    specification.loader.exec_module(module)
    embedder = module.TextEmbedder.__new__(module.TextEmbedder)
    def unexpected_conversion(_):
        raise AssertionError("Numeric vocabulary IDs must not be re-tokenized")
    embedder.model = SimpleNamespace(
        encode=lambda *_args, **_kwargs: {
            "dense_vecs": SimpleNamespace(tolist=lambda: [0.1, 0.9]),
            "lexical_weights": {"123": 0.25, "567": 0.75}},
        tokenizer=SimpleNamespace(convert_tokens_to_ids=unexpected_conversion))
    result = embedder.encode_query("Synthetic query")
    assert result.sparse_indices == [123, 567]
    assert result.sparse_values == [0.25, 0.75]


def test_existing_public_json_and_demo_routes_remain_accessible(monkeypatch):
    class FakeSearch:
        def search(self, query, **_kwargs):
            return {"query": query, "results": [], "latency_ms": 1}
    monkeypatch.setitem(sys.modules, "services.search_service",
                        SimpleNamespace(HybridSearchService=FakeSearch))
    from api.v1.search import router as search_router
    app = FastAPI()
    app.state.search_service = FakeSearch()
    app.state.inference_runner = ImmediateRunner()
    app.add_middleware(ArchiveSecurityMiddleware,
                       configuration=Settings(_env_file=None, ARCHIVE_INFERENCE_API_KEY=""))
    app.include_router(search_router, prefix="/api/v1")
    client = TestClient(app)
    response = client.post("/api/v1/search/", json={"query": "Synthetic public query"})
    assert response.status_code == 200
    assert response.json() == {"query": "Synthetic public query", "results": [], "latency_ms": 1}
    assert client.get("/api/v1/search/demo").status_code == 200
    response = client.post("/api/v1/search/demo", data={"query": "Synthetic public query"})
    assert response.status_code == 200


def test_embedding_revision_is_resolved_before_flagembedding_load(monkeypatch):
    calls = {}
    def snapshot(**options):
        calls["snapshot"] = options
        return "synthetic-fixed-snapshot"
    def model(location, **options):
        calls["model_location"] = location
        calls["model_options"] = options
        return object()
    monkeypatch.setitem(sys.modules, "huggingface_hub", SimpleNamespace(snapshot_download=snapshot))
    monkeypatch.setitem(sys.modules, "FlagEmbedding", SimpleNamespace(BGEM3FlagModel=model))
    monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(
        cuda=SimpleNamespace(is_available=lambda: False),
        backends=SimpleNamespace(mps=SimpleNamespace(is_available=lambda: False))))
    path = Path(__file__).resolve().parents[1] / "models" / "embedder.py"
    specification = importlib.util.spec_from_file_location("synthetic_pinned_embedder_test", path)
    module = importlib.util.module_from_spec(specification)
    specification.loader.exec_module(module)
    monkeypatch.setattr(module.TextEmbedder, "_warmup", lambda *_: None)
    revision = "1" * 40
    embedder = module.TextEmbedder("BAAI/bge-m3", use_fp16=True, revision=revision)
    assert calls["snapshot"]["revision"] == revision
    assert calls["model_location"] == "synthetic-fixed-snapshot"
    assert calls["model_options"]["trust_remote_code"] is False
    assert calls["model_options"]["use_fp16"] is False
    assert embedder.compatibility_id == "BAAI/bge-m3@" + revision + ":fp32:unit-cls-v1"