|
|
@@ -0,0 +1,229 @@
|
|
|
+"""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"<root/>"),
|
|
|
+ ("word/document.xml", b"<broken"),
|
|
|
+ ("[Content_Types].xml", b"<Types/>"),
|
|
|
+ ("_rels/.rels", b"<Relationships/>"),
|
|
|
+ ("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()
|
