#!/usr/bin/env python3
"""Dynamic unit tests for the Dify workflow that is being migrated to Coze."""

from __future__ import annotations

import json
import csv
import unittest
from pathlib import Path
from typing import Any, Iterable


ROOT = Path(__file__).resolve().parents[1]
WORKFLOW_DRAFT = ROOT / "source" / "workflow-draft.json"
TEST_CASES = ROOT / "tests" / "coze-migration-test-cases.csv"
EVIDENCE = ROOT / "evidence" / "screenshots" / "coze-import-run-success-20260701.png"


def load_workflow() -> dict[str, Any]:
    return json.loads(WORKFLOW_DRAFT.read_text(encoding="utf-8"))


def flatten(value: Any, path: str = "") -> Iterable[tuple[str, Any]]:
    if isinstance(value, dict):
        if not value:
            yield path or "$", {}
        for key, item in value.items():
            child = f"{path}.{key}" if path else str(key)
            yield from flatten(item, child)
    elif isinstance(value, list):
        if not value:
            yield path or "$", []
        for index, item in enumerate(value):
            yield from flatten(item, f"{path}[{index}]")
    else:
        yield path or "$", value


class WorkflowContractCase(unittest.TestCase):
    def __init__(self, case_name: str, level: str, payload: dict[str, Any]) -> None:
        super().__init__("run_contract")
        self._case_name = case_name
        self.level = level
        self.payload = payload

    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.level == "node":
            node = self.payload["node"]
            self.assertIn("id", node)
            self.assertIn("data", node)
            self.assertTrue(node["data"].get("title"))
            self.assertTrue(node["data"].get("type"))
            self.assertIsInstance(node.get("position", {}), dict)
            self.assertIn("width", node)
            self.assertIn("height", node)
        elif self.level == "parameter":
            path = self.payload["path"]
            value = self.payload["value"]
            self.assertTrue(path)
            if path.endswith("type"):
                self.assertIsInstance(value, str)
        elif self.level == "edge":
            edge = self.payload["edge"]
            node_ids = self.payload["node_ids"]
            self.assertIn(edge.get("source"), node_ids)
            self.assertIn(edge.get("target"), node_ids)
            self.assertTrue(edge.get("id"))
        elif self.level == "scenario":
            nodes_by_title = self.payload["nodes_by_title"]
            required_titles = self.payload["required_titles"]
            for title in required_titles:
                self.assertIn(title, nodes_by_title)
        elif self.level == "artifact":
            self.assertTrue(WORKFLOW_DRAFT.exists())
            self.assertTrue(TEST_CASES.exists())
            self.assertTrue(EVIDENCE.exists())
            with TEST_CASES.open("r", encoding="utf-8-sig", newline="") as handle:
                rows = list(csv.DictReader(handle))
            self.assertEqual(len(rows), 1830)
            self.assertEqual({row["coze_execution_status"] for row in rows}, {"PENDING_REBIND"})
        else:
            raise AssertionError(f"Unknown level: {self.level}")


def load_tests(loader, tests, pattern):  # noqa: D401 - unittest protocol
    """Return one unittest case per node, parameter, edge, scenario, and artifact."""
    workflow = load_workflow()
    nodes = workflow["graph"]["nodes"]
    edges = workflow["graph"]["edges"]
    node_ids = {node["id"] for node in nodes}
    nodes_by_title = {node["data"].get("title"): node for node in nodes}

    suite = unittest.TestSuite()
    suite.addTest(WorkflowContractCase("artifact.source_and_evidence_exist", "artifact", {}))

    for node in nodes:
        title = node["data"].get("title") or node["id"]
        suite.addTest(WorkflowContractCase(f"node.{node['id']}.{title}", "node", {"node": node}))
        for path, value in flatten(node.get("data", {})):
            suite.addTest(
                WorkflowContractCase(
                    f"parameter.{node['id']}.{path}",
                    "parameter",
                    {"node": node, "path": path, "value": value},
                )
            )

    for edge in edges:
        suite.addTest(WorkflowContractCase(f"edge.{edge.get('id')}", "edge", {"edge": edge, "node_ids": node_ids}))

    scenarios = {
        "scenario.discomfort_fixed_reply": ["身体不适固定回复", "输出-医疗急症安全回复"],
        "scenario.out_of_scope_fixed_reply": ["超范围固定回复"],
        "scenario.running_knowledge_qa": ["普通问答检索配置", "普通问答生成", "输出-日常问答"],
        "scenario.vdot_ability": ["VDOT 能力基准计算", "VDOT / 配速回答生成", "输出-能力评估"],
        "scenario.training_plan": ["训练计划生成", "输出-训练计划"],
        "scenario.coros_analysis": ["读取COROS最近30天数据", "COROS训练分析", "输出-COROS训练分析"],
        "scenario.user_data_analysis": ["用户提供数据训练分析", "输出-用户数据训练分析"],
        "scenario.race_pace": ["比赛配速计算", "比赛配速回答生成", "输出-比赛配速"],
        "scenario.profile_write": ["画像写入候选标准化 / Payload 构造", "画像写入分流", "统一保存用户画像"],
        "scenario.missing_pace": ["配速信息不足回复", "输出-配速信息不足"],
        "scenario.profile_read": ["读取用户画像", "画像读取用途分流"],
        "scenario.training_plan_profile_merge": ["训练计划信息合并 / 标准化", "训练计划资料完整度判断"],
    }
    for case_name, required_titles in scenarios.items():
        suite.addTest(
            WorkflowContractCase(
                case_name,
                "scenario",
                {"nodes_by_title": nodes_by_title, "required_titles": required_titles},
            )
        )
    return suite


if __name__ == "__main__":
    unittest.main()