diff --git a/graphify/prs.py b/graphify/prs.py index 9534e6c00..a1ef3b6c5 100644 --- a/graphify/prs.py +++ b/graphify/prs.py @@ -142,6 +142,7 @@ def _gh(*args: str) -> list | dict | None: try: result = subprocess.run( ["gh", *args], + stdin=subprocess.DEVNULL, # Decode gh's output as UTF-8, not the Windows cp1252 locale codec: gh # emits UTF-8 JSON with non-Latin1 titles/logins (emoji, فارسی), and the # default text=True decode crashes on those (#1505 fixed the same in llm). @@ -157,24 +158,27 @@ def _gh(*args: str) -> list | dict | None: def _detect_default_branch(repo: str | None = None) -> str: """Auto-detect the repo's default branch via gh, then git, then fall back to 'main'.""" # Try gh first — works for any repo, not just the current directory - args = ["repo", "view", "--json", "defaultBranchRef"] + args = ["repo", "view"] if repo: - args += ["--repo", repo] + args.append(repo) + args += ["--json", "defaultBranchRef"] data = _gh(*args) if data and data.get("defaultBranchRef", {}).get("name"): return data["defaultBranchRef"]["name"] - # Fall back to git symbolic-ref for the current repo - try: - result = subprocess.run( - ["git", "symbolic-ref", "refs/remotes/origin/HEAD"], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5 - ) - if result.returncode == 0: - # refs/remotes/origin/main → main - ref = result.stdout.strip() - return ref.split("/")[-1] if ref else "main" - except (subprocess.TimeoutExpired, FileNotFoundError): - pass + # Fall back to git symbolic-ref for the current repo (only when repo is not specified) + if not repo: + try: + result = subprocess.run( + ["git", "symbolic-ref", "refs/remotes/origin/HEAD"], + stdin=subprocess.DEVNULL, + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5 + ) + if result.returncode == 0: + # refs/remotes/origin/main → main + ref = result.stdout.strip() + return ref.split("/")[-1] if ref else "main" + except (subprocess.TimeoutExpired, FileNotFoundError): + pass return "main" @@ -232,7 +236,11 @@ def fetch_pr_files(number: int, repo: str | None = None) -> list[str]: if repo: args += ["--repo", repo] try: - result = subprocess.run(["gh", *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30) + result = subprocess.run( + ["gh", *args], + stdin=subprocess.DEVNULL, + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30 + ) if result.returncode != 0: return [] return [l.strip() for l in result.stdout.splitlines() if l.strip()] @@ -303,6 +311,7 @@ def fetch_worktrees() -> dict[str, str]: try: result = subprocess.run( ["git", "worktree", "list", "--porcelain"], + stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10 ) if result.returncode != 0: diff --git a/tests/test_prs.py b/tests/test_prs.py index 4acc343e6..2bf4e9154 100644 --- a/tests/test_prs.py +++ b/tests/test_prs.py @@ -376,6 +376,27 @@ class TestDetectDefaultBranch: ): 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 ───────────────────────────────────────────────────── @@ -462,3 +483,39 @@ class TestSubprocessOutputEncoding: _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