harness_coding_docs/scripts/test_gitea_claim_race.py

#!/usr/bin/env python3
"""Compatibility smoke for Gitea's same-name claim branch race behavior."""

from __future__ import annotations

import argparse
import os
import sys
import threading
import urllib.parse
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Any

from setup_gitea_labels import ApiError, GiteaClient, validate_config


PROBE_PREFIX = "claims/__probe__/race-"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="并发创建唯一临时 claim 分支，smoke 期望一个 201、一个 409。"
    )
    parser.add_argument("--repo", help="目标 owner/repo；也可设置 GITEA_REPOSITORY。")
    parser.add_argument(
        "--apply",
        action="store_true",
        help="执行两次写入并清理临时分支；省略时只读并打印计划。",
    )
    return parser.parse_args()


def new_probe_branch() -> str:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return f"{PROBE_PREFIX}{stamp}-{uuid.uuid4().hex[:12]}"


def branch_commit(client: GiteaClient, branch: str) -> str | None:
    encoded = urllib.parse.quote(branch, safe="")
    try:
        response = client.request("GET", f"/branches/{encoded}")
    except ApiError as exc:
        if exc.status == 404:
            return None
        raise
    if not isinstance(response, dict):
        raise RuntimeError("Gitea branch 响应格式异常。")
    commit = response.get("commit")
    if not isinstance(commit, dict) or not isinstance(commit.get("id"), str):
        raise RuntimeError("Gitea branch 响应缺少 commit.id。")
    return commit["id"]


def repository_base(client: GiteaClient) -> tuple[str, str]:
    repository = client.request("GET", "")
    if not isinstance(repository, dict) or not isinstance(
        repository.get("default_branch"), str
    ):
        raise RuntimeError("Gitea repository 响应缺少 default_branch。")
    default_branch = repository["default_branch"]
    commit = branch_commit(client, default_branch)
    if commit is None:
        raise RuntimeError("默认分支不存在。")
    return default_branch, commit


def create_once(client: GiteaClient, barrier: threading.Barrier, branch: str, ref: str) -> int:
    barrier.wait(timeout=10)
    try:
        client.request(
            "POST",
            "/branches",
            {"new_branch_name": branch, "old_ref_name": ref},
        )
        return 201
    except ApiError as exc:
        return exc.status


def cleanup_probe(client: GiteaClient, branch: str, expected_sha: str) -> None:
    if not branch.startswith(PROBE_PREFIX):
        raise RuntimeError("拒绝清理非探针分支。")
    actual_sha = branch_commit(client, branch)
    if actual_sha is None:
        return
    if actual_sha != expected_sha:
        raise RuntimeError("探针分支 SHA 与预期不一致，已保留供人工检查。")
    encoded = urllib.parse.quote(branch, safe="")
    client.request("DELETE", f"/branches/{encoded}")
    if branch_commit(client, branch) is not None:
        raise RuntimeError("探针分支清理后仍然存在。")


def client_for(root: str, owner: str, repo: str, token: str) -> GiteaClient:
    return GiteaClient(root, owner, repo, token)


def run_probe(root: str, owner: str, repo: str, token: str, branch: str, sha: str) -> list[int]:
    barrier = threading.Barrier(2)
    clients = [client_for(root, owner, repo, token) for _ in range(2)]
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(create_once, client, barrier, branch, sha) for client in clients
        ]
        return sorted(future.result(timeout=40) for future in futures)


def main() -> int:
    args = parse_args()
    repo_value = args.repo or os.environ.get("GITEA_REPOSITORY")
    if not repo_value:
        print("ERROR: 需要 --repo owner/repo 或 GITEA_REPOSITORY。", file=sys.stderr)
        return 2
    token = os.environ.get("GITEA_TOKEN", "")
    try:
        root, owner, repo = validate_config(
            os.environ.get("GITEA_URL", ""), token, repo_value
        )
        control = client_for(root, owner, repo, token)
        default_branch, sha = repository_base(control)
        branch = new_probe_branch()
        print(f"目标仓库：{repo_value}")
        print(f"基准分支：{default_branch} @ {sha}")
        print(f"临时分支：{branch}")
        if not args.apply:
            print("dry-run：未写入；追加 --apply 才会执行竞态探针和受控清理。")
            return 0

        results: list[int] = []
        probe_error: Exception | None = None
        try:
            results = run_probe(root, owner, repo, token, branch, sha)
        except Exception as exc:  # cleanup still has to run after partial writes
            probe_error = exc
        try:
            cleanup_probe(control, branch, sha)
        except (ApiError, RuntimeError) as cleanup_error:
            print(f"ERROR: 清理失败：{cleanup_error}", file=sys.stderr)
            return 2
        if probe_error is not None:
            print("ERROR: 竞态请求未完整返回；临时分支已安全清理。", file=sys.stderr)
            return 2
        print("竞态结果：" + ", ".join(str(status) for status in results))
        if results != [201, 409]:
            print(
                "ERROR: 未得到恰好一个 201 和一个 409；目标实例不符合预期 smoke，临时分支已安全清理。",
                file=sys.stderr,
            )
            return 1
        print("claim 并发兼容性 smoke 通过；这不证明线性化，临时分支已删除并确认 404。")
        return 0
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    except ApiError as exc:
        print(f"ERROR: Gitea 探针失败（HTTP {exc.status}）。", file=sys.stderr)
        return 2
    except RuntimeError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())