vrsub/tests/sdk/test_models/test_models.py

"""src/wov_sdk/models.py 的模块级测试（数据 → 测试过程 → 验证结果）。

被测模块：`src/wov_sdk/models.py`（协议数据模型：节点清单、调用请求/响应、
工作流 DAG），是全部节点与平台之间的长期稳定契约，可独立调用。
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from wov_sdk.models import (
    HealthResponse,
    InvokeRequest,
    InvokeResponse,
    NodeManifest,
    ProgressEvent,
    WorkflowDefinition,
    WorkflowEdge,
    WorkflowNode,
)

# 仓库根与真实清单目录。
WORKSPACE = Path(__file__).resolve().parents[3]
MANIFESTS_DIR = WORKSPACE / "manifests"


# ---------------------------------------------------------------------------
# NodeManifest
# ---------------------------------------------------------------------------


def test_manifest_loads_from_real_json_file() -> None:
    """真实 manifests/*.json 可被加载为清单对象（含必填字段）。"""
    # 数据：仓库真实清单（echo 节点）。
    path = MANIFESTS_DIR / "echo.json"
    assert path.is_file()

    # 测试过程
    manifest = NodeManifest.load(str(path))

    # 验证结果
    assert manifest.id == "echo"
    assert manifest.name
    assert manifest.version
    assert manifest.capability


@pytest.mark.parametrize("manifest_file", ["echo.json", "whisper.json", "vlm.json", "ass.json"])
def test_all_real_manifests_validate(manifest_file: str) -> None:
    """全部真实节点清单通过协议校验（注册表依赖此校验）。"""
    # 数据：真实清单文件。
    # 测试过程
    manifest = NodeManifest.load(str(MANIFESTS_DIR / manifest_file))

    # 验证结果：校验不抛异常且 ID 非空。
    manifest.validate()
    assert manifest.id


def test_manifest_validate_rejects_empty_id() -> None:
    """空 ID 视为非法（节点注册表以 ID 为键）。"""
    # 数据：ID 为空的清单。
    manifest = NodeManifest(id="", name="x", version="1", capability="c", command=["python"])

    # 测试过程与验证结果
    with pytest.raises(ValueError):
        manifest.validate()


def test_manifest_validate_rejects_empty_version() -> None:
    """空版本号非法（版本与节点仓库 tag 绑定）。"""
    # 数据：版本为空的清单。
    manifest = NodeManifest(id="n", name="x", version="", capability="c", command=["python"])

    # 测试过程与验证结果
    with pytest.raises(ValueError):
        manifest.validate()


def test_manifest_round_trip_dict() -> None:
    """清单 to_dict/from_dict 往返保持字段一致（协议序列化稳定）。"""
    # 数据：完整清单。
    manifest = NodeManifest(
        id="n1", name="节点", version="1.2.3", capability="cap",
        command=["python", "-m", "n1"], repo_dir="nodes",
        env={"A": "1"}, input_schema={"in": "file"}, output_schema={"out": "file"},
        max_concurrency=2, idle_ttl_seconds=100, health_timeout_seconds=5, keep_warm=True,
    )

    # 测试过程
    restored = NodeManifest.from_dict(manifest.to_dict())

    # 验证结果
    assert restored.to_dict() == manifest.to_dict()


def test_manifest_rejects_unknown_required_field() -> None:
    """缺少必填字段的字典加载时报错（不静默使用默认值）。"""
    # 数据：缺 capability。
    payload = {"id": "n", "name": "x", "version": "1", "command": ["python"]}

    # 测试过程与验证结果
    with pytest.raises((KeyError, TypeError, ValueError)):
        NodeManifest.from_dict(payload)


# ---------------------------------------------------------------------------
# InvokeRequest / InvokeResponse
# ---------------------------------------------------------------------------


def test_invoke_request_defaults_and_round_trip() -> None:
    """调用请求的默认输出目录为当前目录，序列化往返一致。"""
    # 数据：最小请求。
    request = InvokeRequest(run_id="r1", node_instance_id="n1")

    # 测试过程
    restored = InvokeRequest.from_dict(request.to_dict())

    # 验证结果
    assert request.output_dir == "."
    assert restored.inputs == {} and restored.params == {}
    assert restored.run_id == "r1"


def test_invoke_response_success_and_failure_shapes() -> None:
    """响应分成功（outputs）与失败（error）两种形态，序列化保留状态。"""
    # 数据：成功与失败各一。
    ok = InvokeResponse(status="completed", outputs={"text": "hi"})
    bad = InvokeResponse(status="failed", error="boom")

    # 测试过程
    ok_round = InvokeResponse.from_dict(ok.to_dict())
    bad_round = InvokeResponse.from_dict(bad.to_dict())

    # 验证结果
    assert ok_round.outputs == {"text": "hi"} and ok_round.error is None
    assert bad_round.status == "failed" and bad_round.error == "boom"


def test_invoke_response_defaults() -> None:
    """响应默认字段为空（避免 None 参与下游拼接）。"""
    # 数据：仅给状态。
    response = InvokeResponse(status="completed")

    # 测试过程与验证结果
    assert response.outputs == {}
    assert response.error is None


# ---------------------------------------------------------------------------
# HealthResponse / ProgressEvent
# ---------------------------------------------------------------------------


def test_health_response_round_trip() -> None:
    """健康响应字段往返一致（节点身份与版本，无 from_dict 时按字典比对）。"""
    # 数据：健康响应。
    health = HealthResponse(status="ok", node_id="echo", version="0.1.0")

    # 测试过程
    payload = health.to_dict()

    # 验证结果：包含状态、节点 ID 与版本。
    assert payload == {"status": "ok", "node_id": "echo", "version": "0.1.0"}


def test_progress_event_round_trip() -> None:
    """进度事件含运行、节点、进度值与消息。"""
    # 数据：一个进度事件。
    event = ProgressEvent(run_id="r1", node_id="asr", progress=0.5, message="转写中")

    # 测试过程
    payload = event.to_dict()

    # 验证结果
    assert payload == {
        "run_id": "r1", "node_id": "asr", "progress": 0.5, "message": "转写中",
    }


# ---------------------------------------------------------------------------
# WorkflowNode / WorkflowEdge / WorkflowDefinition
# ---------------------------------------------------------------------------


def test_workflow_node_preserves_all_params_keys() -> None:
    """节点的 params 完整保留（含 `_note_*` 参数说明键，不被清洗）。"""
    # 数据：含说明键的参数。
    payload = {
        "id": "asr", "node_type": "faster-whisper",
        "inputs": {"audio_uri": "extract.audio_uri"},
        "params": {"chunk_seconds": 60, "_note_chunk_seconds": "分块理由", "_node_help": "手册"},
    }

    # 测试过程
    node = WorkflowNode.from_dict(payload)

    # 验证结果：参数与说明键都在。
    assert node.params["chunk_seconds"] == 60
    assert node.params["_note_chunk_seconds"] == "分块理由"
    assert node.params["_node_help"] == "手册"
    assert node.to_dict()["params"] == payload["params"]


def test_workflow_edge_round_trip() -> None:
    """边记录来源与目标节点，输出使用 from/to 短字段名。"""
    # 数据：一条边。
    edge = WorkflowEdge.from_dict({"from": "a", "to": "b"})

    # 测试过程与验证结果
    assert edge.from_node == "a"
    assert edge.to_node == "b"
    assert edge.to_dict() == {"from": "a", "to": "b"}


def test_workflow_definition_validate_accepts_valid_dag() -> None:
    """合法 DAG 通过结构校验。"""
    # 数据：a → b 线性链。
    definition = WorkflowDefinition.from_dict({
        "name": "流程", "version": 1,
        "nodes": [
            {"id": "a", "node_type": "echo", "inputs": {}},
            {"id": "b", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
        ],
        "edges": [{"from": "a", "to": "b"}],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"out": "b.file_uri"},
    })

    # 测试过程与验证结果：不抛异常。
    definition.validate()


def test_workflow_definition_validate_rejects_duplicate_node_ids() -> None:
    """节点 ID 重复被拒绝。"""
    # 数据：两个同 ID 节点。
    definition = WorkflowDefinition.from_dict({
        "name": "流程", "version": 1,
        "nodes": [{"id": "a", "node_type": "echo", "inputs": {}}, {"id": "a", "node_type": "echo", "inputs": {}}],
        "edges": [],
    })

    # 测试过程与验证结果
    with pytest.raises(ValueError):
        definition.validate()


def test_workflow_definition_validate_rejects_edge_to_unknown_node() -> None:
    """边引用不存在的节点被拒绝。"""
    # 数据：边指向 ghost。
    definition = WorkflowDefinition.from_dict({
        "name": "流程", "version": 1,
        "nodes": [{"id": "a", "node_type": "echo", "inputs": {}}],
        "edges": [{"from": "a", "to": "ghost"}],
    })

    # 测试过程与验证结果
    with pytest.raises(ValueError):
        definition.validate()


def test_workflow_definition_validate_rejects_empty_name() -> None:
    """名称为空被拒绝。"""
    # 数据：空名称。
    definition = WorkflowDefinition.from_dict({
        "name": "", "version": 1, "nodes": [], "edges": [],
    })

    # 测试过程与验证结果
    with pytest.raises(ValueError):
        definition.validate()


def test_workflow_definition_round_trip_is_json_safe() -> None:
    """定义可安全 JSON 序列化（工作流以 JSON 存库/传输）。"""
    # 数据：含中文名称与参数的完整定义。
    definition = WorkflowDefinition.from_dict(json.loads(
        (WORKSPACE / "workflows" / "ocr-subtitle.json").read_text(encoding="utf-8")
    )["definition"])

    # 测试过程
    payload = definition.to_dict()
    text = json.dumps(payload, ensure_ascii=False)

    # 验证结果：往返一致且是合法 JSON 字符串。
    assert WorkflowDefinition.from_dict(json.loads(text)).to_dict() == payload


def test_real_builtin_workflows_are_valid() -> None:
    """仓库全部内置工作流定义通过协议校验（数据资产可用）。"""
    # 数据：真实 workflows/*.json。
    files = sorted((WORKSPACE / "workflows").glob("*.json"))
    assert files, "仓库应包含内置工作流定义"

    # 测试过程与验证结果
    for path in files:
        payload = json.loads(path.read_text(encoding="utf-8"))
        definition = WorkflowDefinition.from_dict(payload["definition"])
        definition.validate()
        assert definition.nodes, f"{path.name} 不应为空定义"