diff --git a/tools/validate_gitea_pr_harness_v1.py b/tools/validate_gitea_pr_harness_v1.py new file mode 100644 index 0000000..142c87a --- /dev/null +++ b/tools/validate_gitea_pr_harness_v1.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_BASE_URL = "http://kjh2064.synology.me:8418" +DEFAULT_OWNER = "KimJaeHyun" +DEFAULT_REPO = "myfinance" + + +def _token() -> str: + # GITEA_TOKEN 환경 변수가 우선 + token = os.environ.get("GITEA_TOKEN", "") + if not token.strip(): + # 없으면 GITEA_TOKEN_HOME fallback + token = os.environ.get("GITEA_TOKEN_HOME", "") + if not token.strip(): + raise SystemExit("GITEA_TOKEN 또는 GITEA_TOKEN_HOME 환경 변수가 설정되어 있지 않습니다.") + return token.strip() + + +def _request_json(url: str, token: str, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, str, Any]: + data = None + headers = { + "Authorization": f"token {token}", + "Accept": "application/json", + } + if body is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(body).encode("utf-8") + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read().decode("utf-8", errors="replace") + payload = json.loads(raw) if raw else None + return resp.status, resp.reason, payload + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", errors="replace") + try: + payload = json.loads(raw) if raw else None + except Exception: + payload = raw + return exc.code, exc.reason or "", payload + + +def _repo_base(args: argparse.Namespace) -> str: + return f"{args.base_url.rstrip('/')}/api/v1/repos/{args.owner}/{args.repo}" + + +def main() -> int: + ap = argparse.ArgumentParser(description="Create and Validate Gitea Pull Requests automatically.") + ap.add_argument("--base-url", default=DEFAULT_BASE_URL) + ap.add_argument("--owner", default=DEFAULT_OWNER) + ap.add_argument("--repo", default=DEFAULT_REPO) + ap.add_argument("--head", required=True, help="Branch containing changes (e.g., feature/kis-token-db-cache)") + ap.add_argument("--base-branch", default="main", help="Target branch to merge changes into") + ap.add_argument("--title", default=None, help="PR Title") + ap.add_argument("--body", default="Automated PR generated by quant engine client.", help="PR Description") + args = ap.parse_args() + + token = _token() + base = _repo_base(args) + + title = args.title or f"Merge {args.head} into {args.base_branch}" + + evidence: dict[str, Any] = { + "base_url": args.base_url, + "owner": args.owner, + "repo": args.repo, + "head": args.head, + "base_branch": args.base_branch, + } + + # 1. 원격 저장소 접근 가능 여부 테스트 (Basic Auth/Token Check) + status, reason, repo_payload = _request_json(base, token) + evidence["repo_check_status"] = status + if status != 200: + print(json.dumps({"gate": "FAIL", "evidence": evidence, "error": "Repository inaccessible. Check credentials/token."}, ensure_ascii=False, indent=2)) + return 1 + + # 2. 이미 동일한 Head -> Base PR이 열려 있는지 검사 + status, reason, prs_payload = _request_json(f"{base}/pulls?state=open", token) + if status == 200 and isinstance(prs_payload, list): + for pr in prs_payload: + if pr.get("head", {}).get("ref") == args.head and pr.get("base", {}).get("ref") == args.base_branch: + evidence["pr_exists"] = True + evidence["pr_url"] = pr.get("html_url") + print(json.dumps({"gate": "PASS", "message": "PR already exists and is active.", "evidence": evidence}, ensure_ascii=False, indent=2)) + return 0 + + # 3. 신규 Pull Request 작성 API 요청 + pr_body = { + "title": title, + "body": args.body, + "head": args.head, + "base": args.base_branch, + } + + status, reason, pr_payload = _request_json(f"{base}/pulls", token, method="POST", body=pr_body) + evidence["create_status"] = status + evidence["create_reason"] = reason + + if status == 201: + evidence["pr_url"] = pr_payload.get("html_url") + result = { + "gate": "PASS", + "message": f"Successfully created Pull Request for branch: {args.head}", + "evidence": evidence + } + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + else: + # PR 생성 실패 (동일한 브랜치에 변경 사항이 없거나 권한 부족 등) + print(json.dumps({"gate": "FAIL", "evidence": evidence, "error": pr_payload}, ensure_ascii=False, indent=2)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())