Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions linear-audit/dashboard_render.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,8 +165,8 @@ def generate_dashboard_content(

total_done = 0
total_sub_issues = 0
total_done_effort = 0
total_effort = 0
total_done_effort = 0.0
total_effort = 0.0

def _get_state_type(k: dict) -> str:
st = k.get("state")
Expand Down Expand Up @@ -235,7 +235,7 @@ def _get_estimate(k: dict) -> int | float:

now = now_str or datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")

if total_done == 0 and sum(ws["in_progress_count"] for ws in ws_entries) == 0:
if total_done == 0 and sum(int(ws.get("in_progress_count", 0)) for ws in ws_entries) == 0:
banner = (
"> **\U0001f504 All sub-issues are in Triage \u2014 execution has not yet begun.**\n"
"> The 0% completion across all workstreams is accurate for this starting state. "
Expand Down
33 changes: 32 additions & 1 deletion linear-audit/fetch_issues.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
import sys
import time
from pathlib import Path
from typing import Any

try:
import requests
Expand Down Expand Up @@ -178,6 +179,36 @@ def transform_to_flat(node: dict) -> dict:
}


def gql(
query: str,
variables: dict | None = None,
api_key: str | None = None,
api_url: str = LINEAR_GRAPHQL_URL,
) -> dict | None:
"""Execute a GraphQL query against the Linear API. Returns None on failure."""
if not api_url.startswith(("https://", "http://")):
raise ValueError("Invalid URL scheme: only https/http supported")
key = api_key or os.environ.get("LINEAR_API_KEY", "")
payload: dict[str, Any] = {"query": query}
if variables:
payload["variables"] = variables
try:
resp = requests.post(
api_url,
json=payload,
headers={"Content-Type": "application/json", "Authorization": key},
timeout=30,
)
result = resp.json()
if "errors" in result:
print(f" API Error: {result['errors']}", file=sys.stderr)
return None
return result
except requests.RequestException as e:
print(f" Connection/HTTP error from Linear API: {e}", file=sys.stderr)
return None


def fetch_all_issues(api_key: str, team_id: str, page_size: int = 50, max_pages: int | None = 100) -> list[dict]:
"""Fetch all issues with pagination, returned as v2 flat-shape dicts."""
# Linear API keys must NOT use the "Bearer" prefix — that prefix is
Expand All @@ -198,7 +229,7 @@ def fetch_all_issues(api_key: str, team_id: str, page_size: int = 50, max_pages:
print(f"Reached maximum page limit ({max_pages}), stopping pagination.", file=sys.stderr)
break

variables = {"teamId": team_id, "first": page_size, "after": cursor}
variables: dict[str, Any] = {"teamId": team_id, "first": page_size, "after": cursor}
resp = requests.post(
LINEAR_GRAPHQL_URL,
headers=headers,
Expand Down
1 change: 1 addition & 0 deletions linear-audit/refresh_dashboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ def gql(query: str, api_key: str | None = None, api_url: str = DEFAULT_API_URL)
else:
print(f" Failed after 3 retries: {e}", file=sys.stderr)
return None
return None


def fetch_project_issues(
Expand Down
3 changes: 2 additions & 1 deletion linear-audit/register_webhook.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import os
import secrets
import sys
from typing import Any

import requests

Expand All @@ -42,7 +43,7 @@ def gql(
if not api_url.startswith(("https://", "http://")):
raise ValueError("Invalid URL scheme: only https/http supported")
key = api_key or os.environ.get("LINEAR_API_KEY", "")
payload: dict[str, object] = {"query": query}
payload: dict[str, Any] = {"query": query}
if variables:
payload["variables"] = variables

Expand Down
4 changes: 3 additions & 1 deletion requirements-dev.txt
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
requests==2.34.2
pytest==9.1.1
ruff==0.16.4
ruff==0.16.6
pydantic==2.13.5
pytest-cov==5.0.0
bandit==1.9.4
mypy>=1.10.0
pip-audit>=2.7.0
86 changes: 67 additions & 19 deletions tests/test_comprehensive.py
Original file line number Diff line number Diff line change
@@ -1,35 +1,83 @@

import sys
sys.path.insert(0, 'linear-audit')
import json, tempfile, pathlib, os

sys.path.insert(0, "linear-audit")
import pathlib
import tempfile
from unittest.mock import Mock, patch
import fetch_issues, run_audit, remediate, refresh_dashboard, register_webhook

import fetch_issues
import refresh_dashboard
import register_webhook
import remediate
import run_audit


def test_fetch():
with patch('fetch_issues.requests.post') as m:
with patch("fetch_issues.requests.post") as m:

def se(*a, **kw):
r=Mock(); r.raise_for_status=Mock()
r.json.return_value={"data": {"team": {"name": "T", "issues": {"edges": [{"node": {"id": "1", "identifier": "LIN-1", "title": "t", "state": {"type": "backlog"}}}], "pageInfo": {"hasNextPage": False, "endCursor": None}}}}}
r = Mock()
r.raise_for_status = Mock()
r.json.return_value = {
"data": {
"team": {
"name": "T",
"issues": {
"edges": [
{"node": {"id": "1", "identifier": "LIN-1", "title": "t", "state": {"type": "backlog"}}}
],
"pageInfo": {"hasNextPage": False, "endCursor": None},
},
}
}
}
return r
m.side_effect=se

m.side_effect = se
result = fetch_issues.fetch_all_issues(api_key="k", team_id="team1", max_pages=1)
assert len(result)==1
assert len(result) == 1
from fetch_issues import transform_to_flat
flat=transform_to_flat({"id": "1", "identifier": "LIN-1", "title": "t", "state": {"type": "backlog"}, "priority": 1})
assert flat.get("identifier")=="LIN-1" or flat.get("id")=="LIN-1"

flat = transform_to_flat(
{"id": "1", "identifier": "LIN-1", "title": "t", "state": {"type": "backlog"}, "priority": 1}
)
assert flat.get("identifier") == "LIN-1" or flat.get("id") == "LIN-1"


def test_refresh():
md, entries, stats = refresh_dashboard.generate_dashboard_content([], now_str="2026-01-01 00:00 UTC")
md, _entries, _stats = refresh_dashboard.generate_dashboard_content([], now_str="2026-01-01 00:00 UTC")
assert isinstance(md, str)


def test_register():
with patch('register_webhook.gql', return_value={"data": {"webhooks": {"nodes": []}}}):
with patch("register_webhook.gql", return_value={"data": {"webhooks": {"nodes": []}}}):
assert register_webhook.list_webhooks(api_key="k") == []


def test_remediate():
c=remediate.LinearClient(api_key="k")
c = remediate.LinearClient(api_key="k")
assert "Authorization" in c.headers


def test_run_audit():
import tempfile, pathlib, json as js
import json as js

with tempfile.TemporaryDirectory() as td:
p=pathlib.Path(td)/"issues.json"
p.write_text(js.dumps([{"id": "1", "identifier": "LIN-1", "title": "t", "state": {"type": "backlog"}, "project": "p1", "estimate": 1}]))
issues=run_audit.load_issues(p)
result=run_audit.run_audit(issues)
p = pathlib.Path(td) / "issues.json"
p.write_text(
js.dumps(
[
{
"id": "1",
"identifier": "LIN-1",
"title": "t",
"state": {"type": "backlog"},
"project": "p1",
"estimate": 1,
}
]
)
)
issues = run_audit.load_issues(p)
result = run_audit.run_audit(issues)
assert isinstance(result, dict)
Loading