"""Tests for graphify/prs.py.""" from __future__ import annotations import json import subprocess from datetime import datetime, timedelta, timezone from unittest.mock import patch, MagicMock import networkx as nx import pytest from graphify.prs import ( PRInfo, _classify, _gh, _parse_ci, _path_match, build_community_labels, compute_pr_impact, fetch_pr_files, fetch_worktrees, format_prs_text, _detect_default_branch, _parse_pr_dict, fetch_pr, fetch_prs, attach_graph_impact, cmd_prs, ) # ── Helpers ─────────────────────────────────────────────────────────────────── def make_pr( number: int = 1, title: str = "Test PR", branch: str = "feature", base_branch: str = "v8", author: str = "alice", is_draft: bool = False, review_decision: str = "", ci_status: str = "SUCCESS", updated_at: datetime | None = None, expected_base: str = "v8", ) -> PRInfo: """Build a minimal PRInfo with sensible defaults.""" if updated_at is None: updated_at = datetime.now(timezone.utc) - timedelta(days=1) return PRInfo( number=number, title=title, branch=branch, base_branch=base_branch, author=author, is_draft=is_draft, review_decision=review_decision, ci_status=ci_status, updated_at=updated_at, expected_base=expected_base, ) # ── _classify ───────────────────────────────────────────────────────────────── class TestClassify: def test_ready(self): pr = make_pr(ci_status="SUCCESS", review_decision="", is_draft=False) assert _classify(pr, base="v8") == "READY" def test_ci_fail(self): pr = make_pr(ci_status="FAILURE") assert _classify(pr, base="v8") == "CI-FAIL" def test_changes_req(self): pr = make_pr(ci_status="SUCCESS", review_decision="CHANGES_REQUESTED") assert _classify(pr, base="v8") == "CHANGES-REQ" def test_draft(self): pr = make_pr(ci_status="SUCCESS", is_draft=True) assert _classify(pr, base="v8") == "DRAFT" def test_stale(self): old = datetime.now(timezone.utc) - timedelta(days=20) pr = make_pr(ci_status="SUCCESS", updated_at=old, is_draft=False) assert _classify(pr, base="v8") == "STALE" def test_draft_not_marked_stale(self): # Drafts show as DRAFT even when old — stale-detection only applies to non-drafts old = datetime.now(timezone.utc) - timedelta(days=20) pr = make_pr(ci_status="SUCCESS", updated_at=old, is_draft=True) assert _classify(pr, base="v8") == "DRAFT" def test_pending(self): pr = make_pr(ci_status="PENDING", is_draft=False, review_decision="") assert _classify(pr, base="v8") == "PENDING" def test_wrong_base(self): # WRONG-BASE takes precedence over everything else pr = make_pr(base_branch="master", ci_status="FAILURE") assert _classify(pr, base="v8") == "WRONG-BASE" # ── _parse_ci ───────────────────────────────────────────────────────────────── class TestParseCi: def test_empty_rollup_returns_none(self): assert _parse_ci([]) == "NONE" def test_failure_conclusion(self): rollup = [{"conclusion": "FAILURE", "status": "COMPLETED"}] assert _parse_ci(rollup) == "FAILURE" def test_cancelled_is_failure(self): rollup = [{"conclusion": "CANCELLED", "status": "COMPLETED"}] assert _parse_ci(rollup) == "FAILURE" def test_timed_out_is_failure(self): rollup = [{"conclusion": "TIMED_OUT", "status": "COMPLETED"}] assert _parse_ci(rollup) == "FAILURE" def test_in_progress_is_pending(self): rollup = [{"conclusion": None, "status": "IN_PROGRESS"}] assert _parse_ci(rollup) == "PENDING" def test_success(self): rollup = [{"conclusion": "SUCCESS", "status": "COMPLETED"}] assert _parse_ci(rollup) == "SUCCESS" def test_mixed_success_and_failure_is_failure(self): rollup = [ {"conclusion": "SUCCESS", "status": "COMPLETED"}, {"conclusion": "FAILURE", "status": "COMPLETED"}, ] assert _parse_ci(rollup) == "FAILURE" # ── _path_match ─────────────────────────────────────────────────────────────── class TestPathMatch: def test_exact_match(self): assert _path_match("src/auth/api.py", "src/auth/api.py") is True def test_graph_path_longer_with_boundary(self): # graph_src is longer, ends with "/" + pr_file assert _path_match("src/auth/api.py", "api.py") is True def test_no_false_positive_on_partial_filename(self): # "config.py" should NOT match "g.py" — must be at path boundary assert _path_match("config.py", "g.py") is False assert _path_match("g.py", "config.py") is False def test_both_directions_work(self): # pr_file longer than graph_src assert _path_match("api.py", "src/auth/api.py") is True # graph_src longer than pr_file assert _path_match("src/auth/api.py", "api.py") is True # ── compute_pr_impact ───────────────────────────────────────────────────────── class TestComputePrImpact: def _make_graph(self) -> nx.Graph: """3 nodes across 2 communities, 2 distinct source files.""" G = nx.Graph() G.add_node("n1", source_file="src/auth/api.py", community=0) G.add_node("n2", source_file="src/auth/api.py", community=0) G.add_node("n3", source_file="src/utils/helpers.py", community=1) return G def test_matching_files_returns_correct_communities_and_count(self): G = self._make_graph() comms, nodes = compute_pr_impact(["src/auth/api.py"], G) assert comms == [0] assert nodes == 2 def test_matching_both_files(self): G = self._make_graph() comms, nodes = compute_pr_impact( ["src/auth/api.py", "src/utils/helpers.py"], G ) assert comms == [0, 1] assert nodes == 3 def test_empty_files_returns_empty(self): G = self._make_graph() comms, nodes = compute_pr_impact([], G) assert comms == [] assert nodes == 0 def test_no_matching_files_returns_empty(self): G = self._make_graph() comms, nodes = compute_pr_impact(["docs/README.md"], G) assert comms == [] assert nodes == 0 def test_no_double_counting_when_basename_matches_multiple_paths(self): # "api.py" should NOT match both src/auth/api.py AND src/admin/api.py G = nx.Graph() G.add_node("a1", source_file="src/auth/api.py", community=0) G.add_node("a2", source_file="src/admin/api.py", community=1) comms, nodes = compute_pr_impact(["src/auth/api.py"], G) # Only src/auth/api.py matches by exact path — not src/admin/api.py assert nodes == 1 assert comms == [0] def test_no_double_counting_same_graph_file_matched_by_two_pr_files(self): # If PR diff lists both "api.py" and "src/auth/api.py", the graph node # for src/auth/api.py should only be counted once G = nx.Graph() G.add_node("n1", source_file="src/auth/api.py", community=0) G.add_node("n2", source_file="src/auth/api.py", community=0) comms, nodes = compute_pr_impact(["src/auth/api.py", "api.py"], G) assert nodes == 2 # 2 nodes in that file, counted once assert comms == [0] # ── fetch_worktrees ─────────────────────────────────────────────────────────── class TestFetchWorktrees: def test_normal_case_maps_branch_to_path(self): porcelain = ( "worktree /home/user/proj\n" "HEAD abc123\n" "branch refs/heads/main\n" "\n" "worktree /home/user/proj-feature\n" "HEAD def456\n" "branch refs/heads/feature-x\n" "\n" ) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = porcelain with patch("graphify.prs.subprocess.run", return_value=mock_result): mapping = fetch_worktrees() assert mapping == { "main": "/home/user/proj", "feature-x": "/home/user/proj-feature", } def test_detached_head_does_not_leak_into_next_record(self): """A detached HEAD (no branch line) must not associate its path with the next record's branch — the blank line separator resets state.""" porcelain = ( "worktree /home/user/detached\n" "HEAD abc123\n" "detached\n" "\n" "worktree /home/user/proj-feature\n" "HEAD def456\n" "branch refs/heads/feature-x\n" "\n" ) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = porcelain with patch("graphify.prs.subprocess.run", return_value=mock_result): mapping = fetch_worktrees() # Only feature-x should be mapped, and it should point to its own worktree assert mapping == {"feature-x": "/home/user/proj-feature"} assert "/home/user/detached" not in mapping.values() def test_empty_output_returns_empty_dict(self): mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" with patch("graphify.prs.subprocess.run", return_value=mock_result): mapping = fetch_worktrees() assert mapping == {} def test_nonzero_returncode_returns_empty_dict(self): mock_result = MagicMock() mock_result.returncode = 1 mock_result.stdout = "" with patch("graphify.prs.subprocess.run", return_value=mock_result): mapping = fetch_worktrees() assert mapping == {} def test_subprocess_failure_returns_empty_dict(self): with patch( "graphify.prs.subprocess.run", side_effect=FileNotFoundError("git not found"), ): mapping = fetch_worktrees() assert mapping == {} # ── format_prs_text ─────────────────────────────────────────────────────────── class TestFormatPrsText: def test_contains_pr_metadata_and_count_header(self): prs = [ make_pr( number=101, title="Add awesome feature", base_branch="v8", expected_base="v8", ci_status="SUCCESS", ), make_pr( number=102, title="Fix flaky test", base_branch="v8", expected_base="v8", ci_status="FAILURE", ), make_pr( number=103, title="Wrong base PR", base_branch="master", expected_base="v8", ), ] out = format_prs_text(prs, base="v8") # Count header: 2 actionable, 1 on wrong base assert "Open PRs targeting v8: 2" in out assert "(1 on wrong base, not shown)" in out # PR numbers and titles included assert "#101" in out assert "Add awesome feature" in out assert "#102" in out assert "Fix flaky test" in out # Statuses included assert "[READY]" in out assert "[CI-FAIL]" in out # Wrong-base PR should be filtered out of body assert "#103" not in out def test_empty_pr_list(self): out = format_prs_text([], base="v8") assert "Open PRs targeting v8: 0" in out assert "(0 on wrong base, not shown)" in out # ── _detect_default_branch ──────────────────────────────────────────────────── class TestDetectDefaultBranch: def test_gh_returns_main(self): with patch( "graphify.prs._gh", return_value={"defaultBranchRef": {"name": "main"}}, ): assert _detect_default_branch() == "main" def test_falls_back_to_git_symbolic_ref(self): mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "refs/remotes/origin/develop\n" with patch("graphify.prs._gh", return_value=None), patch( "graphify.prs.subprocess.run", return_value=mock_result ): assert _detect_default_branch() == "develop" def test_both_fail_returns_main(self): mock_result = MagicMock() mock_result.returncode = 1 mock_result.stdout = "" with patch("graphify.prs._gh", return_value=None), patch( "graphify.prs.subprocess.run", return_value=mock_result ): assert _detect_default_branch() == "main" def test_gh_returns_empty_dict_falls_back(self): """gh returns data but with no defaultBranchRef — should still fall back.""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "refs/remotes/origin/trunk\n" with patch("graphify.prs._gh", return_value={}), patch( "graphify.prs.subprocess.run", return_value=mock_result ): assert _detect_default_branch() == "trunk" def test_git_timeout_returns_main(self): with patch("graphify.prs._gh", return_value=None), patch( "graphify.prs.subprocess.run", side_effect=subprocess.TimeoutExpired("git", 5), ): assert _detect_default_branch() == "main" def test_repo_positional_arg_in_gh_repo_view(self): """gh repo view takes repo positionally; --repo is not a valid flag (#3318).""" with patch( "graphify.prs._gh", return_value={"defaultBranchRef": {"name": "main"}}, ) as mock_gh: branch = _detect_default_branch(repo="owner/repo") assert branch == "main" args, _ = mock_gh.call_args assert args == ("repo", "view", "owner/repo", "--json", "defaultBranchRef") assert "--repo" not in args def test_explicit_repo_gh_fails_returns_main_without_local_git(self): """When an explicit repo is supplied and gh fails, return 'main' without inspecting the local repository (#3318).""" with patch("graphify.prs._gh", return_value=None), \ patch("graphify.prs.subprocess.run") as mock_git: branch = _detect_default_branch(repo="owner/repo") assert branch == "main" mock_git.assert_not_called() # ── build_community_labels ───────────────────────────────────────────────────── class TestBuildCommunityLabels: def test_basic_grouping(self): data = { "nodes": [ {"id": "a", "label": "Alpha", "community": 0}, {"id": "b", "label": "Beta", "community": 0}, {"id": "c", "label": "Gamma", "community": 1}, ] } labels = build_community_labels(data) assert set(labels[0]) == {"Alpha", "Beta"} assert labels[1] == ["Gamma"] def test_top_n_capped(self): nodes = [{"id": str(i), "label": f"Node{i}", "community": 0} for i in range(10)] labels = build_community_labels({"nodes": nodes}, top_n=4) assert len(labels[0]) == 4 def test_no_community_field_skipped(self): data = {"nodes": [{"id": "x", "label": "X"}]} assert build_community_labels(data) == {} def test_empty_nodes(self): assert build_community_labels({}) == {} assert build_community_labels({"nodes": []}) == {} # ── Windows cp1252 subprocess-decode hardening (decode-side sibling of #1505) ── class TestSubprocessOutputEncoding: """prs.py reads gh/git/claude output via subprocess.run(text=True). Without an explicit encoding= the stdout is decoded with the locale codec (cp1252 on Windows), so non-Latin1 output (emoji, non-ASCII PR titles/logins/paths) fails: the reader thread raises UnicodeDecodeError, subprocess.run returns stdout=None, and the caller then dies on json.loads(None) / None.splitlines(). Every call must pass encoding="utf-8" so decoding matches Linux/macOS. #1505 fixed the same defect on the llm.py claude-cli subprocess. """ _NON_LATIN1 = "docs: add Persian (فارسی) 🏆" def test_fixture_is_cp1252_undecodable(self): """Guard: the fixture's UTF-8 bytes must be undecodable as cp1252, else these tests would prove nothing about the failure surface.""" raw = self._NON_LATIN1.encode("utf-8") with pytest.raises(UnicodeDecodeError): raw.decode("cp1252") assert raw.decode("utf-8") == self._NON_LATIN1 def test_gh_decodes_output_as_utf8(self): completed = MagicMock( returncode=0, stdout=json.dumps([{"title": self._NON_LATIN1}]), stderr="" ) with patch("subprocess.run", return_value=completed) as mock_run: data = _gh("pr", "list", "--json", "title") _args, kwargs = mock_run.call_args assert kwargs.get("encoding") == "utf-8", ( f"_gh subprocess must use encoding='utf-8'; got {kwargs.get('encoding')!r}" ) assert data == [{"title": self._NON_LATIN1}] def test_fetch_pr_files_decodes_output_as_utf8(self): completed = MagicMock(returncode=0, stdout="src/café.py\n", stderr="") with patch("subprocess.run", return_value=completed) as mock_run: fetch_pr_files(1) _args, kwargs = mock_run.call_args assert kwargs.get("encoding") == "utf-8" def test_fetch_worktrees_decodes_output_as_utf8(self): completed = MagicMock(returncode=0, stdout="", stderr="") with patch("subprocess.run", return_value=completed) as mock_run: fetch_worktrees() _args, kwargs = mock_run.call_args assert kwargs.get("encoding") == "utf-8" def test_detect_default_branch_decodes_output_as_utf8(self): # Force the git symbolic-ref fallback: gh returns None -> git subprocess runs. completed = MagicMock(returncode=0, stdout="refs/remotes/origin/v8\n", stderr="") with patch("graphify.prs._gh", return_value=None), \ patch("subprocess.run", return_value=completed) as mock_run: _detect_default_branch() _args, kwargs = mock_run.call_args assert kwargs.get("encoding") == "utf-8" # ── Subprocess stdin isolation (#3318) ───────────────────────────────────────── class TestSubprocessStdinIsolation: """Subprocesses in prs.py must isolate stdin with subprocess.DEVNULL so they never inherit an open stdin pipe from the MCP stdio server (#3318).""" def test_gh_passes_devnull_stdin(self): completed = MagicMock(returncode=0, stdout="[]", stderr="") with patch("subprocess.run", return_value=completed) as mock_run: _gh("repo", "view") _args, kwargs = mock_run.call_args assert kwargs.get("stdin") == subprocess.DEVNULL def test_fetch_pr_files_passes_devnull_stdin(self): completed = MagicMock(returncode=0, stdout="", stderr="") with patch("subprocess.run", return_value=completed) as mock_run: fetch_pr_files(1) _args, kwargs = mock_run.call_args assert kwargs.get("stdin") == subprocess.DEVNULL def test_fetch_worktrees_passes_devnull_stdin(self): completed = MagicMock(returncode=0, stdout="", stderr="") with patch("subprocess.run", return_value=completed) as mock_run: fetch_worktrees() _args, kwargs = mock_run.call_args assert kwargs.get("stdin") == subprocess.DEVNULL def test_detect_default_branch_git_fallback_passes_devnull_stdin(self): completed = MagicMock(returncode=0, stdout="refs/remotes/origin/main\n", stderr="") with patch("graphify.prs._gh", return_value=None), \ patch("subprocess.run", return_value=completed) as mock_run: _detect_default_branch() _args, kwargs = mock_run.call_args assert kwargs.get("stdin") == subprocess.DEVNULL # ── _parse_pr_dict & fetch_pr (#3824) ────────────────────────────────────────── class TestParsePrDict: def test_parse_pr_dict_complete(self): # A recent updatedAt so status stays APPROVED (a hardcoded date eventually # crosses the staleness window and _classify returns STALE first). recent = datetime.now(timezone.utc) - timedelta(days=1) raw = { "number": 42, "title": "Add feature", "headRefName": "feat-42", "baseRefName": "v8", "author": {"login": "alice"}, "isDraft": False, "reviewDecision": "APPROVED", "statusCheckRollup": [{"conclusion": "SUCCESS", "status": "COMPLETED"}], "updatedAt": recent.strftime("%Y-%m-%dT%H:%M:%SZ"), } pr = _parse_pr_dict(raw, expected_base="v8") assert pr.number == 42 assert pr.title == "Add feature" assert pr.branch == "feat-42" assert pr.base_branch == "v8" assert pr.author == "alice" assert pr.is_draft is False assert pr.review_decision == "APPROVED" assert pr.ci_status == "SUCCESS" assert pr.status == "APPROVED" assert pr.updated_at.year == recent.year def test_parse_pr_dict_defaults_on_missing_optional_fields(self): raw = { "number": 7, "title": "Minimal PR", "headRefName": "patch-1", "baseRefName": "main", } pr = _parse_pr_dict(raw, expected_base="main") assert pr.author == "?" assert pr.is_draft is False assert pr.review_decision == "" assert pr.ci_status == "NONE" class TestFetchPr: def test_fetch_pr_success(self): raw = { "number": 105, "title": "Fix something", "state": "OPEN", "headRefName": "fix-105", "baseRefName": "v8", "author": {"login": "bob"}, "isDraft": False, "reviewDecision": "", "statusCheckRollup": [], "updatedAt": "2026-09-24T12:00:00Z", } with patch("graphify.prs._gh", return_value=raw) as mock_gh: pr = fetch_pr(105, base="v8") assert pr is not None assert pr.number == 105 assert pr.title == "Fix something" args, _ = mock_gh.call_args assert args[0:3] == ("pr", "view", "105") def test_fetch_pr_with_repo(self): with patch("graphify.prs._gh", return_value=None) as mock_gh: fetch_pr(105, repo="owner/repo", base="v8") args, _ = mock_gh.call_args assert "--repo" in args assert "owner/repo" in args def test_fetch_pr_not_found_returns_none(self): with patch("graphify.prs._gh", return_value=None): pr = fetch_pr(999, base="v8") assert pr is None @pytest.mark.parametrize("state", ["CLOSED", "MERGED"]) def test_fetch_pr_closed_or_merged_returns_none(self, state: str): raw = { "number": 105, "title": "Old PR", "state": state, "headRefName": "feature", "baseRefName": "v8", "updatedAt": "2026-09-24T12:00:00Z", } with patch("graphify.prs._gh", return_value=raw): pr = fetch_pr(105, base="v8") assert pr is None # ── attach_graph_impact & include_wrong_base (#3824) ────────────────────────── class TestAttachGraphImpact: def _sample_graph_data(self): return { "nodes": [ {"id": "n1", "source_file": "src/auth.py", "community": 0, "label": "auth"}, {"id": "n2", "source_file": "src/auth.py", "community": 0, "label": "login"}, {"id": "n3", "source_file": "src/api.py", "community": 1, "label": "api"}, ] } def test_attach_graph_impact_skips_wrong_base_by_default(self, tmp_path): graph_file = tmp_path / "graph.json" graph_file.write_text(json.dumps(self._sample_graph_data()), encoding="utf-8") pr_ready = make_pr(number=1, base_branch="v8", expected_base="v8") pr_wrong = make_pr(number=2, base_branch="develop", expected_base="v8") with patch("graphify.prs.fetch_pr_files", return_value=["src/auth.py"]) as mock_files: attach_graph_impact([pr_ready, pr_wrong], graph_file, include_wrong_base=False) mock_files.assert_called_once_with(1, None) assert pr_ready.nodes_affected == 2 assert pr_ready.communities_touched == [0] assert pr_wrong.nodes_affected == 0 assert pr_wrong.communities_touched == [] def test_attach_graph_impact_includes_wrong_base_when_requested(self, tmp_path): graph_file = tmp_path / "graph.json" graph_file.write_text(json.dumps(self._sample_graph_data()), encoding="utf-8") pr_wrong = make_pr(number=2, base_branch="develop", expected_base="v8") assert pr_wrong.status == "WRONG-BASE" with patch("graphify.prs.fetch_pr_files", return_value=["src/auth.py"]) as mock_files: attach_graph_impact([pr_wrong], graph_file, include_wrong_base=True) mock_files.assert_called_once_with(2, None) assert pr_wrong.nodes_affected == 2 assert pr_wrong.communities_touched == [0] assert pr_wrong.status == "WRONG-BASE" assert pr_wrong.blast_radius == "2 nodes / 1 community" # ── cmd_prs single-PR detail mode (#3824) ───────────────────────────────────── class TestCmdPrsDetail: def test_detail_only_fetches_diff_for_requested_pr(self, tmp_path, capsys): graph_file = tmp_path / "graph.json" graph_data = { "nodes": [ {"id": "n1", "source_file": "src/app.py", "community": 0, "label": "app"}, ] } graph_file.write_text(json.dumps(graph_data), encoding="utf-8") pr1 = make_pr(number=10, base_branch="v8", expected_base="v8") pr2 = make_pr(number=20, base_branch="v8", expected_base="v8") pr3 = make_pr(number=30, base_branch="v8", expected_base="v8") with patch("graphify.prs.fetch_prs", return_value=[pr1, pr2, pr3]), \ patch("graphify.prs.fetch_worktrees", return_value={}), \ patch("graphify.prs.fetch_pr_files", return_value=["src/app.py"]) as mock_diff: cmd_prs(["20", f"--graph={graph_file}"]) # Crucial check for #3824: only PR 20's diff must be fetched, NOT 10 or 30 assert mock_diff.call_count == 1 assert mock_diff.call_args[0][0] == 20 captured = capsys.readouterr() assert "PR #20" in captured.out assert "Graph impact:" in captured.out def test_detail_calculates_impact_for_wrong_base_pr(self, tmp_path, capsys): graph_file = tmp_path / "graph.json" graph_data = { "nodes": [ {"id": "n1", "source_file": "src/app.py", "community": 0, "label": "app"}, ] } graph_file.write_text(json.dumps(graph_data), encoding="utf-8") pr_wrong = make_pr(number=42, base_branch="feature-x", expected_base="v8") assert pr_wrong.status == "WRONG-BASE" with patch("graphify.prs.fetch_prs", return_value=[pr_wrong]), \ patch("graphify.prs.fetch_worktrees", return_value={}), \ patch("graphify.prs.fetch_pr_files", return_value=["src/app.py"]) as mock_diff: cmd_prs(["42", f"--graph={graph_file}"]) assert mock_diff.call_count == 1 assert mock_diff.call_args[0][0] == 42 captured = capsys.readouterr() assert "PR #42" in captured.out assert "WRONG-BASE" in captured.out assert "Graph impact:" in captured.out def test_detail_falls_back_to_fetch_pr_when_outside_first_50(self, tmp_path, capsys): graph_file = tmp_path / "graph.json" graph_data = { "nodes": [ {"id": "n1", "source_file": "src/app.py", "community": 0, "label": "app"}, ] } graph_file.write_text(json.dumps(graph_data), encoding="utf-8") # 50 PRs that do not include PR 99 prs_50 = [make_pr(number=i, base_branch="v8") for i in range(1, 51)] pr_99 = make_pr(number=99, branch="feat-99", base_branch="v8", expected_base="v8") with patch("graphify.prs.fetch_prs", return_value=prs_50), \ patch("graphify.prs._detect_default_branch", return_value="v8"), \ patch("graphify.prs.fetch_worktrees", return_value={"feat-99": "/path/to/wt"}), \ patch("graphify.prs.fetch_pr", return_value=pr_99) as mock_fetch_single, \ patch("graphify.prs.fetch_pr_files", return_value=["src/app.py"]): cmd_prs(["99", f"--graph={graph_file}"]) mock_fetch_single.assert_called_once_with(99, repo=None, base="v8") assert pr_99.worktree_path == "/path/to/wt" captured = capsys.readouterr() assert "PR #99" in captured.out assert "worktree:" in captured.out assert "/path/to/wt" in captured.out def test_detail_exits_when_not_found_in_list_or_view(self, capsys): prs_50 = [make_pr(number=i) for i in range(1, 51)] with patch("graphify.prs.fetch_prs", return_value=prs_50), \ patch("graphify.prs.fetch_worktrees", return_value={}), \ patch("graphify.prs.fetch_pr", return_value=None): with pytest.raises(SystemExit) as exc_info: cmd_prs(["999"]) assert exc_info.value.code == 1 captured = capsys.readouterr() assert "PR #999 not found in open PRs." in captured.err