381 lines
16 KiB
Python
381 lines
16 KiB
Python
"""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"]
|