diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..59a901e --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,15 @@ +name: Tests + +on: + push: + pull_request: + +jobs: + unittest: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.x' + - run: python3 -m unittest -v diff --git a/linear b/linear index 413cc92..96da376 100755 --- a/linear +++ b/linear @@ -456,6 +456,42 @@ def _suggest(value: str, names: list[str]) -> str: return f" Did you mean: {', '.join(close)}?" if close else "" +def _sort_by_recent(nodes: list[dict]) -> list[dict]: + return sorted( + nodes, + key=lambda n: n.get("updatedAt") or n.get("startsAt") or "", + reverse=True, + ) + + +def _select_named_node(nodes: list[dict], value: str, kind: str, + strict: bool = False, suggest: bool = False, + skip_word: bool = False) -> dict | None: + """Select a node from already-fetched data: exact name, then substring, + newest first on ambiguity. This keeps API wrappers thin and testable.""" + matches = [n for n in nodes if (n.get("name") or "").lower() == value.lower()] + if not matches: + matches = [n for n in nodes if value.lower() in (n.get("name") or "").lower()] + if not matches: + msg = f"{kind} '{value}' not found." + if suggest: + msg += _suggest(value, [n.get("name") or "" for n in nodes]) + if strict: + raise LookupError(msg) + suffix = " Skipping." if skip_word else "" + print(f"Warning: {msg}{suffix}", file=sys.stderr) + return None + matches = _sort_by_recent(matches) + if len(matches) > 1: + suffix = "" + if kind == "milestone": + project_name = matches[0].get("_projectName", "") + suffix = f" in project '{project_name}'" if project_name else "" + print(f"Note: '{value}' matched {len(matches)} {kind}s; " + f"picked '{matches[0].get('name')}'{suffix} (most recent).", file=sys.stderr) + return matches[0] + + def resolve_project_id(api_key: str, team_id: str, value: str, strict: bool = False) -> str | None: """Resolve --project to a projectId. Accepts UUID or name (exact match @@ -472,20 +508,11 @@ def resolve_project_id(api_key: str, team_id: str, value: str, if len(value) == 36 and value.count("-") == 4: return value projects = list_team_projects(api_key, team_id) - matches = [p for p in projects if p["name"].lower() == value.lower()] - if not matches: - # Substring fallback so 'phoenix' matches 'Phoenix Horizon'. - matches = [p for p in projects if value.lower() in p["name"].lower()] - if not matches: - msg = f"project '{value}' not found.{_suggest(value, [p['name'] for p in projects])}" - if strict: - raise LookupError(msg) - print(f"Warning: {msg} Skipping.", file=sys.stderr) + match = _select_named_node(projects, value, "project", strict=strict, + suggest=True, skip_word=True) + if not match: return None - if len(matches) > 1: - print(f"Note: '{value}' matched {len(matches)} projects; " - f"picked '{matches[0]['name']}' (most recent).", file=sys.stderr) - return matches[0]["id"] + return match["id"] def resolve_milestone_id(api_key: str, team_id: str, name: str, @@ -539,22 +566,11 @@ def resolve_milestone_id(api_key: str, team_id: str, name: str, m["_projectName"] = p["name"] nodes.append(m) - matches = [m for m in nodes if m["name"].lower() == name.lower()] - if not matches: - matches = [m for m in nodes if name.lower() in m["name"].lower()] - if not matches: - msg = f"milestone '{name}' not found.{_suggest(name, [m['name'] for m in nodes])}" - if strict: - raise LookupError(msg) - print(f"Warning: {msg} Skipping.", file=sys.stderr) + match = _select_named_node(nodes, name, "milestone", strict=strict, + suggest=True, skip_word=True) + if not match: return None - matches.sort(key=lambda m: m.get("updatedAt") or "", reverse=True) - if len(matches) > 1: - proj = matches[0].get("_projectName", "") - suffix = f" in project '{proj}'" if proj else "" - print(f"Note: '{name}' matched {len(matches)} milestones; " - f"picked '{matches[0]['name']}'{suffix} (most recent).", file=sys.stderr) - return matches[0]["id"] + return match["id"] def list_team_cycles(api_key: str, team_id: str) -> list[dict]: @@ -569,7 +585,7 @@ def list_team_cycles(api_key: str, team_id: str) -> list[dict]: pageInfo { hasNextPage endCursor } nodes { id number name startsAt endsAt completedAt - issueCountHistory + issueCountHistory updatedAt } } }""" @@ -961,74 +977,65 @@ def list_team_labels(api_key: str, team_id: str) -> list[dict]: after: $cursor ) { pageInfo { hasNextPage endCursor } - nodes { id name description color } + nodes { id name description color updatedAt } } }""" return paginate_connection(api_key, query, ["issueLabels"], {"teamId": team_id}) or [] -def resolve_cycle_by_value(api_key: str, team_id: str, value: str) -> dict | None: - """Resolve a non-keyword --cycle value to a cycle node {id, name, ...}. - Accepts a UUID (passthrough), a cycle number, or a fuzzy name match. - Warns and returns None when nothing matches.""" - if not value: - return None - if len(value) == 36 and value.count("-") == 4: - return {"id": value, "name": None, "startsAt": None, "endsAt": None} - cycles = list_team_cycles(api_key, team_id) +def _select_cycle_node(cycles: list[dict], value: str, strict: bool = False) -> dict | None: matches = [c for c in cycles if (c.get("name") or "").lower() == value.lower()] if not matches and value.isdigit(): matches = [c for c in cycles if str(c.get("number")) == value] if not matches: matches = [c for c in cycles if value.lower() in (c.get("name") or "").lower()] if not matches: - print(f"Warning: cycle '{value}' not found.", file=sys.stderr) + msg = f"cycle '{value}' not found." + if strict: + raise LookupError(msg) + print(f"Warning: {msg}", file=sys.stderr) return None + matches = _sort_by_recent(matches) if len(matches) > 1: - matches.sort(key=lambda c: c.get("startsAt") or "", reverse=True) print(f"Note: '{value}' matched {len(matches)} cycles; " f"picked '{matches[0].get('name')}' (most recent).", file=sys.stderr) return matches[0] -def resolve_label_by_value(api_key: str, team_id: str, value: str) -> dict | None: +def resolve_cycle_by_value(api_key: str, team_id: str, value: str, + strict: bool = False) -> dict | None: + """Resolve a non-keyword --cycle value to a cycle node {id, name, ...}. + Accepts a UUID (passthrough), a cycle number, or a fuzzy name match. + Warns and returns None when nothing matches.""" + if not value: + return None + if len(value) == 36 and value.count("-") == 4: + return {"id": value, "name": None, "startsAt": None, "endsAt": None} + return _select_cycle_node(list_team_cycles(api_key, team_id), value, strict=strict) + + +def resolve_label_by_value(api_key: str, team_id: str, value: str, + strict: bool = False) -> dict | None: """Resolve a label id-or-name to a label node {id, name, ...}. UUID passes - through; otherwise exact (case-insensitive) name wins, then a unique - substring match. Ambiguous or missing -> warn and return None.""" + through; otherwise exact (case-insensitive) name wins, then substring. + Ambiguity picks the most-recently-updated match and warns.""" if not value: return None if len(value) == 36 and value.count("-") == 4: return {"id": value, "name": value} - labels = list_team_labels(api_key, team_id) - exact = [l for l in labels if l["name"].lower() == value.lower()] - if exact: - return exact[0] - subs = [l for l in labels if value.lower() in l["name"].lower()] - if len(subs) == 1: - return subs[0] - if len(subs) > 1: - names = ", ".join(l["name"] for l in subs[:5]) - print(f"Warning: label '{value}' is ambiguous ({names}). " - f"Use the exact name or ID.", file=sys.stderr) - return None - print(f"Warning: label '{value}' not found.", file=sys.stderr) - return None + return _select_named_node(list_team_labels(api_key, team_id), value, "label", + strict=strict) -def build_cycle_scope(api_key: str, team_id: str, - scope: str | None) -> tuple[str | None, str | None, dict | None]: - """Translate a --cycle value into an issues() filter fragment + display label. - - Returns (fragment, label, cycle_meta): - - active/next -> concrete cycle id filter, cycle name, meta dict - - none/backlog -> `cycle: { null: true }`, "Backlog (no cycle)", None - - all -> "" (no cycle constraint), "All issues", None - - / -> concrete cycle id filter, matched name, node - On an unresolvable active/next/named cycle, returns (None, None, None). - """ +def build_cycle_scope_from_resolved( + scope: str | None, + active_meta: dict | None = None, + next_meta: dict | None = None, + cycle_node: dict | None = None, +) -> tuple[str | None, str | None, dict | None]: s = (scope or "active").lower() if s in ("active", "next"): - meta = resolve_cycle_meta(api_key, team_id, s) + meta = active_meta if s == "active" else next_meta if not meta: return None, None, None return f'cycle: {{ id: {{ eq: "{meta["id"]}" }} }}', cycle_label(meta), meta @@ -1036,15 +1043,39 @@ def build_cycle_scope(api_key: str, team_id: str, return "cycle: { null: true }", "Backlog (no cycle)", None if s == "all": return "", "All issues", None - node = resolve_cycle_by_value(api_key, team_id, scope) - if not node: + if not cycle_node: return None, None, None - # A raw-UUID lookup yields a node with neither name nor number; keep showing - # the user's input there rather than collapsing to "no cycle". - label = cycle_label(node) + label = cycle_label(cycle_node) if label == "no cycle": label = scope - return f'cycle: {{ id: {{ eq: "{node["id"]}" }} }}', label, node + return f'cycle: {{ id: {{ eq: "{cycle_node["id"]}" }} }}', label, cycle_node + + +def build_cycle_scope(api_key: str, team_id: str, + scope: str | None) -> tuple[str | None, str | None, dict | None]: + """Translate a --cycle value into an issues() filter fragment + display label. + + Returns (fragment, label, cycle_meta): + - active/next -> concrete cycle id filter, cycle name, meta dict + - none/backlog -> `cycle: { null: true }`, "Backlog (no cycle)", None + - all -> "" (no cycle constraint), "All issues", None + - / -> concrete cycle id filter, matched name, node + On an unresolvable active/next/named cycle, returns (None, None, None). + """ + s = (scope or "active").lower() + if s == "active": + return build_cycle_scope_from_resolved( + scope, active_meta=resolve_cycle_meta(api_key, team_id, s) + ) + if s == "next": + return build_cycle_scope_from_resolved( + scope, next_meta=resolve_cycle_meta(api_key, team_id, s) + ) + if s in ("none", "backlog", "all"): + return build_cycle_scope_from_resolved(scope) + return build_cycle_scope_from_resolved( + scope, cycle_node=resolve_cycle_by_value(api_key, team_id, scope) + ) def cycle_date_to_iso(value: str, flag: str) -> str: @@ -1476,6 +1507,47 @@ def show_issue_detail(args, api_key, team_id): # update - all mutations on an existing issue # --------------------------------------------------------------------------- +def collect_update_identifiers(identifiers, stdin_lines=None) -> list[str]: + collected = list(identifiers or []) + for line in stdin_lines or []: + collected.extend(line.split()) + return list(dict.fromkeys(collected)) + + +def run_bulk_update(args, cfg, api_key, team_id, identifiers: list[str], + project_id=_UNSET, milestone_id=_UNSET, + resolve_issue_fn=None, apply_update_fn=None): + resolve_issue_fn = resolve_issue_fn or resolve_issue + apply_update_fn = apply_update_fn or _apply_update + total = len(identifiers) + succeeded = 0 + failed: list[str] = [] + for i, ident in enumerate(identifiers, 1): + issue = resolve_issue_fn(api_key, team_id, ident) + if not issue: + print(f"[{i}/{total}] {ident} -> not found") + failed.append(ident) + continue + print(f"[{i}/{total}] {issue['identifier']}") + try: + if apply_update_fn(args, cfg, api_key, team_id, issue, single=False, + project_id=project_id, milestone_id=milestone_id): + succeeded += 1 + else: + failed.append(ident) + except Exception as e: # never let one ticket abort the batch + print(f" error: {e}", file=sys.stderr) + failed.append(ident) + + summary = f"\n{succeeded}/{total} updated." + if failed: + summary += f" Failed: {', '.join(failed)}" + sys.stdout.flush() # keep per-issue progress ahead of the summary when piped + print(summary, file=sys.stderr) + if failed: + sys.exit(1) + + def cmd_update(args, cfg, api_key, team_id): """Dispatcher: apply the same set of changes to one or many issues. @@ -1492,12 +1564,8 @@ def cmd_update(args, cfg, api_key, team_id): print(' --proof "Sent 15/30 emails" (text)', file=sys.stderr) sys.exit(1) - identifiers = list(args.identifier or []) - if getattr(args, "stdin", False): - for line in sys.stdin: - identifiers.extend(line.split()) - # De-dupe while preserving order. - identifiers = list(dict.fromkeys(identifiers)) + stdin_lines = sys.stdin if getattr(args, "stdin", False) else None + identifiers = collect_update_identifiers(args.identifier, stdin_lines) if not identifiers: print("No issue identifier given.", file=sys.stderr) sys.exit(1) @@ -1533,33 +1601,8 @@ def cmd_update(args, cfg, api_key, team_id): sys.exit(1) return - total = len(identifiers) - succeeded = 0 - failed: list[str] = [] - for i, ident in enumerate(identifiers, 1): - issue = resolve_issue(api_key, team_id, ident) - if not issue: - print(f"[{i}/{total}] {ident} -> not found") - failed.append(ident) - continue - print(f"[{i}/{total}] {issue['identifier']}") - try: - if _apply_update(args, cfg, api_key, team_id, issue, single=False, - project_id=project_id, milestone_id=milestone_id): - succeeded += 1 - else: - failed.append(ident) - except Exception as e: # never let one ticket abort the batch - print(f" error: {e}", file=sys.stderr) - failed.append(ident) - - summary = f"\n{succeeded}/{total} updated." - if failed: - summary += f" Failed: {', '.join(failed)}" - sys.stdout.flush() # keep per-issue progress ahead of the summary when piped - print(summary, file=sys.stderr) - if failed: - sys.exit(1) + run_bulk_update(args, cfg, api_key, team_id, identifiers, + project_id=project_id, milestone_id=milestone_id) def _apply_update(args, cfg, api_key, team_id, issue, single=True, diff --git a/test_linear.py b/test_linear.py index ea5f97e..6723c97 100644 --- a/test_linear.py +++ b/test_linear.py @@ -1,14 +1,15 @@ #!/usr/bin/env python3 -"""Regression tests for the `linear` CLI. Dependency-free (stdlib unittest) so it -runs anywhere with `python3 test_linear.py` — this repo ships no pytest harness. +"""Regression tests for the `linear` CLI. -Focus: cycle_label(), the fix for numbered (nameless) Linear cycles rendering as -'None' / 'no cycle'. Linear auto-numbered cycles have name=None and only a -number, so display must fall back to `Cycle {number}` rather than collapsing a -present-but-nameless cycle to "no cycle".""" +The CLI is intentionally dependency-free, so tests use only stdlib unittest. +""" +import contextlib import importlib.util +import io import os +import sys +import types import unittest from importlib.machinery import SourceFileLoader @@ -30,12 +31,9 @@ def test_named_cycle_uses_name(self): ) def test_numbered_nameless_cycle_falls_back_to_number(self): - # The core bug: a present cycle with name=None must render as its number, - # not "no cycle". self.assertEqual(linear_cli.cycle_label({"name": None, "number": 20}), "Cycle 20") def test_number_zero_is_not_treated_as_missing(self): - # `or number` would drop cycle 0; the guard is `number is not None`. self.assertEqual(linear_cli.cycle_label({"name": None, "number": 0}), "Cycle 0") def test_none_node_is_no_cycle(self): @@ -48,5 +46,202 @@ def test_empty_name_falls_back_to_number(self): self.assertEqual(linear_cli.cycle_label({"name": "", "number": 7}), "Cycle 7") +class BulkUpdateTest(unittest.TestCase): + def test_collect_update_identifiers_dedupes_positional_and_stdin_in_order(self): + self.assertEqual( + linear_cli.collect_update_identifiers( + ["RUSH-1", "RUSH-2", "RUSH-1"], + ["RUSH-3 RUSH-2\n", "RUSH-4\n"], + ), + ["RUSH-1", "RUSH-2", "RUSH-3", "RUSH-4"], + ) + + def test_bulk_update_continues_after_missing_issue_and_rolls_up_errors(self): + args = types.SimpleNamespace() + seen = [] + + def resolve_issue(_api_key, _team_id, ident): + seen.append(ident) + if ident == "RUSH-404": + return None + return {"id": ident.lower(), "identifier": ident} + + def apply_update(_args, _cfg, _api_key, _team_id, issue, **_kwargs): + if issue["identifier"] == "RUSH-3": + raise RuntimeError("boom") + return True + + stdout = io.StringIO() + stderr = io.StringIO() + with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as raised: + linear_cli.run_bulk_update( + args, + {}, + "api-key", + "team-id", + ["RUSH-1", "RUSH-404", "RUSH-3"], + resolve_issue_fn=resolve_issue, + apply_update_fn=apply_update, + ) + + self.assertEqual(raised.exception.code, 1) + self.assertEqual(seen, ["RUSH-1", "RUSH-404", "RUSH-3"]) + self.assertIn("[1/3] RUSH-1", stdout.getvalue()) + self.assertIn("[2/3] RUSH-404 -> not found", stdout.getvalue()) + self.assertIn("[3/3] RUSH-3", stdout.getvalue()) + self.assertIn("error: boom", stderr.getvalue()) + self.assertIn("1/3 updated. Failed: RUSH-404, RUSH-3", stderr.getvalue()) + + def test_cmd_update_dedupes_stdin_and_keeps_going_after_bad_ticket(self): + args = types.SimpleNamespace( + done=False, + proof=[], + identifier=["RUSH-1", "RUSH-1", "RUSH-404"], + stdin=True, + project=None, + milestone=None, + ) + seen = [] + original_resolve_issue = linear_cli.resolve_issue + original_apply_update = linear_cli._apply_update + original_stdin = sys.stdin + + def resolve_issue(_api_key, _team_id, ident): + seen.append(ident) + if ident == "RUSH-404": + return None + return {"id": ident.lower(), "identifier": ident} + + def apply_update(_args, _cfg, _api_key, _team_id, _issue, **_kwargs): + return True + + linear_cli.resolve_issue = resolve_issue + linear_cli._apply_update = apply_update + sys.stdin = io.StringIO("RUSH-2 RUSH-2\n") + stdout = io.StringIO() + stderr = io.StringIO() + try: + with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as raised: + linear_cli.cmd_update(args, {}, "api-key", "team-id") + finally: + linear_cli.resolve_issue = original_resolve_issue + linear_cli._apply_update = original_apply_update + sys.stdin = original_stdin + + self.assertEqual(raised.exception.code, 1) + self.assertEqual(seen, ["RUSH-1", "RUSH-404", "RUSH-2"]) + self.assertIn("[1/3] RUSH-1", stdout.getvalue()) + self.assertIn("[2/3] RUSH-404 -> not found", stdout.getvalue()) + self.assertIn("[3/3] RUSH-2", stdout.getvalue()) + self.assertIn("2/3 updated. Failed: RUSH-404", stderr.getvalue()) + + +class CycleScopeTest(unittest.TestCase): + def test_active_scope_uses_active_cycle_id(self): + meta = {"id": "active-id", "name": "Active", "number": 10} + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("active", active_meta=meta), + ('cycle: { id: { eq: "active-id" } }', "Active", meta), + ) + + def test_next_scope_uses_next_cycle_id(self): + meta = {"id": "next-id", "name": None, "number": 11} + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("next", next_meta=meta), + ('cycle: { id: { eq: "next-id" } }', "Cycle 11", meta), + ) + + def test_all_scope_has_no_cycle_constraint(self): + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("all"), + ("", "All issues", None), + ) + + def test_none_and_backlog_scope_filter_for_null_cycle(self): + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("none"), + ("cycle: { null: true }", "Backlog (no cycle)", None), + ) + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("backlog"), + ("cycle: { null: true }", "Backlog (no cycle)", None), + ) + + def test_named_scope_uses_resolved_cycle_id(self): + node = {"id": "named-id", "name": "Launch", "number": 12} + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("Launch", cycle_node=node), + ('cycle: { id: { eq: "named-id" } }', "Launch", node), + ) + + def test_raw_id_scope_keeps_user_supplied_label(self): + node = {"id": "raw-id", "name": None} + self.assertEqual( + linear_cli.build_cycle_scope_from_resolved("raw-id", cycle_node=node), + ('cycle: { id: { eq: "raw-id" } }', "raw-id", node), + ) + + +class NameResolutionTest(unittest.TestCase): + def test_exact_match_beats_newer_substring_match(self): + nodes = [ + {"id": "new-substring", "name": "Launch Plan", "updatedAt": "2026-07-17T00:00:00Z"}, + {"id": "old-exact", "name": "Launch", "updatedAt": "2026-07-01T00:00:00Z"}, + ] + self.assertEqual( + linear_cli._select_named_node(nodes, "Launch", "project")["id"], + "old-exact", + ) + + def test_unknown_strict_name_raises_lookup_error(self): + with self.assertRaisesRegex(LookupError, "project 'Missing' not found"): + linear_cli._select_named_node([], "Missing", "project", strict=True) + + def test_ambiguous_name_picks_most_recent_and_warns(self): + nodes = [ + {"id": "old", "name": "Launch A", "updatedAt": "2026-07-01T00:00:00Z"}, + {"id": "new", "name": "Launch B", "updatedAt": "2026-07-16T00:00:00Z"}, + ] + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + match = linear_cli._select_named_node(nodes, "Launch", "milestone") + self.assertEqual(match["id"], "new") + self.assertIn("matched 2 milestones", stderr.getvalue()) + self.assertIn("picked 'Launch B'", stderr.getvalue()) + + def test_cycle_resolution_accepts_number_and_uses_updated_at_for_ambiguity(self): + cycles = [ + {"id": "old", "number": 1, "name": "Sprint Alpha", "updatedAt": "2026-07-01T00:00:00Z"}, + {"id": "number", "number": 42, "name": "Other", "updatedAt": "2026-07-02T00:00:00Z"}, + {"id": "new", "number": 2, "name": "Sprint Beta", "updatedAt": "2026-07-17T00:00:00Z"}, + ] + self.assertEqual(linear_cli._select_cycle_node(cycles, "42")["id"], "number") + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + match = linear_cli._select_cycle_node(cycles, "Sprint") + self.assertEqual(match["id"], "new") + self.assertIn("matched 2 cycles", stderr.getvalue()) + + def test_label_wrapper_uses_prefetched_labels_without_network_logic(self): + original = linear_cli.list_team_labels + labels = [ + {"id": "old", "name": "bug urgent", "updatedAt": "2026-07-01T00:00:00Z"}, + {"id": "new", "name": "bug backlog", "updatedAt": "2026-07-17T00:00:00Z"}, + ] + linear_cli.list_team_labels = lambda _api_key, _team_id: labels + try: + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + match = linear_cli.resolve_label_by_value("api-key", "team-id", "bug") + self.assertEqual(match["id"], "new") + self.assertIn("matched 2 labels", stderr.getvalue()) + with self.assertRaisesRegex(LookupError, "label 'missing' not found"): + linear_cli.resolve_label_by_value("api-key", "team-id", "missing", strict=True) + finally: + linear_cli.list_team_labels = original + + if __name__ == "__main__": unittest.main()