delivery-desk/tests/test_upload_skill.py

#!/usr/bin/env python3
"""Regression tests for the project-embedded safe upload Skill."""

from __future__ import annotations

import argparse
import contextlib
import importlib.util
import io
import json
import sys
import tempfile
from pathlib import Path


sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / ".agents" / "skills" / "upload-delivery-desk-work" / "scripts" / "delivery_desk_upload.py"
SPEC = importlib.util.spec_from_file_location("delivery_desk_upload", SCRIPT)
assert SPEC and SPEC.loader
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)

PROJECT = {"group_id": 7, "group_name": "精确运营组", "id": 12, "name": "光影项目", "slug": "light-project", "status": "active"}
IDENTITY = MODULE.project_identity(PROJECT)
CONTENT = {"title": "作品标题", "description": "正文", "tags": ["#测试"], "images": ["https://cdn.example.com/01.jpg"]}


def make_plan(operation: str) -> dict:
    value = {"schema_version": 1, "operation": operation, "base_url": "http://example.invalid", "target": IDENTITY, "content": CONTENT}
    if operation == "create_work":
        value["external_id"] = "agent-test-001"
    else:
        value["work"] = {"work_id": 34, "external_id": "existing-34", "expected_version_number": 2, "current_round": 2, "next_round": 3}
    value["confirmation_code"] = MODULE.confirmation_code(value)
    return value


def write_plan(folder: Path, value: dict) -> Path:
    path = folder / "plan.json"
    path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
    return path


def detail(version: int, *, work_id: int = 88, image_url: str | None = None) -> dict:
    return {"id": work_id, "project": {"id": 12}, "external_id": "agent-test-001", "version_number": version, "title": CONTENT["title"], "description": CONTENT["description"], "tags": CONTENT["tags"], "images": [{"url": image_url or CONTENT["images"][0]}], "rounds": [{"round_number": version}]}


def apply_silently(path: Path, code: str) -> dict:
    output = io.StringIO()
    with contextlib.redirect_stdout(output):
        MODULE.cmd_apply(argparse.Namespace(plan=str(path), confirm=code))
    return json.loads(output.getvalue())


def main() -> None:
    original_exact_project = MODULE.exact_project
    original_request_json = MODULE.request_json
    original_get_work = MODULE.get_work
    MODULE.exact_project = lambda _base, _project_id: PROJECT
    try:
        with tempfile.TemporaryDirectory() as folder_name:
            folder = Path(folder_name)
            work_plan = make_plan("create_work")
            work_path = write_plan(folder, work_plan)

            calls: list[tuple] = []
            MODULE.request_json = lambda *_args, **_kwargs: calls.append((_args, _kwargs))
            try:
                apply_silently(work_path, "WRONG")
                raise AssertionError("错误确认码未被阻止")
            except MODULE.UploadError as error:
                assert "确认码不匹配" in str(error)
            assert not calls

            MODULE.request_json = lambda *_args, **_kwargs: (200, [{"id": 88, "external_id": "agent-test-001"}])
            result = apply_silently(work_path, work_plan["confirmation_code"])
            assert result["idempotent"] is True and result["work"]["id"] == 88

            create_calls: list[tuple[str, str]] = []
            def create_request(_base: str, path: str, *, method: str = "GET", body=None):
                create_calls.append((method, path))
                return (200, []) if method == "GET" else (201, {"id": 88})
            MODULE.request_json = create_request
            MODULE.get_work = lambda *_args, **_kwargs: detail(1, image_url="https://cos.example.com/imports/normalized.jpg")
            result = apply_silently(work_path, work_plan["confirmation_code"])
            assert result["work_id"] == 88 and result["image_count"] == 1
            assert [method for method, _path in create_calls].count("POST") == 1

            round_plan = make_plan("create_round")
            round_path = write_plan(folder, round_plan)
            MODULE.get_work = lambda *_args, **_kwargs: detail(3, work_id=34)
            post_calls: list[tuple] = []
            MODULE.request_json = lambda *_args, **_kwargs: post_calls.append((_args, _kwargs))
            try:
                apply_silently(round_path, round_plan["confirmation_code"])
                raise AssertionError("版本漂移未被阻止")
            except MODULE.UploadError as error:
                assert "当前版本已变化" in str(error)
            assert not post_calls

            work_reads = iter([detail(2, work_id=34), detail(3, work_id=34)])
            MODULE.get_work = lambda *_args, **_kwargs: next(work_reads)
            MODULE.request_json = lambda *_args, **_kwargs: (201, {"id": 34})
            result = apply_silently(round_path, round_plan["confirmation_code"])
            assert result["work_id"] == 34 and result["round_number"] == 3

            try:
                MODULE.assert_work_project({"project": {"id": 99}}, 12)
                raise AssertionError("错误项目归属未被阻止")
            except MODULE.UploadError as error:
                assert "不是已确认项目" in str(error)
    finally:
        MODULE.exact_project = original_exact_project
        MODULE.request_json = original_request_json
        MODULE.get_work = original_get_work
    print("Upload Skill regression tests passed")


if __name__ == "__main__":
    main()