From 678e0cd301455c1738187221fe54208438f3b025 Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Fri, 24 Jul 2026 13:17:41 +0900 Subject: [PATCH] feat(harness): standardize Gitea API token priority - add GITEA_TOKEN_BAIK alias across all harness tools [WBS-10] --- tools/inspect_gitea_actions_run_v1.py | 102 +++++++++++++++ tools/inspect_gitea_actions_run_v2.py | 179 ++++++++++++++++++++++++++ tools/validate_gitea_pr_harness_v1.py | 17 +-- tools/validate_gitea_token_home_v1.py | 12 +- 4 files changed, 293 insertions(+), 17 deletions(-) create mode 100644 tools/inspect_gitea_actions_run_v1.py create mode 100644 tools/inspect_gitea_actions_run_v2.py diff --git a/tools/inspect_gitea_actions_run_v1.py b/tools/inspect_gitea_actions_run_v1.py new file mode 100644 index 00000000..dd411a72 --- /dev/null +++ b/tools/inspect_gitea_actions_run_v1.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_BASE_URL = "https://gitea.taxbaik.com" +DEFAULT_OWNER = "kjh2064" +DEFAULT_REPO = "QuantEngineByItz" + + +def _get_token(cli_token: str = "") -> str: + """토큰 우선순위: CLI → GITEA_TOKEN_BAIK → GITEA_TOKEN_TAXBAIK → GITEA_TOKEN → GITEA_TOKEN_HOME""" + if cli_token and cli_token.strip(): + return cli_token.strip() + for env_key in ("GITEA_TOKEN_BAIK", "GITEA_TOKEN_TAXBAIK", "GITEA_TOKEN", "GITEA_TOKEN_HOME"): + val = os.environ.get(env_key, "").strip() + if val: + return val + return "" + + +def _request_json(url: str, token: str = "", method: str = "GET") -> tuple[int, str, Any]: + headers = { + "Accept": "application/json", + "User-Agent": "QuantEngine-Harness/1.0", + } + if token: + headers["Authorization"] = f"token {token}" + + req = urllib.request.Request(url, 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 main() -> int: + ap = argparse.ArgumentParser(description="Gitea Actions API Inspector Harness Tool") + 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("--run-id", type=int, default=2545) + ap.add_argument("--token", default=None, help="Optional Gitea API Token") + args = ap.parse_args() + + token = args.token or _get_token() + repo_url = f"{args.base_url.rstrip('/')}/api/v1/repos/{args.owner}/{args.repo}" + + # 1. Fetch Repository Info + status, reason, repo_info = _request_json(repo_url, token=token) + + # 2. Fetch Actions Runs List + runs_url = f"{repo_url}/actions/runs" + status_runs, reason_runs, runs_payload = _request_json(runs_url, token=token) + + target_run = None + if isinstance(runs_payload, dict): + workflow_runs = runs_payload.get("workflow_runs") or [] + for r in workflow_runs: + if r.get("id") == args.run_id or r.get("run_number") == args.run_id: + target_run = r + break + if not target_run and workflow_runs: + target_run = workflow_runs[0] # Fallback to latest + + result = { + "formula_id": "GITEA_ACTIONS_RUN_INSPECTOR_V1", + "gate": "PASS" if status == 200 and status_runs == 200 else "FAIL", + "repository": f"{args.owner}/{args.repo}", + "target_run_id": args.run_id, + "token_provided": bool(token), + "run_details": target_run or runs_payload, + "meta": { + "api_status": status, + "runs_status": status_runs, + "runs_reason": reason_runs + } + } + + out = ROOT / "Temp" / "gitea_actions_run_2545_v1.json" + out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if (status == 200 and status_runs == 200) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/inspect_gitea_actions_run_v2.py b/tools/inspect_gitea_actions_run_v2.py new file mode 100644 index 00000000..701d3052 --- /dev/null +++ b/tools/inspect_gitea_actions_run_v2.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +""" +Gitea Actions Run Detail Inspector v2 +AGENTS.md 지침 준수: Gitea API를 하네스로 조회. 직접 서버 접근 금지. +formula_id: GITEA_ACTIONS_RUN_DETAIL_V2 + +토큰 우선순위 (높은 것부터): + 1. --token CLI 인자 + 2. GITEA_TOKEN_BAIK (사용자 지정 별칭) + 3. GITEA_TOKEN_TAXBAIK (기존 표준) + 4. GITEA_TOKEN (일반 fallback) + 5. GITEA_TOKEN_HOME (레거시) +""" +from __future__ import annotations + +import argparse +import json +import os +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_BASE_URL = "https://gitea.taxbaik.com" +DEFAULT_OWNER = "kjh2064" +DEFAULT_REPO = "QuantEngineByItz" + + +def _get_token(cli_token: str = "") -> str: + """토큰 우선순위: CLI → GITEA_TOKEN_BAIK → GITEA_TOKEN_TAXBAIK → GITEA_TOKEN → GITEA_TOKEN_HOME""" + if cli_token and cli_token.strip(): + return cli_token.strip() + for env_key in ("GITEA_TOKEN_BAIK", "GITEA_TOKEN_TAXBAIK", "GITEA_TOKEN", "GITEA_TOKEN_HOME"): + val = os.environ.get(env_key, "").strip() + if val: + return val + return "" + + +def _request_json(url: str, token: str = "") -> tuple[int, str, Any]: + headers = { + "Accept": "application/json", + "User-Agent": "QuantEngine-Harness/2.0", + } + if token: + headers["Authorization"] = f"token {token}" + req = urllib.request.Request(url, headers=headers, method="GET") + 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) + except Exception: + payload = raw + return exc.code, exc.reason or "", payload + + +def _summary_run(run: dict) -> dict: + return { + "id": run.get("id"), + "run_number": run.get("run_number"), + "display_title": run.get("display_title"), + "workflow_path": run.get("path"), + "event": run.get("event"), + "status": run.get("status"), + "conclusion": run.get("conclusion"), + "head_branch": run.get("head_branch"), + "head_sha": run.get("head_sha"), + "started_at": run.get("started_at"), + "completed_at": run.get("completed_at"), + "actor": (run.get("actor") or {}).get("login"), + } + + +def main() -> int: + ap = argparse.ArgumentParser(description="Gitea Actions Run Detail Inspector v2") + 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("--run-id", type=int, default=2545, help="Gitea Actions run ID") + ap.add_argument("--token", default="", help="Gitea API Personal Access Token") + ap.add_argument("--all-runs", action="store_true", help="List recent N runs") + ap.add_argument("--limit", type=int, default=10, help="Number of recent runs to list when --all-runs") + args = ap.parse_args() + + token = _get_token(args.token) + base = f"{args.base_url.rstrip('/')}/api/v1/repos/{args.owner}/{args.repo}" + errors: list[str] = [] + evidence: dict[str, Any] = { + "owner": args.owner, + "repo": args.repo, + "target_run_id": args.run_id, + "token_provided": bool(token), + } + + # 1. Target run details + run_url = f"{base}/actions/runs/{args.run_id}" + s1, _, run_payload = _request_json(run_url, token=token) + evidence["target_run_http_status"] = s1 + + if s1 != 200: + errors.append(f"Run #{args.run_id} fetch failed: HTTP {s1}") + target_run_summary = None + jobs_data = None + else: + target_run_summary = _summary_run(run_payload) + + # 2. Fetch jobs for this run + jobs_url = f"{base}/actions/runs/{args.run_id}/jobs" + s2, _, jobs_payload = _request_json(jobs_url, token=token) + evidence["jobs_http_status"] = s2 + + if s2 == 200 and isinstance(jobs_payload, dict): + raw_jobs = jobs_payload.get("workflow_jobs") or [] + jobs_data = [] + for job in raw_jobs: + steps = job.get("steps") or [] + failed_steps = [ + { + "step_number": st.get("number"), + "name": st.get("name"), + "conclusion": st.get("conclusion"), + "started_at": st.get("started_at"), + "completed_at": st.get("completed_at"), + } + for st in steps + if st.get("conclusion") not in ("success", "skipped", None) + ] + jobs_data.append({ + "job_id": job.get("id"), + "job_name": job.get("name"), + "status": job.get("status"), + "conclusion": job.get("conclusion"), + "started_at": job.get("started_at"), + "completed_at": job.get("completed_at"), + "runner_name": job.get("runner_name"), + "total_steps": len(steps), + "failed_steps": failed_steps, + }) + else: + jobs_data = None + if s2 != 200: + errors.append(f"Jobs fetch failed: HTTP {s2}") + + # 3. Recent runs list (optional) + recent_runs = None + if args.all_runs: + runs_url = f"{base}/actions/runs?limit={args.limit}" + s3, _, runs_payload = _request_json(runs_url, token=token) + evidence["runs_list_http_status"] = s3 + if s3 == 200 and isinstance(runs_payload, dict): + raw_runs = runs_payload.get("workflow_runs") or [] + recent_runs = [_summary_run(r) for r in raw_runs] + else: + errors.append(f"Runs list fetch failed: HTTP {s3}") + + result = { + "formula_id": "GITEA_ACTIONS_RUN_DETAIL_V2", + "gate": "PASS" if not errors else "FAIL", + "errors": errors, + "evidence": evidence, + "run_summary": target_run_summary, + "jobs": jobs_data, + "recent_runs": recent_runs, + } + + out = ROOT / "Temp" / f"gitea_actions_run_{args.run_id}_detail_v2.json" + out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if not errors else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/validate_gitea_pr_harness_v1.py b/tools/validate_gitea_pr_harness_v1.py index 117e4de7..0bf9de27 100644 --- a/tools/validate_gitea_pr_harness_v1.py +++ b/tools/validate_gitea_pr_harness_v1.py @@ -17,17 +17,12 @@ DEFAULT_REPO = "myfinance" def _token() -> str: - # GITEA_TOKEN 환경 변수가 우선 - token = os.environ.get("GITEA_TOKEN", "") - if not token.strip(): - # 없으면 GITEA_TOKEN_TAXBAIK fallback - token = os.environ.get("GITEA_TOKEN_TAXBAIK", "") - if not token.strip(): - # legacy fallback - token = os.environ.get("GITEA_TOKEN_HOME", "") - if not token.strip(): - raise SystemExit("GITEA_TOKEN 또는 GITEA_TOKEN_TAXBAIK 환경 변수가 설정되어 있지 않습니다.") - return token.strip() + # 토큰 우선순위: GITEA_TOKEN_BAIK → GITEA_TOKEN_TAXBAIK → GITEA_TOKEN → GITEA_TOKEN_HOME + for env_key in ("GITEA_TOKEN_BAIK", "GITEA_TOKEN_TAXBAIK", "GITEA_TOKEN", "GITEA_TOKEN_HOME"): + val = os.environ.get(env_key, "").strip() + if val: + return val + raise SystemExit("GITEA_TOKEN_BAIK 또는 GITEA_TOKEN_TAXBAIK 환경 변수가 설정되어 있지 않습니다.") def _request_json(url: str, token: str, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, str, Any]: diff --git a/tools/validate_gitea_token_home_v1.py b/tools/validate_gitea_token_home_v1.py index 048474ac..a5f34ae4 100644 --- a/tools/validate_gitea_token_home_v1.py +++ b/tools/validate_gitea_token_home_v1.py @@ -18,12 +18,12 @@ DEFAULT_REPO = "myfinance" def _token() -> str: - token = os.environ.get("GITEA_TOKEN_TAXBAIK", "") - if not token.strip(): - token = os.environ.get("GITEA_TOKEN_HOME", "") - if not token.strip(): - raise SystemExit("GITEA_TOKEN_TAXBAIK missing or empty") - return token.strip() + # 토큰 우선순위: GITEA_TOKEN_BAIK → GITEA_TOKEN_TAXBAIK → GITEA_TOKEN_HOME + for env_key in ("GITEA_TOKEN_BAIK", "GITEA_TOKEN_TAXBAIK", "GITEA_TOKEN", "GITEA_TOKEN_HOME"): + val = os.environ.get(env_key, "").strip() + if val: + return val + raise SystemExit("GITEA_TOKEN_BAIK missing or empty") def _request_json(url: str, token: str, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, str, Any]: