from io import BytesIO import json import shutil import subprocess import tempfile import unittest from pathlib import Path from unittest.mock import patch from docx import Document from fastapi.testclient import TestClient import pymupdf import api_worker import http_api def _write_docx(path: Path, text: str = "ok") -> None: path.parent.mkdir(parents=True, exist_ok=True) doc = Document() doc.add_paragraph(text) doc.save(path) def _docx_bytes(text: str = "document") -> bytes: stream = BytesIO() doc = Document() doc.add_paragraph(text) doc.save(stream) return stream.getvalue() def _pdf_bytes(text: str = "tender text") -> bytes: document = pymupdf.open() page = document.new_page() if text: page.insert_text((72, 72), text) content = document.tobytes() document.close() return content def _valid_files(tender_name: str = "招标文件.pdf") -> dict: return { "TENDER_FILE": (tender_name, _pdf_bytes(), "application/pdf"), "PROCUREMENT_FILE": ( "采购需求.docx", _docx_bytes("procurement"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ), "REFERENCE_BID": ( "参考投书.docx", _docx_bytes("reference"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ), } def _write_json_input_files(root: Path): tender = root / "招标文件.pdf" procurement = root / "采购需求.docx" reference = root / "参考投书.docx" tender.write_bytes(_pdf_bytes()) procurement.write_bytes(_docx_bytes("procurement")) reference.write_bytes(_docx_bytes("reference")) return tender, procurement, reference class RecordingExecutor: def __init__(self): self.submitted = [] def submit(self, function, *args): self.submitted.append((function, args)) return object() class ApiWorkerTests(unittest.TestCase): def test_worker_runs_test_step1_through_test_step6_with_fixed_local_assets(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) tender = root / "tender.pdf" procurement = root / "procurement.docx" reference = root / "reference.docx" step6 = root / "step6.docx" for path in (tender, procurement, reference): path.write_bytes(b"input") calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) script_name = Path(command[1]).name if script_name == "test_step6.py": _write_docx(step6, "step6") return subprocess.CompletedProcess(command, 0) with patch("api_worker.subprocess.run", side_effect=fake_run): result = api_worker.run_test_pipeline( tender, procurement, reference, step6 ) self.assertEqual(result, step6) self.assertEqual( [Path(call[0][1]).name for call in calls], [ "test_step1.py", "test_step2.py", "test_step3.py", "test_step4.py", "test_step5.py", "test_step6.py", ], ) env = calls[0][1]["env"] self.assertEqual( Path(env["PROPOSA_COMPANY_INFO_DIR"]), api_worker.COMPANY_INFO_DIR.resolve(), ) self.assertEqual( Path(env["PROPOSA_TEMPLATE_PATH"]), api_worker.TEMPLATE_PATH.resolve(), ) self.assertEqual( Path(env["PROPOSA_STEP1_ITEMS_FILE"]), (step6.parent / "step1_extracted_items.pkl").resolve(), ) progress_path = step6.parent / api_worker.PROGRESS_FILE_NAME self.assertTrue(progress_path.is_file()) progress = json.loads(progress_path.read_text(encoding="utf-8")) self.assertEqual(progress["current_step"], 6) self.assertEqual(progress["status"], "completed") self.assertEqual(progress["completed_steps"], [1, 2, 3, 4, 5, 6]) class FinalReviewApiTests(unittest.TestCase): def setUp(self): with http_api._JOBS_LOCK: http_api._JOBS.clear() def _post( self, files=None, callback_url="127.0.0.1:9999/callback", txb_id="dms-tender-001", ): return TestClient(http_api.app).post( "/api/v1/final-review", data={"CALLBACK_URL": callback_url, "txbId": txb_id}, files=files or _valid_files(), ) def test_valid_request_returns_202_before_pipeline_runs(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) work_root = root / "jobs" executor = RecordingExecutor() with patch.object(http_api, "API_WORK_ROOT", work_root), patch.object( http_api, "API_WORK_SETTING", "output/api_jobs" ), patch.object(http_api, "_JOB_EXECUTOR", executor), patch.object( http_api, "_run_pipeline_worker" ) as worker: response = self._post( _valid_files("上海市群众艺术馆物业管理服务采购项目招标文件.pdf") ) self.assertEqual(response.status_code, 202) body = response.json() self.assertEqual(body["status"], "processing") self.assertEqual(body["txbId"], "dms-tender-001") self.assertEqual(body["message"], "三份文件校验通过，已开始处理") self.assertEqual( body["resultPath"], f"output/api_jobs/{body['request_id']}/" "上海市群众艺术馆物业管理服务采购项目招标文件.docx", ) self.assertEqual( body["output_path"], f"output/api_jobs/{body['request_id']}/" "上海市群众艺术馆物业管理服务采购项目招标文件.docx", ) self.assertEqual(len(executor.submitted), 1) worker.assert_not_called() shutil.rmtree(work_root, ignore_errors=True) def test_background_success_publishes_file_and_sends_callback(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) executor = RecordingExecutor() callback_payloads = [] def fake_worker(_t, _p, _r, step6, _log): _write_docx(step6, "final review") with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker", side_effect=fake_worker ), patch.object( http_api, "_send_callback", side_effect=lambda _url, payload: callback_payloads.append(payload), ): response = self._post(_valid_files("项目A.pdf")) function, args = executor.submitted[0] job = args[0] function(*args) request_id = response.json()["request_id"] status = TestClient(http_api.app).get( f"/api/v1/jobs/{request_id}" ) download = TestClient(http_api.app).get( f"/api/v1/jobs/{request_id}/file" ) self.assertEqual(status.status_code, 200) self.assertEqual(status.json()["status"], "completed") self.assertTrue(status.json()["callback_delivered"]) self.assertEqual(callback_payloads[0]["status"], "completed") self.assertEqual(callback_payloads[0]["txbId"], "dms-tender-001") self.assertEqual( callback_payloads[0]["files"]["TENDER_FILE"]["filename"], "项目A.pdf", ) self.assertTrue( callback_payloads[0]["resultPath"].endswith( "项目A.docx" ) ) self.assertEqual(job["final_output"].parent.name, request_id) self.assertNotIn("final_review_path", callback_payloads[0]) self.assertEqual(download.status_code, 200) self.assertTrue(download.content.startswith(b"PK")) def test_progress_endpoint_returns_step_progress(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) executor = RecordingExecutor() def fake_worker(_t, _p, _r, step6, _log): _write_docx(step6, "final review") progress_path = step6.parent / api_worker.PROGRESS_FILE_NAME progress_path.write_text( json.dumps( { "current_step": 6, "total_steps": 6, "status": "completed", "message": "全部步骤处理完成", "completed_steps": [1, 2, 3, 4, 5, 6], }, ensure_ascii=False, ), encoding="utf-8", ) with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker", side_effect=fake_worker ), patch.object( http_api, "_send_callback", side_effect=lambda _url, payload: None, ): response = self._post(_valid_files("项目A.pdf")) request_id = response.json()["request_id"] function, args = executor.submitted[0] function(*args) progress_response = TestClient(http_api.app).get( f"/api/v1/jobs/{request_id}/progress" ) self.assertEqual(response.status_code, 202) self.assertEqual(progress_response.status_code, 200) self.assertEqual(progress_response.json()["current_step"], 6) self.assertEqual(progress_response.json()["status"], "completed") shutil.rmtree(root / "jobs", ignore_errors=True) def test_optional_cla_file_is_recorded_and_does_not_affect_worker_submission(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) executor = RecordingExecutor() files = _valid_files() files["CLA_FILE"] = ( "澄清公告.docx", _docx_bytes("clarification"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ) with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker" ) as worker: response = self._post(files=files) function, args = executor.submitted[0] job = args[0] self.assertEqual(response.status_code, 202) self.assertEqual(job["files"]["CLA_FILE"]["filename"], "澄清公告.docx") self.assertGreater(job["files"]["CLA_FILE"]["size"], 0) worker.assert_not_called() shutil.rmtree(root / "jobs", ignore_errors=True) def test_json_body_accepts_server_local_paths_and_records_files(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) tender, procurement, reference = _write_json_input_files(root) executor = RecordingExecutor() payload = { "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": 1, "TENDER_FILE": str(tender), "PROCUREMENT_FILE": str(procurement), "REFERENCE_BID": str(reference), "CLA_FILE": "", } with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker" ) as worker: response = TestClient(http_api.app).post( "/api/v1/final-review", json=payload, ) function, args = executor.submitted[0] job = args[0] self.assertEqual(response.status_code, 202) self.assertEqual(job["txbId"], "1") self.assertEqual(job["files"]["TENDER_FILE"]["filename"], "招标文件.pdf") self.assertEqual(job["files"]["PROCUREMENT_FILE"]["filename"], "采购需求.docx") self.assertEqual(job["files"]["REFERENCE_BID"]["filename"], "参考投书.docx") self.assertNotIn("CLA_FILE", job["files"]) worker.assert_not_called() shutil.rmtree(root / "jobs", ignore_errors=True) def test_json_relative_reference_bid_joins_reference_dir(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) reference_dir = root / "reference_root" reference_dir.mkdir() reference_file = reference_dir / "参考投书.docx" reference_file.write_bytes(_docx_bytes("reference")) tender = root / "招标文件.pdf" procurement = root / "采购需求.docx" tender.write_bytes(_pdf_bytes()) procurement.write_bytes(_docx_bytes("procurement")) executor = RecordingExecutor() payload = { "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": 1, "TENDER_FILE": str(tender), "PROCUREMENT_FILE": str(procurement), "REFERENCE_BID": "参考投书.docx", "REFERENCE_IS_ABSOLUTE": False, "CLA_FILE": "", } with patch.object(http_api, "REFERENCE_DIR", str(reference_dir)), patch.object( http_api, "API_WORK_ROOT", root / "jobs" ), patch.object(http_api, "_JOB_EXECUTOR", executor), patch.object( http_api, "_run_pipeline_worker" ) as worker: response = TestClient(http_api.app).post( "/api/v1/final-review", json=payload, ) function, args = executor.submitted[0] job = args[0] self.assertEqual(response.status_code, 202) self.assertEqual(job["files"]["REFERENCE_BID"]["filename"], "参考投书.docx") worker.assert_not_called() shutil.rmtree(root / "jobs", ignore_errors=True) def test_invalid_reference_is_absolute_value_is_rejected(self): with patch.object(http_api, "_JOB_EXECUTOR") as executor: response = TestClient(http_api.app).post( "/api/v1/final-review", json={ "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": "bad-ref", "TENDER_FILE": "tender.pdf", "PROCUREMENT_FILE": "procurement.docx", "REFERENCE_BID": "reference.docx", "REFERENCE_IS_ABSOLUTE": "maybe", }, ) self.assertEqual(response.status_code, 400) executor.submit.assert_not_called() def test_multipart_accepts_reference_is_absolute_without_path_join(self): with tempfile.TemporaryDirectory() as temp_dir, patch.object( http_api, "API_WORK_ROOT", Path(temp_dir) ), patch.object(http_api, "_JOB_EXECUTOR") as executor, patch.object( http_api, "_run_pipeline_worker" ) as worker: response = TestClient(http_api.app).post( "/api/v1/final-review", data={ "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": "multipart-ref", "REFERENCE_IS_ABSOLUTE": "false", }, files=_valid_files(), ) self.assertEqual(response.status_code, 202) executor.submit.assert_called_once() worker.assert_not_called() shutil.rmtree(Path(temp_dir), ignore_errors=True) def test_clean_intermediate_false_keeps_job_directory(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) executor = RecordingExecutor() def fake_worker(_t, _p, _r, step6, _log): _write_docx(step6, "final review") with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker", side_effect=fake_worker ), patch.object( http_api, "_send_callback", side_effect=lambda _url, payload: None, ): response = TestClient(http_api.app).post( "/api/v1/final-review", data={ "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": "keep-001", "CLEAN_INTERMEDIATE": "false", }, files=_valid_files("项目A.pdf"), ) function, args = executor.submitted[0] job = args[0] function(*args) self.assertEqual(response.status_code, 202) self.assertIn("intermediate_dir", response.json()) self.assertTrue(Path(job["job_dir"]).exists()) status = http_api._job_snapshot(response.json()["request_id"]) self.assertEqual(status["intermediate_dir"], str(job["job_dir"])) shutil.rmtree(root / "jobs", ignore_errors=True) def test_invalid_clean_intermediate_value_is_rejected(self): with patch.object(http_api, "_JOB_EXECUTOR") as executor: response = TestClient(http_api.app).post( "/api/v1/final-review", data={ "CALLBACK_URL": "http://127.0.0.1:9999/callback", "txbId": "bad-001", "CLEAN_INTERMEDIATE": "maybe", }, files=_valid_files(), ) self.assertEqual(response.status_code, 400) executor.submit.assert_not_called() def test_background_failure_sends_error_callback(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) executor = RecordingExecutor() callback_payloads = [] with patch.object(http_api, "API_WORK_ROOT", root / "jobs"), patch.object( http_api, "_JOB_EXECUTOR", executor ), patch.object( http_api, "_run_pipeline_worker", side_effect=subprocess.CalledProcessError(7, ["worker"]), ), patch.object( http_api, "_send_callback", side_effect=lambda _url, payload: callback_payloads.append(payload), ): response = self._post() function, args = executor.submitted[0] function(*args) request_id = response.json()["request_id"] status = http_api._job_snapshot(request_id) self.assertEqual(status["status"], "failed") self.assertTrue(status["callback_delivered"]) self.assertEqual(callback_payloads[0]["status"], "failed") self.assertEqual(callback_payloads[0]["txbId"], "dms-tender-001") self.assertEqual( callback_payloads[0]["resultPath"], "error: Step1-6 处理进程异常退出（退出码 7）", ) self.assertIn("退出码 7", callback_payloads[0]["error"]) def test_callback_ip_without_scheme_defaults_to_http(self): self.assertEqual( http_api._normalize_callback_url("127.0.0.1:9000/callback"), "http://127.0.0.1:9000/callback", ) def test_callback_url_with_scheme_but_missing_slashes_is_normalized(self): self.assertEqual( http_api._normalize_callback_url( "http:121.43.55.7:10026/shenqin/tender/callback" ), "http://121.43.55.7:10026/shenqin/tender/callback", ) def test_callback_posts_json_without_authorization_token(self): captured = {} class Response: status = 200 def __enter__(self): return self def __exit__(self, *_args): return False def fake_urlopen(request, timeout): captured["request"] = request captured["timeout"] = timeout return Response() payload = { "txbId": "dms-tender-001", "resultPath": "output/result.docx", } with patch("http_api.urlopen", side_effect=fake_urlopen): http_api._send_callback( "http://121.43.55.7:10026/shenqin/tender/callback", payload ) request = captured["request"] self.assertEqual(json.loads(request.data.decode("utf-8")), payload) self.assertNotIn( "authorization", {key.lower() for key, _value in request.header_items()} ) def test_txb_id_is_required_and_cannot_be_blank(self): client = TestClient(http_api.app) missing = client.post( "/api/v1/final-review", data={"CALLBACK_URL": "127.0.0.1:9999/callback"}, files=_valid_files(), ) blank = self._post(txb_id=" ") self.assertEqual(missing.status_code, 422) self.assertEqual(blank.status_code, 400) self.assertEqual(blank.json()["detail"], "txbId 不能为空") def test_invalid_callback_is_rejected_before_job_submission(self): with patch.object(http_api, "_JOB_EXECUTOR") as executor: response = self._post(callback_url="ftp://127.0.0.1/callback") self.assertEqual(response.status_code, 400) executor.submit.assert_not_called() def test_missing_required_upload_is_rejected(self): response = self._post( files={ "TENDER_FILE": ("tender.pdf", _pdf_bytes()), "REFERENCE_BID": ("reference.docx", _docx_bytes()), } ) self.assertEqual(response.status_code, 422) def test_invalid_extension_is_rejected_before_job_submission(self): files = _valid_files() files["TENDER_FILE"] = ("tender.exe", b"bad") with patch.object(http_api, "_JOB_EXECUTOR") as executor: response = self._post(files) self.assertEqual(response.status_code, 400) executor.submit.assert_not_called() def test_procurement_and_reference_bid_must_be_docx(self): for field_name in ("PROCUREMENT_FILE", "REFERENCE_BID"): files = _valid_files() files[field_name] = ("wrong.pdf", _pdf_bytes()) with self.subTest(field_name=field_name), patch.object( http_api, "_JOB_EXECUTOR" ) as executor: response = self._post(files) self.assertEqual(response.status_code, 400) self.assertIn("只能上传 .DOCX", response.json()["detail"]) executor.submit.assert_not_called() def test_empty_tender_is_rejected(self): with tempfile.TemporaryDirectory() as temp_dir, patch.object( http_api, "API_WORK_ROOT", Path(temp_dir) ), patch.object(http_api, "_JOB_EXECUTOR") as executor: files = _valid_files() files["TENDER_FILE"] = ("tender.pdf", b"") response = self._post(files) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()["detail"], "TENDER_FILE 不能为空") executor.submit.assert_not_called() self.assertEqual(list(Path(temp_dir).iterdir()), []) def test_scanned_pdf_without_extractable_text_is_rejected(self): with tempfile.TemporaryDirectory() as temp_dir, patch.object( http_api, "API_WORK_ROOT", Path(temp_dir) ), patch.object(http_api, "_JOB_EXECUTOR") as executor: files = _valid_files() files["TENDER_FILE"] = ("tender.pdf", _pdf_bytes(text="")) response = self._post(files) self.assertEqual(response.status_code, 400) self.assertIn("文字类 PDF", response.json()["detail"]) executor.submit.assert_not_called() def test_docx_extension_with_invalid_content_is_rejected(self): with tempfile.TemporaryDirectory() as temp_dir, patch.object( http_api, "API_WORK_ROOT", Path(temp_dir) ), patch.object(http_api, "_JOB_EXECUTOR") as executor: files = _valid_files() files["PROCUREMENT_FILE"] = ("procurement.docx", b"not a docx") response = self._post(files) self.assertEqual(response.status_code, 400) self.assertIn("不是有效的 DOCX", response.json()["detail"]) executor.submit.assert_not_called() if __name__ == "__main__": unittest.main()