✨ [+feature] Added GitHub CI/CD (#22)

Author: davidbrownell

TODO.md

@@ -0,0 +1,4 @@
+## Archived Status - Whether the repo is archived
+
+Already available (archived field)
+Useful visual indicator (maybe gray out row)

src/AllGitStatus/MainApp.py

@@ -44,6 +44,7 @@ class Column:
 IssuesColumn = Column(8, "Issues", "center")
 PullRequestsColumn = Column(9, "PRs", "center")
 SecurityAlertsColumn = Column(10, "Security", "center")
+CICDStatusColumn = Column(11, "CI/CD", "center")
 
 COLUMN_MAP: dict[
     tuple[
@@ -63,6 +64,7 @@ class Column:
     (GitHubSource.__name__, "issues"): IssuesColumn,
     (GitHubSource.__name__, "pull_requests"): PullRequestsColumn,
     (GitHubSource.__name__, "security_alerts"): SecurityAlertsColumn,
+    (GitHubSource.__name__, "cicd_status"): CICDStatusColumn,
 }
 
 
@@ -329,7 +331,7 @@ def _PopulateCell(self, repository_index: int, info: ResultInfo | ErrorInfo) ->
         column = COLUMN_MAP[info.key]
 
         if isinstance(info, ErrorInfo):
-            display_value = "❌"
+            display_value = "💥"
             additional_info = Traceback.from_exception(
                 type(info.error),
                 info.error,

src/AllGitStatus/Sources/GitHubSource.py

@@ -1,5 +1,7 @@
 # noqa: D100
 import re
+import textwrap
+
 from collections.abc import AsyncGenerator
 
 import aiohttp
@@ -45,7 +47,9 @@ async def Query(self, repo: Repository) -> AsyncGenerator[ResultInfo | ErrorInfo
 
         github_url = repo.remote_url.removesuffix(".git")
 
-        async for info in self._GenerateStandardInfo(repo, github_url):
+        persisted_info: dict[str, object] = {}
+
+        async for info in self._GenerateStandardInfo(repo, github_url, persisted_info):
             yield info
 
         async for info in self._GenerateIssueInfo(repo, github_url):
@@ -57,13 +61,21 @@ async def Query(self, repo: Repository) -> AsyncGenerator[ResultInfo | ErrorInfo
         async for info in self._GenerateSecurityAlertInfo(repo, github_url):
             yield info
 
+        default_branch = persisted_info.get("default_branch")
+
+        # Only generate CI/CD info if we have a default branch (repo API succeeded)
+        if isinstance(default_branch, str):
+            async for info in self._GenerateCICDInfo(repo, github_url, default_branch):
+                yield info
+
     # ----------------------------------------------------------------------
     # ----------------------------------------------------------------------
     # ----------------------------------------------------------------------
     async def _GenerateStandardInfo(
         self,
         repo: Repository,
         github_url: str,
+        persisted_info: dict[str, object],
     ) -> AsyncGenerator[ResultInfo | ErrorInfo]:
         try:
             async with self._session.get(
@@ -72,6 +84,9 @@ async def _GenerateStandardInfo(
                 response.raise_for_status()
                 result = await response.json()
 
+                # Capture default_branch for use by CI/CD status
+                persisted_info["default_branch"] = result.get("default_branch")
+
                 yield ResultInfo(
                     repo,
                     (self.__class__.__name__, "stars"),
@@ -135,7 +150,7 @@ async def _GenerateIssueInfo(
 
                 label_str = f" [{', '.join(issue_labels)}]" if issue_labels else ""
 
-                issue_data.append(f"  • #{issue_number}{label_str} {issue_title} (by {issue_author})")
+                issue_data.append(f"• #{issue_number}{label_str} {issue_title} (by {issue_author})")
 
             # Build additional info with issue details
             additional_info_lines = [
@@ -191,7 +206,7 @@ async def _GeneratePullRequestInfo(
 
                 draft_indicator = "[DRAFT] " if pr_draft else ""
 
-                pr_data.append(f"  • #{pr_number} {draft_indicator}{pr_title} (by {pr_author})")
+                pr_data.append(f"• #{pr_number} {draft_indicator}{pr_title} (by {pr_author})")
 
             additional_info_lines = [
                 f"Pull Requests: {github_url}/pulls",
@@ -246,7 +261,7 @@ async def _GenerateSecurityAlertInfo(
                 package = alert.get("security_vulnerability", {}).get("package", {})
 
                 alert_data.append(
-                    f"  • [{advisory.get('severity', 'unknown').upper()}] {package.get('name', 'unknown')}: {advisory.get('summary', 'No summary')}"
+                    f"• [{advisory.get('severity', 'unknown').upper()}] {package.get('name', 'unknown')}: {advisory.get('summary', 'No summary')}"
                 )
 
             # Build display value with icon based on severity
@@ -290,6 +305,118 @@ async def _GenerateSecurityAlertInfo(
                 ex,
             )
 
+    # ----------------------------------------------------------------------
+    async def _GenerateCICDInfo(
+        self,
+        repo: Repository,
+        github_url: str,
+        default_branch: str,
+    ) -> AsyncGenerator[ResultInfo | ErrorInfo]:
+        info_key = (self.__class__.__name__, "cicd_status")
+
+        try:
+            url = f"https://api.github.com/repos/{repo.github_owner}/{repo.github_repo}/actions/runs?branch={default_branch}&per_page=100"
+
+            async with self._session.get(url) as response:
+                response.raise_for_status()
+                result = await response.json()
+
+                workflow_runs = result.get("workflow_runs", [])
+
+                if not workflow_runs:
+                    yield ResultInfo(
+                        repo,
+                        info_key,
+                        "🔘",
+                        textwrap.dedent(
+                            """\
+                                CI/CD Status: {github_url}/actions
+
+                                No workflow runs found for branch '{default_branch}'
+                                """,
+                        ).format(
+                            github_url=github_url,
+                            default_branch=default_branch,
+                        ),
+                    )
+                    return
+
+                # Group by workflow name and take only the most recent run for each
+                latest_per_workflow: dict[str, dict] = {}
+
+                for run in workflow_runs:
+                    workflow_id = run["workflow_id"]
+
+                    if workflow_id not in latest_per_workflow:
+                        latest_per_workflow[workflow_id] = run
+
+                # Count statuses based on most recent run per workflow
+                status_counts = {
+                    "success": 0,
+                    "failure": 0,
+                    "in_progress": 0,
+                }
+
+                run_details: list[str] = []
+
+                for run in latest_per_workflow.values():
+                    conclusion = run.get("conclusion")
+                    status = run.get("status")
+
+                    # Determine the effective status
+                    if status in ("in_progress", "queued", "pending", "waiting"):
+                        status_counts["in_progress"] += 1
+                        status_label = "IN PROG"
+                    elif conclusion == "success":
+                        status_counts["success"] += 1
+                        status_label = " PASS  "
+                    elif conclusion in ("failure", "cancelled", "timed_out"):
+                        status_counts["failure"] += 1
+                        status_label = " FAIL  "
+                    else:
+                        status_label = conclusion or status or "UNKNOWN"
+
+                    run_details.append(f"• [{status_label}] {run['created_at']} {run['path']}: {run['name']}")
+
+                # Determine display icon based on priority: failure > in_progress > success
+                if status_counts["failure"] > 0:
+                    display_icon = "❌"
+                elif status_counts["in_progress"] > 0:
+                    display_icon = "⏳"
+                elif status_counts["success"] > 0:
+                    display_icon = "✅"
+                else:
+                    display_icon = "🔘"
+
+                # Build additional info
+                additional_info_lines = [
+                    f"CI/CD Status: {github_url}/actions",
+                    "",
+                    f"Default Branch: {default_branch}",
+                    "",
+                    "Summary (latest run per workflow):",
+                    f"  Successful:  {status_counts['success']}",
+                    f"  Failed:      {status_counts['failure']}",
+                    f"  In Progress: {status_counts['in_progress']}",
+                    "",
+                ]
+
+                additional_info_lines.extend(run_details)
+
+                yield ResultInfo(
+                    repo,
+                    info_key,
+                    display_icon,
+                    "\n".join(additional_info_lines),
+                )
+
+        except Exception as ex:
+            yield ErrorInfo(
+                repo,
+                info_key,
+                ex,
+            )
+
     # ----------------------------------------------------------------------
     async def _GeneratePaginatedResults(self, raw_url: str) -> AsyncGenerator[dict]:
         url: str | None = f"{raw_url}?state=open&per_page=100"

tests/MainApp_integration_test.py

@@ -17,6 +17,7 @@
 from AllGitStatus.MainApp import (
     COLUMN_MAP,
     BranchColumn,
+    CICDStatusColumn,
     Column,
     ForksColumn,
     IssuesColumn,
@@ -655,7 +656,7 @@ async def mock_enum(wd):
 
                 # Verify the cell was updated with error indicator
                 cell_value = app._data_table.get_cell_at(Coordinate(0, BranchColumn.value))
-                assert "❌" in str(cell_value)
+                assert "💥" in str(cell_value)
 
                 # The additional_info_data should contain a Traceback
                 assert 0 in app._additional_info_data
@@ -958,6 +959,7 @@ def test_column_map_contains_all_columns(self) -> None:
             IssuesColumn,
             PullRequestsColumn,
             SecurityAlertsColumn,
+            CICDStatusColumn,
         }
 
         assert unique_columns == expected_columns

tests/Sources/GitHubSource_integration_test.py

@@ -194,7 +194,7 @@ async def test_query_returns_valid_results(
 
         # Verify all display values contain numeric counts (for ResultInfo items)
         for result in results:
-            if isinstance(result, ResultInfo):
+            if isinstance(result, ResultInfo) and result.key[1] != "cicd_status":
                 number_part = result.display_value.split()[0]
                 assert number_part.isdigit(), f"Expected number in '{result.display_value}'"
 
@@ -221,13 +221,15 @@ async def test_query_nonexistent_repo_returns_error(self, session: aiohttp.Clien
 
         results = [info async for info in source.Query(repo)]
 
-        # All four API calls fail for nonexistent repos (stars, issues, pull_requests, security_alerts)
+        # Four API calls fail for nonexistent repos (stars, issues, pull_requests, security_alerts)
+        # CI/CD is skipped because we don't have default_branch when repo API fails
         assert len(results) == 4
-        assert all(isinstance(r, ErrorInfo) for r in results)
         assert results[0].key == ("GitHubSource", "stars")
         assert results[1].key == ("GitHubSource", "issues")
         assert results[2].key == ("GitHubSource", "pull_requests")
         assert results[3].key == ("GitHubSource", "security_alerts")
+        # All results are errors
+        assert all(isinstance(r, ErrorInfo) for r in results)
 
 
 # ----------------------------------------------------------------------