"""Upload preflight uses real synthetic documents, without LLM or callbacks.""" import hashlib from io import BytesIO import json from pathlib import Path import tempfile import unittest from unittest.mock import patch import zipfile import pymupdf from fastapi.testclient import TestClient from fastapi import HTTPException import http_api from scripts.tests.test_api import _docx_bytes, _pdf_bytes class FileValidationTests(unittest.TestCase): def setUp(self): temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) self.root = Path(temp.name) self.log = self.root / "requests.log" for target, value in [ ("request_logging.LOG_PATH", self.log), ("http_api.tempfile.tempdir", temp.name), ]: patcher = patch(target, value) patcher.start() self.addCleanup(patcher.stop) self.client = TestClient(http_api.app) def post(self, name, content, mime="application/octet-stream"): jobs = dict(http_api._JOBS) with patch.object(http_api._JOB_EXECUTOR, "submit") as submit: result = self.client.post("/api/v1/files/validate", files={"file": (name, content, mime)}) submit.assert_not_called() self.assertEqual(http_api._JOBS, jobs, "检测接口不得创建任务") self.assertEqual(list(self.root.glob("proposa-validate-*")), [], "检测副本必须清理") return result def test_valid_docx_and_text_pdf_metadata_and_audit(self): for name, content, kind in [("文件.DOCX", _docx_bytes(), "docx"), ("文件.PDF", _pdf_bytes(), "pdf")]: with self.subTest(kind=kind): response = self.post(name, content, "image/png") self.assertEqual(response.status_code, 200, response.text) data = response.json() self.assertTrue(data["valid"]) self.assertEqual(data["file_type"], kind) self.assertEqual(data["filename"], name) self.assertEqual(data["size"], len(content)) self.assertEqual(data["sha256"], hashlib.sha256(content).hexdigest()) self.assertIs(data["is_text_pdf"], True if kind == "pdf" else None) records = [json.loads(line) for line in self.log.read_text(encoding="utf-8").splitlines()] request = next(record for record in records if record["event"] == "request") self.assertEqual(request["body"][0]["name"], "file") self.assertNotIn("tender text", self.log.read_text(encoding="utf-8")) def test_invalid_empty_disguised_and_unsupported(self): for name, content in [ ("a.doc", _docx_bytes()), ("a.txt", b"text"), ("a.pdf", b""), ("a.docx", b""), ("a.pdf", b"broken"), ("a.docx", b"broken"), ("a.pdf", _docx_bytes()), ("a.docx", _pdf_bytes()), ]: with self.subTest(name=name, size=len(content)): response = self.post(name, content) self.assertEqual(response.status_code, 400, response.text) self.assertIn("detail", response.json()) def test_pdf_blank_image_only_and_encrypted(self): with pymupdf.open() as doc: page = doc.new_page() pixmap = pymupdf.Pixmap(pymupdf.csRGB, (0, 0, 20, 20), False) pixmap.clear_with(255) page.insert_image(page.rect, pixmap=pixmap) scanned = doc.tobytes() with pymupdf.open(stream=_pdf_bytes(), filetype="pdf") as doc: encrypted = doc.tobytes(encryption=pymupdf.PDF_ENCRYPT_AES_256, owner_pw="owner", user_pw="user") for content in [_pdf_bytes(""), scanned, encrypted]: self.assertEqual(self.post("a.pdf", content).status_code, 400) def test_mixed_pdf_uses_existing_at_least_one_text_page_rule(self): with pymupdf.open() as doc: doc.new_page() doc.new_page().insert_text((72, 72), "extractable text") content = doc.tobytes() self.assertEqual(self.post("mixed.pdf", content).status_code, 200) def test_docx_container_xml_content_type_and_relationships(self): for part, replacement in [ ("word/document.xml", b""), ("word/document.xml", b""), ("_rels/.rels", b""), ("word/document.xml", None), ]: with self.subTest(part=part, replacement=replacement): output = BytesIO() with zipfile.ZipFile(BytesIO(_docx_bytes())) as source, zipfile.ZipFile(output, "w") as target: for name in source.namelist(): if name == part: if replacement is not None: target.writestr(name, replacement) else: target.writestr(name, source.read(name)) self.assertEqual(self.post("a.docx", output.getvalue()).status_code, 400) def test_size_limit_and_exact_boundary(self): content = _pdf_bytes() with patch.object(http_api, "MAX_UPLOAD_BYTES", len(content)): self.assertEqual(self.post("a.pdf", content).status_code, 200) with patch.object(http_api, "MAX_UPLOAD_BYTES", len(content) - 1): self.assertEqual(self.post("a.pdf", content).status_code, 413) def test_request_contract_and_openapi(self): url = "/api/v1/files/validate" self.assertEqual(self.client.post(url, content="text", headers={"Content-Type": "text/plain"}).status_code, 415) for files in [ [("other", ("a.pdf", _pdf_bytes()))], [("file", (None, "not an upload"))], [("file", ("a.pdf", _pdf_bytes())), ("file", ("b.pdf", _pdf_bytes()))], [("file", ("a.pdf", _pdf_bytes())), ("other", ("b.pdf", _pdf_bytes()))], ]: self.assertEqual(self.client.post(url, files=files).status_code, 422) schema = self.client.get("/openapi.json").json()["paths"][url]["post"] self.assertIn("multipart/form-data", schema["requestBody"]["content"]) self.assertEqual(schema["requestBody"]["content"]["application/json"]["schema"]["required"], ["file_path", "file_role", "is_absolute"]) def post_path(self, path, role="TENDER_FILE", absolute=False): jobs = dict(http_api._JOBS) with patch.object(http_api._JOB_EXECUTOR, "submit") as submit: response = self.client.post("/api/v1/files/validate", json={ "file_path": str(path), "file_role": role, "is_absolute": absolute, }) submit.assert_not_called() self.assertEqual(http_api._JOBS, jobs) self.assertEqual(list(self.root.glob("proposa-validate-*")), []) return response def test_json_tpc_roles_and_absolute_static_paths(self): for role in ["TENDER_FILE", "PROCUREMENT_FILE", "CLA_FILE"]: with self.subTest(role=role), patch.object(http_api, "TPC_DIR", str(self.root)): path = self.root / "files" / "a.pdf" path.parent.mkdir(exist_ok=True) content = _pdf_bytes() path.write_bytes(content) response = self.post_path("/static/files/a.pdf", role) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["resolved_path"], str(path.resolve())) self.assertEqual(response.json()["file_role"], role) self.assertEqual(path.read_bytes(), content) self.assertEqual(self.post_path(path.parent / "static" / path.name, role, True).status_code, 200) def test_json_reference_fallback_and_priority(self): first = self.root / "first" second = self.root / "second" first.mkdir() second.mkdir() content = _docx_bytes() with patch.object(http_api, "REFERENCE_DIR", str(first)), patch.object(http_api, "REFERENCE_DIR1", str(second)): response = self.post_path("static/missing.docx", "REFERENCE_BID") self.assertEqual(response.status_code, 400) expected = [str((directory / "missing.docx").resolve()) for directory in (first, second)] self.assertEqual(response.json()["candidate_paths"], expected) self.assertIsNone(response.json()["resolved_path"]) for path in expected: self.assertIn(path, response.json()["message"]) (second / "a.docx").write_bytes(content) response = self.post_path("static/a.docx", "REFERENCE_BID") self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["resolved_path"], str((second / "a.docx").resolve())) (first / "a.docx").write_bytes(b"broken") response = self.post_path("static/a.docx", "REFERENCE_BID") self.assertEqual(response.status_code, 400) self.assertIn(str((first / "a.docx").resolve()), response.json()["message"]) self.assertNotIn(str(second), response.json()["message"]) self.assertEqual(self.post_path(second / "a.docx", "REFERENCE_BID", True).status_code, 200) def test_json_all_file_failures_include_source_absolute_path(self): for name, content in [("missing.pdf", None), ("empty.pdf", b""), ("bad.pdf", b"broken"), ("scan.pdf", _pdf_bytes("")), ("bad.docx", b"broken"), ("wrong.txt", b"text")]: with self.subTest(name=name), patch.object(http_api, "TPC_DIR", str(self.root)): path = self.root / name if content is not None: path.write_bytes(content) response = self.post_path("static/" + name) self.assertEqual(response.status_code, 400, response.text) body = response.json() self.assertFalse(body["valid"]) self.assertIn(str(path.resolve()), body["message"]) self.assertEqual(body["resolved_path"], str(path.resolve())) self.assertNotIn("proposa-validate-", body["message"]) path = self.root / "large.pdf" path.write_bytes(_pdf_bytes()) with patch.object(http_api, "MAX_UPLOAD_BYTES", 1): response = self.post_path(path, absolute=True) self.assertEqual(response.status_code, 413) self.assertIn(str(path.resolve()), response.json()["message"]) with patch.object(http_api, "_save_local_path", side_effect=HTTPException(400, "无法读取")): response = self.post_path(path, absolute=True) self.assertIn(str(path.resolve()), response.json()["message"]) def test_json_invalid_parameters_and_unconfigured_roots(self): url = "/api/v1/files/validate" valid = {"file_path": "a.pdf", "file_role": "TENDER_FILE", "is_absolute": False} for body in [None, [], {}, {**valid, "file_path": 3}, {**valid, "file_path": " "}, {**valid, "file_role": "unknown"}, {**valid, "is_absolute": "false"}, {key: value for key, value in valid.items() if key != "is_absolute"}]: response = self.client.post(url, content=json.dumps(body), headers={"Content-Type": "application/json"}) self.assertEqual(response.status_code, 422, response.text) self.assertIn("message", response.json()) response = self.client.post(url, content="{", headers={"Content-Type": "application/json"}) self.assertEqual(response.status_code, 400) with patch.object(http_api, "TPC_DIR", ""), patch.object(http_api, "REFERENCE_DIR", ""), patch.object(http_api, "REFERENCE_DIR1", ""): for role in ("TENDER_FILE", "REFERENCE_BID"): response = self.post_path("a.pdf", role) self.assertEqual(response.status_code, 400) self.assertIn("未配置", response.json()["message"]) self.assertEqual(response.json()["candidate_paths"], []) if __name__ == "__main__": unittest.main()