#!/usr/bin/env python3 """Unit tests for Coze/Running workflow package format compatibility.""" from __future__ import annotations import csv import re import unittest import zipfile from pathlib import Path from collections import Counter import yaml ROOT = Path(__file__).resolve().parents[1] COZE_SAMPLE = ROOT / "coze" / "sample-export" / "Workflow-dish_recognition_more_simple-draft-5725.zip" RUNNING_SAMPLE = Path("/Users/jack/Downloads/dify-running-ai-assistant-coze-upload-candidate.zip") CONVERTED = ROOT / "coze" / "converted" CONVERTED_ZIP = CONVERTED / "Workflow-running_ai_assistant_5_compat-draft-0001.zip" CONVERTED_ROOT = CONVERTED / "Workflow-running_ai_assistant_5_compat-draft-0001" CONVERTED_MANIFEST = CONVERTED_ROOT / "MANIFEST.yml" CONVERTED_WORKFLOW = CONVERTED_ROOT / "workflow" / "running_ai_assistant_5_compat-draft.yaml" NODE_MAPPING = CONVERTED / "node-mapping.csv" EDGE_MAPPING = CONVERTED / "edge-mapping.csv" REPORT = CONVERTED / "format-compatibility-report.md" OLD_PLACEHOLDER = "Dify/Running compatibility package imported" NEW_PLACEHOLDER = "NOT_DIFY_BUSINESS_OUTPUT" def zip_names(path: Path) -> list[str]: with zipfile.ZipFile(path) as zf: return zf.namelist() def csv_rows(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def converted_workflow() -> dict: return yaml.safe_load(CONVERTED_WORKFLOW.read_text(encoding="utf-8")) class FormatCompatibilityCase(unittest.TestCase): def __init__(self, case_name: str, kind: str, payload: dict[str, str] | None = None) -> None: super().__init__("run_contract") self._case_name = case_name self.kind = kind self.payload = payload or {} def id(self) -> str: # pragma: no cover - unittest reporting hook return self._case_name def shortDescription(self) -> str: # pragma: no cover - unittest reporting hook return self._case_name def run_contract(self) -> None: if self.kind == "coze_sample_shape": names = zip_names(COZE_SAMPLE) self.assertTrue(any(name.endswith("/MANIFEST.yml") for name in names)) self.assertTrue(any("/workflow/" in name and name.endswith(".yaml") for name in names)) elif self.kind == "running_sample_shape": names = zip_names(RUNNING_SAMPLE) self.assertIn("dify-export-no-secret.yml", names) self.assertIn("workflow-draft.json", names) self.assertFalse(any(name.endswith("/MANIFEST.yml") for name in names)) self.assertFalse(any("/workflow/" in name and name.endswith(".yaml") for name in names)) elif self.kind == "converted_shape": names = zip_names(CONVERTED_ZIP) self.assertIn("Workflow-running_ai_assistant_5_compat-draft-0001/MANIFEST.yml", names) self.assertIn( "Workflow-running_ai_assistant_5_compat-draft-0001/workflow/running_ai_assistant_5_compat-draft.yaml", names, ) elif self.kind == "converted_yaml_contract": text = CONVERTED_WORKFLOW.read_text(encoding="utf-8") self.assertIn('schema_version: "1.0.0"', text) self.assertIn('mode: "workflow"', text) self.assertIn(NEW_PLACEHOLDER, text) self.assertNotIn(OLD_PLACEHOLDER, text) self.assertEqual(len(re.findall(r"^ - id:", text, re.MULTILINE)), 59) self.assertEqual(len(re.findall(r"^ - source_node:", text, re.MULTILINE)), 58) elif self.kind == "converted_linear_import_edges": workflow = converted_workflow() nodes = workflow["nodes"] edges = workflow["edges"] node_ids = {node["id"] for node in nodes} outgoing = Counter(edge["source_node"] for edge in edges) self.assertEqual(len(nodes), 59) self.assertEqual(len(edges), 58) self.assertFalse([(node_id, count) for node_id, count in outgoing.items() if count > 1]) for edge in edges: self.assertIn(edge["source_node"], node_ids) self.assertIn(edge["target_node"], node_ids) self.assertEqual(edges[-1]["target_node"], "900001") elif self.kind == "edge_mapping_preserves_original_graph": self.assertEqual(len(csv_rows(EDGE_MAPPING)), 72) elif self.kind == "manifest_contract": text = CONVERTED_MANIFEST.read_text(encoding="utf-8") self.assertIn('type: "Workflow"', text) self.assertIn("main:", text) self.assertIn('name: "running_ai_assistant_5_compat"', text) elif self.kind == "node_mapping_row": row = self.payload self.assertTrue(row["dify_node_id"]) self.assertTrue(row["coze_node_id"]) self.assertTrue(row["dify_title"]) self.assertIn(row["import_package_type"], {"start", "code"}) self.assertTrue(row["runtime_target_type"]) self.assertTrue(row["manual_rebind_note"]) self.assertTrue(row["dify_data_json"]) elif self.kind == "edge_mapping_row": row = self.payload self.assertTrue(row["dify_source"]) self.assertTrue(row["dify_target"]) self.assertTrue(row["coze_source_node"]) self.assertTrue(row["coze_target_node"]) self.assertEqual(row["coze_import_edge"], "source_node -> target_node") elif self.kind == "report_contract": text = REPORT.read_text(encoding="utf-8") self.assertIn("根因", text) self.assertIn("本次兼容策略", text) self.assertIn("仍需人工重绑", text) self.assertIn("业务语义测试：PENDING_REBIND", text) self.assertIn(NEW_PLACEHOLDER, text) self.assertNotIn("业务语义测试：PASS", text) self.assertNotIn(OLD_PLACEHOLDER, text) else: raise AssertionError(f"Unknown compatibility test kind: {self.kind}") def load_tests(loader, tests, pattern): # noqa: D401 - unittest protocol """Return artifact, node mapping, and edge mapping compatibility tests.""" suite = unittest.TestSuite() suite.addTest(FormatCompatibilityCase("format.coze_sample_has_manifest_and_workflow_yaml", "coze_sample_shape")) suite.addTest(FormatCompatibilityCase("format.running_sample_lacks_coze_manifest", "running_sample_shape")) suite.addTest(FormatCompatibilityCase("format.converted_zip_matches_coze_layout", "converted_shape")) suite.addTest(FormatCompatibilityCase("format.converted_yaml_has_expected_graph_counts", "converted_yaml_contract")) suite.addTest(FormatCompatibilityCase("format.converted_yaml_uses_linear_import_edges", "converted_linear_import_edges")) suite.addTest(FormatCompatibilityCase("format.edge_mapping_preserves_original_dify_graph", "edge_mapping_preserves_original_graph")) suite.addTest(FormatCompatibilityCase("format.converted_manifest_is_workflow", "manifest_contract")) suite.addTest(FormatCompatibilityCase("format.compatibility_report_explains_rebinds", "report_contract")) for row in csv_rows(NODE_MAPPING): suite.addTest( FormatCompatibilityCase( f"format.node_mapping.{row['dify_node_id']}.{row['dify_type']}", "node_mapping_row", row, ) ) for row in csv_rows(EDGE_MAPPING): suite.addTest( FormatCompatibilityCase( f"format.edge_mapping.{row['dify_source']}.{row['dify_target']}.{row['dify_source_handle']}", "edge_mapping_row", row, ) ) return suite if __name__ == "__main__": unittest.main()