🐛 fix(gitea-fix-ci): trace workflow_run CI chains

2026-08-20 15:56:40 +08:00
parent 5a8d95ebcf
commit 77ab8211b1
6 changed files with 429 additions and 194 deletions
@@ -0,0 +1,151 @@
+import importlib.util
+import io
+import sys
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from types import SimpleNamespace
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SCRIPT_PATH = REPO_ROOT / "skills/gitea-fix-ci/scripts/fetch_ci_logs.py"
+SPEC = importlib.util.spec_from_file_location("gitea_fetch_ci_logs", SCRIPT_PATH)
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+class FakeClient:
+    def __init__(self, payload):
+        self.payload = payload
+        self.urls = []
+
+    def get_json(self, url):
+        self.urls.append(url)
+        return self.payload
+
+
+class FetchCILogsRunsTests(unittest.TestCase):
+    def setUp(self):
+        self.target = MODULE.RepoTarget(
+            base_url="https://gitea.example.test",
+            owner="owner",
+            repo="repo",
+        )
+
+    def test_run_row_preserves_workflow_provenance(self):
+        sha = "a" * 40
+        row = MODULE._run_row(
+            {
+                "id": 42,
+                "path": "checks.yml@refs/heads/main",
+                "display_title": "CI title",
+                "event": "workflow_run",
+                "status": "completed",
+                "conclusion": "failure",
+                "head_branch": "main",
+                "head_sha": sha,
+                "started_at": "2026-08-20T10:00:00+08:00",
+                "completed_at": "2026-08-20T10:01:00+08:00",
+                "run_number": 9,
+                "run_attempt": 2,
+            }
+        )
+
+        self.assertEqual(row["name"], "checks.yml")
+        self.assertEqual(row["workflow"], "checks.yml")
+        self.assertEqual(row["path"], "checks.yml@refs/heads/main")
+        self.assertEqual(row["title"], "CI title")
+        self.assertEqual(row["sha"], sha)
+        self.assertEqual(row["run_number"], 9)
+        self.assertEqual(row["run_attempt"], 2)
+
+    def test_workflow_filter_uses_workflow_endpoint_and_all_omits_status(self):
+        sha = "b" * 40
+        args = SimpleNamespace(
+            workflow="checks.yml",
+            branch=None,
+            event="workflow_run",
+            status="all",
+            sha=None,
+            limit=30,
+            json=True,
+        )
+        client = FakeClient(
+            {
+                "workflow_runs": [
+                    {
+                        "id": 7,
+                        "path": "checks.yml@refs/heads/main",
+                        "event": "workflow_run",
+                        "conclusion": "success",
+                        "head_sha": sha,
+                    }
+                ]
+            }
+        )
+
+        stdout = io.StringIO()
+        with redirect_stdout(stdout):
+            result = MODULE.cmd_runs(args, self.target, client)
+
+        self.assertEqual(result, 0)
+        self.assertIn("/actions/workflows/checks.yml/runs?", client.urls[0])
+        self.assertIn("event=workflow_run", client.urls[0])
+        self.assertIn("limit=30", client.urls[0])
+        self.assertNotIn("status=", client.urls[0])
+        self.assertIn(sha, stdout.getvalue())
+
+    def test_sha_filter_sends_and_preserves_full_sha(self):
+        sha = "c" * 40
+        args = SimpleNamespace(
+            workflow="prepare.yml",
+            branch=None,
+            event="pull_request",
+            status="all",
+            sha=sha,
+            limit=20,
+            json=True,
+        )
+        client = FakeClient(
+            {
+                "workflow_runs": [
+                    {
+                        "id": 8,
+                        "path": "prepare.yml@refs/pull/42/head",
+                        "event": "pull_request",
+                        "conclusion": "success",
+                        "head_sha": sha,
+                    }
+                ]
+            }
+        )
+
+        stdout = io.StringIO()
+        with redirect_stdout(stdout):
+            result = MODULE.cmd_runs(args, self.target, client)
+
+        self.assertEqual(result, 0)
+        self.assertIn(f"head_sha={sha}", client.urls[0])
+        self.assertIn(sha, stdout.getvalue())
+
+    def test_sha_filter_rejects_abbreviated_sha(self):
+        args = SimpleNamespace(
+            workflow="prepare.yml",
+            branch=None,
+            event="pull_request",
+            status="all",
+            sha="abc123",
+            limit=20,
+            json=True,
+        )
+        client = FakeClient({"workflow_runs": []})
+
+        with self.assertRaisesRegex(MODULE.ConfigError, "full 40-character"):
+            MODULE.cmd_runs(args, self.target, client)
+
+        self.assertEqual(client.urls, [])
+
+
+if __name__ == "__main__":
+    unittest.main()