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 os
import sys
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
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 make_file_plan(images: list[dict]) -> dict:
    value = {
        "schema_version": 2,
        "operation": "create_work",
        "base_url": "http://example.invalid",
        "target": IDENTITY,
        "external_id": "agent-file-001",
        "content": {**CONTENT, "image_source": "file", "images": images},
    }
    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:
    previous_base_url = os.environ.get("DELIVERY_DESK_BASE_URL")
    os.environ["DELIVERY_DESK_BASE_URL"] = "http://should-not-override.invalid"
    try:
        assert MODULE.base_url() == "http://192.168.30.90:3010"
        assert MODULE.base_url("https://delivery.example.com/") == "https://delivery.example.com"
    finally:
        if previous_base_url is None:
            os.environ.pop("DELIVERY_DESK_BASE_URL", None)
        else:
            os.environ["DELIVERY_DESK_BASE_URL"] = previous_base_url

    skill_text = (SCRIPT.parents[1] / "SKILL.md").read_text(encoding="utf-8")
    assert "请回复“确认目标”" in skill_text
    assert "Do not generate `plan-work` or `plan-round` before it." in skill_text

    original_exact_project = MODULE.exact_project
    original_request_json = MODULE.request_json
    original_request_multipart = MODULE.request_multipart
    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 error.action == "revise" and "确认码" in error.next_step
            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

            local_image = folder / "generated.png"
            local_image.write_bytes(b"local-image-content")
            descriptors = MODULE.describe_image_files([str(local_image)])
            captured: dict = {}

            class Handler(BaseHTTPRequestHandler):
                def do_POST(self):
                    captured["authorization"] = self.headers.get("Authorization")
                    captured["content_type"] = self.headers.get("Content-Type")
                    captured["body"] = self.rfile.read(int(self.headers["Content-Length"]))
                    response = json.dumps({"id": 88}).encode()
                    self.send_response(201)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(response)))
                    self.end_headers()
                    self.wfile.write(response)

                def log_message(self, _format, *_args):
                    return

            server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            previous_key = os.environ.get("DELIVERY_DESK_API_KEY")
            os.environ["DELIVERY_DESK_API_KEY"] = "dd_test"
            thread.start()
            try:
                status, payload = MODULE.request_multipart(
                    f"http://127.0.0.1:{server.server_port}",
                    "/api/projects/12/works",
                    fields={"title": "作品标题", "description": "正文", "tags": "[]"},
                    files=descriptors,
                )
                local_image.write_bytes(b"x" * len(b"local-image-content"))
                try:
                    MODULE.request_multipart(
                        f"http://127.0.0.1:{server.server_port}",
                        "/api/projects/12/works",
                        fields={"title": "作品标题", "description": "正文", "tags": "[]"},
                        files=descriptors,
                    )
                    raise AssertionError("同尺寸但哈希变化的文件未被阻止")
                except MODULE.UploadError as error:
                    assert error.action == "revise" and "发生变化" in str(error)
                local_image.write_bytes(b"local-image-content")
            finally:
                server.shutdown()
                server.server_close()
                thread.join()
                if previous_key is None:
                    os.environ.pop("DELIVERY_DESK_API_KEY", None)
                else:
                    os.environ["DELIVERY_DESK_API_KEY"] = previous_key
            assert status == 201 and payload["id"] == 88
            assert captured["authorization"] == "Bearer dd_test"
            assert "multipart/form-data" in captured["content_type"]
            assert b"local-image-content" in captured["body"] and b'name="images"' in captured["body"]

            file_plan = make_file_plan(descriptors)
            file_path = write_plan(folder, file_plan)
            MODULE.request_json = lambda *_args, **_kwargs: (200, [])
            multipart_calls: list[tuple] = []
            MODULE.request_multipart = lambda *_args, **_kwargs: (multipart_calls.append((_args, _kwargs)) or (201, {"id": 88}))
            MODULE.get_work = lambda *_args, **_kwargs: {**detail(1, image_url="https://cos.example.com/originals/generated.png"), "external_id": "agent-file-001"}
            result = apply_silently(file_path, file_plan["confirmation_code"])
            assert result["work_id"] == 88 and len(multipart_calls) == 1
            assert multipart_calls[0][1]["files"][0]["sha256"] == descriptors[0]["sha256"]

            local_image.write_bytes(b"changed-after-confirmation")
            multipart_calls.clear()
            try:
                apply_silently(file_path, file_plan["confirmation_code"])
                raise AssertionError("确认后变化的文件未被阻止")
            except MODULE.UploadError as error:
                assert error.action == "revise" and "发生变化" in str(error)
            assert not multipart_calls

            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 error.action == "ask_operator"
            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)

            revised = MODULE.http_failure("POST", "/api/projects/12/works", 422, "图片不可用")
            assert revised.action == "revise" and "重新生成计划" in revised.next_step
            unauthorized = MODULE.http_failure("GET", "/api/projects", 401, "无效 Key")
            assert unauthorized.action == "ask_operator" and "DELIVERY_DESK_API_KEY" in unauthorized.next_step
            transient_read = MODULE.http_failure("GET", "/api/projects", 503, "暂时不可用")
            assert transient_read.action == "retry" and "只读" in transient_read.next_step
            uncertain_write = MODULE.http_failure("POST", "/api/works/34/rounds", 503, "暂时不可用")
            assert uncertain_write.action == "ask_operator" and "禁止自动重试" in uncertain_write.next_step
            private_url = MODULE.http_failure("POST", "/api/projects/12/works", 400, "图片地址不允许访问私有网络")
            assert private_url.action == "revise" and "修改输入" in private_url.next_step
    finally:
        MODULE.exact_project = original_exact_project
        MODULE.request_json = original_request_json
        MODULE.request_multipart = original_request_multipart
        MODULE.get_work = original_get_work
    print("Upload Skill regression tests passed")


if __name__ == "__main__":
    main()