diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 70074661..f1d2ea8f 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -5,6 +5,32 @@ on: pull_request: jobs: + # ruff and mypy are configured in pyproject.toml but nothing ran them, so + # violations accumulated silently. This job keeps that config honest. + lint: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install lint deps + run: | + python -m pip install --upgrade pip + pip install -c constraints.txt -e .[dev] + + - name: Ruff + run: | + ruff check . + + - name: Mypy + run: | + mypy agent_reach + test: runs-on: ubuntu-latest strategy: diff --git a/agent_reach/channels/web.py b/agent_reach/channels/web.py index 9d10dfe1..9c772438 100644 --- a/agent_reach/channels/web.py +++ b/agent_reach/channels/web.py @@ -2,6 +2,7 @@ """Web — any URL via Jina Reader. Always available.""" import urllib.request + from .base import Channel _UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" diff --git a/agent_reach/cli.py b/agent_reach/cli.py index 7e662383..7b117a53 100644 --- a/agent_reach/cli.py +++ b/agent_reach/cli.py @@ -9,10 +9,10 @@ agent-reach setup """ -import sys import argparse import json import os +import sys import time from agent_reach import __version__ @@ -20,6 +20,10 @@ # Pinned to the 0.4.2 state — PyPI still only has 0.4.1 (upstream issue #10). _RDT_GIT_SOURCE = "git+https://github.com/public-clis/rdt-cli.git@5e4fb3720d5c174e976cd425ccc3b879d52cac66" +# Module-level so tests can redirect them instead of touching the real system. +_GH_KEYRING_PATH = "/usr/share/keyrings/githubcli-archive-keyring.gpg" +_GH_APT_LIST_PATH = "/etc/apt/sources.list.d/github-cli.list" + def _ensure_utf8_console(): """Best-effort Windows console UTF-8 setup for CLI runtime only.""" @@ -277,9 +281,9 @@ def _cmd_install(args): env = _detect_environment() if env == "server": - print(f"Environment: Server/VPS (auto-detected)") + print("Environment: Server/VPS (auto-detected)") else: - print(f"Environment: Local computer (auto-detected)") + print("Environment: Local computer (auto-detected)") server_skipped_opencli_channels = set() if env == "server" and requested_channels: @@ -295,7 +299,7 @@ def _cmd_install(args): else: config.set("proxy", args.proxy) config.set("bilibili_proxy", args.proxy) # legacy key - print(f"✅ 代理已保存(Agent 访问受限网络时使用)") + print("✅ 代理已保存(Agent 访问受限网络时使用)") # ── Install core system dependencies (lightweight, always) ── print() @@ -412,9 +416,9 @@ def _cmd_install(args): def _install_skill(force: bool = True): """Install Agent Reach as an agent skill (OpenClaw / Claude Code / .agents).""" + import importlib.resources import os import shutil - import importlib.resources def _is_english_locale(value: str) -> bool: normalized = value.strip().lower() @@ -587,60 +591,200 @@ def _cmd_format(args): print(json.dumps(cleaned, ensure_ascii=False, indent=2)) +def _run_checked(cmd, *, timeout, **kwargs) -> bool: + """Run a command and report whether it actually succeeded. + + Installer steps chain: an apt source naming a keyring that the previous + step failed to fetch breaks `apt-get update` for the whole machine, not + just for Agent Reach, and the breakage surfaces long after this installer + exits. Every step therefore has to prove it exited cleanly before the next + one is allowed to run. + """ + import subprocess + + try: + result = subprocess.run(cmd, capture_output=True, timeout=timeout, **kwargs) + except (OSError, subprocess.TimeoutExpired): + return False + return result.returncode == 0 + + +def _can_modify_system(action: str) -> bool: + """Return whether this process may make privileged system-wide changes. + + Writing to /etc and running apt-get needs root. Without this check a + non-root run turns every step into a PermissionError that the caller + reports as a generic "install failed", hiding the one thing the user + needs to know. + """ + geteuid = getattr(os, "geteuid", None) + if geteuid is None: # Windows has no euid; nothing here targets it anyway + return True + if geteuid() == 0: + return True + print(f" -- Skipping {action}: needs root. Re-run with sudo, or install manually.") + return False + + +def _install_gh_apt() -> bool: + """Set up the official GitHub apt source and install gh. + + Returns False leaving apt as it found it. The keyring is fetched to a + sibling temp file and renamed into place, so a failed download can never + truncate a working keyring; the source list is written only once that + keyring exists, and is rolled back if `apt-get update` then fails. + """ + import shutil + import subprocess + + if not shutil.which("apt-get"): + print(" [!] gh CLI not found and this is not an apt-based system. Install: https://cli.github.com") + return False + if not _can_modify_system("gh CLI apt install"): + return False + + keyring_path = _GH_KEYRING_PATH + list_path = _GH_APT_LIST_PATH + keyring_tmp = keyring_path + ".agent-reach.tmp" + + try: + arch_result = subprocess.run( + ["dpkg", "--print-architecture"], + capture_output=True, encoding="utf-8", errors="replace", timeout=10, + ) + except (OSError, subprocess.TimeoutExpired): + return False + arch = (arch_result.stdout or "").strip() or "amd64" + + try: + os.makedirs(os.path.dirname(keyring_path), mode=0o755, exist_ok=True) + downloaded = _run_checked( + ["curl", "-fsSL", + "https://cli.github.com/packages/githubcli-archive-keyring.gpg", + "-o", keyring_tmp], + timeout=60, + ) + if not downloaded or os.path.getsize(keyring_tmp) == 0: + print(" -- Could not download the GitHub apt keyring; apt sources left untouched") + return False + os.chmod(keyring_tmp, 0o644) + os.replace(keyring_tmp, keyring_path) + except OSError: + print(" -- Could not install the GitHub apt keyring; apt sources left untouched") + return False + finally: + try: + os.unlink(keyring_tmp) + except OSError: + pass + + # Only now that the keyring is really on disk may anything reference it. + created_list = not os.path.exists(list_path) + repo_line = ( + f"deb [arch={arch} signed-by={keyring_path}] " + "https://cli.github.com/packages stable main\n" + ) + try: + os.makedirs(os.path.dirname(list_path), mode=0o755, exist_ok=True) + with open(list_path, "w", encoding="utf-8") as f: + f.write(repo_line) + except OSError: + print(" -- Could not write the GitHub apt source") + return False + + if not _run_checked(["apt-get", "update", "-qq"], timeout=120): + print(" -- `apt-get update` failed; removing the source Agent Reach just added") + if created_list: + try: + os.unlink(list_path) + except OSError: + pass + return False + + return _run_checked(["apt-get", "install", "-y", "-qq", "gh"], timeout=120) + + +def _install_brew_formula(formula: str) -> bool: + """Install one Homebrew formula, reporting whether it actually succeeded. + + No privilege check here: Homebrew refuses to run as root by design, so the + apt paths' root requirement would be exactly backwards. The timeout is + generous because a first `brew install` may update the formula index or + build from source before it installs anything. + """ + import shutil + + brew = shutil.which("brew") + if not brew: + return False + return _run_checked([brew, "install", formula], timeout=600) + + +def _install_node_apt() -> bool: + """Install Node.js via the NodeSource setup script, verifying each step.""" + import shutil + import tempfile + + if not shutil.which("apt-get"): + print(" [!] Node.js not found and this is not an apt-based system. Install: https://nodejs.org") + return False + if not _can_modify_system("Node.js apt install"): + return False + + script_path = None + try: + fd, script_path = tempfile.mkstemp(suffix=".sh") + os.close(fd) + downloaded = _run_checked( + ["curl", "-fsSL", "https://deb.nodesource.com/setup_22.x", "-o", script_path], + timeout=60, + ) + # An unchecked curl leaves an empty file that bash exits 0 on, which + # would read as "setup succeeded" right before apt-get finds no package. + if not downloaded or os.path.getsize(script_path) == 0: + print(" -- Could not download the NodeSource setup script") + return False + if not _run_checked(["bash", script_path], timeout=180): + print(" -- NodeSource setup failed; apt sources unchanged by Agent Reach") + return False + except OSError: + return False + finally: + if script_path: + try: + os.unlink(script_path) + except OSError: + pass + + return _run_checked(["apt-get", "install", "-y", "-qq", "nodejs"], timeout=180) + + def _install_system_deps(): """Install system-level dependencies: gh CLI, Node.js (for mcporter).""" + import platform import shutil import subprocess - import platform - import tempfile print("Checking system dependencies...") + os_type = platform.system().lower() # ── gh CLI ── if shutil.which("gh"): print(" ✅ gh CLI already installed") else: print(" Installing gh CLI...") - os_type = platform.system().lower() if os_type == "linux": - try: - # Official GitHub apt source setup without invoking a shell. - keyring_path = "/usr/share/keyrings/githubcli-archive-keyring.gpg" - list_path = "/etc/apt/sources.list.d/github-cli.list" - arch = subprocess.run( - ["dpkg", "--print-architecture"], - capture_output=True, encoding="utf-8", errors="replace", timeout=10, - ).stdout.strip() or "amd64" - subprocess.run( - ["curl", "-fsSL", "https://cli.github.com/packages/githubcli-archive-keyring.gpg", "-o", keyring_path], - capture_output=True, timeout=60, - ) - repo_line = ( - f"deb [arch={arch} signed-by={keyring_path}] " - "https://cli.github.com/packages stable main\n" - ) - with open(list_path, "w", encoding="utf-8") as f: - f.write(repo_line) - subprocess.run(["apt-get", "update", "-qq"], capture_output=True, timeout=60) - subprocess.run(["apt-get", "install", "-y", "-qq", "gh"], capture_output=True, timeout=60) - if shutil.which("gh"): - print(" ✅ gh CLI installed") - else: - print(" [!] gh CLI install failed. You can try: snap install gh, or download from https://github.com/cli/cli/releases") - except Exception: + if _install_gh_apt() and shutil.which("gh"): + print(" ✅ gh CLI installed") + else: print(" [!] gh CLI install failed. You can try: snap install gh, or download from https://github.com/cli/cli/releases") elif os_type == "darwin": - if shutil.which("brew"): - try: - subprocess.run(["brew", "install", "gh"], capture_output=True, timeout=120) - if shutil.which("gh"): - print(" ✅ gh CLI installed") - else: - print(" [!] gh CLI install failed. Try: brew install gh") - except Exception: - print(" [!] gh CLI install failed. Try: brew install gh") - else: + if not shutil.which("brew"): print(" [!] gh CLI not found. Install: https://cli.github.com") + elif _install_brew_formula("gh") and shutil.which("gh"): + print(" ✅ gh CLI installed") + else: + print(" [!] gh CLI install failed. Try: brew install gh") else: print(" [!] gh CLI not found. Install: https://cli.github.com") @@ -649,32 +793,20 @@ def _install_system_deps(): print(" ✅ Node.js already installed") else: print(" Installing Node.js...") - try: - # Use NodeSource setup script without invoking a shell pipeline. - with tempfile.NamedTemporaryFile(delete=False, suffix=".sh") as tf: - script_path = tf.name - subprocess.run( - ["curl", "-fsSL", "https://deb.nodesource.com/setup_22.x", "-o", script_path], - capture_output=True, timeout=60, - ) - subprocess.run( - ["bash", script_path], - capture_output=True, timeout=120, - ) - try: - os.unlink(script_path) - except Exception: - pass - subprocess.run( - ["apt-get", "install", "-y", "-qq", "nodejs"], - capture_output=True, timeout=120, - ) - if shutil.which("node"): + if os_type == "linux": + if _install_node_apt() and shutil.which("node"): print(" ✅ Node.js installed") else: print(" [!] Node.js install failed. Try: apt install nodejs npm, or nvm install 22, or download from https://nodejs.org") - except Exception: - print(" [!] Node.js install failed. Try: apt install nodejs npm, or nvm install 22, or download from https://nodejs.org") + elif os_type == "darwin": + if not shutil.which("brew"): + print(" [!] Node.js not found. Try: nvm install 22, or download from https://nodejs.org") + elif _install_brew_formula("node") and shutil.which("node"): + print(" ✅ Node.js installed") + else: + print(" [!] Node.js install failed. Try: brew install node, or nvm install 22, or download from https://nodejs.org") + else: + print(" [!] Node.js not found. Install: https://nodejs.org") # ── undici (proxy support for Node.js fetch) ── npm_cmd = shutil.which("npm") @@ -776,6 +908,7 @@ def _install_system_deps(): def _install_xiaoyuzhou_deps(): """Install Xiaoyuzhou podcast transcription script.""" import shutil + from agent_reach.config import Config config = Config() @@ -1311,15 +1444,15 @@ def _cmd_configure(args): elif args.key == "github-token": config.set("github_token", value) - print(f"✅ GitHub token configured!") + print("✅ GitHub token configured!") elif args.key == "groq-key": config.set("groq_api_key", value) - print(f"✅ Groq key configured!") + print("✅ Groq key configured!") elif args.key == "openai-key": config.set("openai_api_key", value) - print(f"✅ OpenAI key configured!") + print("✅ OpenAI key configured!") def _cmd_transcribe(args): @@ -1739,12 +1872,11 @@ def _cmd_uninstall(args): def _cmd_doctor(args=None): + from rich import print as rich_print + from agent_reach.config import Config from agent_reach.doctor import check_all, format_report - try: - from rich import print as rprint - except ImportError: - rprint = print + config = Config(read_only=True) results = check_all(config) @@ -1752,7 +1884,7 @@ def _cmd_doctor(args=None): print(json.dumps(results, ensure_ascii=False, indent=2)) return - rprint(format_report(results)) + rich_print(format_report(results)) def _cmd_setup(): @@ -1826,7 +1958,7 @@ def _cmd_setup(): print(" 获取: https://github.com/settings/tokens (无需任何权限)") current = config.get("github_token") if current: - print(f" 当前状态: ✅ 已配置") + print(" 当前状态: ✅ 已配置") else: key = input(" GITHUB_TOKEN (回车跳过): ").strip() if key: @@ -1847,7 +1979,7 @@ def _cmd_setup(): print(" 免费额度,注册: https://console.groq.com") current = config.get("groq_api_key") if current: - print(f" 当前状态: ✅ 已配置") + print(" 当前状态: ✅ 已配置") else: key = input(" GROQ_API_KEY (回车跳过): ").strip() if key: @@ -1978,10 +2110,10 @@ def parse(v): except ValueError: return None - r, l = parse(remote), parse(local) - if r is None or l is None: + remote_parts, local_parts = parse(remote), parse(local) + if remote_parts is None or local_parts is None: return remote != local # unparseable — fall back to old behavior - return r > l + return remote_parts > local_parts def _cmd_check_update(): @@ -2014,7 +2146,7 @@ def _cmd_check_update(): print() print(_UPDATE_INSTRUCTIONS) return "update_available" - print(f"✅ 已是最新版本") + print("✅ 已是最新版本") return "up_to_date" release_err = _classify_github_response_error(resp) @@ -2051,9 +2183,9 @@ def _cmd_watch(): Only outputs problems. If everything is fine, outputs a single line. """ + from agent_reach import __version__ from agent_reach.config import Config from agent_reach.doctor import check_all - from agent_reach import __version__ config = Config(read_only=True) issues = [] @@ -2092,8 +2224,8 @@ def _cmd_watch(): print(f"Agent Reach: 全部正常 ({ok}/{total} 渠道可用,v{__version__} 已是最新)") return - print(f"Agent Reach 监控报告") - print(f"=" * 40) + print("Agent Reach 监控报告") + print("=" * 40) print(f"版本: v{__version__} | 渠道: {ok}/{total}") if issues: diff --git a/tests/test_p0_cli.py b/tests/test_p0_cli.py index 1003b0e9..f502a95d 100644 --- a/tests/test_p0_cli.py +++ b/tests/test_p0_cli.py @@ -901,3 +901,341 @@ def fake_run(args, **_kwargs): output = capsys.readouterr().out assert not any("remove" in call for call in calls) assert "来源无法证明" in output + + +# ── apt source safety ─────────────────────────────── +# +# Writing /etc/apt/sources.list.d/github-cli.list while its `signed-by` keyring +# is missing breaks `apt-get update` for the whole machine, not just for Agent +# Reach — and the breakage surfaces long after the installer exits. These tests +# pin every path that must leave apt exactly as it was found. + + +@pytest.fixture +def apt_paths(monkeypatch, tmp_path): + """Redirect the privileged apt paths and grant fake root.""" + keyring = tmp_path / "keyrings" / "githubcli-archive-keyring.gpg" + listfile = tmp_path / "sources.list.d" / "github-cli.list" + monkeypatch.setattr(cli, "_GH_KEYRING_PATH", str(keyring)) + monkeypatch.setattr(cli, "_GH_APT_LIST_PATH", str(listfile)) + monkeypatch.setattr(cli.os, "geteuid", lambda: 0, raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: "/usr/bin/apt-get" if name == "apt-get" else None, + ) + return keyring, listfile + + +def test_gh_apt_source_not_written_when_keyring_download_fails( + apt_paths, monkeypatch, capsys +): + """A failed keyring fetch must not leave a source referencing it.""" + keyring, listfile = apt_paths + + def fake_run(args, **_kwargs): + if args[0] == "dpkg": + return _docker_result(args, stdout="amd64\n") + if args[0] == "curl": + return _docker_result(args, returncode=22) # curl -f on HTTP error + pytest.fail(f"apt must not run after a failed keyring download: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_gh_apt() is False + assert not listfile.exists() + assert not keyring.exists() + assert "apt sources left untouched" in capsys.readouterr().out + + +def test_gh_apt_source_not_written_when_keyring_download_is_empty( + apt_paths, monkeypatch +): + """A zero-byte keyring is a failed download even when curl exits 0.""" + keyring, listfile = apt_paths + + def fake_run(args, **_kwargs): + if args[0] == "dpkg": + return _docker_result(args, stdout="amd64\n") + if args[0] == "curl": + target = Path(args[args.index("-o") + 1]) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(b"") + return _docker_result(args) + pytest.fail(f"apt must not run for an empty keyring: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_gh_apt() is False + assert not listfile.exists() + + +def test_gh_apt_source_rolled_back_when_update_fails(apt_paths, monkeypatch): + """If our new source breaks `apt-get update`, we remove it again.""" + keyring, listfile = apt_paths + + def fake_run(args, **_kwargs): + if args[0] == "dpkg": + return _docker_result(args, stdout="amd64\n") + if args[0] == "curl": + target = Path(args[args.index("-o") + 1]) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(b"keyring-bytes") + return _docker_result(args) + if args[:2] == ["apt-get", "update"]: + return _docker_result(args, returncode=100) + pytest.fail(f"install must not run after a failed update: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_gh_apt() is False + assert not listfile.exists() + assert keyring.read_bytes() == b"keyring-bytes" + + +def test_gh_apt_keyring_download_failure_preserves_existing_keyring( + apt_paths, monkeypatch +): + """A retry that fails must not truncate a keyring from an earlier install.""" + keyring, listfile = apt_paths + keyring.parent.mkdir(parents=True) + keyring.write_bytes(b"already-working") + + def fake_run(args, **_kwargs): + if args[0] == "dpkg": + return _docker_result(args, stdout="amd64\n") + if args[0] == "curl": + return _docker_result(args, returncode=6) + pytest.fail(f"apt must not run after a failed download: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_gh_apt() is False + assert keyring.read_bytes() == b"already-working" + assert not listfile.exists() + + +def test_gh_apt_install_writes_source_only_on_the_happy_path( + apt_paths, monkeypatch +): + """With every step succeeding, the source names the keyring we fetched.""" + keyring, listfile = apt_paths + + def fake_run(args, **_kwargs): + if args[0] == "dpkg": + return _docker_result(args, stdout="arm64\n") + if args[0] == "curl": + target = Path(args[args.index("-o") + 1]) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(b"keyring-bytes") + return _docker_result(args) + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_gh_apt() is True + contents = listfile.read_text(encoding="utf-8") + assert f"arch=arm64 signed-by={keyring}" in contents + assert keyring.exists() + + +def test_gh_apt_install_refuses_without_root(apt_paths, monkeypatch, capsys): + """A non-root run must say so instead of failing generically in /etc.""" + keyring, listfile = apt_paths + monkeypatch.setattr(cli.os, "geteuid", lambda: 1000, raising=False) + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **k: pytest.fail("no system command may run without root"), + ) + + assert cli._install_gh_apt() is False + assert not listfile.exists() + assert "needs root" in capsys.readouterr().out + + +def test_gh_apt_install_skipped_on_non_apt_system(monkeypatch, capsys): + """Fedora/Arch must not get an /etc/apt source written for them.""" + monkeypatch.setattr("shutil.which", lambda _name: None) + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **k: pytest.fail("no command may run without apt-get"), + ) + + assert cli._install_gh_apt() is False + assert "not an apt-based system" in capsys.readouterr().out + + +def test_nodesource_script_removed_even_when_it_fails(monkeypatch, capsys): + """The downloaded setup script must never leak into the temp dir.""" + script_paths = [] + monkeypatch.setattr(cli.os, "geteuid", lambda: 0, raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: "/usr/bin/apt-get" if name == "apt-get" else None, + ) + + def fake_run(args, **_kwargs): + if args[0] == "curl": + target = Path(args[args.index("-o") + 1]) + target.write_text("#!/bin/bash\nexit 1\n", encoding="utf-8") + script_paths.append(target) + return _docker_result(args) + if args[0] == "bash": + return _docker_result(args, returncode=1) + pytest.fail(f"apt must not run after a failed setup script: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_node_apt() is False + assert script_paths and not script_paths[0].exists() + assert "apt sources unchanged" in capsys.readouterr().out + + +def test_nodesource_empty_script_is_not_treated_as_success(monkeypatch, capsys): + """bash exits 0 on an empty file — that must not count as setup succeeding.""" + monkeypatch.setattr(cli.os, "geteuid", lambda: 0, raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: "/usr/bin/apt-get" if name == "apt-get" else None, + ) + + def fake_run(args, **_kwargs): + if args[0] == "curl": + Path(args[args.index("-o") + 1]).write_bytes(b"") + return _docker_result(args) + pytest.fail(f"nothing may run for an empty setup script: {args}") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_node_apt() is False + assert "Could not download the NodeSource setup script" in capsys.readouterr().out + + +# ── Homebrew paths ────────────────────────────────── +# +# `brew install` exit codes were ignored the same way apt's were, so a failed +# formula install printed the success line. macOS also has no apt, so the +# Node.js branch must never reach the NodeSource/apt-get path. + + +def _darwin_which(available): + """shutil.which stub: only names in *available* resolve.""" + return lambda name: f"/opt/homebrew/bin/{name}" if name in available else None + + +def test_brew_formula_install_reports_failure(monkeypatch): + """A non-zero brew exit is a failure, not a success.""" + monkeypatch.setattr("shutil.which", _darwin_which({"brew"})) + monkeypatch.setattr( + subprocess, "run", lambda args, **k: _docker_result(args, returncode=1) + ) + + assert cli._install_brew_formula("gh") is False + + +def test_brew_formula_install_reports_timeout(monkeypatch): + """A brew install that hangs past the timeout is a failure, not a crash.""" + monkeypatch.setattr("shutil.which", _darwin_which({"brew"})) + + def fake_run(args, **kwargs): + raise subprocess.TimeoutExpired(args, kwargs.get("timeout", 0)) + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_brew_formula("gh") is False + + +def test_brew_formula_install_without_brew(monkeypatch): + """No brew means no install attempt at all.""" + monkeypatch.setattr("shutil.which", lambda _name: None) + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **k: pytest.fail("nothing may run without brew"), + ) + + assert cli._install_brew_formula("gh") is False + + +def test_brew_formula_install_happy_path(monkeypatch): + """The formula name is passed through to brew verbatim.""" + calls = [] + monkeypatch.setattr("shutil.which", _darwin_which({"brew"})) + + def fake_run(args, **_kwargs): + calls.append(args) + return _docker_result(args) + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert cli._install_brew_formula("node") is True + assert calls == [["/opt/homebrew/bin/brew", "install", "node"]] + + +def test_darwin_gh_install_failure_is_not_reported_as_success( + monkeypatch, capsys +): + """A failed `brew install gh` must not print the success line.""" + import platform + + monkeypatch.setattr(platform, "system", lambda: "Darwin") + monkeypatch.setattr("shutil.which", _darwin_which({"brew", "node", "npm"})) + + def fake_run(args, **_kwargs): + if args[1:2] == ["install"] and args[0].endswith("brew"): + return _docker_result(args, returncode=1) + return _docker_result( + args, stdout="/opt/homebrew/lib\n" if args[1:3] == ["root", "-g"] else "" + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + + cli._install_system_deps() + + out = capsys.readouterr().out + assert "gh CLI install failed" in out + assert "✅ gh CLI installed" not in out + + +def test_darwin_node_install_never_touches_apt(monkeypatch, capsys): + """macOS has no apt — the Node.js branch must go through brew only.""" + import platform + + calls = [] + monkeypatch.setattr(platform, "system", lambda: "Darwin") + monkeypatch.setattr("shutil.which", _darwin_which({"brew", "gh"})) + + def fake_run(args, **_kwargs): + calls.append(args) + return _docker_result(args, returncode=1) + + monkeypatch.setattr(subprocess, "run", fake_run) + + cli._install_system_deps() + + programs = {Path(call[0]).name for call in calls} + assert not programs & {"apt-get", "curl", "bash", "dpkg"} + assert ["/opt/homebrew/bin/brew", "install", "node"] in calls + out = capsys.readouterr().out + assert "apt install nodejs" not in out + assert "brew install node" in out + + +def test_darwin_without_brew_gives_macos_advice(monkeypatch, capsys): + """Missing brew must not fall through to apt-flavoured instructions.""" + import platform + + monkeypatch.setattr(platform, "system", lambda: "Darwin") + monkeypatch.setattr("shutil.which", lambda _name: None) + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **k: pytest.fail("no installer may run without brew"), + ) + + cli._install_system_deps() + + out = capsys.readouterr().out + assert "apt install nodejs" not in out + assert "nvm install 22" in out + assert "https://cli.github.com" in out diff --git a/tests/test_v2ex_channel.py b/tests/test_v2ex_channel.py index 2379ed62..9e63586e 100644 --- a/tests/test_v2ex_channel.py +++ b/tests/test_v2ex_channel.py @@ -16,7 +16,6 @@ from agent_reach.channels import v2ex as v2 from agent_reach.channels.v2ex import V2EXChannel - # --- can_handle --- def test_can_handle_matches_v2ex_hosts(): diff --git a/tests/test_web_channel.py b/tests/test_web_channel.py index 2529c784..4b0d0283 100644 --- a/tests/test_web_channel.py +++ b/tests/test_web_channel.py @@ -10,7 +10,7 @@ from unittest.mock import MagicMock, patch -from agent_reach.channels.web import WebChannel, _UA +from agent_reach.channels.web import _UA, WebChannel def _resp(body=b"# Example\nfull text\n"):