"""Export historical protocol fixtures, not domain scores or actual model usage.""" import copy import hashlib import json from datetime import timedelta from uuid import uuid4 import pytest from sqlalchemy import event, func, select from sqlalchemy.orm import Session from sqlalchemy.orm.attributes import flag_modified from test_rag import prepare from test_retrieval import add_item, search from test_runtime import server from test_vectors import ToyProvider, text_version, vector_build, vector_publish from app.models.auth import User from app.models.catalog import utc_now from app.models.rag import RagContext from app.models.retrieval import RetrievalRun from app.schemas.rag import CompletionOptions from app.services import exports from app.services.completions import canonical_json from app.services.completions import request_body as legacy_body from app.services.embedding_runtime import get_embedding_provider PATH = "/api/retrieval/exports" def download(client, runs, **fields): return client.post(PATH, json={"run_ids": [r["run_id"] for r in runs], **fields}) def lines(response): assert response.status_code == 200, response.text assert response.content.endswith(b"\n") return [json.loads(line) for line in response.content.rstrip(b"\n").split(b"\n")] def chunk_run(client): add_item(client, content="alpha query 合成占位正文") text_version(client) response = client.post("/api/retrieval/chunks", json={"query": "alpha"}) assert response.status_code == 200, response.text return response.json() def test_mixed_modes_exact_snapshots_manifest_digest_and_optional_contexts( client, migrated_database ): add_item(client, content="alpha query\n合成测试正文") text_ = text_version(client) client.app.dependency_overrides[get_embedding_provider] = ToyProvider vector_publish(client, vector_build(client, text_)) runs = [] for path in ["", "/chunks", "/vector", "/hybrid"]: response = client.post("/api/retrieval" + path, json={"query": "alpha\nquery"}) assert response.status_code == 200, response.text runs.append(response.json()) context = prepare(client, runs[-1]["run_id"]).json() client.app.dependency_overrides.pop(get_embedding_provider) response = download(client, list(reversed(runs))) records = lines(response) manifest = records[0] assert manifest["schema_version"] == "nalati-retrieval-export-v1" assert manifest["counts"] == {"retrieval_runs": 4, "rag_contexts": 0} assert manifest["selection"] == { "run_ids": sorted(r["run_id"] for r in runs), "include_contexts": False, } assert manifest["snapshot_semantics"] == "stored_history_not_current_corpus" assert manifest["generation_measurements"] == { "status": "not_called", "usage": None, "cost": None, "currency": None, } data = response.content.split(b"\n", 1)[1] assert hashlib.sha256(data).hexdigest() == manifest["records_sha256"] assert response.headers["X-Records-SHA256"] == manifest["records_sha256"] assert response.headers["content-type"] == "application/x-ndjson; charset=utf-8" assert response.headers["content-disposition"] == ( f'attachment; filename="nalati-retrieval-{manifest["export_id"]}.jsonl"' ) assert response.headers["cache-control"] == "no-store" assert response.headers["x-content-type-options"] == "nosniff" assert [r["payload"] for r in records[1:]] == runs assert [r["ranking_unit"] for r in records[1:]] == ["knowledge_item", "chunk", "chunk", "chunk"] repeated = download(client, runs) assert repeated.content.split(b"\n", 1)[1] == data assert lines(repeated)[0]["export_id"] != manifest["export_id"] included = download(client, runs, include_contexts=True) all_records = lines(included) assert all_records[0]["counts"] == {"retrieval_runs": 4, "rag_contexts": 1} assert all_records[-1] == { "record_type": "rag_context", "id": context["id"], "payload": context, } for forbidden in [ b"created_by", b"password_hash", b"test_admin", b"csrf_token", b"nalati_session", ]: assert forbidden not in included.content with Session(migrated_database) as db: assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 4 assert db.scalar(select(func.count()).select_from(RagContext)) == 1 @pytest.mark.parametrize("profile", ["openai", "deepseek", "bailian", "compatible"]) def test_responses_and_preexisting_chat_snapshots_export_without_rewriting( client, migrated_database, profile ): run = chunk_run(client) new = prepare( client, run["run_id"], completion={"model": "contract-model", "provider_profile": profile} ).json() assert new["completion"]["protocol"] == "responses" assert "messages" not in new["request_body"] old = prepare(client, run["run_id"]).json() options = CompletionOptions(model="historical-contract-model") old["completion"] = options.model_dump(mode="json") old["request_body"] = legacy_body(old["query"], old["evidence"], options) old["request_sha256"] = hashlib.sha256(canonical_json(old["request_body"]).encode()).hexdigest() old["input_characters"] = sum(len(m["content"]) for m in old["request_body"]["messages"]) with Session(migrated_database) as db: row = db.get(RagContext, old["id"]) row.response = copy.deepcopy(old) db.commit() assert client.get(f"/api/rag/contexts/{old['id']}").json() == old assert client.get(f"/api/rag/contexts/{new['id']}").json() == new history = client.get("/api/rag/contexts").json()["items"] assert {x["id"]: x for x in history} == {old["id"]: old, new["id"]: new} first = download(client, [run], include_contexts=True) records = lines(first) snapshots = {x["id"]: x["payload"] for x in records if x["record_type"] == "rag_context"} assert snapshots == {old["id"]: old, new["id"]: new} assert "protocol" not in snapshots[old["id"]]["completion"] assert ( download(client, [run], include_contexts=True).content.split(b"\n", 1)[1] == first.content.split(b"\n", 1)[1] ) def test_missing_selection_is_atomic_and_no_evidence_is_exportable(client): run = search(client, query="没有语料的合成问题") assert run["status"] == "no_evidence" missing = download(client, [run, {"run_id": str(uuid4())}]) assert missing.status_code == 404 and "Content-Disposition" not in missing.headers assert "manifest" not in missing.text assert lines(download(client, [run]))[1]["payload"] == run @pytest.mark.parametrize( "payload", [ {}, {"run_ids": []}, {"run_ids": ["not-a-uuid"]}, {"run_ids": [str(uuid4()) for _ in range(101)]}, {"run_ids": [str(uuid4())], "include_contexts": "true"}, {"run_ids": [str(uuid4())], "api_key": "test-secret-not-echoed"}, ], ) def test_selection_validation_rejects_unsafe_or_unbounded_input(client, payload): response = client.post(PATH, json=payload) assert response.status_code == 422 assert "test-secret-not-echoed" not in response.text def test_duplicate_uuids_are_rejected_after_normalization(client): run = search(client) response = client.post(PATH, json={"run_ids": [run["run_id"], run["run_id"].upper()]}) assert response.status_code == 422 def test_admin_origin_csrf_and_no_auth_data_in_errors(anonymous_client, migrated_database): payload = {"run_ids": [str(uuid4())]} assert anonymous_client.post(PATH, json=payload).status_code == 401 credentials = {"username": "export_user", "password": "test-password-123"} assert anonymous_client.post("/api/auth/register", json=credentials).status_code == 201 login = anonymous_client.post("/api/auth/login", json=credentials).json() anonymous_client.headers["X-CSRF-Token"] = login["csrf_token"] assert anonymous_client.post(PATH, json=payload).status_code == 403 with Session(migrated_database) as db: user = db.scalar(select(User).where(User.username == credentials["username"])) user.role = "admin" db.commit() token = anonymous_client.headers.pop("X-CSRF-Token") assert anonymous_client.post(PATH, json=payload).status_code == 403 anonymous_client.headers["X-CSRF-Token"] = token assert ( anonymous_client.post( PATH, json=payload, headers={"Origin": "https://example.com"} ).status_code == 403 ) assert anonymous_client.post(PATH, json=payload).status_code == 404 def test_export_retains_history_after_catalog_deletion_and_context_is_no_generation(client): add_item(client, content="alpha query 合成正文") text_version(client) run = client.post("/api/retrieval/chunks", json={"query": "不存在的词"}).json() context = prepare(client, run["run_id"]).json() assert context["status"] == "insufficient_evidence" item = client.get("/api/knowledge").json()[0] assert ( client.delete(f"/api/knowledge/{item['id']}?expected_version={item['version']}").status_code == 204 ) result = lines(download(client, [run], include_contexts=True)) assert result[1]["payload"] == run and result[2]["payload"] == context assert result[2]["payload"]["model_called"] is False @pytest.mark.parametrize( "change", ["id", "mode", "query", "fingerprint", "created_at", "unknown", "score_type"] ) def test_corrupt_run_fails_without_partial_download_or_private_error( client, migrated_database, change ): add_item(client, content="alpha 合成正文") run = search(client, query="alpha") with Session(migrated_database) as db: row = db.get(RetrievalRun, run["run_id"]) raw = copy.deepcopy(row.response) if change == "id": raw["run_id"] = str(uuid4()) elif change == "mode": row.mode = "vector" elif change == "query": row.query = "private-test-data" elif change == "fingerprint": row.corpus_fingerprint = "f" * 64 elif change == "created_at": row.created_at += timedelta(seconds=1) elif change == "unknown": raw["api_key"] = "private-test-data" else: raw["results"][0]["score"] = "NaN" row.response = raw db.commit() response = download(client, [run]) assert response.status_code == 409 assert ( "Content-Disposition" not in response.headers and "private-test-data" not in response.text ) @pytest.mark.parametrize( "change", [ "request", "request_numeric", "hash", "link", "catalog_revision", "model_called", "creator_extra", ], ) def test_corrupt_context_fails_atomically(client, migrated_database, change): run = chunk_run(client) context = prepare(client, run["run_id"]).json() with Session(migrated_database) as db: row = db.get(RagContext, context["id"]) raw = copy.deepcopy(row.response) if change == "request": raw["request_body"]["api_key"] = "private-test-data" elif change == "request_numeric": # Python equality alone treats True == 1; the actual wire body/hash must not. raw["request_body"]["n"] = True elif change == "hash": raw["request_sha256"] = "f" * 64 elif change == "link": raw["retrieval_run_id"] = str(uuid4()) elif change == "catalog_revision": raw["catalog_revision"] += 1 elif change == "model_called": raw["model_called"] = True else: raw["created_by"] = "private-test-data" row.response = raw # SQLAlchemy's Python comparison otherwise considers JSON true equal to 1. flag_modified(row, "response") db.commit() if change == "request_numeric": db.refresh(row) assert row.response["request_body"]["n"] is True response = download(client, [run], include_contexts=True) assert response.status_code == 409 and "Content-Disposition" not in response.headers assert "private-test-data" not in response.text assert download(client, [run]).status_code == 200 @pytest.mark.parametrize("boundary", ["record", "manifest"]) def test_utf8_byte_cap_rejects_whole_file_including_manifest(client, monkeypatch, boundary): run = search(client, query="多字节中文合成问题") original = download(client, [run]) full_size = len(original.content) record_size = len(original.content.split(b"\n", 1)[1]) monkeypatch.setattr( exports, "MAX_BYTES", record_size - 1 if boundary == "record" else full_size - 1 ) response = download(client, [run]) assert response.status_code == 413 and "Content-Disposition" not in response.headers assert client.get(f"/api/retrieval/runs/{run['run_id']}").json() == run monkeypatch.setattr(exports, "MAX_BYTES", full_size) assert len(download(client, [run]).content) == full_size def copied_context(context, run_id, actor): raw = copy.deepcopy(context) raw.update(id=str(uuid4()), created_at=utc_now().isoformat()) return RagContext( id=raw["id"], retrieval_run_id=run_id, created_by=actor, response=raw, created_at=raw["created_at"], ) def test_context_capacity_rejects_instead_of_truncating(client, migrated_database): run = chunk_run(client) context = prepare(client, run["run_id"]).json() with Session(migrated_database) as db: actor = db.scalar(select(RagContext.created_by)) db.add_all(copied_context(context, run["run_id"], actor) for _ in range(100)) db.commit() assert download(client, [run]).status_code == 200 response = download(client, [run], include_contexts=True) assert response.status_code == 413 and "Content-Disposition" not in response.headers with Session(migrated_database) as db: assert db.scalar(select(func.count()).select_from(RagContext)) == 101 def test_one_statement_snapshot_excludes_concurrent_context_insert(client, migrated_database): run = chunk_run(client) context = prepare(client, run["run_id"]).json() with Session(migrated_database) as db: # Separate fixture actor avoids the authentication row lock held by the export request. actor = User(username="snapshot_fixture", password_hash="not-a-login-hash", role="user") db.add(actor) db.commit() actor_id = actor.id inserted = False def concurrent_insert(connection, cursor, statement, parameters, context_, many): nonlocal inserted if not inserted and "UNION ALL" in statement and "rag_contexts" in statement: inserted = True with Session(migrated_database) as db: db.add(copied_context(context, run["run_id"], actor_id)) db.commit() event.listen(migrated_database, "after_cursor_execute", concurrent_insert) try: result = lines(download(client, [run], include_contexts=True)) finally: event.remove(migrated_database, "after_cursor_execute", concurrent_insert) assert inserted and result[0]["counts"]["rag_contexts"] == 1 assert lines(download(client, [run], include_contexts=True))[0]["counts"]["rag_contexts"] == 2 def test_http_export_records_and_digest_survive_process_restart(migrated_database): with server(migrated_database) as client: run = search(client, query="HTTP导出合成问题") first = download(client, [run]) assert first.status_code == 200 with server(migrated_database) as client: second = download(client, [run]) assert second.status_code == 200 assert second.content.split(b"\n", 1)[1] == first.content.split(b"\n", 1)[1] assert second.headers["X-Records-SHA256"] == first.headers["X-Records-SHA256"]