278 lines
9.1 KiB
Python
278 lines
9.1 KiB
Python
import io
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from PIL import Image
|
|
|
|
from apps.tickets.materials import MaterialParseError, scan_bytes
|
|
from apps.tickets.models import TicketDraftMaterial
|
|
from apps.tickets.tests.test_ticket_drafts import create_ticket_agent
|
|
|
|
|
|
def _create_draft(admin_client):
|
|
create_ticket_agent(capability="vision_chat", approved=True)
|
|
response = admin_client.post(
|
|
"/api/ticket-drafts",
|
|
data={"source_text": "材料解析测试"},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 200
|
|
return response.json()
|
|
|
|
|
|
def _docx_bytes():
|
|
from docx import Document
|
|
|
|
output = io.BytesIO()
|
|
document = Document()
|
|
document.add_paragraph("采集任务从上午十点起没有更新")
|
|
document.save(output)
|
|
return output.getvalue()
|
|
|
|
|
|
def _xlsx_bytes():
|
|
from openpyxl import Workbook
|
|
|
|
output = io.BytesIO()
|
|
workbook = Workbook()
|
|
workbook.active.append(["任务编号", "问题"])
|
|
workbook.active.append(["JOB-1024", "采集任务未更新"])
|
|
workbook.save(output)
|
|
workbook.close()
|
|
return output.getvalue()
|
|
|
|
|
|
def _pptx_bytes():
|
|
from pptx import Presentation
|
|
|
|
output = io.BytesIO()
|
|
presentation = Presentation()
|
|
slide = presentation.slides.add_slide(presentation.slide_layouts[5])
|
|
slide.shapes.title.text = "生产采集任务未更新"
|
|
presentation.save(output)
|
|
return output.getvalue()
|
|
|
|
|
|
def _pdf_bytes():
|
|
from pypdf import PdfWriter
|
|
|
|
output = io.BytesIO()
|
|
writer = PdfWriter()
|
|
writer.add_blank_page(width=72, height=72)
|
|
writer.write(output)
|
|
return output.getvalue()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("file_name", "content_type", "content", "material_type"),
|
|
[
|
|
("chat.txt", "text/plain", b"job failed with error 500", "text"),
|
|
("service.log", "text/x-log", b"2026-07-25 ERROR timeout", "log"),
|
|
("report.pdf", "application/pdf", _pdf_bytes(), "document"),
|
|
(
|
|
"report.docx",
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
_docx_bytes(),
|
|
"document",
|
|
),
|
|
(
|
|
"report.xlsx",
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
_xlsx_bytes(),
|
|
"spreadsheet",
|
|
),
|
|
(
|
|
"report.pptx",
|
|
"application/vnd.openxmlformats-officedocument.presentationml.presentation",
|
|
_pptx_bytes(),
|
|
"presentation",
|
|
),
|
|
],
|
|
ids=["txt", "log", "pdf", "docx", "xlsx", "pptx"],
|
|
)
|
|
def test_material_upload_scans_and_parses_supported_documents(
|
|
admin_client,
|
|
db,
|
|
monkeypatch,
|
|
file_name,
|
|
content_type,
|
|
content,
|
|
material_type,
|
|
):
|
|
created = _create_draft(admin_client)
|
|
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
|
|
|
|
response = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile(file_name, content, content_type=content_type)},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
material = response.json()["materials"][0]
|
|
assert material["material_type"] == material_type
|
|
assert material["scan_status"] == "clean"
|
|
assert material["parse_status"] == "parsed"
|
|
assert TicketDraftMaterial.objects.get(public_id=material["id"]).storage_path
|
|
|
|
|
|
def test_material_rejects_disguised_extension_and_mime(admin_client, db, monkeypatch):
|
|
created = _create_draft(admin_client)
|
|
stored = []
|
|
monkeypatch.setattr(
|
|
"apps.tickets.draft_api.put_object",
|
|
lambda *args: stored.append(args),
|
|
)
|
|
|
|
signature = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("fake.pdf", b"MZ executable", content_type="application/pdf")},
|
|
)
|
|
output = io.BytesIO()
|
|
Image.new("RGB", (8, 8), color="red").save(output, format="PNG")
|
|
mime = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("proof.png", output.getvalue(), content_type="text/plain")},
|
|
)
|
|
|
|
assert signature.status_code == 415
|
|
assert mime.status_code == 415
|
|
assert stored == []
|
|
assert TicketDraftMaterial.objects.count() == 0
|
|
|
|
|
|
def test_infected_material_is_retained_and_not_parsed(admin_client, db, monkeypatch):
|
|
created = _create_draft(admin_client)
|
|
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
|
|
monkeypatch.setattr(
|
|
"apps.tickets.materials.scan_bytes",
|
|
lambda _raw: {
|
|
"status": "infected",
|
|
"engine": "clamav",
|
|
"signature": "Eicar-Test-Signature",
|
|
},
|
|
)
|
|
|
|
response = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("sample.txt", b"EICAR sample", content_type="text/plain")},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
material = response.json()["materials"][0]
|
|
assert material["scan_status"] == "infected"
|
|
assert material["parse_status"] == "blocked"
|
|
stored = TicketDraftMaterial.objects.get(public_id=material["id"])
|
|
assert stored.storage_path
|
|
assert stored.parsed_text == ""
|
|
|
|
|
|
def test_clamav_instream_response_marks_eicar_as_infected(monkeypatch, settings):
|
|
class FakeConnection:
|
|
def __init__(self):
|
|
self.sent = []
|
|
self.responses = [b"stream: Eicar-Test-Signature FOUND\0"]
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return None
|
|
|
|
def sendall(self, data):
|
|
self.sent.append(data)
|
|
|
|
def recv(self, _size):
|
|
return self.responses.pop(0) if self.responses else b""
|
|
|
|
connection = FakeConnection()
|
|
settings.TICKET_MATERIAL_SCANNER_BACKEND = "clamav"
|
|
monkeypatch.setattr(
|
|
"apps.tickets.materials.socket.create_connection",
|
|
lambda *_args, **_kwargs: connection,
|
|
)
|
|
|
|
result = scan_bytes(b"standard antivirus test payload")
|
|
|
|
assert result["status"] == "infected"
|
|
assert result["signature"] == "Eicar-Test-Signature"
|
|
assert connection.sent[0] == b"zINSTREAM\0"
|
|
|
|
|
|
def test_parse_failure_isolated_to_material_and_keeps_original(admin_client, db, monkeypatch):
|
|
created = _create_draft(admin_client)
|
|
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
|
|
monkeypatch.setattr(
|
|
"apps.tickets.materials.parse_material",
|
|
lambda *_args: (_ for _ in ()).throw(MaterialParseError("文档损坏")),
|
|
)
|
|
|
|
response = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("chat.txt", b"valid text", content_type="text/plain")},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
material = response.json()["materials"][0]
|
|
assert material["scan_status"] == "clean"
|
|
assert material["parse_status"] == "failed"
|
|
assert material["parse_error"] == "文档损坏"
|
|
assert TicketDraftMaterial.objects.filter(public_id=material["id"]).exists()
|
|
|
|
|
|
def test_material_limits_are_server_controlled(admin_client, db, monkeypatch, settings):
|
|
settings.TICKET_MATERIAL_MAX_FILE_BYTES = 4
|
|
settings.TICKET_MATERIAL_MAX_FILES = 1
|
|
created = _create_draft(admin_client)
|
|
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
|
|
|
|
oversized = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("large.txt", b"12345", content_type="text/plain")},
|
|
)
|
|
first = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("one.txt", b"1234", content_type="text/plain")},
|
|
)
|
|
second = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{"file": SimpleUploadedFile("two.txt", b"12", content_type="text/plain")},
|
|
)
|
|
|
|
assert oversized.status_code == 413
|
|
assert first.status_code == 200
|
|
assert second.status_code == 413
|
|
|
|
|
|
def test_parsed_document_can_start_intake_without_text(admin_client, db, monkeypatch):
|
|
create_ticket_agent(capability="text_chat")
|
|
created = admin_client.post(
|
|
"/api/ticket-drafts",
|
|
data={"source_text": ""},
|
|
content_type="application/json",
|
|
).json()
|
|
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
|
|
uploaded = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/materials",
|
|
{
|
|
"file": SimpleUploadedFile(
|
|
"service.log",
|
|
b"2026-07-25 ERROR timeout",
|
|
content_type="text/plain",
|
|
)
|
|
},
|
|
).json()
|
|
monkeypatch.setattr(
|
|
"apps.tickets.draft_api.execute_ticket_draft_analysis.delay",
|
|
lambda *_args: SimpleNamespace(id="task-1"),
|
|
)
|
|
|
|
response = admin_client.post(
|
|
f"/api/ticket-drafts/{created['public_id']}/analyze",
|
|
data={"expected_revision": uploaded["revision"]},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "analyzing"
|