270 lines
10 KiB
Python
270 lines
10 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.orm.exc import StaleDataError
|
|
|
|
from app.models import KnowledgeItem
|
|
|
|
COLLECTED = "2026-10-15T09:00:00+08:00"
|
|
|
|
|
|
def create(client, path, data):
|
|
response = client.post("/api/" + path, json=data)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def source(client):
|
|
return create(client, "sources", {"name": "示例来源", "source_url": "https://example.com/"})
|
|
|
|
|
|
def document(client, source_id):
|
|
return create(
|
|
client,
|
|
"documents",
|
|
{
|
|
"source_id": source_id,
|
|
"title": "示例文档",
|
|
"content": "仅用于自动化测试的占位内容",
|
|
"collected_at": COLLECTED,
|
|
},
|
|
)
|
|
|
|
|
|
def knowledge(client, **overrides):
|
|
return create(
|
|
client,
|
|
"knowledge",
|
|
{
|
|
"title": "示例知识",
|
|
"content": "测试资料，不代表文旅事实",
|
|
"source_url": "https://example.com/",
|
|
"collected_at": COLLECTED,
|
|
**overrides,
|
|
},
|
|
)
|
|
|
|
|
|
def review(client, item, status="verified"):
|
|
return client.post(
|
|
f"/api/knowledge/{item['id']}/review",
|
|
json={
|
|
"expected_version": item["version"],
|
|
"verification_status": status,
|
|
"reviewer": "test-reviewer",
|
|
"verification_notes": "仅验证审核流程，非真实核验结论",
|
|
},
|
|
)
|
|
|
|
|
|
def test_health_docs_and_stub(client):
|
|
assert client.get("/api/health").json()["status"] == "ok"
|
|
assert client.get("/api/ready").json()["database"] == "connected"
|
|
assert client.get("/docs").status_code == 200
|
|
assert "/api/knowledge/{item_id}/review" in client.get("/openapi.json").json()["paths"]
|
|
assert client.post("/api/chat", json={"question": "测试"}).json()["status"] == "stub"
|
|
|
|
|
|
def test_legacy_create_and_missing_evidence(client):
|
|
item = create(client, "knowledge", {"title": "旧版调用", "content": "示例"})
|
|
assert item["verification_status"] == "pending"
|
|
assert item["collected_at"] is None
|
|
assert isinstance(client.get("/api/knowledge").json(), list)
|
|
assert review(client, item).status_code == 422
|
|
assert client.get(f"/api/knowledge/{item['id']}/history").json()["total"] == 1
|
|
|
|
|
|
def test_optional_empty_url_matches_existing_frontend(client):
|
|
item = create(client, "knowledge", {"title": "示例", "content": "示例", "source_url": ""})
|
|
assert item["source_url"] is None
|
|
|
|
|
|
def test_reassign_document_resolves_new_source(client):
|
|
first = source(client)
|
|
first_doc = document(client, first["id"])
|
|
second = source(client)
|
|
second_doc = document(client, second["id"])
|
|
item = knowledge(client, document_id=first_doc["id"])
|
|
response = client.patch(
|
|
f"/api/knowledge/{item['id']}",
|
|
json={
|
|
"expected_version": item["version"],
|
|
"document_id": second_doc["id"],
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["source_id"] == second["id"]
|
|
|
|
|
|
def test_browser_preflight_and_document_source_lists(client):
|
|
response = client.options(
|
|
"/api/knowledge",
|
|
headers={
|
|
"Origin": "http://localhost:5173",
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "content-type",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.headers["access-control-allow-origin"] == "http://localhost:5173"
|
|
src = source(client)
|
|
doc = document(client, src["id"])
|
|
assert client.get("/api/sources?q=示例").json()["items"][0]["id"] == src["id"]
|
|
result = client.get(f"/api/documents?source_id={src['id']}&q=占位").json()
|
|
assert result["total"] == 1 and result["items"][0]["id"] == doc["id"]
|
|
|
|
|
|
def test_review_edit_and_history(client):
|
|
item = knowledge(client)
|
|
verified = review(client, item)
|
|
assert verified.status_code == 200
|
|
verified = verified.json()
|
|
assert verified["version"] == 2
|
|
assert datetime.fromisoformat(verified["reviewed_at"]).tzinfo is not None
|
|
assert review(client, item).status_code == 409
|
|
response = client.patch(
|
|
f"/api/knowledge/{item['id']}",
|
|
json={
|
|
"expected_version": verified["version"],
|
|
"content": "修订后的示例正文",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
updated = response.json()
|
|
assert updated["verification_status"] == "pending"
|
|
assert updated["reviewer"] is None and updated["reviewed_at"] is None
|
|
events = client.get(f"/api/knowledge/{item['id']}/history").json()
|
|
assert [event["action"] for event in events["items"]] == ["update", "review", "create"]
|
|
assert events["items"][1]["snapshot"]["item"]["verification_status"] == "verified"
|
|
|
|
|
|
def test_source_document_changes_invalidate_evidence(client):
|
|
src = source(client)
|
|
doc = document(client, src["id"])
|
|
item = knowledge(client, source_id=src["id"], document_id=doc["id"])
|
|
assert review(client, item).status_code == 200
|
|
response = client.patch(
|
|
f"/api/documents/{doc['id']}",
|
|
json={
|
|
"expected_version": doc["version"],
|
|
"content": "修订后的文档",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
current = client.get(f"/api/knowledge/{item['id']}").json()
|
|
assert current["verification_status"] == "pending"
|
|
assert review(client, current).status_code == 200
|
|
response = client.patch(
|
|
f"/api/sources/{src['id']}",
|
|
json={
|
|
"expected_version": src["version"],
|
|
"source_url": "https://example.com/new",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
current = client.get(f"/api/knowledge/{item['id']}").json()
|
|
assert current["verification_status"] == "pending"
|
|
assert current["source_url"] == "https://example.com/new"
|
|
events = client.get(f"/api/knowledge/{item['id']}/history").json()["items"]
|
|
reviewed = next(e for e in events if e["action"] == "review")
|
|
assert reviewed["snapshot"]["source"]["source_url"] == "https://example.com/"
|
|
assert reviewed["snapshot"]["document"]["content"] == "修订后的文档"
|
|
|
|
|
|
def test_delete_preserves_history_and_guards_references(client):
|
|
src = source(client)
|
|
doc = document(client, src["id"])
|
|
item = knowledge(client, document_id=doc["id"])
|
|
assert item["source_id"] == src["id"]
|
|
assert client.delete(f"/api/sources/{src['id']}?expected_version=1").status_code == 409
|
|
assert client.delete(f"/api/documents/{doc['id']}?expected_version=1").status_code == 409
|
|
assert client.delete(f"/api/knowledge/{item['id']}?expected_version=2").status_code == 409
|
|
assert client.delete(f"/api/knowledge/{item['id']}?expected_version=1").status_code == 204
|
|
assert client.get(f"/api/knowledge/{item['id']}").status_code == 404
|
|
assert client.get("/api/knowledge").json() == []
|
|
assert (
|
|
client.get(f"/api/knowledge/{item['id']}/history").json()["items"][0]["action"] == "delete"
|
|
)
|
|
assert client.delete(f"/api/documents/{doc['id']}?expected_version=1").status_code == 204
|
|
assert client.delete(f"/api/sources/{src['id']}?expected_version=1").status_code == 204
|
|
assert client.get("/api/sources").json()["total"] == 0
|
|
|
|
|
|
def test_filter_pagination_and_literal_search(client):
|
|
knowledge(client, title="示例%资料", category="交通出行")
|
|
other = knowledge(client, title="其他资料", category="民宿服务")
|
|
assert review(client, other, "rejected").status_code == 200
|
|
page = client.get("/api/knowledge/page?page_size=1").json()
|
|
assert page["total"] == 2 and len(page["items"]) == 1
|
|
assert page["items"][0]["id"] == other["id"]
|
|
assert client.get("/api/knowledge/page?page=3&page_size=1").json()["items"] == []
|
|
assert client.get("/api/knowledge/page?q=%25").json()["total"] == 1
|
|
assert len(client.get("/api/knowledge?category=交通出行").json()) == 1
|
|
assert client.get("/api/knowledge/page?verification_status=verified").json()["total"] == 0
|
|
assert client.get("/api/knowledge/page?verification_status=rejected").json()["total"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"patch",
|
|
[
|
|
{"title": " "},
|
|
{"title": None},
|
|
{"content": ""},
|
|
{"category": ""},
|
|
{"collected_at": "2026-10-15T09:00:00"},
|
|
{"source_url": "javascript:alert(1)"},
|
|
{"verification_status": "verified"},
|
|
{},
|
|
],
|
|
)
|
|
def test_invalid_update_cannot_mutate_or_audit(client, patch):
|
|
item = knowledge(client)
|
|
response = client.patch(f"/api/knowledge/{item['id']}", json={"expected_version": 1, **patch})
|
|
assert response.status_code == 422
|
|
assert client.get(f"/api/knowledge/{item['id']}").json()["version"] == 1
|
|
assert client.get(f"/api/knowledge/{item['id']}/history").json()["total"] == 1
|
|
|
|
|
|
def test_invalid_relationship_and_query(client):
|
|
src = source(client)
|
|
doc = document(client, src["id"])
|
|
second = source(client)
|
|
response = client.post(
|
|
"/api/knowledge",
|
|
json={
|
|
"title": "示例",
|
|
"content": "示例",
|
|
"source_id": second["id"],
|
|
"document_id": doc["id"],
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
assert (
|
|
client.post(
|
|
"/api/knowledge", json={"title": "示例", "content": "示例", "document_id": 9999}
|
|
).status_code
|
|
== 404
|
|
)
|
|
for path in ["knowledge", "sources", "documents"]:
|
|
assert client.get(f"/api/{path}/9999").status_code == 404
|
|
assert client.get(f"/api/{path}/9999/history").status_code == 404
|
|
assert client.get("/api/knowledge/page?page_size=101").status_code == 422
|
|
assert client.get("/api/knowledge/page?verification_status=unknown").status_code == 422
|
|
assert client.get("/api/knowledge/page?page=0").status_code == 422
|
|
|
|
|
|
def test_optimistic_lock_prevents_concurrent_overwrite(client, migrated_database):
|
|
item = knowledge(client)
|
|
with Session(migrated_database) as first, Session(migrated_database) as second:
|
|
first_item = first.scalar(select(KnowledgeItem).where(KnowledgeItem.id == item["id"]))
|
|
second_item = second.get(KnowledgeItem, item["id"])
|
|
first_item.content = "先提交的内容"
|
|
first.commit()
|
|
second_item.content = "过期编辑"
|
|
with pytest.raises(StaleDataError):
|
|
second.commit()
|
|
second.rollback()
|
|
assert client.get(f"/api/knowledge/{item['id']}").json()["content"] == "先提交的内容"
|