proposa-ai/scripts/tests/test_api.py

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),
                "TPC_IS_ABSOLUTE": True,
                "REFERENCE_BID": str(reference),
                "REFERENCE_IS_ABSOLUTE": True,
                "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),
                "TPC_IS_ABSOLUTE": True,
                "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_json_relative_tpc_files_join_tpc_dir(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            tpc_dir = root / "tpc_root"
            tpc_dir.mkdir()
            tender = tpc_dir / "招标文件.pdf"
            procurement = tpc_dir / "采购需求.docx"
            clarification = tpc_dir / "澄清公告.docx"
            reference = root / "参考投书.docx"
            tender.write_bytes(_pdf_bytes())
            procurement.write_bytes(_docx_bytes("procurement"))
            clarification.write_bytes(_docx_bytes("clarification"))
            reference.write_bytes(_docx_bytes("reference"))
            executor = RecordingExecutor()
            payload = {
                "CALLBACK_URL": "http://127.0.0.1:9999/callback",
                "txbId": 1,
                "TENDER_FILE": f"/{tender.name}",
                "PROCUREMENT_FILE": f"/{procurement.name}",
                "REFERENCE_BID": str(reference),
                "REFERENCE_IS_ABSOLUTE": True,
                "CLA_FILE": f"/{clarification.name}",
                "TPC_IS_ABSOLUTE": False,
            }
            with patch.object(http_api, "TPC_DIR", str(tpc_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"]["TENDER_FILE"]["filename"], tender.name)
            self.assertEqual(
                job["files"]["PROCUREMENT_FILE"]["filename"], procurement.name
            )
            self.assertEqual(
                job["files"]["CLA_FILE"]["filename"], clarification.name
            )
            worker.assert_not_called()
            shutil.rmtree(root / "jobs", ignore_errors=True)

    def test_relative_reference_bid_prefers_reference_dir_when_both_exist(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            reference_dir = root / "reference_root"
            reference_dir1 = root / "dms_upload"
            reference_dir.mkdir()
            reference_dir1.mkdir()
            primary = reference_dir / "reference.docx"
            secondary = reference_dir1 / "reference.docx"
            primary.write_bytes(_docx_bytes("primary"))
            secondary.write_bytes(_docx_bytes("secondary"))

            with patch.object(http_api, "REFERENCE_DIR", str(reference_dir)), patch.object(
                http_api, "REFERENCE_DIR1", str(reference_dir1)
            ):
                resolved = http_api._resolve_reference_path(
                    "/reference.docx", is_absolute=False
                )

            self.assertEqual(Path(resolved), primary)

    def test_json_relative_reference_bid_falls_back_to_reference_dir1(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            reference_dir = root / "reference_root"
            reference_dir1 = root / "dms_upload"
            reference_dir.mkdir()
            reference_dir1.mkdir()
            reference_file = reference_dir1 / "参考投书.docx"
            reference_file.write_bytes(_docx_bytes("reference-dir1"))
            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),
                "TPC_IS_ABSOLUTE": True,
                "REFERENCE_BID": "/参考投书.docx",
                "REFERENCE_IS_ABSOLUTE": False,
                "CLA_FILE": "",
            }
            with patch.object(http_api, "REFERENCE_DIR", str(reference_dir)), patch.object(
                http_api, "REFERENCE_DIR1", str(reference_dir1)
            ), 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"], reference_file.name
            )
            worker.assert_not_called()
            shutil.rmtree(root / "jobs", ignore_errors=True)

    def test_json_relative_reference_bid_rejects_when_both_candidates_missing(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            reference_dir = root / "reference_root"
            reference_dir1 = root / "dms_upload"
            reference_dir.mkdir()
            reference_dir1.mkdir()
            with patch.object(http_api, "REFERENCE_DIR", str(reference_dir)), patch.object(
                http_api, "REFERENCE_DIR1", str(reference_dir1)
            ), 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": "missing-reference",
                        "TENDER_FILE": "tender.pdf",
                        "PROCUREMENT_FILE": "procurement.docx",
                        "TPC_IS_ABSOLUTE": True,
                        "REFERENCE_BID": "/missing.docx",
                        "REFERENCE_IS_ABSOLUTE": False,
                    },
                )

            self.assertEqual(response.status_code, 400)
            self.assertEqual(
                response.json()["detail"],
                "REFERENCE_BID 在 REFERENCE_DIR 和 REFERENCE_DIR1 下均不存在: missing.docx",
            )
            executor.submit.assert_not_called()

    def test_invalid_tpc_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-tpc",
                    "TENDER_FILE": "tender.pdf",
                    "PROCUREMENT_FILE": "procurement.docx",
                    "REFERENCE_BID": "reference.docx",
                    "TPC_IS_ABSOLUTE": "false",
                    "REFERENCE_IS_ABSOLUTE": True,
                },
            )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json()["detail"],
            "TPC_IS_ABSOLUTE 必须是布尔值 true/false",
        )
        executor.submit.assert_not_called()

    def test_relative_tpc_files_require_tpc_dir(self):
        with patch.object(http_api, "TPC_DIR", ""), 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": "missing-tpc-dir",
                    "TENDER_FILE": "tender.pdf",
                    "PROCUREMENT_FILE": "procurement.docx",
                    "REFERENCE_BID": "reference.docx",
                    "TPC_IS_ABSOLUTE": False,
                    "REFERENCE_IS_ABSOLUTE": True,
                },
            )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json()["detail"],
            "TENDER_FILE 为相对路径，但未配置 TPC_DIR",
        )
        executor.submit.assert_not_called()

    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",
                    "TPC_IS_ABSOLUTE": True,
                    "REFERENCE_BID": "reference.docx",
                    "REFERENCE_IS_ABSOLUTE": "maybe",
                },
            )
        self.assertEqual(response.status_code, 400)
        executor.submit.assert_not_called()

    def test_missing_absolute_flags_are_rejected(self):
        base_payload = {
            "CALLBACK_URL": "http://127.0.0.1:9999/callback",
            "txbId": "missing-flags",
            "TENDER_FILE": "tender.pdf",
            "PROCUREMENT_FILE": "procurement.docx",
            "REFERENCE_BID": "reference.docx",
        }
        with patch.object(http_api, "_JOB_EXECUTOR") as executor:
            missing_tpc = TestClient(http_api.app).post(
                "/api/v1/final-review", json=base_payload
            )
            missing_reference = TestClient(http_api.app).post(
                "/api/v1/final-review",
                json={**base_payload, "TPC_IS_ABSOLUTE": True},
            )
        self.assertEqual(missing_tpc.status_code, 400)
        self.assertEqual(
            missing_tpc.json()["detail"],
            "TPC_IS_ABSOLUTE 必须是布尔值 true/false",
        )
        self.assertEqual(missing_reference.status_code, 400)
        self.assertEqual(
            missing_reference.json()["detail"],
            "REFERENCE_IS_ABSOLUTE 必须是布尔值 true/false",
        )
        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()