diff --git a/tests/test_regulatory_info_package_field_extract.py b/tests/test_regulatory_info_package_field_extract.py new file mode 100644 index 0000000..0d50569 --- /dev/null +++ b/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"] + diff --git a/tests/test_regulatory_info_package_field_merge.py b/tests/test_regulatory_info_package_field_merge.py new file mode 100644 index 0000000..18192ed --- /dev/null +++ b/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 diff --git a/tests/test_regulatory_info_package_frontend.py b/tests/test_regulatory_info_package_frontend.py new file mode 100644 index 0000000..2b10f0b --- /dev/null +++ b/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 + diff --git a/tests/test_regulatory_info_package_input_select.py b/tests/test_regulatory_info_package_input_select.py new file mode 100644 index 0000000..a580aa5 --- /dev/null +++ b/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"] + diff --git a/tests/test_regulatory_info_package_instruction_extract.py b/tests/test_regulatory_info_package_instruction_extract.py new file mode 100644 index 0000000..93b9e78 --- /dev/null +++ b/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 + diff --git a/tests/test_regulatory_info_package_legacy_doc.py b/tests/test_regulatory_info_package_legacy_doc.py new file mode 100644 index 0000000..951b609 --- /dev/null +++ b/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"} + diff --git a/tests/test_regulatory_info_package_models.py b/tests/test_regulatory_info_package_models.py new file mode 100644 index 0000000..e100935 --- /dev/null +++ b/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="准备资料", + ) diff --git a/tests/test_regulatory_info_package_notification.py b/tests/test_regulatory_info_package_notification.py new file mode 100644 index 0000000..6b69ac8 --- /dev/null +++ b/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 diff --git a/tests/test_regulatory_info_package_package_generate.py b/tests/test_regulatory_info_package_package_generate.py new file mode 100644 index 0000000..fb8badc --- /dev/null +++ b/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) diff --git a/tests/test_regulatory_info_package_summary.py b/tests/test_regulatory_info_package_summary.py new file mode 100644 index 0000000..6575a96 --- /dev/null +++ b/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") + diff --git a/tests/test_regulatory_info_package_template_config.py b/tests/test_regulatory_info_package_template_config.py new file mode 100644 index 0000000..506f9ab --- /dev/null +++ b/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() + diff --git a/tests/test_regulatory_info_package_traceability.py b/tests/test_regulatory_info_package_traceability.py new file mode 100644 index 0000000..e80fac8 --- /dev/null +++ b/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" + diff --git a/tests/test_regulatory_info_package_trigger.py b/tests/test_regulatory_info_package_trigger.py new file mode 100644 index 0000000..2402e0a --- /dev/null +++ b/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 + diff --git a/tests/test_regulatory_info_package_views.py b/tests/test_regulatory_info_package_views.py new file mode 100644 index 0000000..9836eae --- /dev/null +++ b/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" diff --git a/tests/test_regulatory_info_package_workflow.py b/tests/test_regulatory_info_package_workflow.py new file mode 100644 index 0000000..fc1331f --- /dev/null +++ b/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) + diff --git a/tests/test_regulatory_info_package_zip.py b/tests/test_regulatory_info_package_zip.py new file mode 100644 index 0000000..60e9235 --- /dev/null +++ b/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"]