Add markdown output and tests

Author: Yann-P

.github/workflows/ci.yml

@@ -0,0 +1,19 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: astral-sh/setup-uv@v5
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Run tests
+        run: uv run pytest tests/

.gitignore

@@ -0,0 +1,8 @@
+.claude
+.vscode
+.venv
+.pytest_cache
+.ruff_cache
+__pycache__
+*.egg-info
+task*.md

pyproject.toml

@@ -0,0 +1,16 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "security-overview"
+version = "0.1.0"
+requires-python = ">=3.10"
+dependencies = ["httpx", "trio"]
+
+[project.optional-dependencies]
+dev = ["pytest", "pytest-trio", "respx"]
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["security_overview*"]

security-overview

@@ -1,257 +1,5 @@
 #!/usr/bin/env python3
-import argparse
-import os
-import sys
-from datetime import datetime, timezone
-
-import httpx
 import trio
+from security_overview.cli import main
 
-ALL_STATES = ["published", "draft", "triage", "closed"]
-
-RESET = "\033[0m"
-BOLD = "\033[1m"
-GREY = "\033[2m"
-
-STATE_COLORS = {
-    "draft":     "\033[36m",  # cyan
-    "triage":    "\033[33m",  # yellow
-    "published": "\033[32m",  # green
-    "closed":    GREY,
-}
-
-PR_STATE_COLORS = {
-    "open":   "\033[32m",  # green
-    "merged": "\033[35m",  # purple
-    "closed": "\033[31m",  # red
-}
-
-
-def pr_state(pull):
-    if pull is None:
-        return None
-    if pull.get("merged_at"):
-        return "merged"
-    return pull.get("state", "?")  # "open" or "closed"
-
-
-def plasma_rgb(t):
-    """Plasma color: t=0 → purple (old), t=1 → yellow (recent)."""
-    r = int(13 + (253 - 13) * t)
-    g = int(8 + (231 - 8) * t)
-    b = int(135 + (37 - 135) * t)
-    return r, g, b
-
-
-def age_t(date_str):
-    dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
-    age = (datetime.now(timezone.utc) - dt).days
-    return max(0.0, 1.0 - age / 365)
-
-
-def age_badge(date_str, text):
-    """Render text on a plasma background; foreground chosen for legibility."""
-    t = age_t(date_str) if date_str else 0.0
-    r, g, b = plasma_rgb(t)
-    luminance = 0.299 * r + 0.587 * g + 0.114 * b
-    fg = "\033[30m" if luminance > 128 else "\033[97m"
-    bg = f"\033[48;2;{r};{g};{b}m"
-    return f"{bg}{fg}{text}{RESET}"
-
-
-def redact_str(s):
-    """Replace alphanumeric characters with '•', preserving spaces and punctuation."""
-    return "".join("•" if c.isalnum() else c for c in s)
-
-
-GITHUB_API = "https://api.github.com"
-
-
-def check_token():
-    token = os.environ.get("GITHUB_TOKEN")
-    if not token:
-        print("error: GITHUB_TOKEN environment variable not set", file=sys.stderr)
-        sys.exit(1)
-    return token
-
-
-async def fetch_first_pull(client, fork_html_url, results):
-    """Fetch the first pull request (by creation date) for a private fork repo."""
-    path = fork_html_url.split("github.com/", 1)[-1].rstrip("/")
-    url = f"{GITHUB_API}/repos/{path}/pulls"
-    params = {"per_page": 1, "state": "all", "sort": "created", "direction": "asc"}
-    try:
-        req = client.build_request("GET", url, params=params)
-        resp = await client.send(req)
-        if resp.status_code == 200:
-            pulls = resp.json()
-            results[fork_html_url] = pulls[0] if pulls else None
-        else:
-            results[fork_html_url] = None
-    except httpx.HTTPError:
-        results[fork_html_url] = None
-
-
-async def fetch(client, org, state, results):
-    """Fetch all advisories for (org, state), following pagination, storing in results[(org, state)]."""
-    url = f"{GITHUB_API}/orgs/{org}/security-advisories"
-    params = {"state": state, "per_page": 100}
-    advisories = []
-    while url is not None:
-        try:
-            req = client.build_request("GET", url, params=params)
-            resp = await client.send(req)
-        except httpx.HTTPError as e:
-            print(f"[error] {org} ({state}): {e}", file=sys.stderr)
-            break
-        if resp.status_code != 200:
-            print(
-                f"[error] {org} ({state}): HTTP {resp.status_code} {resp.text.strip()}",
-                file=sys.stderr,
-            )
-            break
-        try:
-            page = resp.json()
-        except ValueError:
-            break
-        if isinstance(page, list):
-            advisories.extend(page)
-        next_link = resp.links.get("next")
-        url = next_link["url"] if next_link else None
-        params = None  # next URL already has the pagination cursor
-    results[(org, state)] = advisories
-
-
-def link(text, target):
-    return f"\033[4m\033]8;;{target}\033\\{text}\033]8;;\033\\\033[24m"
-
-
-def color(text, color):
-    return f"{color}{text}{RESET}"
-
-
-def print_org(org, states, results_by_key, pull_results=None, redact=False):
-    advisories = []
-    for state in states:
-        advisories.extend(results_by_key.get((org, state), []))
-    print(f"\n{BOLD}=== {org if not redact else 'REDACTED ORG'} ==={RESET}")
-    if not advisories:
-        if not redact:
-            print("  (no advisories)")
-        return
-
-    by_repo = {}
-    for advisory in advisories:
-        repo = (advisory.get("html_url") or "").split("/")[4] or "unknown"
-        by_repo.setdefault(repo, []).append(advisory)
-
-    advisories_by_repo = sorted(
-        by_repo.items(),
-        key=lambda x: max(a.get("updated_at", "") for a in x[1]),
-        reverse=True,
-    )
-
-    for repo, items in advisories_by_repo:
-        items = sorted(items, key=lambda a: a.get("updated_at", ""), reverse=True)
-        print(f"{BOLD}{repo if not redact else 'REDACTED REPO'}{RESET}")
-        for advisory in items:
-            id = advisory.get("ghsa_id", "") if not redact else "GHSA-xxxx-yyyy-zzzz"
-            url = advisory.get("html_url", "")
-            date = advisory.get("updated_at", "")
-            title = (advisory.get("summary") or "")[:40].ljust(40)
-            state_val = advisory.get("state") or "?"
-            state_str = color(state_val.ljust(9), STATE_COLORS.get(state_val, ""))
-            cve = advisory.get("cve_id") or ""
-            fork = advisory.get("private_fork")
-            fork_html_url = fork.get("html_url") if fork else None
-            cve_str = cve[:15].ljust(15)
-            cve_str = color(cve_str, GREY)
-            first_pull = (
-                (pull_results or {}).get(fork_html_url) if fork_html_url else None
-            )
-            ps = pr_state(first_pull)
-            pull_str = (
-                color(link("PR " + ps.ljust(6), first_pull["html_url"]), PR_STATE_COLORS.get(ps, GREY))
-                if first_pull else " " * 9
-            )
-            days_ago = (
-                (
-                    datetime.now(timezone.utc)
-                    - datetime.fromisoformat(date.replace("Z", "+00:00"))
-                ).days
-                if date
-                else "?"
-            )
-            badge = age_badge(date, (str(days_ago) + 'd').rjust(5))
-            if redact:
-                print(
-                    f" {badge}\t{link(id, url)}  {state_str}"
-                )
-            else:
-                print(
-                    f" {badge}\t{state_str}  {color(link(id, url), GREY)}  {title}  {cve_str}  {pull_str}"
-                )
-
-
-def parse_states(value):
-    """Accept a comma-separated list of states; validate each against ALL_STATES."""
-    states = [s.strip() for s in value.split(",") if s.strip()]
-    for s in states:
-        if s not in ALL_STATES:
-            raise argparse.ArgumentTypeError(
-                f"invalid state {s!r} (choose from {', '.join(ALL_STATES)})"
-            )
-    return states
-
-
-async def main():
-    token = check_token()
-    parser = argparse.ArgumentParser(prog="security-overview")
-    parser.add_argument("orgs", nargs="+", metavar="org")
-    parser.add_argument(
-        "--state",
-        type=parse_states,
-        action="append",
-        metavar="STATE[,STATE...]",
-        help="filter by state; repeatable and/or comma-separated (default: all states)",
-    )
-    parser.add_argument(
-        "--redact",
-        action="store_true",
-        help="hide org/repo names (as ORG/REPO), GHSA random chars, and drop title + headers",
-    )
-    args = parser.parse_args()
-    states = [s for group in (args.state or []) for s in group] or ALL_STATES
-
-    headers = {
-        "Authorization": f"Bearer {token}",
-        "Accept": "application/vnd.github+json",
-        "X-GitHub-Api-Version": "2026-03-10",
-        "User-Agent": "security-overview",
-    }
-
-    results = {}
-    pull_results = {}
-    async with httpx.AsyncClient(headers=headers, timeout=30.0) as client:
-        async with trio.open_nursery() as nursery:
-            for org in args.orgs:
-                for state in states:
-                    nursery.start_soon(fetch, client, org, state, results)
-
-        fork_urls = {
-            advisory["private_fork"]["html_url"]
-            for advisories in results.values()
-            for advisory in advisories
-            if advisory.get("private_fork") and advisory["private_fork"].get("html_url")
-        }
-        async with trio.open_nursery() as nursery:
-            for fork_url in fork_urls:
-                nursery.start_soon(fetch_first_pull, client, fork_url, pull_results)
-
-    for org in args.orgs:
-        print_org(org, states, results, pull_results=pull_results, redact=args.redact)
-    print()
-
-
-if __name__ == "__main__":
-    trio.run(main)
+trio.run(main)

