#!/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: 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()