feat(harness): standardize Gitea API token priority - add GITEA_TOKEN_BAIK alias across all harness tools [WBS-10]
Validators (Pushes and Pull Requests) / validate-ui-and-storage (push) Failing after 11s
Validators (Pushes and Pull Requests) / validate-core (push) Failing after 18s

This commit is contained in:
2026-07-24 13:17:41 +09:00
parent 482dedbe22
commit 678e0cd301
4 changed files with 293 additions and 17 deletions
+102
View File
@@ -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())
+179
View File
@@ -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())
+6 -11
View File
@@ -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]:
+6 -6
View File
@@ -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]: