test(regulatory-info-package): 覆盖材料包主链路

tests/test_regulatory_info_package_field_extract.py

@@ -0,0 +1,36 @@
+from review_agent.regulatory_info_package.schemas import InstructionExtractResult
+from review_agent.regulatory_info_package.services.field_extract import extract_fields_by_rules, run_parallel_extract
+
+
+def test_extract_fields_by_rules_finds_product_name_and_storage():
+    instruction = InstructionExtractResult(
+        source_file_name="目标产品说明书.docx",
+        paragraphs=["产品名称：新型冠状病毒检测试剂盒", "储存条件：2-8℃保存"],
+        sections={},
+        tables=[],
+        component_tables=[],
+        front_text="产品名称：新型冠状病毒检测试剂盒\n储存条件：2-8℃保存",
+    )
+
+    result = extract_fields_by_rules(instruction)
+
+    assert result["product_name"]["value"] == "新型冠状病毒检测试剂盒"
+    assert result["storage_condition"]["value"] == "2-8℃保存"
+
+
+def test_run_parallel_extract_keeps_rule_result_when_llm_fails():
+    instruction = InstructionExtractResult(
+        source_file_name="目标产品说明书.docx",
+        paragraphs=["产品名称：测试产品"],
+        sections={},
+        tables=[],
+        component_tables=[],
+        front_text="产品名称：测试产品",
+    )
+
+    result = run_parallel_extract(instruction, llm_extract_func=lambda _instruction: (_ for _ in ()).throw(ValueError("bad llm")))
+
+    assert result["regex_results"]["product_name"]["value"] == "测试产品"
+    assert result["llm_results"] == {}
+    assert result["llm_error"]
+

tests/test_regulatory_info_package_field_merge.py

@@ -0,0 +1,24 @@
+from review_agent.regulatory_info_package.services.field_merge import merge_fields
+
+
+def test_merge_fields_marks_missing_llm_only_and_conflict():
+    merged, summary = merge_fields(
+        {
+            "product_name": {"value": "规则产品", "evidence": "说明书", "confidence": 0.8, "label": "产品名称"},
+            "applicant_name": {"value": "", "evidence": "", "confidence": 0.0, "label": "申请人名称"},
+            "package_specification": {"value": "24人份/盒", "evidence": "表格", "confidence": 0.7, "label": "包装规格"},
+        },
+        {
+            "intended_use": {"value": "用于检测", "evidence": "LLM", "confidence": 0.6, "label": "预期用途"},
+            "package_specification": {"value": "48人份/盒", "evidence": "LLM", "confidence": 0.6, "label": "包装规格"},
+        },
+    )
+
+    assert merged["applicant_name"].value == "/"
+    assert merged["applicant_name"].highlight_reason == "missing"
+    assert merged["intended_use"].highlight_reason == "llm_only"
+    assert merged["package_specification"].value == "24人份/盒"
+    assert merged["package_specification"].highlight_reason == "conflict"
+    assert any(item["field_key"] == "applicant_name" for item in summary["missing_fields"])
+    assert len(summary["llm_only_fields"]) == 1
+    assert len(summary["conflict_fields"]) == 1

tests/test_regulatory_info_package_frontend.py

@@ -0,0 +1,45 @@
+import pytest
+from django.urls import reverse
+
+from review_agent.models import Conversation, RegulatoryInfoPackageBatch, WorkflowNodeRun
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_workspace_renders_regulatory_info_package_chip_and_card(client, django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="RIP-CARD",
+        status=RegulatoryInfoPackageBatch.Status.SUCCESS,
+        generated_files=[{"status": "success"} for _ in range(7)],
+    )
+    WorkflowNodeRun.objects.create(
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        node_group="regulatory_info_package",
+        node_code="zip_export",
+        node_name="打包下载",
+        status=WorkflowNodeRun.Status.SUCCESS,
+        progress=100,
+    )
+    client.force_login(user)
+
+    response = client.get(f"{reverse('chat')}?conversation={conversation.pk}")
+    content = response.content.decode("utf-8")
+
+    assert "第1章监管信息" in content
+    assert 'data-workflow-type="regulatory_info_package"' in content
+    assert "data-regulatory-info-package-status-url-template" in content
+    assert "RIP-CARD" in content
+
+
+def test_frontend_selects_regulatory_info_package_status_url():
+    script = open("static/js/app.js", encoding="utf-8").read()
+
+    assert 'workflow_type === "regulatory_info_package"' in script
+    assert "data-regulatory-info-package-status-url-template" in script
+

tests/test_regulatory_info_package_input_select.py

@@ -0,0 +1,48 @@
+import pytest
+
+from review_agent.models import Conversation, FileAttachment
+from review_agent.regulatory_info_package.services.input_select import select_instruction_input
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_select_instruction_input_prefers_message_filename(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    selected = FileAttachment.objects.create(
+        conversation=conversation,
+        user=user,
+        original_name="目标产品说明书.docx",
+        storage_path="uploads/target.docx",
+    )
+    FileAttachment.objects.create(
+        conversation=conversation,
+        user=user,
+        original_name="其他说明书.docx",
+        storage_path="uploads/other.docx",
+    )
+
+    result = select_instruction_input(conversation, "请使用目标产品说明书生成第1章监管信息")
+
+    assert result.status == "selected"
+    assert result.attachment == selected
+    assert result.file_name == "目标产品说明书.docx"
+
+
+def test_select_instruction_input_waits_on_multiple_candidates(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    for name in ["A说明书.docx", "B说明书.docx"]:
+        FileAttachment.objects.create(
+            conversation=conversation,
+            user=user,
+            original_name=name,
+            storage_path=f"uploads/{name}",
+        )
+
+    result = select_instruction_input(conversation, "生成第1章监管信息")
+
+    assert result.status == "waiting_user"
+    assert result.candidates == ["A说明书.docx", "B说明书.docx"]
+

tests/test_regulatory_info_package_instruction_extract.py

@@ -0,0 +1,16 @@
+from pathlib import Path
+
+from review_agent.regulatory_info_package.services.instruction_extract import parse_instruction_docx
+
+
+def test_parse_instruction_docx_extracts_paragraphs_and_tables():
+    path = Path("docs/0.原始材料/目标产品说明书.docx")
+
+    result = parse_instruction_docx(path)
+
+    assert result.source_file_name == "目标产品说明书.docx"
+    assert result.paragraphs
+    assert isinstance(result.sections, dict)
+    assert isinstance(result.tables, list)
+    assert result.front_text
+

tests/test_regulatory_info_package_legacy_doc.py

@@ -0,0 +1,9 @@
+from review_agent.regulatory_info_package.services.legacy_doc_document import detect_legacy_doc_capability
+
+
+def test_detect_legacy_doc_capability_is_stable():
+    capability = detect_legacy_doc_capability()
+
+    assert capability.status in {"available", "unavailable"}
+    assert capability.adapter in {"WordComDocAdapter", "UnavailableLegacyDocAdapter"}
+

tests/test_regulatory_info_package_models.py

@@ -0,0 +1,109 @@
+import pytest
+from django.db import IntegrityError
+
+from review_agent.models import (
+    Conversation,
+    ExportedSummaryFile,
+    FileAttachment,
+    RegulatoryInfoPackageArtifact,
+    RegulatoryInfoPackageBatch,
+    RegulatoryInfoPackageNotificationRecord,
+    WorkflowNodeRun,
+)
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_regulatory_info_package_batch_defaults(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    attachment = FileAttachment.objects.create(
+        conversation=conversation,
+        user=user,
+        original_name="目标产品说明书.docx",
+        storage_path="uploads/instruction.docx",
+    )
+
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        source_attachment=attachment,
+        batch_no="RIP-20260610153000-abcdef",
+        source_file_name=attachment.original_name,
+        source_storage_path=attachment.storage_path,
+    )
+
+    assert batch.status == RegulatoryInfoPackageBatch.Status.PENDING
+    assert batch.output_zip_name == "第1章 监管信息(预生成版).zip"
+    assert batch.generated_files == []
+    assert batch.missing_fields == []
+    assert batch.llm_only_fields == []
+    assert batch.conflict_fields == []
+    assert batch.risk_notes == []
+    assert batch.adapter_summary == {}
+    assert str(batch) == "RIP-20260610153000-abcdef"
+
+
+def test_regulatory_info_package_artifact_and_notification(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="RIP-20260610153100-abcdef",
+    )
+
+    artifact = RegulatoryInfoPackageArtifact.objects.create(
+        batch=batch,
+        artifact_type=RegulatoryInfoPackageArtifact.ArtifactType.ZIP_PACKAGE,
+        file_format=RegulatoryInfoPackageArtifact.FileFormat.ZIP,
+        name="主下载包",
+        file_name="第1章 监管信息(预生成版).zip",
+        storage_path="media/regulatory_info_package/package.zip",
+    )
+    notification = RegulatoryInfoPackageNotificationRecord.objects.create(
+        batch=batch,
+        recipient=user,
+        export_ids=[1, 2],
+        message_summary="材料包已生成",
+        send_status=RegulatoryInfoPackageNotificationRecord.SendStatus.SUCCESS,
+    )
+
+    assert artifact.metadata == {}
+    assert artifact.is_deleted is False
+    assert notification.channel == RegulatoryInfoPackageNotificationRecord.Channel.MOCK
+    assert notification.retry_count == 0
+
+
+def test_exported_summary_file_supports_zip_type():
+    values = {value for value, _label in ExportedSummaryFile.ExportType.choices}
+
+    assert "zip" in values
+
+
+def test_workflow_node_run_unique_for_workflow_batch(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="RIP-20260610153200-abcdef",
+    )
+
+    WorkflowNodeRun.objects.create(
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        node_group="regulatory_info_package",
+        node_code="prepare",
+        node_name="准备资料",
+    )
+
+    with pytest.raises(IntegrityError):
+        WorkflowNodeRun.objects.create(
+            workflow_type="regulatory_info_package",
+            workflow_batch_id=batch.pk,
+            node_group="regulatory_info_package",
+            node_code="prepare",
+            node_name="准备资料",
+        )

tests/test_regulatory_info_package_notification.py

@@ -0,0 +1,17 @@
+import pytest
+
+from review_agent.models import Conversation, RegulatoryInfoPackageBatch, RegulatoryInfoPackageNotificationRecord
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_regulatory_info_package_notification_record_defaults(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(conversation=conversation, user=user, batch_no="RIP-NOTIFY")
+
+    record = RegulatoryInfoPackageNotificationRecord.objects.create(batch=batch, recipient=user)
+
+    assert record.channel == RegulatoryInfoPackageNotificationRecord.Channel.MOCK
+    assert record.send_status == RegulatoryInfoPackageNotificationRecord.SendStatus.PENDING

tests/test_regulatory_info_package_package_generate.py

@@ -0,0 +1,31 @@
+import zipfile
+
+import pytest
+
+from review_agent.models import Conversation, RegulatoryInfoPackageBatch
+from review_agent.regulatory_info_package.services.field_merge import merge_fields
+from review_agent.regulatory_info_package.services.package_generate import generate_package_documents
+from review_agent.regulatory_info_package.services.template_config import load_template_config
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_generate_package_documents_creates_seven_results(django_user_model, tmp_path):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="RIP-20260610154000-abcdef",
+        work_dir=str(tmp_path),
+    )
+    merged, _summary = merge_fields({"product_name": {"value": "测试产品", "label": "产品名称"}}, {})
+
+    results = generate_package_documents(batch, load_template_config(), merged)
+
+    assert len(results) == 7
+    assert all(result.status in {"success", "fallback_success"} for result in results), [
+        (result.template_code, result.status, result.error_message) for result in results
+    ]
+    assert all(result.path for result in results)

tests/test_regulatory_info_package_summary.py

@@ -0,0 +1,13 @@
+from review_agent.regulatory_info_package.services.summary import build_assistant_summary
+
+
+def test_build_assistant_summary_puts_zip_first():
+    exports = [
+        {"file_name": "CH1.4 申请表.docx", "download_url": "/docx"},
+        {"file_name": "第1章 监管信息(预生成版).zip", "download_url": "/zip", "export_type": "zip"},
+    ]
+
+    summary = build_assistant_summary(batch_no="RIP-1", exports=exports, failed_files=[])
+
+    assert summary.index("第1章 监管信息(预生成版).zip") < summary.index("CH1.4 申请表.docx")
+

tests/test_regulatory_info_package_template_config.py

@@ -0,0 +1,48 @@
+from pathlib import Path
+
+import pytest
+
+from review_agent.regulatory_info_package.constants import DEFAULT_ZIP_NAME
+from review_agent.regulatory_info_package.services.template_config import (
+    compute_config_hash,
+    load_template_config,
+    validate_template_config,
+)
+
+
+def test_template_config_loads_seven_templates():
+    config = load_template_config()
+
+    assert config["version"] == "regulatory_info_package_templates_v1"
+    assert config["zip_name"] == DEFAULT_ZIP_NAME
+    assert len(config["templates"]) == 7
+    assert {template["code"] for template in config["templates"]} == {
+        "ch1_2_directory",
+        "ch1_4_application_form",
+        "ch1_5_product_list",
+        "ch1_9_pre_submission",
+        "ch1_11_1_standards",
+        "ch1_11_5_authenticity",
+        "ch1_11_6_conformity",
+    }
+    assert validate_template_config(config) == []
+    assert compute_config_hash()
+
+
+def test_template_config_rejects_duplicate_codes():
+    config = load_template_config()
+    config["templates"].append(dict(config["templates"][0]))
+
+    errors = validate_template_config(config)
+
+    assert any("重复" in error for error in errors)
+
+
+def test_template_config_sources_exist():
+    config = load_template_config()
+    source_dir = Path(config["source_dir"])
+
+    assert source_dir.exists()
+    for template in config["templates"]:
+        assert (source_dir / template["source_file"]).exists()
+

tests/test_regulatory_info_package_traceability.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+from openpyxl import load_workbook
+
+from review_agent.regulatory_info_package.schemas import MergedField
+from review_agent.regulatory_info_package.services.traceability_export import save_traceability_exports
+
+
+def test_save_traceability_exports_writes_excel_and_json(tmp_path):
+    fields = {
+        "product_name": MergedField(
+            key="product_name",
+            label="产品名称",
+            value="测试产品",
+            source="rule",
+            evidence="说明书",
+            confidence=0.9,
+        )
+    }
+
+    excel_path, json_path = save_traceability_exports(tmp_path, fields)
+
+    assert excel_path.name == "traceability.xlsx"
+    assert json_path.name == "traceability.json"
+    assert json_path.exists()
+    workbook = load_workbook(excel_path)
+    assert workbook.active["A1"].value == "target_file"
+

tests/test_regulatory_info_package_trigger.py

@@ -0,0 +1,19 @@
+import pytest
+
+from review_agent.models import Conversation
+from review_agent.skill_router import route_message_intent
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_fixed_keyword_routes_to_regulatory_info_package(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+
+    route = route_message_intent(conversation, "请根据说明书生成第1章监管信息")
+
+    assert route.action == "regulatory_info_package"
+    assert route.workflow_type == "regulatory_info_package"
+    assert route.starts_regulatory_info_package is True
+

tests/test_regulatory_info_package_views.py

@@ -0,0 +1,140 @@
+from pathlib import Path
+
+import pytest
+
+from review_agent.models import (
+    Conversation,
+    ExportedSummaryFile,
+    RegulatoryInfoPackageBatch,
+    WorkflowNodeRun,
+)
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_regulatory_info_package_export_download_checks_owner(client, django_user_model, tmp_path):
+    owner = django_user_model.objects.create_user(username="owner", password="pass")
+    other = django_user_model.objects.create_user(username="other", password="pass")
+    conversation = Conversation.objects.create(user=owner, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=owner,
+        batch_no="RIP-20260610153300-abcdef",
+    )
+    path = tmp_path / "第1章 监管信息(预生成版).zip"
+    path.write_bytes(b"zip-content")
+    exported = ExportedSummaryFile.objects.create(
+        batch=None,
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        export_category="regulatory_info_package",
+        export_type=ExportedSummaryFile.ExportType.ZIP,
+        file_name=path.name,
+        storage_path=str(path),
+    )
+
+    client.force_login(other)
+    denied = client.get(f"/api/review-agent/file-summary/exports/{exported.pk}/download/")
+    assert denied.status_code == 404
+
+    client.force_login(owner)
+    allowed = client.get(f"/api/review-agent/file-summary/exports/{exported.pk}/download/")
+    assert allowed.status_code == 200
+    assert allowed["Content-Type"] == "application/zip"
+
+
+@pytest.mark.parametrize(
+    ("file_name", "export_type", "expected"),
+    [
+        ("CH1.9 产品申报前沟通的说明.doc", ExportedSummaryFile.ExportType.WORD, "application/msword"),
+        (
+            "CH1.4 申请表.docx",
+            ExportedSummaryFile.ExportType.WORD,
+            "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+        ),
+        ("第1章 监管信息(预生成版).zip", ExportedSummaryFile.ExportType.ZIP, "application/zip"),
+    ],
+)
+def test_regulatory_info_package_download_mime_by_extension(
+    client,
+    django_user_model,
+    tmp_path,
+    file_name,
+    export_type,
+    expected,
+):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no=f"RIP-20260610153400-{Path(file_name).suffix[1:] or 'zip'}",
+    )
+    path = tmp_path / file_name
+    path.write_bytes(b"content")
+    exported = ExportedSummaryFile.objects.create(
+        batch=None,
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        export_category="generated_document",
+        export_type=export_type,
+        file_name=file_name,
+        storage_path=str(path),
+    )
+    client.force_login(user)
+
+    response = client.get(f"/api/review-agent/file-summary/exports/{exported.pk}/download/")
+
+    assert response.status_code == 200
+    assert response["Content-Type"] == expected
+
+
+def test_regulatory_info_package_status_returns_nodes_and_zip_first(client, django_user_model, tmp_path):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = RegulatoryInfoPackageBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="RIP-20260610153500-abcdef",
+        status=RegulatoryInfoPackageBatch.Status.SUCCESS,
+    )
+    WorkflowNodeRun.objects.create(
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        node_group="regulatory_info_package",
+        node_code="zip_export",
+        node_name="打包下载",
+        status=WorkflowNodeRun.Status.SUCCESS,
+        progress=100,
+    )
+    doc = tmp_path / "CH1.4 申请表.docx"
+    zip_file = tmp_path / "第1章 监管信息(预生成版).zip"
+    doc.write_bytes(b"doc")
+    zip_file.write_bytes(b"zip")
+    ExportedSummaryFile.objects.create(
+        batch=None,
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        export_category="generated_document",
+        export_type=ExportedSummaryFile.ExportType.WORD,
+        file_name=doc.name,
+        storage_path=str(doc),
+    )
+    ExportedSummaryFile.objects.create(
+        batch=None,
+        workflow_type="regulatory_info_package",
+        workflow_batch_id=batch.pk,
+        export_category="regulatory_info_package",
+        export_type=ExportedSummaryFile.ExportType.ZIP,
+        file_name=zip_file.name,
+        storage_path=str(zip_file),
+    )
+    client.force_login(user)
+
+    response = client.get(f"/api/review-agent/regulatory-info-package/{batch.pk}/status/")
+
+    payload = response.json()
+    assert payload["batch"]["workflow_type"] == "regulatory_info_package"
+    assert payload["nodes"][0]["node_code"] == "zip_export"
+    assert payload["exports"][0]["export_type"] == "zip"

tests/test_regulatory_info_package_workflow.py

@@ -0,0 +1,62 @@
+import pytest
+
+from review_agent.models import Conversation, RegulatoryInfoPackageBatch, WorkflowNodeRun
+from review_agent.regulatory_info_package.constants import (
+    REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS,
+    WORKFLOW_TYPE,
+)
+from review_agent.regulatory_info_package.workflow import (
+    create_regulatory_info_package_batch,
+    start_regulatory_info_package_workflow,
+)
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_create_regulatory_info_package_batch_initializes_nodes(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+
+    batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
+
+    assert batch.batch_no.startswith("RIP-")
+    assert batch.work_dir
+    nodes = WorkflowNodeRun.objects.filter(
+        workflow_type=WORKFLOW_TYPE,
+        workflow_batch_id=batch.pk,
+    ).order_by("id")
+    assert [node.node_code for node in nodes] == [
+        code for code, _name, _group in REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS
+    ]
+
+
+def test_create_regulatory_info_package_batch_is_node_idempotent(django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
+
+    create_regulatory_info_package_batch(conversation=conversation, user=user, existing_batch=batch)
+
+    assert WorkflowNodeRun.objects.filter(
+        workflow_type=WORKFLOW_TYPE,
+        workflow_batch_id=batch.pk,
+    ).count() == len(REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS)
+
+
+def test_empty_workflow_skeleton_completes(django_user_model, settings):
+    settings.REGULATORY_INFO_PACKAGE_ASYNC = False
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
+
+    start_regulatory_info_package_workflow(batch, async_run=False)
+    batch.refresh_from_db()
+
+    assert batch.status == RegulatoryInfoPackageBatch.Status.SUCCESS
+    assert WorkflowNodeRun.objects.filter(
+        workflow_type=WORKFLOW_TYPE,
+        workflow_batch_id=batch.pk,
+        status=WorkflowNodeRun.Status.SUCCESS,
+    ).count() == len(REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS)
+

tests/test_regulatory_info_package_zip.py

@@ -0,0 +1,22 @@
+import zipfile
+
+from review_agent.regulatory_info_package.schemas import GeneratedFileResult
+from review_agent.regulatory_info_package.services.zip_export import create_zip_package
+
+
+def test_create_zip_package_includes_only_success_files(tmp_path):
+    success = tmp_path / "ok.docx"
+    failed = tmp_path / "bad.docx"
+    success.write_bytes(b"ok")
+    failed.write_bytes(b"bad")
+
+    zip_path = create_zip_package(
+        tmp_path,
+        [
+            GeneratedFileResult("ok", "ok.docx", "docx", "docx", "success", path=str(success)),
+            GeneratedFileResult("bad", "bad.docx", "docx", "docx", "failed", path=str(failed)),
+        ],
+    )
+
+    with zipfile.ZipFile(zip_path) as archive:
+        assert archive.namelist() == ["ok.docx"]