321 lines
14 KiB
Python
321 lines
14 KiB
Python
"""Versioned chunks remain evidence snapshots; publication uses real PostgreSQL locks."""
|
|
|
|
import hashlib
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import func, select, text
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.orm import Session
|
|
from test_retrieval import add_item
|
|
|
|
from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion
|
|
from app.services import indexing
|
|
from app.services.indexing import locked_state
|
|
|
|
|
|
def build(client, **payload):
|
|
response = client.post("/api/indexes", json=payload)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def publish(client, version, generation=0):
|
|
return client.post(
|
|
f"/api/indexes/{version['id']}/publish", json={"expected_generation": generation}
|
|
)
|
|
|
|
|
|
def chunks(client, version):
|
|
return client.get(f"/api/indexes/{version['id']}/chunks?page_size=100").json()["items"]
|
|
|
|
|
|
def test_build_preserves_original_offsets_content_hash_and_provenance(client):
|
|
content = ("Ａ字符\n那拉提🙂占位资料。" * 37).strip()
|
|
item = add_item(client, content=content)
|
|
first = build(client, chunk_size=100, overlap=20)
|
|
second = build(client, chunk_size=100, overlap=20)
|
|
assert first["status"] == "ready" and first["item_count"] == 1
|
|
assert first["corpus_fingerprint"] == second["corpus_fingerprint"]
|
|
assert first["config"] == second["config"] and first["id"] != second["id"]
|
|
actual = chunks(client, first)
|
|
assert len(actual) == first["chunk_count"]
|
|
assert [c["ordinal"] for c in actual] == list(range(len(actual)))
|
|
assert actual[0]["start_offset"] == 0 and actual[-1]["end_offset"] == len(content)
|
|
restored = [""] * len(content)
|
|
for chunk in actual:
|
|
start, end = chunk["start_offset"], chunk["end_offset"]
|
|
assert chunk["text"] == content[start:end]
|
|
assert chunk["text_sha256"] == hashlib.sha256(chunk["text"].encode()).hexdigest()
|
|
assert chunk["provenance"]["knowledge_version"] == item["version"]
|
|
assert chunk["provenance"]["source_url"] == "https://example.com/"
|
|
restored[start:end] = list(chunk["text"])
|
|
assert "".join(restored) == content
|
|
assert [(c["text"], c["start_offset"], c["end_offset"]) for c in actual] == [
|
|
(c["text"], c["start_offset"], c["end_offset"]) for c in chunks(client, second)
|
|
]
|
|
state = client.get("/api/indexes/active").json()
|
|
assert state["generation"] == 0 and state["active_version"] is None and not state["is_current"]
|
|
|
|
|
|
def test_only_valid_reviewed_evidence_is_chunked(client):
|
|
good = add_item(client)
|
|
add_item(client, status="pending")
|
|
add_item(client, status="rejected")
|
|
removed = add_item(client)
|
|
assert (
|
|
client.delete(
|
|
f"/api/knowledge/{removed['id']}?expected_version={removed['version']}"
|
|
).status_code
|
|
== 204
|
|
)
|
|
version = build(client)
|
|
assert version["item_count"] == 1
|
|
assert {c["knowledge_id"] for c in chunks(client, version)} == {good["id"]}
|
|
|
|
|
|
def test_publication_is_atomic_optimistic_and_audited(client):
|
|
add_item(client)
|
|
first, second = build(client), build(client)
|
|
result = publish(client, first)
|
|
assert result.status_code == 200 and result.json()["generation"] == 1
|
|
assert result.json()["is_current"]
|
|
assert publish(client, second).status_code == 409
|
|
assert publish(client, second, 1).json()["generation"] == 2
|
|
assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "superseded"
|
|
assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "published"
|
|
# Same active version with the current generation is an idempotent no-op.
|
|
assert publish(client, second, 2).json()["generation"] == 2
|
|
history = client.get("/api/indexes/publications").json()
|
|
assert history["total"] == 2
|
|
assert history["items"][0]["previous_version_id"] == first["id"]
|
|
assert history["items"][0]["version_id"] == second["id"]
|
|
assert client.get("/api/indexes/active").json()["active_version"]["id"] == second["id"]
|
|
|
|
|
|
def test_catalog_mutation_marks_active_stale_and_rejects_old_build(client):
|
|
item = add_item(client)
|
|
version = build(client)
|
|
assert publish(client, version).status_code == 200
|
|
old = chunks(client, version)
|
|
updated = client.patch(
|
|
f"/api/knowledge/{item['id']}",
|
|
json={"expected_version": item["version"], "content": "修订的占位正文"},
|
|
)
|
|
assert updated.status_code == 200
|
|
active = client.get("/api/indexes/active").json()
|
|
assert not active["is_current"] and active["generation"] == 1
|
|
assert active["active_version"]["id"] == version["id"]
|
|
assert chunks(client, version) == old
|
|
assert publish(client, version, 1).status_code == 409
|
|
replacement = build(client)
|
|
assert replacement["item_count"] == 0 # revised body is pending, not approved evidence
|
|
assert publish(client, replacement, 1).json()["is_current"]
|
|
assert chunks(client, replacement) == []
|
|
|
|
|
|
def test_failed_catalog_write_does_not_invalidate_current_version(client):
|
|
item = add_item(client)
|
|
version = build(client)
|
|
publish(client, version)
|
|
before = client.get("/api/indexes/active").json()
|
|
assert (
|
|
client.patch(
|
|
f"/api/knowledge/{item['id']}", json={"expected_version": 999, "content": "占位"}
|
|
).status_code
|
|
== 409
|
|
)
|
|
assert client.post("/api/knowledge", json={"title": "", "content": "占位"}).status_code == 422
|
|
assert client.get("/api/indexes/active").json() == before
|
|
|
|
|
|
def test_source_and_document_mutations_preserve_old_provenance(client):
|
|
src = client.post(
|
|
"/api/sources", json={"name": "原始占位来源", "source_url": "https://example.com/"}
|
|
).json()
|
|
doc = client.post(
|
|
"/api/documents",
|
|
json={
|
|
"source_id": src["id"],
|
|
"title": "占位文档",
|
|
"content": "占位正文",
|
|
"collected_at": "2026-10-16T00:00:00Z",
|
|
},
|
|
).json()
|
|
add_item(client, document_id=doc["id"])
|
|
version = build(client)
|
|
publish(client, version)
|
|
snapshot = chunks(client, version)
|
|
assert snapshot[0]["provenance"]["document_version"] == doc["version"]
|
|
assert snapshot[0]["provenance"]["source_name"] == src["name"]
|
|
assert (
|
|
client.patch(
|
|
f"/api/documents/{doc['id']}", json={"expected_version": 1, "content": "文档新正文"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert not client.get("/api/indexes/active").json()["is_current"]
|
|
empty = build(client)
|
|
publish(client, empty, 1)
|
|
assert (
|
|
client.patch(
|
|
f"/api/sources/{src['id']}", json={"expected_version": 1, "name": "新来源名"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert not client.get("/api/indexes/active").json()["is_current"]
|
|
assert chunks(client, version) == snapshot
|
|
assert publish(client, version, 2).status_code == 409
|
|
|
|
|
|
def test_fingerprint_recheck_rejects_out_of_band_catalog_changes(client, migrated_database):
|
|
item = add_item(client)
|
|
version = build(client)
|
|
publish(client, version)
|
|
with migrated_database.begin() as c:
|
|
c.execute(
|
|
text("UPDATE knowledge_items SET content=:body WHERE id=:id"),
|
|
{"body": "直接修改的占位正文", "id": item["id"]},
|
|
)
|
|
assert not client.get("/api/indexes/active").json()["is_current"]
|
|
assert publish(client, version, 1).status_code == 409
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["chunk_budget", "corpus_budget", "database_write"])
|
|
def test_failed_build_keeps_previous_publication_and_leaves_no_partial_rows(
|
|
client, migrated_database, monkeypatch, failure
|
|
):
|
|
add_item(client, content="占位正文" * 100)
|
|
previous = build(client)
|
|
publish(client, previous)
|
|
if failure == "chunk_budget":
|
|
monkeypatch.setattr(indexing, "MAX_CHUNKS", 1)
|
|
response = client.post("/api/indexes", json={"chunk_size": 100, "overlap": 0})
|
|
assert response.status_code == 503
|
|
elif failure == "corpus_budget":
|
|
from app.services import retrieval
|
|
|
|
monkeypatch.setattr(retrieval, "MAX_ITEMS", 0)
|
|
assert client.post("/api/indexes", json={}).status_code == 503
|
|
monkeypatch.undo()
|
|
else:
|
|
# Real PostgreSQL constraint failure during chunk insert; the new version rolls back too.
|
|
with migrated_database.begin() as c:
|
|
c.execute(
|
|
text("ALTER TABLE chunks ADD CONSTRAINT test_reject_chunk CHECK (false) NOT VALID")
|
|
)
|
|
assert client.post("/api/indexes", json={}).status_code == 409
|
|
with Session(migrated_database) as db:
|
|
assert db.scalar(select(func.count()).select_from(IndexVersion)) == 1
|
|
assert db.scalar(select(func.count()).select_from(Chunk)) == previous["chunk_count"]
|
|
assert db.get(IndexState, 1).active_version_id == previous["id"]
|
|
assert db.scalar(select(func.count()).select_from(IndexPublication)) == 1
|
|
assert client.get("/api/indexes/active").json()["active_version"]["id"] == previous["id"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"chunk_size": 99},
|
|
{"chunk_size": 2001},
|
|
{"overlap": -1},
|
|
{"chunk_size": 100, "overlap": 51},
|
|
{"unexpected": 1},
|
|
],
|
|
)
|
|
def test_build_validation_has_no_side_effects(client, payload):
|
|
assert client.post("/api/indexes", json=payload).status_code == 422
|
|
assert client.get("/api/indexes").json()["total"] == 0
|
|
|
|
|
|
def test_index_management_permissions_csrf_and_pagination(client, anonymous_client):
|
|
# Same TestClient fixture is authenticated as admin here; first verify authorized validation.
|
|
version = build(client)
|
|
assert client.post("/api/indexes", json={}, headers={"X-CSRF-Token": ""}).status_code == 403
|
|
assert client.get("/api/indexes?page_size=101").status_code == 422
|
|
assert client.get(f"/api/indexes/{uuid4()}").status_code == 404
|
|
assert client.get(f"/api/indexes/{uuid4()}/chunks").status_code == 404
|
|
assert client.get("/api/indexes/invalid").status_code == 422
|
|
assert client.get(f"/api/indexes/{version['id']}/chunks?page=2").json()["items"] == []
|
|
client.cookies.clear()
|
|
for method, path in [
|
|
("GET", "/api/indexes"),
|
|
("GET", "/api/indexes/active"),
|
|
("GET", "/api/indexes/publications"),
|
|
("POST", "/api/indexes"),
|
|
("GET", f"/api/indexes/{version['id']}/chunks"),
|
|
("POST", f"/api/indexes/{version['id']}/publish"),
|
|
]:
|
|
assert client.request(method, path, json={}).status_code == 401
|
|
assert (
|
|
client.post(
|
|
"/api/auth/register", json={"username": "normal_user", "password": "test-password-123"}
|
|
).status_code
|
|
== 201
|
|
)
|
|
auth = client.post(
|
|
"/api/auth/login", json={"username": "normal_user", "password": "test-password-123"}
|
|
).json()
|
|
client.headers["X-CSRF-Token"] = auth["csrf_token"]
|
|
assert client.post("/api/indexes", json={}).status_code == 403
|
|
assert client.get("/api/indexes/active").status_code == 403
|
|
|
|
|
|
def test_publication_lock_serializes_competing_publishers(client, migrated_database):
|
|
add_item(client)
|
|
first, second = build(client), build(client)
|
|
actor = first["created_by"]
|
|
with Session(migrated_database) as owner, Session(migrated_database) as competitor:
|
|
locked_state(owner)
|
|
competitor.execute(text("SET LOCAL lock_timeout = '100ms'"))
|
|
with pytest.raises(OperationalError) as blocked:
|
|
indexing.publish_index(competitor, UUID(second["id"]), 0, actor)
|
|
assert blocked.value.orig.sqlstate == "55P03"
|
|
competitor.rollback()
|
|
assert indexing.publish_index(owner, UUID(first["id"]), 0, actor).generation == 1
|
|
with pytest.raises(HTTPException) as stale:
|
|
indexing.publish_index(competitor, UUID(second["id"]), 0, actor)
|
|
assert stale.value.status_code == 409
|
|
competitor.rollback()
|
|
assert client.get("/api/indexes/publications").json()["total"] == 1
|
|
assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"]
|
|
|
|
|
|
def test_catalog_writes_use_the_publication_lock(client, migrated_database):
|
|
# The guard must run before a catalog mutation, even for creation with no parent row.
|
|
with Session(migrated_database) as owner:
|
|
locked_state(owner)
|
|
from app.db.session import get_db
|
|
|
|
factory = client.app.dependency_overrides[get_db]
|
|
|
|
def short_timeout():
|
|
for db in factory():
|
|
db.execute(text("SET LOCAL lock_timeout = '100ms'"))
|
|
yield db
|
|
|
|
client.app.dependency_overrides[get_db] = short_timeout
|
|
response = client.post("/api/knowledge", json={"title": "占位", "content": "占位"})
|
|
assert response.status_code == 503
|
|
client.app.dependency_overrides[get_db] = factory
|
|
owner.rollback()
|
|
assert client.get("/api/knowledge").json() == []
|
|
|
|
|
|
def test_publication_failure_rolls_back_old_status_pointer_and_audit(client, migrated_database):
|
|
add_item(client)
|
|
first, second = build(client), build(client)
|
|
publish(client, first)
|
|
with migrated_database.begin() as c:
|
|
c.execute(
|
|
text(
|
|
"ALTER TABLE index_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID"
|
|
)
|
|
)
|
|
assert publish(client, second, 1).status_code == 409
|
|
assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"]
|
|
assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "published"
|
|
assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "ready"
|
|
assert client.get("/api/indexes/publications").json()["total"] == 1
|