326 lines
11 KiB
Python
326 lines
11 KiB
Python
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import override_settings
|
|
|
|
from apps.knowledge.documents import process_knowledge_document
|
|
from apps.knowledge.models import (
|
|
GovernedAsset,
|
|
KnowledgeBase,
|
|
KnowledgeChunk,
|
|
KnowledgeDocument,
|
|
KnowledgeItem,
|
|
)
|
|
from apps.tickets.materials import MaterialScanError
|
|
|
|
|
|
def test_create_and_list_knowledge_base(admin_client, db):
|
|
response = admin_client.post(
|
|
"/api/knowledge/bases",
|
|
data={
|
|
"name": "税务采集知识库",
|
|
"description": "税局采集、日志和常见问题",
|
|
"scope": "数据采集",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["name"] == "税务采集知识库"
|
|
assert payload["enabled"] is True
|
|
|
|
response = admin_client.get("/api/knowledge/bases")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[0]["name"] == "税务采集知识库"
|
|
|
|
|
|
def test_create_knowledge_item_always_creates_governed_draft(admin_client, db):
|
|
base = KnowledgeBase.objects.create(name="税务采集知识库", scope="数据采集")
|
|
|
|
response = admin_client.post(
|
|
"/api/knowledge/items",
|
|
data={
|
|
"knowledge_base_id": base.id,
|
|
"title": "税局采集进度查询",
|
|
"problem": "客户询问采集进度长时间不更新",
|
|
"applicable_scope": "生产环境税局采集",
|
|
"error_code": "",
|
|
"root_cause": "税局服务响应不稳定",
|
|
"troubleshooting_steps": "查询 Python 日志并确认采集任务状态",
|
|
"solution": "告知客户税局不稳定，任务仍在采集中，稍后复查。",
|
|
"risk_note": "不要重复触发采集任务。",
|
|
"status": "published",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
item = KnowledgeItem.objects.get(id=response.json()["id"])
|
|
assert item.status == "draft"
|
|
assert KnowledgeChunk.objects.filter(item=item).count() == 0
|
|
assert GovernedAsset.objects.filter(
|
|
asset_type="knowledge_item",
|
|
object_id=item.id,
|
|
versions__status="draft",
|
|
).exists()
|
|
|
|
|
|
def test_create_knowledge_item_rejects_unknown_category(admin_client, db):
|
|
base = KnowledgeBase.objects.create(name="税务采集知识库", scope="数据采集")
|
|
|
|
response = admin_client.post(
|
|
"/api/knowledge/items",
|
|
data={
|
|
"knowledge_base_id": base.id,
|
|
"title": "未知分类知识",
|
|
"problem": "分类编码不存在",
|
|
"category_code": "unknown_category",
|
|
"solution": "不应写入知识库",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "知识适用分类不存在或已停用"
|
|
assert not KnowledgeItem.objects.filter(title="未知分类知识").exists()
|
|
|
|
|
|
def test_ingest_document_text_creates_document_and_chunks(admin_client, db):
|
|
base = KnowledgeBase.objects.create(name="日志知识库", scope="技术支持")
|
|
|
|
response = admin_client.post(
|
|
"/api/knowledge/documents",
|
|
data={
|
|
"knowledge_base_id": base.id,
|
|
"file_name": "tax-log.md",
|
|
"file_type": "markdown",
|
|
"content": "税局接口不稳定时，采集任务会停留在处理中。需要查询 Python 日志确认。",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
document = KnowledgeDocument.objects.get(id=response.json()["id"])
|
|
assert document.parse_status == "success"
|
|
assert KnowledgeChunk.objects.filter(document=document).count() >= 1
|
|
|
|
|
|
def test_upload_document_preserves_original_and_indexes_text(
|
|
admin_client,
|
|
db,
|
|
monkeypatch,
|
|
):
|
|
base = KnowledgeBase.objects.create(name="文件知识库", scope="技术支持")
|
|
objects = {}
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.put_object",
|
|
lambda path, raw, mime_type: objects.update({path: raw}),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.get_object",
|
|
lambda path: objects[path],
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.api.dispatch_knowledge_document_processing",
|
|
lambda document_id: process_knowledge_document(document_id),
|
|
)
|
|
|
|
response = admin_client.post(
|
|
f"/api/knowledge/documents/upload?knowledge_base_id={base.id}",
|
|
data={
|
|
"file": SimpleUploadedFile(
|
|
"support.log",
|
|
"接口超时，请检查连接池。".encode(),
|
|
content_type="text/plain",
|
|
)
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
document = KnowledgeDocument.objects.get(public_id=payload["public_id"])
|
|
assert document.scan_status == "clean"
|
|
assert document.parse_status == "success"
|
|
assert document.storage_path in objects
|
|
assert document.content_text == "接口超时，请检查连接池。"
|
|
assert document.sha256
|
|
assert document.file_size == len(objects[document.storage_path])
|
|
assert document.processing_attempts == 1
|
|
assert document.indexed_at is not None
|
|
assert document.chunks.count() >= 1
|
|
assert payload["has_original"] is True
|
|
|
|
listed = admin_client.get(f"/api/knowledge/documents?knowledge_base_id={base.id}")
|
|
assert listed.status_code == 200
|
|
assert listed.json()[0]["public_id"] == str(document.public_id)
|
|
|
|
|
|
def test_upload_document_rejects_disguised_file(admin_client, db):
|
|
base = KnowledgeBase.objects.create(name="文件知识库")
|
|
|
|
response = admin_client.post(
|
|
f"/api/knowledge/documents/upload?knowledge_base_id={base.id}",
|
|
data={
|
|
"file": SimpleUploadedFile(
|
|
"fake.pdf",
|
|
b"not a pdf",
|
|
content_type="application/pdf",
|
|
)
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 415
|
|
assert not KnowledgeDocument.objects.exists()
|
|
|
|
|
|
def test_document_scan_failure_keeps_original_and_can_retry(
|
|
admin_client,
|
|
db,
|
|
monkeypatch,
|
|
):
|
|
base = KnowledgeBase.objects.create(name="文件知识库")
|
|
objects = {}
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.put_object",
|
|
lambda path, raw, mime_type: objects.update({path: raw}),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.get_object",
|
|
lambda path: objects[path],
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.scan_bytes",
|
|
lambda raw: (_ for _ in ()).throw(MaterialScanError("扫描服务不可用")),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.api.dispatch_knowledge_document_processing",
|
|
lambda document_id: process_knowledge_document(document_id),
|
|
)
|
|
response = admin_client.post(
|
|
f"/api/knowledge/documents/upload?knowledge_base_id={base.id}",
|
|
data={
|
|
"file": SimpleUploadedFile(
|
|
"support.txt",
|
|
"重试后可解析".encode(),
|
|
content_type="text/plain",
|
|
)
|
|
},
|
|
)
|
|
document = KnowledgeDocument.objects.get(public_id=response.json()["public_id"])
|
|
|
|
assert response.status_code == 200
|
|
assert document.scan_status == "failed"
|
|
assert document.parse_status == "failed"
|
|
assert document.storage_path in objects
|
|
assert document.chunks.count() == 0
|
|
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.scan_bytes",
|
|
lambda raw: {"status": "clean", "engine": "test"},
|
|
)
|
|
retried = admin_client.post(f"/api/knowledge/documents/{document.public_id}/retry")
|
|
|
|
assert retried.status_code == 200
|
|
document.refresh_from_db()
|
|
assert document.parse_status == "success"
|
|
assert document.processing_attempts == 2
|
|
assert document.chunks.count() >= 1
|
|
process_knowledge_document(document.id)
|
|
document.refresh_from_db()
|
|
assert document.processing_attempts == 2
|
|
|
|
|
|
def test_upload_document_storage_failure_does_not_create_record(
|
|
admin_client,
|
|
db,
|
|
monkeypatch,
|
|
):
|
|
base = KnowledgeBase.objects.create(name="文件知识库")
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.documents.put_object",
|
|
lambda *args: (_ for _ in ()).throw(OSError("storage down")),
|
|
)
|
|
|
|
response = admin_client.post(
|
|
f"/api/knowledge/documents/upload?knowledge_base_id={base.id}",
|
|
data={
|
|
"file": SimpleUploadedFile(
|
|
"support.txt",
|
|
b"content",
|
|
content_type="text/plain",
|
|
)
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert not KnowledgeDocument.objects.exists()
|
|
|
|
|
|
def test_download_uploaded_document_returns_original(
|
|
admin_client,
|
|
db,
|
|
monkeypatch,
|
|
):
|
|
base = KnowledgeBase.objects.create(name="文件知识库")
|
|
document = KnowledgeDocument.objects.create(
|
|
knowledge_base=base,
|
|
file_name="支持手册.txt",
|
|
file_type="text",
|
|
extension="txt",
|
|
mime_type="text/plain",
|
|
file_size=7,
|
|
sha256="a" * 64,
|
|
storage_path="knowledge-documents/manual.txt",
|
|
scan_status="clean",
|
|
parse_status="success",
|
|
)
|
|
monkeypatch.setattr("apps.knowledge.api.get_object", lambda path: b"content")
|
|
|
|
response = admin_client.get(f"/api/knowledge/documents/{document.public_id}/download")
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == b"content"
|
|
assert response["Content-Type"] == "text/plain"
|
|
assert "attachment" in response["Content-Disposition"]
|
|
|
|
|
|
def test_download_infected_document_is_blocked(admin_client, db, monkeypatch):
|
|
base = KnowledgeBase.objects.create(name="文件知识库")
|
|
document = KnowledgeDocument.objects.create(
|
|
knowledge_base=base,
|
|
file_name="可疑文件.txt",
|
|
storage_path="knowledge-documents/infected.txt",
|
|
scan_status="infected",
|
|
parse_status="failed",
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.knowledge.api.get_object",
|
|
lambda path: (_ for _ in ()).throw(AssertionError("隔离文件不应从对象存储读取")),
|
|
)
|
|
|
|
response = admin_client.get(f"/api/knowledge/documents/{document.public_id}/download")
|
|
|
|
assert response.status_code == 423
|
|
|
|
|
|
@override_settings(KNOWLEDGE_EMBEDDING_ALLOW_HASH_FALLBACK=False)
|
|
def test_knowledge_draft_creation_does_not_require_embedding_deployment(admin_client, db):
|
|
base = KnowledgeBase.objects.create(name="正式知识库")
|
|
|
|
response = admin_client.post(
|
|
"/api/knowledge/items",
|
|
data={
|
|
"knowledge_base_id": base.id,
|
|
"title": "订单超时",
|
|
"problem": "订单接口超时",
|
|
"solution": "检查下游连接池",
|
|
"status": "published",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
item = KnowledgeItem.objects.get(title="订单超时")
|
|
assert item.status == "draft"
|
|
assert item.chunks.count() == 0
|