security_overview/__init__.py

@@ -0,0 +1,79 @@
+import argparse
+
+import httpx
+import trio
+
+from . import render_md, render_terminal
+from .constants import ALL_STATES
+from .fetch import check_token, fetch, fetch_first_pull
+
+RENDERERS = {
+    "terminal": render_terminal,
+    "md": render_md,
+}
+
+
+def parse_states(value):
+    states = [s.strip() for s in value.split(",") if s.strip()]
+    for s in states:
+        if s not in ALL_STATES:
+            raise argparse.ArgumentTypeError(
+                f"invalid state {s!r} (choose from {', '.join(ALL_STATES)})"
+            )
+    return states
+
+
+async def main():
+    token = check_token()
+    parser = argparse.ArgumentParser(prog="security-overview")
+    parser.add_argument("orgs", nargs="+", metavar="org")
+    parser.add_argument(
+        "--state",
+        type=parse_states,
+        action="append",
+        metavar="STATE[,STATE...]",
+        help="filter by state; repeatable and/or comma-separated (default: all states)",
+    )
+    parser.add_argument(
+        "--redact",
+        action="store_true",
+        help="hide org/repo names, GHSA random chars, and drop title + headers",
+    )
+    parser.add_argument(
+        "--format",
+        choices=["terminal", "md"],
+        default="terminal",
+        help="output format (default: terminal)",
+    )
+    args = parser.parse_args()
+    states = [s for group in (args.state or []) for s in group] or ALL_STATES
+    renderer = RENDERERS[args.format]
+
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2026-03-10",
+        "User-Agent": "security-overview",
+    }
+
+    results = {}
+    pull_results = {}
+    async with httpx.AsyncClient(headers=headers, timeout=30.0) as client:
+        async with trio.open_nursery() as nursery:
+            for org in args.orgs:
+                for state in states:
+                    nursery.start_soon(fetch, client, org, state, results)
+
+        fork_urls = {
+            advisory["private_fork"]["html_url"]
+            for advisories in results.values()
+            for advisory in advisories
+            if advisory.get("private_fork") and advisory["private_fork"].get("html_url")
+        }
+        async with trio.open_nursery() as nursery:
+            for fork_url in fork_urls:
+                nursery.start_soon(fetch_first_pull, client, fork_url, pull_results)
+
+    for org in args.orgs:
+        print(renderer.render_org(org, states, results, pull_results=pull_results, redact=args.redact))
+    print()