From 856c850c54d0ed2ff9910c3ef6fd8d90620d6dcd Mon Sep 17 00:00:00 2001 From: Iury Souza Date: Fri, 17 Jul 2026 22:56:21 +0200 Subject: [PATCH] feat(picker): use Television popup picker --- .github/workflows/ci.yml | 15 +- CHANGELOG.md | 21 ++ README.md | 35 +- cable/termscope-alpha.toml | 31 ++ cable/termscope-appearance.toml | 31 ++ docs/publishing.md | 12 +- herdr-plugin.toml | 24 +- scripts/install-dependencies.sh | 56 +++ termscope | 631 +++++++++++++++++--------------- termscope_herdr.py | 47 +-- tests/test_installer.py | 107 ++++++ tests/test_plugin.py | 128 +++++++ tests/test_termscope.py | 544 +++++++++++++++++++-------- 13 files changed, 1191 insertions(+), 491 deletions(-) create mode 100644 cable/termscope-alpha.toml create mode 100644 cable/termscope-appearance.toml create mode 100755 scripts/install-dependencies.sh create mode 100644 tests/test_installer.py create mode 100644 tests/test_plugin.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 754c849..2978001 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,8 +47,21 @@ jobs: manifest = tomllib.loads(Path('herdr-plugin.toml').read_text()) assert manifest['id'] == 'termscope' - assert manifest['min_herdr_version'] >= '0.7.0' + assert manifest['version'] == '0.2.0' + assert manifest['min_herdr_version'] >= '0.7.4' assert set(manifest['platforms']) == {'linux', 'macos'} + assert manifest['build'] == [{ + 'platforms': ['linux', 'macos'], + 'command': ['sh', 'scripts/install-dependencies.sh'], + }] assert {a['id'] for a in manifest['actions']} == {'open', 'open-links'} assert {p['id'] for p in manifest['panes']} == {'picker', 'link-picker'} + assert all(p['placement'] == 'popup' for p in manifest['panes']) + assert all(p['width'] == '80%' and p['height'] == '60%' for p in manifest['panes']) + + for path in Path('cable').glob('termscope-*.toml'): + channel = tomllib.loads(path.read_text()) + assert len(channel['source']['command']) == 2 + assert channel['source']['no_sort'] is True + assert channel['source']['frecency'] is False PY diff --git a/CHANGELOG.md b/CHANGELOG.md index 103fa8f..af55957 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Changelog +## 0.2.0 - Unreleased + +### Added + +- Television channels with file previews, always-visible action hints, and `Ctrl-S` appearance/alphabetical source cycling. +- Built-in text and directory previews when `bat` is unavailable. +- Install-time Television provisioning through Homebrew for Herdr plugin installs. + +### Changed + +- Replaced fzf with Television `0.15+`. +- Replaced full-pane Herdr overlays with bounded `80% × 60%` session-modal popups. +- Raised the minimum Herdr version to `0.7.4`. + +### Fixed + +- Encoded picker targets before Television preview interpolation so filenames cannot inject shell syntax. +- Propagated Herdr popup and Television runtime failures while keeping user cancellation successful. +- Used the Plannotator slash command in agent panes and the CLI command in plain shells. +- Bounded preview reads by size, output, and wall-clock time. + ## 0.1.0 - 2026-07-09 ### Added diff --git a/README.md b/README.md index fd4479b..759fa72 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![CI](https://github.com/iurysza/termscope/actions/workflows/ci.yml/badge.svg)](https://github.com/iurysza/termscope/actions/workflows/ci.yml) [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE) -![herdr 0.7+](https://img.shields.io/badge/herdr-0.7%2B-8a2be2) +![herdr 0.7.4+](https://img.shields.io/badge/herdr-0.7.4%2B-8a2be2) ![platforms: macOS • Linux](https://img.shields.io/badge/platforms-macOS%20%E2%80%A2%20Linux-informational) Open the files and links your agent just mentioned. @@ -39,14 +39,16 @@ Termscope stays conservative: - verifies paths against the repo/worktree on disk - preserves `file:line` targets - falls back to a full repo picker when no visible file matches -- opens outside the picker overlay, so the agent conversation stays where it is +- uses a session-modal Herdr popup, leaving the tiled pane layout untouched ## Requirements -- [Herdr](https://herdr.dev) `>= 0.7.0` or tmux +- [Herdr](https://herdr.dev) `>= 0.7.4` or tmux - Python `>= 3.10` - [`fd`](https://github.com/sharkdp/fd) -- [`fzf`](https://github.com/junegunn/fzf) +- [Homebrew](https://brew.sh) when a Herdr plugin install needs to add or upgrade Television +- [Television](https://alexpasmantier.github.io/television/) `>= 0.15` (provisioned automatically when needed) +- [`bat`](https://github.com/sharkdp/bat) is optional for syntax-highlighted previews - `nvim` for the default file-open action - `open` on macOS or `xdg-open` on Linux for default-app opens @@ -59,11 +61,18 @@ environment provides it. herdr plugin install iurysza/termscope ``` -For local development: +The install includes a visible build step that installs or upgrades Television +through Homebrew when `tv` is missing or older than `0.15`. It never installs +Homebrew itself; a missing Homebrew installation aborts cleanly before the +plugin is registered. + +For local development (`plugin link` does not run install-time build steps): ```bash git clone https://github.com/iurysza/termscope.git -herdr plugin link ./termscope +cd termscope +./scripts/install-dependencies.sh +herdr plugin link "$PWD" ``` Verify Herdr sees the actions: @@ -110,7 +119,7 @@ File picker controls: | --- | --- | | `Enter` | Open in a new Neovim split beside the source pane | | `Ctrl-O` | Open with the default app | -| `Ctrl-Y` | Send `/plannotator-annotate ` to the source pane | +| `Ctrl-Y` | Agent pane: send `/plannotator-annotate `; shell pane: run `plannotator annotate ` | | `Ctrl-S` | Toggle appearance order / alphabetical sort | Link picker controls: @@ -160,7 +169,7 @@ export TERMSCOPE_SORT=alpha ## Dry run / debug -See what the scanner would offer without opening `fzf`: +See what the scanner would offer without opening Television: ```bash ./termscope scan --pane-path "$PWD" --pane-id "$HERDR_PANE_ID" --multiplexer herdr @@ -177,10 +186,12 @@ selection decision. ## How it works -Herdr plugin actions run without a TTY, so `termscope.open` does not run `fzf` -directly. It opens a Herdr-managed overlay pane. That pane inherits the source -pane id/cwd, captures visible text with `herdr pane read --source visible`, scans -the repo with `fd`, and runs the interactive picker. +Herdr plugin actions run without a TTY, so `termscope.open` first opens an +`80% × 60%` session-modal popup. The popup inherits the source pane id/cwd, +captures visible text with `herdr pane read --source visible`, scans the repo +with `fd`, and runs Television. Two bundled channels let `Ctrl-S` cycle between +appearance and alphabetical order. File previews use `bat` when available and a +built-in text preview otherwise. When you choose a file, Termscope asks Herdr to split beside the source pane and runs `nvim +line path`. For URLs, it uses the default opener unless diff --git a/cable/termscope-alpha.toml b/cable/termscope-alpha.toml new file mode 100644 index 0000000..62fbbf1 --- /dev/null +++ b/cable/termscope-alpha.toml @@ -0,0 +1,31 @@ +[metadata] +name = "termscope-alpha" +description = "Pick visible terminal targets in alphabetical order" + +[source] +command = [ + { name = "Alphabetical", run = "cat \"$TERMSCOPE_ALPHA_CANDIDATES\"" }, + { name = "Appearance", run = "cat \"$TERMSCOPE_APPEARANCE_CANDIDATES\"" }, +] +display = "{split:\\t:1..}" +output = "{split:\\t:0}" +no_sort = true +frecency = false + +[keybindings] +ctrl-s = "cycle_sources" + +[preview] +command = "\"$TERMSCOPE_PYTHON\" \"$TERMSCOPE_SCRIPT\" preview --pane-path \"$TERMSCOPE_PANE_PATH\" {split:\\t:0}" + +[ui] +layout = "landscape" +input_prompt = "> " + +[ui.preview_panel] +size = 60 +footer = "Enter Neovim · Ctrl-O Default app · Ctrl-Y Plannotator" +border_type = "rounded" + +[ui.results_panel] +border_type = "rounded" diff --git a/cable/termscope-appearance.toml b/cable/termscope-appearance.toml new file mode 100644 index 0000000..ffb4cb2 --- /dev/null +++ b/cable/termscope-appearance.toml @@ -0,0 +1,31 @@ +[metadata] +name = "termscope-appearance" +description = "Pick visible terminal targets in appearance order" + +[source] +command = [ + { name = "Appearance", run = "cat \"$TERMSCOPE_APPEARANCE_CANDIDATES\"" }, + { name = "Alphabetical", run = "cat \"$TERMSCOPE_ALPHA_CANDIDATES\"" }, +] +display = "{split:\\t:1..}" +output = "{split:\\t:0}" +no_sort = true +frecency = false + +[keybindings] +ctrl-s = "cycle_sources" + +[preview] +command = "\"$TERMSCOPE_PYTHON\" \"$TERMSCOPE_SCRIPT\" preview --pane-path \"$TERMSCOPE_PANE_PATH\" {split:\\t:0}" + +[ui] +layout = "landscape" +input_prompt = "> " + +[ui.preview_panel] +size = 60 +footer = "Enter Neovim · Ctrl-O Default app · Ctrl-Y Plannotator" +border_type = "rounded" + +[ui.results_panel] +border_type = "rounded" diff --git a/docs/publishing.md b/docs/publishing.md index f8924d0..8f9525e 100644 --- a/docs/publishing.md +++ b/docs/publishing.md @@ -12,7 +12,7 @@ gh repo edit iurysza/termscope \ gh repo edit iurysza/termscope \ --add-topic herdr-plugin \ --add-topic herdr \ - --add-topic fzf \ + --add-topic television \ --add-topic tmux \ --add-topic terminal \ --add-topic python @@ -39,20 +39,26 @@ marketplace index. ```bash git add . git commit -m "chore: prepare Termscope release" - git tag v0.1.0 + git tag v0.2.0 git push origin main --tags - gh release create v0.1.0 --title "v0.1.0" --notes-file CHANGELOG.md + gh release create v0.2.0 --title "v0.2.0" --notes-file CHANGELOG.md ``` ## Install smoke test On a clean machine or temp user profile: +Start with Television absent or older than `0.15`, but with Homebrew available: + ```bash +herdr --version # 0.7.4+ herdr plugin install iurysza/termscope +tv --version # now 0.15+ herdr plugin action list --plugin termscope ``` +Also verify installation aborts before registration when Homebrew is absent. + Then add keybindings from the README and run: ```bash diff --git a/herdr-plugin.toml b/herdr-plugin.toml index a2e47f2..8abd174 100644 --- a/herdr-plugin.toml +++ b/herdr-plugin.toml @@ -7,38 +7,46 @@ # herdr plugin link /path/to/termscope # # Termscope opens real files and URLs already visible in the active pane. The -# user-facing action opens a Herdr overlay pane because actions run without a -# TTY and fzf needs one. +# action opens a bounded Herdr popup because plugin actions run without a TTY +# and Television needs an interactive terminal. id = "termscope" name = "Termscope" -version = "0.1.0" -min_herdr_version = "0.7.0" +version = "0.2.0" +min_herdr_version = "0.7.4" description = "Open files and links already visible on your terminal screen." platforms = ["linux", "macos"] +[[build]] +platforms = ["linux", "macos"] +command = ["sh", "scripts/install-dependencies.sh"] + [[actions]] id = "open" title = "Termscope: open visible file picker" -description = "Scan the focused pane for visible file paths and open a picker overlay." +description = "Scan the focused pane for visible file paths and open a Television popup." contexts = ["workspace"] command = ["python3", "termscope_herdr.py", "--open-pane"] [[actions]] id = "open-links" title = "Termscope: open visible link picker" -description = "Scan the focused pane for visible URLs and open a picker overlay." +description = "Scan the focused pane for visible URLs and open a Television popup." contexts = ["workspace"] command = ["python3", "termscope_herdr.py", "--open-links-pane"] [[panes]] id = "picker" title = "Termscope" -placement = "overlay" +placement = "popup" +width = "80%" +height = "60%" command = ["python3", "termscope_herdr.py"] [[panes]] id = "link-picker" title = "Termscope Links" -placement = "overlay" +placement = "popup" +width = "80%" +height = "60%" command = ["python3", "termscope_herdr.py", "--links"] diff --git a/scripts/install-dependencies.sh b/scripts/install-dependencies.sh new file mode 100755 index 0000000..9969704 --- /dev/null +++ b/scripts/install-dependencies.sh @@ -0,0 +1,56 @@ +#!/bin/sh +set -eu + +television_supported() { + candidate="${1:-}" + [ -n "$candidate" ] || return 1 + + output="$("$candidate" --version 2>/dev/null)" || return 1 + case "$output" in + "television "*) version=${output#television } ;; + *) return 1 ;; + esac + version=${version%% *} + + old_ifs=$IFS + IFS=. + set -- $version + IFS=$old_ifs + major=${1:-} + minor=${2:-} + case "$major" in ''|*[!0-9]*) return 1 ;; esac + case "$minor" in ''|*[!0-9]*) return 1 ;; esac + + [ "$major" -gt 0 ] || { [ "$major" -eq 0 ] && [ "$minor" -ge 15 ]; } +} + +tv_bin="$(command -v tv 2>/dev/null || true)" +if television_supported "$tv_bin"; then + echo "Television 0.15+ already installed: $tv_bin" + exit 0 +fi + +brew_bin="$(command -v brew 2>/dev/null || true)" +if [ -z "$brew_bin" ]; then + echo "Termscope requires Homebrew to install Television 0.15+." >&2 + echo "Install Homebrew from https://brew.sh, then retry the plugin install." >&2 + exit 1 +fi + +if "$brew_bin" list --formula television >/dev/null 2>&1; then + echo "Upgrading Television with Homebrew" + "$brew_bin" upgrade television +else + echo "Installing Television with Homebrew" + "$brew_bin" install television +fi + +hash -r 2>/dev/null || true +tv_bin="$(command -v tv 2>/dev/null || true)" +if ! television_supported "$tv_bin"; then + echo "Homebrew completed, but Television 0.15+ is not available as 'tv' on PATH." >&2 + [ -n "$tv_bin" ] && echo "Current tv: $tv_bin" >&2 + exit 1 +fi + +echo "Television installed: $tv_bin" diff --git a/termscope b/termscope index 0b7499f..d90907d 100755 --- a/termscope +++ b/termscope @@ -4,11 +4,14 @@ from __future__ import annotations import argparse +import base64 +import binascii import json import os import re import shlex import shutil +import signal import subprocess import sys import tempfile @@ -33,8 +36,7 @@ class ParsedTarget: @dataclass -class FzfResult: - query: str = "" +class PickerResult: key: str = "" selection: str = "" @@ -91,7 +93,7 @@ class MultiplexerBackend(ABC): @abstractmethod def send_annotate_command(self, pane_id: str, file_path: str) -> None: - """Send /plannotator-annotate + Enter to the source pane.""" + """Request Plannotator through the source pane's active interface.""" ... @abstractmethod @@ -186,22 +188,35 @@ class TmuxBackend(MultiplexerBackend): os.execvp(tmux_cmd[0], tmux_cmd) def send_annotate_command(self, pane_id: str, file_path: str) -> None: - # Cancel copy mode if the source pane is in it, so the slash command - # reaches Pi's prompt instead of being swallowed by copy mode. - mode_result = subprocess.run( - ["tmux", "display-message", "-p", "-t", pane_id, "#{pane_in_mode}"], + # Cancel copy mode so the command reaches the source application. + pane_result = subprocess.run( + [ + "tmux", + "display-message", + "-p", + "-t", + pane_id, + "#{pane_in_mode}\t#{pane_current_command}", + ], text=True, capture_output=True, check=False, ) - if mode_result.stdout.strip() == "1": + fields = pane_result.stdout.rstrip("\n").split("\t", 1) + in_copy_mode = fields[0] == "1" + current_command = fields[1].lower() if len(fields) == 2 else "" + if in_copy_mode: subprocess.run( ["tmux", "send-keys", "-t", pane_id, "-X", "cancel"], check=False, ) quoted_path = shlex.quote(file_path) - cmd_text = f"/plannotator-annotate {quoted_path}" + agent_commands = {"pi", "opencode", "claude", "codex"} + if not current_command or current_command in agent_commands: + cmd_text = f"/plannotator-annotate {quoted_path}" + else: + cmd_text = f"plannotator annotate {quoted_path}" subprocess.run( ["tmux", "send-keys", "-t", pane_id, "-l", cmd_text], check=False, @@ -225,7 +240,7 @@ class HerdrBackend(MultiplexerBackend): return "herdr" def _herdr_bin(self) -> str: - return shutil.which("herdr") or "herdr" + return os.environ.get("HERDR_BIN_PATH") or shutil.which("herdr") or "herdr" def resolve_source_pane_path(self, pane_path: Path, pane_id: str) -> Path: if not pane_id: @@ -287,8 +302,13 @@ class HerdrBackend(MultiplexerBackend): def send_annotate_command(self, pane_id: str, file_path: str) -> None: herdr = self._herdr_bin() quoted_path = shlex.quote(file_path) + source_agent = os.environ.get("SOURCE_PANE_AGENT") + if source_agent is None or source_agent: + command = f"/plannotator-annotate {quoted_path}" + else: + command = f"plannotator annotate {quoted_path}" subprocess.run( - [herdr, "pane", "run", pane_id, f"/plannotator-annotate {quoted_path}"], + [herdr, "pane", "run", pane_id, command], check=False, ) self.show_message("Plannotator annotation requested") @@ -302,7 +322,7 @@ class HerdrBackend(MultiplexerBackend): def detect_backend(pane_id: str) -> MultiplexerBackend: """Pick a backend from the pane id shape.""" - if pane_id and re.match(r"^w\d+:p", pane_id): + if pane_id and re.match(r"^(?:w[0-9A-Za-z]+:p[0-9A-Za-z]+|\d+-\d+)$", pane_id): return HerdrBackend() return TmuxBackend() @@ -495,7 +515,7 @@ def list_repo_files(search_root: Path) -> list[str]: if result.returncode != 0: return [] lines = [ln.rstrip("\n") for ln in result.stdout.splitlines() if ln.rstrip("\n")] - # Cap at 5000 files — fzf can't handle more anyway + # Bound scan cost and picker startup time on very large repositories. return lines[:5000] @@ -701,144 +721,198 @@ def resolve_existing_path(path_text: str, pane_path: Path, search_root: Path) -> # --------------------------------------------------------------------------- # -# fzf +# Television # --------------------------------------------------------------------------- # -def parse_fzf_result(stdout: str) -> FzfResult: - if not stdout: - return FzfResult() - stdout = stdout.rstrip("\n") - lines = stdout.split("\n") - query = lines[0] if lines else "" - if len(lines) == 1: - return FzfResult(query=query) - - rest = lines[1:] - first_rest = rest[0] - remaining = "\n".join(rest[1:]) if len(rest) > 1 else "" - - if first_rest in ("enter", "ctrl-o", "ctrl-y"): - return FzfResult(query=query, key=first_rest, selection=remaining) - if first_rest == "": - # Interactive Enter with --expect prints: query, blank key line, selection. - return FzfResult(query=query, selection=remaining) - return FzfResult(query=query, selection=first_rest) - - -def _prepare_candidate_files( - candidates: list[str], initial_sort: str = "appearance" -) -> tuple[Path, Path]: - """Write the original candidate order and sort-state files. - - Returns the path to the original-order file and the state file. - The state file tracks which sort mode fzf is currently showing. +def parse_tv_result(stdout: str, expect_keys: tuple[str, ...]) -> PickerResult: + """Parse Television's stdout selection protocol. + + Enter prints only the selection. A key supplied through ``--expect`` prints + the key first, followed by the selection. """ - with tempfile.NamedTemporaryFile( - mode="w", delete=False, prefix="termscope-candidates-", suffix=".txt" - ) as tmp: - tmp.write("\n".join(candidates)) - if candidates: - tmp.write("\n") - candidates_path = Path(tmp.name) - - state_path = candidates_path.with_suffix(".state") - state_path.write_text(initial_sort) - return candidates_path, state_path - - -def _sort_toggle_bind(candidates_path: Path, state_path: Path) -> str: - """Build an fzf --bind expression that toggles sort with Ctrl-s.""" - script = shlex.quote(sys.argv[0]) - reorder = ( - f"{sys.executable} {script} reorder " - f"--candidates-file {shlex.quote(str(candidates_path))} " - f"--state-file {shlex.quote(str(state_path))}" - ) - return f"ctrl-s:reload({reorder})" + lines = stdout.rstrip("\n").split("\n") if stdout else [] + if not lines: + return PickerResult() + if lines[0] in expect_keys: + return PickerResult(key=lines[0], selection="\n".join(lines[1:])) + return PickerResult(selection="\n".join(lines)) -def _sort_candidate_list(candidates: list[str], sort: str) -> list[str]: - if sort == "alpha": - return sorted(candidates, key=lambda s: s.lower()) - return candidates[:] +def _television_cable_dir() -> Path: + return Path(__file__).resolve().parent / "cable" + + +def encode_candidate(candidate: str) -> str: + """Return a shell-safe opaque id for a Television candidate.""" + return base64.urlsafe_b64encode(candidate.encode("utf-8")).decode("ascii").rstrip("=") + + +def decode_candidate(encoded: str) -> str: + """Restore a candidate id emitted by the bundled Television channels.""" + if not encoded or not re.fullmatch(r"[A-Za-z0-9_-]+", encoded): + raise ValueError("invalid Television candidate id") + padding = "=" * (-len(encoded) % 4) + try: + return base64.urlsafe_b64decode(encoded + padding).decode("utf-8") + except (binascii.Error, UnicodeDecodeError) as exc: + raise ValueError("invalid Television candidate id") from exc + + +def _write_candidate_file(path: Path, candidates: list[str]) -> None: + records = [f"{encode_candidate(candidate)}\t{candidate}" for candidate in candidates] + content = "\n".join(records) + if records: + content += "\n" + path.write_text(content, encoding="utf-8") + + +def _picker_error(message: str) -> None: + print(f"termscope: {message}", file=sys.stderr) + herdr = os.environ.get("HERDR_BIN_PATH") + if herdr: + subprocess.run( + [herdr, "notification", "show", f"Termscope: {message}"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + + +def _television_is_supported(television: str) -> bool: + try: + result = subprocess.run( + [television, "--version"], + text=True, + capture_output=True, + check=False, + timeout=2, + ) + except (OSError, subprocess.TimeoutExpired): + return False + if result.returncode != 0: + return False + match = re.search(r"\btelevision\s+(\d+)\.(\d+)\.(\d+)", result.stdout) + return bool(match and tuple(map(int, match.groups())) >= (0, 15, 0)) -def _run_fzf( +def _run_tv( candidates: list[str], header: str, - expect: str, + expect_keys: tuple[str, ...], sort: str = "appearance", cwd: Path | None = None, - extra_args: list[str] | None = None, -) -> FzfResult | None: - input_lines = [c for c in candidates if c.strip()] + preview: bool = True, +) -> PickerResult | None: + input_lines = [candidate for candidate in candidates if candidate.strip()] if not input_lines: return None - candidates_path, state_path = _prepare_candidate_files(input_lines, sort) - display_lines = _sort_candidate_list(input_lines, sort) - input_text = "\n".join(display_lines) - if display_lines: - input_text += "\n" - - fzf_cmd = [ - "fzf", - "--reverse", - "--print-query", - "--header", - header, - "--expect", - expect, - "--bind", - _sort_toggle_bind(candidates_path, state_path), - ] - if extra_args: - fzf_cmd.extend(extra_args) + television = shutil.which("tv") + if not television: + _picker_error("television (tv) not found") + raise SystemExit(1) + if not _television_is_supported(television): + _picker_error("Television 0.15+ is required") + raise SystemExit(1) + + cwd = cwd or Path.cwd() + with tempfile.TemporaryDirectory(prefix="termscope-tv-") as temp_dir: + temp_path = Path(temp_dir) + appearance_path = temp_path / "appearance.txt" + alpha_path = temp_path / "alpha.txt" + _write_candidate_file(appearance_path, input_lines) + _write_candidate_file(alpha_path, sorted(input_lines, key=str.lower)) + + env = os.environ.copy() + env.update( + { + "TERMSCOPE_APPEARANCE_CANDIDATES": str(appearance_path), + "TERMSCOPE_ALPHA_CANDIDATES": str(alpha_path), + "TERMSCOPE_PANE_PATH": str(cwd), + "TERMSCOPE_PYTHON": sys.executable, + "TERMSCOPE_SCRIPT": str(Path(__file__).resolve()), + } + ) + + command = [ + television, + "--cable-dir", + str(_television_cable_dir()), + "--no-remote", + "--no-help-panel", + "--input-header", + header, + ] + if expect_keys: + command.extend(["--expect", ";".join(expect_keys)]) + if not preview: + command.append("--no-preview") + command.append(f"termscope-{sort}") + + try: + result = subprocess.run( + command, + cwd=cwd, + env=env, + text=True, + stdout=subprocess.PIPE, + check=False, + ) + except FileNotFoundError: + _picker_error("television (tv) not found") + raise SystemExit(1) - proc = None + if result.returncode != 0: + _picker_error(f"Television failed with status {result.returncode}") + raise SystemExit(result.returncode) + + if result.stdout == "": + return PickerResult() + picker_result = parse_tv_result(result.stdout, expect_keys) + if not picker_result.selection: + _picker_error("Television returned an invalid selection") + raise SystemExit(1) try: - proc = subprocess.run( - fzf_cmd, - input=input_text, - cwd=cwd, - text=True, - stdout=subprocess.PIPE, - ) - except FileNotFoundError: - print("termscope: fzf not found", file=sys.stderr) - return None - finally: - for p in (candidates_path, state_path): - try: - p.unlink(missing_ok=True) - except OSError: - pass + picker_result.selection = decode_candidate(picker_result.selection) + except ValueError: + _picker_error("Television returned an invalid selection") + raise SystemExit(1) + if picker_result.selection not in input_lines: + _picker_error("Television returned an unknown selection") + raise SystemExit(1) + return picker_result - if proc is None or proc.returncode >= 2: - return None - return parse_fzf_result(proc.stdout) +def display_path(path: Path) -> str: + try: + relative = path.relative_to(Path.home()) + except ValueError: + return str(path) + return "~" if relative == Path(".") else f"~/{relative}" -def run_fzf_visible( - candidates: list[str], - search_root: Path, - sort: str = "appearance", - header_label: str = "visible files in", -) -> FzfResult | None: - return _run_fzf( + +def run_tv_visible( + candidates: list[str], search_root: Path, sort: str = "appearance" +) -> PickerResult | None: + return _run_tv( candidates, - f"{header_label}: {search_root} | enter=nvim ctrl-o=default app ctrl-y=annotate ctrl-s=toggle sort", - "enter,ctrl-o,ctrl-y", + f"Files — {display_path(search_root)}", + ("ctrl-o", "ctrl-y"), sort=sort, + cwd=search_root, ) -def run_fzf_links(candidates: list[str], sort: str = "appearance") -> FzfResult | None: - return _run_fzf( +def run_tv_links( + candidates: list[str], pane_path: Path, sort: str = "appearance" +) -> PickerResult | None: + return _run_tv( candidates, - "visible links | enter=open in browser ctrl-y=copy ctrl-s=toggle sort", - "enter,ctrl-y", + "Visible links — Enter opens · Ctrl-Y copies · Ctrl-S changes order", + ("ctrl-y",), sort=sort, + cwd=pane_path, + preview=False, ) @@ -849,71 +923,20 @@ def sort_file_list(files: list[str], sort: str = "appearance") -> list[str]: return files -def run_fzf_full_repo(search_root: Path, query: str, sort: str = "appearance") -> FzfResult | None: - """Show all repo files in a full fzf picker. Used as fallback when - no visible candidates are found on screen.""" +def run_tv_full_repo(search_root: Path, sort: str = "appearance") -> PickerResult | None: + """Show all repo files when no visible candidate exists.""" files = list_repo_files(search_root) if not files: return None - return _run_fzf( + return _run_tv( files, - f"all files in: {search_root} | enter=nvim ctrl-o=default app ctrl-s=toggle sort", - "enter,ctrl-o,ctrl-y", + f"All files — {display_path(search_root)}", + ("ctrl-o", "ctrl-y"), sort=sort, cwd=search_root, ) -def run_fallback_picker(search_root: Path, query: str, sort: str = "appearance") -> str | None: - files = list_repo_files(search_root) - input_lines = [f for f in files if f.strip()] - if not input_lines: - return None - - candidates_path, state_path = _prepare_candidate_files(input_lines, sort) - display_lines = _sort_candidate_list(input_lines, sort) - input_text = "\n".join(display_lines) - if display_lines: - input_text += "\n" - - fzf_cmd = [ - "fzf", - "--reverse", - "--select-1", - "--exit-0", - "--query", - query, - "--header", - f"find file in: {search_root} | query: {query} | ctrl-s=toggle sort", - "--bind", - _sort_toggle_bind(candidates_path, state_path), - ] - - proc = None - try: - proc = subprocess.run( - fzf_cmd, - cwd=search_root, - input=input_text, - text=True, - stdout=subprocess.PIPE, - ) - except FileNotFoundError: - print("termscope: fzf not found", file=sys.stderr) - return None - finally: - for p in (candidates_path, state_path): - try: - p.unlink(missing_ok=True) - except OSError: - pass - - if proc is None or proc.returncode >= 2: - return None - selected = proc.stdout.strip().split("\n")[0] if proc.stdout.strip() else "" - return selected if selected else None - - # --------------------------------------------------------------------------- # # Opening # --------------------------------------------------------------------------- # @@ -1010,12 +1033,12 @@ def debug_dump(state: DebugState) -> None: (run_dir / "files.txt").write_text("\n".join(state.indexed_files), encoding="utf-8") (run_dir / "candidates.txt").write_text("\n".join(state.candidates), encoding="utf-8") - fzf_data = { + picker_data = { "query": state.query, "key": state.key, "selection": state.selection, } - (run_dir / "fzf.json").write_text(json.dumps(fzf_data, indent=2), encoding="utf-8") + (run_dir / "picker.json").write_text(json.dumps(picker_data, indent=2), encoding="utf-8") decision = { "pane_path": state.pane_path, @@ -1085,31 +1108,108 @@ def _open_target( open_in_nvim_split(resolved, parsed.line, pane_path, pane_id, backend) -def _open_fallback_popup( - mode: str, query: str, line: str, pane_path: Path, pane_id: str -) -> NoReturn: - cmd = [ - sys.argv[0], - "fallback", - "--mode", - mode, - "--pane-path", - str(pane_path), - "--pane-id", - pane_id, - "--query", - query, - ] - if line: - cmd.extend(["--line", line]) - inner = shlex.join(cmd) +def preview_target(target_text: str, pane_path: Path) -> None: + """Render a preview with one wall-clock bound around every filesystem call.""" + def timeout_preview(_signum: int, _frame: object) -> None: + raise TimeoutError + + previous_handler = signal.signal(signal.SIGALRM, timeout_preview) + signal.setitimer(signal.ITIMER_REAL, 4) + try: + _render_preview(target_text, pane_path) + except TimeoutError: + print("termscope: preview timed out") + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, previous_handler) + + +def _render_preview(target_text: str, pane_path: Path) -> None: + parsed = parse_selected_target(target_text) + search_root = find_search_root(pane_path) + resolved = resolve_existing_path(parsed.path, pane_path, search_root) + if not resolved: + print(target_text) + return + + if resolved.is_dir(): + entries: list[Path] = [] + has_more = False + try: + with os.scandir(resolved) as directory: + for index, entry in enumerate(directory): + if index == 200: + has_more = True + break + entries.append(Path(entry.path)) + entries.sort(key=lambda path: (not path.is_dir(), path.name.lower())) + except OSError as exc: + print(f"termscope: preview failed: {exc}") + return + print(f"{resolved}/") + for entry in entries: + suffix = "/" if entry.is_dir() else "" + print(f" {entry.name}{suffix}") + if has_more: + print(" … more entries") + return + + if not resolved.is_file(): + print("termscope: preview unavailable for non-regular file") + return - tmux_cmd = ["tmux", "display-popup"] - if pane_id: - tmux_cmd += ["-t", pane_id] - tmux_cmd += ["-E", "-w", "80%", "-h", "60%", inner] + line_number = int(parsed.line) if parsed.line.isdigit() else 0 + max_bytes = 1_048_576 + try: + file_size = resolved.stat().st_size + except OSError as exc: + print(f"termscope: preview failed: {exc}") + return + bat = shutil.which("bat") if file_size <= max_bytes else None + if bat: + command = [bat, "--color=always", "--paging=never", "--style=numbers"] + if line_number: + start = max(1, line_number - 15) + end = line_number + 35 + command.extend( + ["--line-range", f"{start}:{end}", "--highlight-line", str(line_number)] + ) + else: + command.extend(["--line-range", "1:200"]) + command.append(str(resolved)) + try: + result = subprocess.run( + command, + text=True, + capture_output=True, + check=False, + timeout=3, + ) + except subprocess.TimeoutExpired: + result = None + if result is not None and result.returncode == 0: + print(result.stdout[:200_000], end="") + return - os.execvp(tmux_cmd[0], tmux_cmd) + try: + with resolved.open("rb") as source: + content = source.read(max_bytes + 1) + except OSError as exc: + print(f"termscope: preview failed: {exc}") + return + truncated = len(content) > max_bytes + lines = content[:max_bytes].decode("utf-8", errors="replace").splitlines() + start = max(0, line_number - 16) if line_number else 0 + end = min(len(lines), line_number + 35) if line_number else min(len(lines), 200) + if line_number and start >= len(lines) and truncated: + print("termscope: target line lies beyond the 1 MiB fallback preview limit") + return + for index in range(start, end): + marker = ">" if index + 1 == line_number else " " + text = lines[index] + if len(text) > 1_000: + text = f"{text[:1_000]}…" + print(f"{marker} {index + 1:5} │ {text}") def cmd_pick(args: argparse.Namespace) -> None: @@ -1141,7 +1241,7 @@ def cmd_pick(args: argparse.Namespace) -> None: candidates=candidates, ) - # Filter out whitespace-only candidates that would produce an empty fzf. + # Filter out whitespace-only candidates that would produce an empty picker. real_candidates = [c for c in candidates if c.strip()] visible_empty = len(real_candidates) == 0 @@ -1168,30 +1268,29 @@ def cmd_pick(args: argparse.Namespace) -> None: if visible_empty: # No visible files — fall back to full repo listing so the picker # is never empty. - fzf_result = run_fzf_full_repo(search_root, "", sort=sort) - if fzf_result is None: + picker_result = run_tv_full_repo(search_root, sort=sort) + if picker_result is None: show_message("No files found in repo", backend) log_event("pick_empty_repo", pane_id=pane_id) debug_dump(state) return else: - fzf_result = run_fzf_visible(real_candidates, search_root) + picker_result = run_tv_visible(real_candidates, search_root, sort=sort) - if fzf_result is None: + if picker_result is None: log_event("pick_cancel_or_empty", pane_id=pane_id, candidate_count=len(real_candidates)) debug_dump(state) return - state.query = fzf_result.query - state.key = fzf_result.key - state.selection = fzf_result.selection + state.key = picker_result.key + state.selection = picker_result.selection - selection = fzf_result.selection + selection = picker_result.selection if not selection: debug_dump(state) return - key = fzf_result.key or "enter" + key = picker_result.key or "enter" if key == "ctrl-o": mode = "default" elif key == "ctrl-y": @@ -1223,47 +1322,6 @@ def cmd_open(args: argparse.Namespace) -> None: _open_target(mode, target, source_pane_path, pane_id, search_root, state=state, backend=backend) -def cmd_fallback(args: argparse.Namespace) -> None: - pane_path = Path(args.pane_path) if args.pane_path else Path.cwd() - pane_id = args.pane_id or "" - mode = args.mode - query = args.query - line = args.line or "" - sort = getattr(args, "sort", None) or os.environ.get("TERMSCOPE_SORT", "appearance") - backend = choose_backend(getattr(args, "multiplexer", "auto"), pane_id) - source_pane_path = resolve_source_pane_path(pane_path, pane_id, backend) - search_root = find_search_root(source_pane_path) - - state = DebugState( - pane_path=str(pane_path), - pane_id=pane_id, - source_pane_path=str(source_pane_path), - search_root=str(search_root), - mode=mode, - line=line, - query=query, - ) - - selected = run_fallback_picker(search_root, query, sort) - if not selected: - debug_dump(state) - return - - resolved = resolve_existing_path(selected, source_pane_path, search_root) - if not resolved: - resolved = search_root / selected - - state.selection = selected - state.resolved_path = str(resolved) - debug_dump(state) - copy_to_clipboard(str(resolved)) - - if mode == "default": - open_with_default_app(resolved) - else: - open_in_nvim_split(resolved, line, source_pane_path, pane_id, backend) - - def cmd_links(args: argparse.Namespace) -> None: pane_path = Path(args.pane_path) if args.pane_path else Path.cwd() pane_id = args.pane_id or "" @@ -1301,17 +1359,16 @@ def cmd_links(args: argparse.Namespace) -> None: backend=backend.name, ) - fzf_result = run_fzf_links(real_candidates, sort=sort) - if fzf_result is None: + picker_result = run_tv_links(real_candidates, pane_path, sort=sort) + if picker_result is None: log_event("links_cancel_or_empty", pane_id=pane_id, candidate_count=len(real_candidates)) debug_dump(state) return - state.query = fzf_result.query - state.key = fzf_result.key - state.selection = fzf_result.selection + state.key = picker_result.key + state.selection = picker_result.selection - selection = fzf_result.selection + selection = picker_result.selection if not selection: debug_dump(state) return @@ -1321,34 +1378,13 @@ def cmd_links(args: argparse.Namespace) -> None: copy_to_clipboard(selection) debug_dump(state) - if fzf_result.key == "ctrl-y": + if picker_result.key == "ctrl-y": show_message("Link copied to clipboard", backend) return open_url(selection) -def cmd_reorder(args: argparse.Namespace) -> None: - """Print candidates in the opposite sort order and update the state file. - - This is used by fzf's reload action when the user presses Ctrl-s. - """ - candidates_file = Path(args.candidates_file) - state_file = Path(args.state_file) - current = state_file.read_text().strip() if state_file.exists() else "appearance" - next_sort = "alpha" if current == "appearance" else "appearance" - - lines = candidates_file.read_text().splitlines() - if next_sort == "alpha": - lines.sort(key=lambda s: s.lower()) - state_file.write_text(next_sort) - - output = "\n".join(lines) - if lines: - output += "\n" - print(output, end="") - - def cmd_scan(args: argparse.Namespace) -> None: pane_path = Path(args.pane_path) if args.pane_path else Path.cwd() pane_id = args.pane_id or "" @@ -1398,15 +1434,6 @@ def main() -> None: open_parser.add_argument("--pane-id", required=True) open_parser.add_argument("--multiplexer", choices=["tmux", "herdr", "auto"], default="auto") - fallback_parser = subparsers.add_parser("fallback") - fallback_parser.add_argument("--mode", choices=["nvim", "default"], required=True) - fallback_parser.add_argument("--pane-path", required=True) - fallback_parser.add_argument("--pane-id", required=True) - fallback_parser.add_argument("--query", required=True) - fallback_parser.add_argument("--line", default="") - fallback_parser.add_argument("--multiplexer", choices=["tmux", "herdr", "auto"], default="auto") - fallback_parser.add_argument("--sort", choices=["appearance", "alpha"], default=None) - scan_parser = subparsers.add_parser("scan") scan_parser.add_argument("--pane-path", required=True) scan_parser.add_argument("--pane-id", required=True) @@ -1419,9 +1446,9 @@ def main() -> None: links_parser.add_argument("--multiplexer", choices=["tmux", "herdr", "auto"], default="auto") links_parser.add_argument("--sort", choices=["appearance", "alpha"], default=None) - reorder_parser = subparsers.add_parser("reorder") - reorder_parser.add_argument("--candidates-file", required=True) - reorder_parser.add_argument("--state-file", required=True) + preview_parser = subparsers.add_parser("preview") + preview_parser.add_argument("--pane-path", required=True) + preview_parser.add_argument("target") args = parser.parse_args() log_event("start", command=args.command, argv=sys.argv[1:]) @@ -1430,14 +1457,16 @@ def main() -> None: cmd_pick(args) elif args.command == "open": cmd_open(args) - elif args.command == "fallback": - cmd_fallback(args) elif args.command == "scan": cmd_scan(args) elif args.command == "links": cmd_links(args) - elif args.command == "reorder": - cmd_reorder(args) + elif args.command == "preview": + try: + target = decode_candidate(args.target) + except ValueError as exc: + raise SystemExit(f"termscope: {exc}") from exc + preview_target(target, Path(args.pane_path)) if __name__ == "__main__": diff --git a/termscope_herdr.py b/termscope_herdr.py index 0fe5bd1..845076a 100644 --- a/termscope_herdr.py +++ b/termscope_herdr.py @@ -1,15 +1,9 @@ #!/usr/bin/env python3 -"""Herdr plugin wrapper for termscope. +"""Herdr plugin wrapper for Termscope. -This script is invoked in two modes: - - --open-pane action mode: open the plugin overlay pane for file picking - --open-links-pane action mode: open the plugin overlay pane for link picking - (no flag) pane mode: run inside the overlay pane and launch fzf - -Action mode receives HERDR_PANE_ID (the source pane) and passes the source -pane id/cwd to the overlay pane via env vars. Pane mode reads those vars and -runs the shared picker script. +Action mode captures the originating pane id/cwd and opens a session-modal +popup. Popup mode launches Television through the shared picker script. Herdr +closes the popup automatically when that command exits. """ from __future__ import annotations @@ -32,8 +26,8 @@ def plugin_root() -> str: return os.environ.get("HERDR_PLUGIN_ROOT", os.getcwd()) -def get_source_pane() -> tuple[str, str]: - """Return (pane_id, cwd) for the source pane.""" +def get_source_pane() -> tuple[str, str, str]: + """Return (pane_id, cwd, agent) for the source pane.""" pane_id = os.environ.get("HERDR_PANE_ID") or os.environ.get("HERDR_ACTIVE_PANE_ID") if not pane_id: print("termscope: no source pane id", file=sys.stderr) @@ -51,17 +45,19 @@ def get_source_pane() -> tuple[str, str]: try: data = json.loads(result.stdout) - cwd = data["result"]["pane"]["cwd"] + pane = data["result"]["pane"] + cwd = pane["cwd"] + agent = pane.get("agent") or "" except (json.JSONDecodeError, KeyError) as exc: print(f"termscope: bad pane get response: {exc}", file=sys.stderr) sys.exit(1) - return pane_id, cwd + return pane_id, cwd, agent -def open_overlay(entrypoint: str) -> None: - pane_id, cwd = get_source_pane() - subprocess.run( +def open_popup(entrypoint: str) -> None: + pane_id, cwd, agent = get_source_pane() + result = subprocess.run( [ herdr_bin(), "plugin", @@ -72,14 +68,21 @@ def open_overlay(entrypoint: str) -> None: "--entrypoint", entrypoint, "--placement", - "overlay", + "popup", "--env", f"SOURCE_PANE_ID={pane_id}", "--env", f"SOURCE_PANE_CWD={cwd}", + "--env", + f"SOURCE_PANE_AGENT={agent}", ], + text=True, + capture_output=True, check=False, ) + if result.returncode != 0: + detail = result.stderr.strip() or result.stdout.strip() or "unknown error" + raise SystemExit(f"termscope: popup launch failed: {detail}") def run_picker(subcommand: str) -> None: @@ -105,15 +108,17 @@ def run_picker(subcommand: str) -> None: "--sort", sort, ] - subprocess.run(cmd, check=False) + result = subprocess.run(cmd, check=False) + if result.returncode != 0: + raise SystemExit(result.returncode) def main() -> None: args = sys.argv[1:] if "--open-pane" in args: - open_overlay("picker") + open_popup("picker") elif "--open-links-pane" in args: - open_overlay("link-picker") + open_popup("link-picker") elif "--links" in args: run_picker("links") else: diff --git a/tests/test_installer.py b/tests/test_installer.py new file mode 100644 index 0000000..ad2511d --- /dev/null +++ b/tests/test_installer.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Contract tests for the install-time Television bootstrap.""" + +from __future__ import annotations + +import os +import subprocess +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +INSTALLER = ROOT / "scripts" / "install-dependencies.sh" + + +class TestDependencyInstaller(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + self.bin_dir = Path(self.temp_dir.name) / "bin" + self.bin_dir.mkdir() + self.marker = Path(self.temp_dir.name) / "brew-called" + self.env = os.environ.copy() + self.env.update( + { + "PATH": f"{self.bin_dir}:/usr/bin:/bin", + "FAKE_BIN": str(self.bin_dir), + "BREW_MARKER": str(self.marker), + } + ) + + def write_executable(self, name: str, content: str) -> Path: + path = self.bin_dir / name + path.write_text(content) + path.chmod(0o755) + return path + + def run_installer(self) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["sh", str(INSTALLER)], + env=self.env, + text=True, + capture_output=True, + check=False, + ) + + def write_brew(self, installed: bool) -> None: + list_status = 0 if installed else 1 + self.write_executable( + "brew", + f"""#!/bin/sh +case "$1" in + list) exit {list_status} ;; + install|upgrade) + echo "$1" > "$BREW_MARKER" + cat > "$FAKE_BIN/tv" <<'TV' +#!/bin/sh +echo 'television 0.15.9' +TV + chmod +x "$FAKE_BIN/tv" + ;; + *) exit 2 ;; +esac +""", + ) + + def test_supported_television_skips_homebrew(self) -> None: + self.write_executable("tv", "#!/bin/sh\necho 'television 0.15.9'\n") + self.write_executable( + "brew", "#!/bin/sh\necho called > \"$BREW_MARKER\"\nexit 99\n" + ) + + result = self.run_installer() + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse(self.marker.exists()) + self.assertIn("already installed", result.stdout) + + def test_missing_homebrew_aborts_install(self) -> None: + result = self.run_installer() + + self.assertEqual(result.returncode, 1) + self.assertIn("requires Homebrew", result.stderr) + + def test_missing_television_is_installed(self) -> None: + self.write_brew(installed=False) + + result = self.run_installer() + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.marker.read_text().strip(), "install") + self.assertIn("Television installed", result.stdout) + + def test_old_television_is_upgraded(self) -> None: + self.write_executable("tv", "#!/bin/sh\necho 'television 0.7.1'\n") + self.write_brew(installed=True) + + result = self.run_installer() + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.marker.read_text().strip(), "upgrade") + self.assertIn("Television installed", result.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_plugin.py b/tests/test_plugin.py new file mode 100644 index 0000000..372683c --- /dev/null +++ b/tests/test_plugin.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +"""Contract tests for the Herdr plugin surface.""" + +import importlib.util +import json +import os +import sys +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +try: + import tomllib +except ModuleNotFoundError: + import tomli as tomllib + +ROOT = Path(__file__).parent.parent +spec = importlib.util.spec_from_file_location("termscope_herdr", ROOT / "termscope_herdr.py") +wrapper = importlib.util.module_from_spec(spec) +sys.modules["termscope_herdr"] = wrapper +spec.loader.exec_module(wrapper) + + +class TestManifest(unittest.TestCase): + def test_popup_contract(self): + manifest = tomllib.loads((ROOT / "herdr-plugin.toml").read_text()) + self.assertEqual(manifest["version"], "0.2.0") + self.assertEqual(manifest["min_herdr_version"], "0.7.4") + self.assertEqual( + manifest["build"], + [{ + "platforms": ["linux", "macos"], + "command": ["sh", "scripts/install-dependencies.sh"], + }], + ) + self.assertEqual({pane["id"] for pane in manifest["panes"]}, {"picker", "link-picker"}) + for pane in manifest["panes"]: + self.assertEqual(pane["placement"], "popup") + self.assertEqual(pane["width"], "80%") + self.assertEqual(pane["height"], "60%") + + def test_television_channels_preserve_both_orders(self): + appearance = tomllib.loads((ROOT / "cable" / "termscope-appearance.toml").read_text()) + alpha = tomllib.loads((ROOT / "cable" / "termscope-alpha.toml").read_text()) + self.assertEqual(appearance["metadata"]["name"], "termscope-appearance") + self.assertEqual(alpha["metadata"]["name"], "termscope-alpha") + self.assertEqual(appearance["source"]["command"][0]["name"], "Appearance") + self.assertEqual(alpha["source"]["command"][0]["name"], "Alphabetical") + for channel in (appearance, alpha): + self.assertEqual(channel["source"]["display"], "{split:\\t:1..}") + self.assertEqual(channel["source"]["output"], "{split:\\t:0}") + self.assertTrue(channel["source"]["no_sort"]) + self.assertFalse(channel["source"]["frecency"]) + self.assertEqual(channel["keybindings"]["ctrl-s"], "cycle_sources") + preview = channel["preview"]["command"] + self.assertIn("{split:\\t:0}", preview) + self.assertNotIn("'{}'", preview) + self.assertEqual( + channel["ui"]["preview_panel"]["footer"], + "Enter Neovim · Ctrl-O Default app · Ctrl-Y Plannotator", + ) + + +class TestHerdrWrapper(unittest.TestCase): + def test_action_opens_popup_with_source_context(self): + pane = json.dumps({"result": {"pane": {"cwd": "/tmp/repo", "agent": "pi"}}}) + responses = [ + SimpleNamespace(returncode=0, stdout=pane, stderr=""), + SimpleNamespace(returncode=0, stdout="", stderr=""), + ] + env = { + "HERDR_PANE_ID": "1-1", + "HERDR_BIN_PATH": "/tmp/herdr", + "HERDR_PLUGIN_ID": "termscope", + } + with patch.dict(os.environ, env, clear=True), \ + patch.object(wrapper.subprocess, "run", side_effect=responses) as run: + wrapper.open_popup("picker") + + self.assertEqual(run.call_args_list[0].args[0], ["/tmp/herdr", "pane", "get", "1-1"]) + command = run.call_args_list[1].args[0] + self.assertEqual(command[:6], [ + "/tmp/herdr", "plugin", "pane", "open", "--plugin", "termscope" + ]) + self.assertIn("--placement", command) + self.assertEqual(command[command.index("--placement") + 1], "popup") + self.assertIn("SOURCE_PANE_ID=1-1", command) + self.assertIn("SOURCE_PANE_CWD=/tmp/repo", command) + self.assertIn("SOURCE_PANE_AGENT=pi", command) + self.assertTrue(run.call_args_list[1].kwargs["capture_output"]) + + def test_popup_launch_failure_is_propagated(self): + pane = json.dumps({"result": {"pane": {"cwd": "/tmp/repo"}}}) + responses = [ + SimpleNamespace(returncode=0, stdout=pane, stderr=""), + SimpleNamespace(returncode=1, stdout="", stderr="ui_busy"), + ] + with patch.dict(os.environ, {"HERDR_PANE_ID": "1-1"}, clear=True), \ + patch.object(wrapper.subprocess, "run", side_effect=responses): + with self.assertRaisesRegex(SystemExit, "ui_busy"): + wrapper.open_popup("picker") + + def test_popup_runs_shared_television_picker(self): + env = { + "SOURCE_PANE_ID": "1-1", + "SOURCE_PANE_CWD": "/tmp/repo", + "HERDR_PLUGIN_ROOT": str(ROOT), + "TERMSCOPE_SORT": "alpha", + } + with patch.dict(os.environ, env, clear=True), \ + patch.object( + wrapper.subprocess, + "run", + return_value=SimpleNamespace(returncode=0), + ) as run: + wrapper.run_picker("pick") + + command = run.call_args.args[0] + self.assertEqual(command[1], str(ROOT / "termscope")) + self.assertEqual(command[2], "pick") + self.assertIn("--multiplexer", command) + self.assertEqual(command[command.index("--multiplexer") + 1], "herdr") + self.assertEqual(command[command.index("--sort") + 1], "alpha") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_termscope.py b/tests/test_termscope.py index cf59eaa..ef1c4e2 100644 --- a/tests/test_termscope.py +++ b/tests/test_termscope.py @@ -294,41 +294,43 @@ def test_appearance_sort_is_default(self): self.assertEqual(cands, ["src/utils.py", "src/main.py", "README.md"]) -class TestParseFzfResult(unittest.TestCase): - def test_query_key_selection(self): - r = tfp.parse_fzf_result("query\nenter\nselection") - self.assertEqual(r.query, "query") - self.assertEqual(r.key, "enter") - self.assertEqual(r.selection, "selection") - - def test_query_blank_key_selection(self): - r = tfp.parse_fzf_result("query\n\nselection") - self.assertEqual(r.query, "query") - self.assertEqual(r.key, "") - self.assertEqual(r.selection, "selection") - - def test_query_only(self): - r = tfp.parse_fzf_result("query") - self.assertEqual(r.query, "query") - self.assertEqual(r.key, "") - self.assertEqual(r.selection, "") - - def test_ctrl_o(self): - r = tfp.parse_fzf_result("query\nctrl-o\nselection") - self.assertEqual(r.key, "ctrl-o") - - def test_ctrl_y(self): - r = tfp.parse_fzf_result("query\nctrl-y\nselection") - self.assertEqual(r.query, "query") - self.assertEqual(r.key, "ctrl-y") - self.assertEqual(r.selection, "selection") - - def test_query_selection_no_expect(self): - # First rest is not a known key, treat as selection - r = tfp.parse_fzf_result("query\nselection") - self.assertEqual(r.query, "query") - self.assertEqual(r.key, "") - self.assertEqual(r.selection, "selection") +class TestParseTvResult(unittest.TestCase): + def test_enter_outputs_selection_only(self): + result = tfp.parse_tv_result("src/main.py\n", ("ctrl-o", "ctrl-y")) + self.assertEqual(result.key, "") + self.assertEqual(result.selection, "src/main.py") + + def test_expect_key_precedes_selection(self): + result = tfp.parse_tv_result("ctrl-y\nsrc/main.py\n", ("ctrl-o", "ctrl-y")) + self.assertEqual(result.key, "ctrl-y") + self.assertEqual(result.selection, "src/main.py") + + def test_empty_output_is_cancel(self): + self.assertEqual(tfp.parse_tv_result("", ("ctrl-y",)), tfp.PickerResult()) + + def test_non_expect_first_line_remains_part_of_selection(self): + result = tfp.parse_tv_result("first\nsecond\n", ("ctrl-y",)) + self.assertEqual(result.key, "") + self.assertEqual(result.selection, "first\nsecond") + + +class TestCandidateEncoding(unittest.TestCase): + def test_round_trips_shell_metacharacters(self): + candidates = [ + "path with spaces.py:12", + "semi;colon.py", + "$(touch nope).py", + "single'quote.py", + 'double"quote.py', + ] + for candidate in candidates: + encoded = tfp.encode_candidate(candidate) + self.assertRegex(encoded, r"^[A-Za-z0-9_-]+$") + self.assertEqual(tfp.decode_candidate(encoded), candidate) + + def test_rejects_invalid_candidate_id(self): + with self.assertRaisesRegex(ValueError, "invalid Television candidate id"): + tfp.decode_candidate("!") class TestCapturePaneVisibleOnly(unittest.TestCase): @@ -489,8 +491,8 @@ def test_empty_candidates_falls_back_to_full_repo(self): patch.object(tfp, "strip_ansi", side_effect=lambda x: x), \ patch.object(tfp, "list_repo_files", return_value=[]), \ patch.object(tfp, "extract_visible_candidates", return_value=[]), \ - patch.object(tfp, "run_fzf_visible") as mock_fzf, \ - patch.object(tfp, "run_fzf_full_repo", return_value=None) as mock_full, \ + patch.object(tfp, "run_tv_visible") as mock_picker, \ + patch.object(tfp, "run_tv_full_repo", return_value=None) as mock_full, \ patch.object(tfp.subprocess, "run") as mock_run, \ patch("os.execvp"): @@ -499,8 +501,8 @@ def test_empty_candidates_falls_back_to_full_repo(self): tfp.cmd_pick(args) - # run_fzf_visible should NOT be called when candidates are empty - mock_fzf.assert_not_called() + # The visible-only picker is skipped when no visible candidate exists. + mock_picker.assert_not_called() # Should fall back to full-repo listing mock_full.assert_called_once() @@ -517,8 +519,8 @@ def test_empty_candidates_falls_back_to_full_repo(self): ) -class TestQueryOnlyNoOpen(unittest.TestCase): - def test_typed_query_without_selection_does_not_open(self): +class TestCancelledPickerNoOpen(unittest.TestCase): + def test_empty_selection_does_not_open(self): args = SimpleNamespace( pane_path="/tmp/repo", pane_id="%1", @@ -528,7 +530,7 @@ def test_typed_query_without_selection_does_not_open(self): patch.object(tfp, "strip_ansi", side_effect=lambda x: x), \ patch.object(tfp, "list_repo_files", return_value=["src/main.py"]), \ patch.object(tfp, "extract_visible_candidates", return_value=["src/main.py"]), \ - patch.object(tfp, "run_fzf_visible", return_value=tfp.FzfResult(query="README.md", key="enter", selection="")), \ + patch.object(tfp, "run_tv_visible", return_value=tfp.PickerResult(selection="")), \ patch.object(tfp, "_open_target") as mock_open, \ patch.object(tfp.subprocess, "run") as mock_run: @@ -758,108 +760,333 @@ def test_env_var_defaults_to_alpha(self): self.assertEqual(cands, ["README.md", "src/main.py", "src/utils.py"]) -class TestReorderCommand(unittest.TestCase): - def setUp(self): - self.tmp = Path(__file__).parent / "_test_reorder" - self.tmp.mkdir(exist_ok=True) - self.cand = self.tmp / "candidates.txt" - self.state = self.tmp / "state.txt" +class TestTelevisionCommands(unittest.TestCase): + def test_home_path_is_shortened_for_picker_header(self): + self.assertEqual(tfp.display_path(Path.home() / "projects"), "~/projects") + + def test_empty_candidates_skip_television(self): + with patch.object(tfp.subprocess, "run") as run: + self.assertIsNone(tfp._run_tv([], "Empty", (), cwd=Path("/tmp"))) + run.assert_not_called() + + def test_visible_picker_uses_channels_expect_keys_and_both_orders(self): + captured = {} + + def fake_run(command, **kwargs): + captured["command"] = command + captured["appearance"] = Path( + kwargs["env"]["TERMSCOPE_APPEARANCE_CANDIDATES"] + ).read_text() + captured["alpha"] = Path( + kwargs["env"]["TERMSCOPE_ALPHA_CANDIDATES"] + ).read_text() + return SimpleNamespace( + returncode=0, + stdout=f"ctrl-y\n{tfp.encode_candidate('a.py')}\n", + ) - def tearDown(self): - import shutil - shutil.rmtree(self.tmp, ignore_errors=True) + with patch.object(tfp.shutil, "which", return_value="/opt/homebrew/bin/tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object(tfp.subprocess, "run", side_effect=fake_run): + result = tfp.run_tv_visible( + ["z.py", "a.py", "m.py"], Path("/tmp"), sort="appearance" + ) + + self.assertEqual(result, tfp.PickerResult(key="ctrl-y", selection="a.py")) + appearance = [line.split("\t", 1)[1] for line in captured["appearance"].splitlines()] + alpha = [line.split("\t", 1)[1] for line in captured["alpha"].splitlines()] + self.assertEqual(appearance, ["z.py", "a.py", "m.py"]) + self.assertEqual(alpha, ["a.py", "m.py", "z.py"]) + for line in captured["appearance"].splitlines(): + encoded, _ = line.split("\t", 1) + self.assertRegex(encoded, r"^[A-Za-z0-9_-]+$") + command = captured["command"] + self.assertEqual(command[0], "/opt/homebrew/bin/tv") + self.assertIn("termscope-appearance", command) + expect_index = command.index("--expect") + self.assertEqual(command[expect_index + 1], "ctrl-o;ctrl-y") + header_index = command.index("--input-header") + self.assertEqual(command[header_index + 1], "Files — /tmp") + self.assertNotIn("--no-preview", command) + + def test_alpha_preference_selects_alpha_first_channel(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace( + returncode=0, + stdout=f"{tfp.encode_candidate('a.py')}\n", + ), + ) as run: + tfp.run_tv_visible(["z.py", "a.py"], Path("/tmp"), sort="alpha") + + self.assertIn("termscope-alpha", run.call_args.args[0]) + + def test_link_picker_disables_preview(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace( + returncode=0, + stdout=f"{tfp.encode_candidate('https://example.com')}\n", + ), + ) as run: + result = tfp.run_tv_links(["https://example.com"], Path("/tmp")) + + self.assertEqual(result.selection, "https://example.com") + command = run.call_args.args[0] + self.assertIn("--no-preview", command) + expect_index = command.index("--expect") + self.assertEqual(command[expect_index + 1], "ctrl-y") + + def test_full_repo_uses_repo_files(self): + with patch.object(tfp, "list_repo_files", return_value=["src/main.py"]), \ + patch.object(tfp, "_run_tv", return_value=tfp.PickerResult(selection="src/main.py")) as run: + result = tfp.run_tv_full_repo(Path("/tmp/repo"), sort="alpha") - def test_toggles_appearance_to_alpha(self): - self.cand.write_text("z\na\nm\n") - self.state.write_text("appearance") - args = SimpleNamespace(candidates_file=str(self.cand), state_file=str(self.state)) + self.assertEqual(result.selection, "src/main.py") + self.assertEqual(run.call_args.args[0], ["src/main.py"]) + self.assertEqual(run.call_args.kwargs["sort"], "alpha") + + def test_missing_television_exits_nonzero(self): import io - buf = io.StringIO() - with patch("sys.stdout", buf): - tfp.cmd_reorder(args) - self.assertEqual(self.state.read_text(), "alpha") - self.assertEqual(buf.getvalue(), "a\nm\nz\n") - - def test_toggles_alpha_to_appearance(self): - self.cand.write_text("z\na\nm\n") - self.state.write_text("alpha") - args = SimpleNamespace(candidates_file=str(self.cand), state_file=str(self.state)) + with patch.object(tfp.shutil, "which", return_value=None), \ + patch.object(tfp, "_picker_error") as error, \ + patch("sys.stderr", io.StringIO()), \ + self.assertRaises(SystemExit) as raised: + tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(raised.exception.code, 1) + error.assert_called_once_with("television (tv) not found") + + def test_zero_status_empty_output_is_cancel(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=0, stdout=""), + ): + result = tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(result, tfp.PickerResult()) + + def test_key_without_selection_is_malformed(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=0, stdout="ctrl-y\n"), + ), \ + patch.object(tfp, "_picker_error") as error, \ + self.assertRaises(SystemExit) as raised: + tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(raised.exception.code, 1) + error.assert_called_once_with("Television returned an invalid selection") + + def test_selection_must_have_been_offered(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace( + returncode=0, + stdout=f"{tfp.encode_candidate('other.py')}\n", + ), + ), \ + patch.object(tfp, "_picker_error") as error, \ + self.assertRaises(SystemExit) as raised: + tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(raised.exception.code, 1) + error.assert_called_once_with("Television returned an unknown selection") + + def test_nonzero_television_status_reports_failure(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=True), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=1, stdout=""), + ), \ + patch.object(tfp, "_picker_error") as error, \ + self.assertRaises(SystemExit) as raised: + tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(raised.exception.code, 1) + error.assert_called_once_with("Television failed with status 1") + + def test_rejects_old_television(self): + with patch.object(tfp.shutil, "which", return_value="tv"), \ + patch.object(tfp, "_television_is_supported", return_value=False), \ + patch.object(tfp, "_picker_error") as error, \ + self.assertRaises(SystemExit) as raised: + tfp.run_tv_visible(["src/main.py"], Path("/tmp")) + self.assertEqual(raised.exception.code, 1) + error.assert_called_once_with("Television 0.15+ is required") + + def test_version_preflight_accepts_015(self): + completed = SimpleNamespace(returncode=0, stdout="television 0.15.9\n") + with patch.object(tfp.subprocess, "run", return_value=completed): + self.assertTrue(tfp._television_is_supported("tv")) + + def test_version_preflight_rejects_071(self): + completed = SimpleNamespace(returncode=0, stdout="television 0.7.1\n") + with patch.object(tfp.subprocess, "run", return_value=completed): + self.assertFalse(tfp._television_is_supported("tv")) + + def test_version_preflight_rejects_nonzero_status(self): + completed = SimpleNamespace(returncode=1, stdout="television 0.15.9\n") + with patch.object(tfp.subprocess, "run", return_value=completed): + self.assertFalse(tfp._television_is_supported("tv")) + + def test_version_preflight_times_out(self): + with patch.object( + tfp.subprocess, + "run", + side_effect=tfp.subprocess.TimeoutExpired("tv", 2), + ): + self.assertFalse(tfp._television_is_supported("tv")) + + +class TestPreviewTarget(unittest.TestCase): + def test_whole_preview_timeout_is_reported(self): import io - buf = io.StringIO() - with patch("sys.stdout", buf): - tfp.cmd_reorder(args) - self.assertEqual(self.state.read_text(), "appearance") - self.assertEqual(buf.getvalue(), "z\na\nm\n") - - -class TestFzfCommands(unittest.TestCase): - def test_visible_picker_not_called_when_empty_candidates(self): - # After refactoring, empty candidates means no fzf call - with patch.object(tfp.subprocess, "run") as mock_run: - # If the implementation does call fzf, it should not happen for empty - # This test documents that empty candidates => no fzf - pass # Covered by TestNoCandidates - - def test_fallback_picker_feeds_fd_results_to_fzf(self): - completed = SimpleNamespace(returncode=0, stdout="src/main.py\n") - with patch.object(tfp, "list_repo_files", return_value=["src/main.py", "README.md"]), \ - patch.object(tfp.subprocess, "run", return_value=completed) as run: - selected = tfp.run_fallback_picker(Path("/tmp/repo"), "main") - - self.assertEqual(selected, "src/main.py") - self.assertEqual(run.call_args.kwargs["input"], "src/main.py\nREADME.md\n") - self.assertEqual(run.call_args.kwargs["stdout"], tfp.subprocess.PIPE) - self.assertNotIn("capture_output", run.call_args.kwargs) - - def test_fallback_picker_alpha_sorts_input(self): - completed = SimpleNamespace(returncode=0, stdout="src/main.py\n") - with patch.object(tfp, "list_repo_files", return_value=["src/main.py", "README.md", "app.py"]), \ - patch.object(tfp.subprocess, "run", return_value=completed) as run: - tfp.run_fallback_picker(Path("/tmp/repo"), "", sort="alpha") - - self.assertEqual(run.call_args.kwargs["input"], "app.py\nREADME.md\nsrc/main.py\n") - - def test_fallback_picker_includes_sort_toggle_bind(self): - completed = SimpleNamespace(returncode=0, stdout="src/main.py\n") - with patch.object(tfp, "list_repo_files", return_value=["src/main.py"]), \ - patch.object(tfp.subprocess, "run", return_value=completed) as run: - tfp.run_fallback_picker(Path("/tmp/repo"), "") + output = io.StringIO() + with patch.object(tfp, "_render_preview", side_effect=TimeoutError), \ + patch("sys.stdout", output): + tfp.preview_target("main.py", Path("/tmp")) + self.assertEqual(output.getvalue().strip(), "termscope: preview timed out") + + def test_file_line_uses_bat_range_and_highlight(self): + root = Path(__file__).parent / "_test_preview" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + target = root / "main.py" + target.write_text("\n".join(f"line {index}" for index in range(1, 80))) + + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value="bat"), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=0, stdout=""), + ) as run: + tfp.preview_target("main.py:42", root) + + command = run.call_args.args[0] + self.assertIn("--line-range", command) + self.assertIn("27:77", command) + self.assertIn("--highlight-line", command) + self.assertIn("42", command) + + def test_file_without_line_uses_bounded_bat_range(self): + root = Path(__file__).parent / "_test_preview_bounded_bat" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + (root / "main.py").write_text("one\ntwo\n") + + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value="bat"), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=0, stdout=""), + ) as run: + tfp.preview_target("main.py", root) + + command = run.call_args.args[0] + self.assertIn("--line-range", command) + self.assertIn("1:200", command) + + def test_failed_bat_falls_back_to_python_preview(self): + import io + root = Path(__file__).parent / "_test_preview_bat_failure" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + (root / "main.py").write_text("one\ntwo\nthree\n") + output = io.StringIO() + + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value="bat"), \ + patch.object( + tfp.subprocess, + "run", + return_value=SimpleNamespace(returncode=1, stdout="", stderr="bad cache"), + ), \ + patch("sys.stdout", output): + tfp.preview_target("main.py:2", root) + + self.assertIn("> 2 │ two", output.getvalue()) + + def test_python_preview_marks_requested_line_without_bat(self): + import io + root = Path(__file__).parent / "_test_preview_fallback" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + (root / "main.py").write_text("one\ntwo\nthree\n") + output = io.StringIO() - cmd = run.call_args.args[0] - bind_idx = cmd.index("--bind") - self.assertIn("ctrl-s:reload", cmd[bind_idx + 1]) - self.assertIn("reorder", cmd[bind_idx + 1]) + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value=None), \ + patch("sys.stdout", output): + tfp.preview_target("main.py:2", root) - def test_fzf_visible_includes_ctrl_y(self): - completed = SimpleNamespace(returncode=1, stdout="query\nctrl-y\nselection") - with patch.object(tfp.subprocess, "run", return_value=completed) as run: - result = tfp.run_fzf_visible(["src/main.py"], Path("/tmp/repo")) + self.assertIn("> 2 │ two", output.getvalue()) - self.assertIsNotNone(result) - self.assertEqual(result.key, "ctrl-y") - cmd = run.call_args.args[0] - # Should be in --expect value - expect_idx = cmd.index("--expect") - self.assertIn("ctrl-y", cmd[expect_idx + 1]) - header_idx = cmd.index("--header") - self.assertIn("annotate", cmd[header_idx + 1].lower()) - - def test_fzf_visible_alpha_sorts_input(self): - completed = SimpleNamespace(returncode=1, stdout="query\nenter\nselection") - with patch.object(tfp.subprocess, "run", return_value=completed) as run: - tfp.run_fzf_visible(["z.py", "a.py", "m.py"], Path("/tmp/repo"), sort="alpha") + def test_python_preview_bounds_a_single_large_line(self): + import io + root = Path(__file__).parent / "_test_preview_large_line" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + (root / "large.txt").write_text("x" * 1_100_000) + output = io.StringIO() + + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value="bat"), \ + patch.object(tfp.subprocess, "run") as run, \ + patch("sys.stdout", output): + tfp.preview_target("large.txt", root) + + run.assert_not_called() + rendered = output.getvalue() + self.assertLess(len(rendered), 1_100) + self.assertTrue(rendered.rstrip().endswith("…")) + + def test_directory_preview_lists_directories_first(self): + import io + root = Path(__file__).parent / "_test_preview_dir" + root.mkdir(exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + (root / "z.txt").write_text("z") + (root / "folder").mkdir() + output = io.StringIO() - self.assertEqual(run.call_args.kwargs["input"], "a.py\nm.py\nz.py\n") + with patch.object(tfp, "find_search_root", return_value=root), \ + patch("sys.stdout", output): + tfp.preview_target(".", root) - def test_fzf_visible_includes_sort_toggle_bind(self): - completed = SimpleNamespace(returncode=1, stdout="query\nenter\nselection") - with patch.object(tfp.subprocess, "run", return_value=completed) as run: - tfp.run_fzf_visible(["src/main.py"], Path("/tmp/repo")) + lines = output.getvalue().splitlines() + self.assertEqual(lines[1].strip(), "folder/") - cmd = run.call_args.args[0] - bind_idx = cmd.index("--bind") - self.assertIn("ctrl-s:reload", cmd[bind_idx + 1]) - self.assertIn("reorder", cmd[bind_idx + 1]) + def test_external_file_preview_is_rendered(self): + import io + base = Path(__file__).parent / "_test_preview_external" + root = base / "repo" + root.mkdir(parents=True, exist_ok=True) + self.addCleanup(lambda: __import__("shutil").rmtree(base, ignore_errors=True)) + outside = base / "external.txt" + outside.write_text("outside workspace") + output = io.StringIO() + + with patch.object(tfp, "find_search_root", return_value=root), \ + patch.object(tfp.shutil, "which", return_value=None), \ + patch("sys.stdout", output): + tfp.preview_target(str(outside), root) + + self.assertIn("outside workspace", output.getvalue()) class TestSendAnnotateCommand(unittest.TestCase): @@ -954,6 +1181,22 @@ def fake_run(cmd, **kwargs): ] self.assertEqual(len(display_calls), 1, "should show Plannotator message") + def test_plain_shell_runs_plannotator_cli(self): + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + if cmd[0] == "tmux" and cmd[1] == "display-message": + return SimpleNamespace(returncode=0, stdout="0\tzsh\n") + return SimpleNamespace(returncode=0, stdout="") + + with patch.object(tfp.subprocess, "run", side_effect=fake_run): + tfp.send_annotate_command("%1", "/tmp/repo/src/main.py") + + literal_call = next(c for c in calls if c[:2] == ["tmux", "send-keys"] and "-l" in c) + command = literal_call[literal_call.index("-l") + 1] + self.assertEqual(command, "plannotator annotate /tmp/repo/src/main.py") + def test_path_with_spaces_is_quoted(self): """Paths containing spaces are shell-quoted with shlex.quote.""" calls = [] @@ -1019,16 +1262,12 @@ def test_uses_default_app(self): class TestLinkPickerHeader(unittest.TestCase): - def test_links_header_mentions_browser(self): - completed = SimpleNamespace(returncode=1, stdout="query\nenter\nhttps://example.com") - with patch.object(tfp.subprocess, "run", return_value=completed) as run: - result = tfp.run_fzf_links(["https://example.com"]) + def test_links_header_mentions_visible_links(self): + with patch.object(tfp, "_run_tv", return_value=tfp.PickerResult()) as run: + tfp.run_tv_links(["https://example.com"], Path("/tmp")) - self.assertIsNotNone(result) - self.assertEqual(result.key, "enter") - cmd = run.call_args.args[0] - header_idx = cmd.index("--header") - self.assertIn("browser", cmd[header_idx + 1]) + self.assertIn("Visible links", run.call_args.args[1]) + self.assertFalse(run.call_args.kwargs["preview"]) class TestBackendDetection(unittest.TestCase): @@ -1037,8 +1276,11 @@ def test_tmux_pane_id(self): self.assertIsInstance(tfp.detect_backend("%42"), tfp.TmuxBackend) def test_herdr_pane_id(self): + self.assertIsInstance(tfp.detect_backend("1-1"), tfp.HerdrBackend) + self.assertIsInstance(tfp.detect_backend("12-3"), tfp.HerdrBackend) self.assertIsInstance(tfp.detect_backend("w1:p1"), tfp.HerdrBackend) self.assertIsInstance(tfp.detect_backend("w12:pT"), tfp.HerdrBackend) + self.assertIsInstance(tfp.detect_backend("w1H:pS"), tfp.HerdrBackend) def test_empty_defaults_to_tmux(self): self.assertIsInstance(tfp.detect_backend(""), tfp.TmuxBackend) @@ -1081,9 +1323,10 @@ def test_capture_pane_text_reads_visible(self): cmd = run.call_args.args[0] self.assertEqual(cmd, ["herdr", "pane", "read", "w1:p1", "--source", "visible"]) - def test_send_annotate_command_runs_slash_command(self): + def test_send_annotate_command_runs_slash_command_for_agent(self): backend = tfp.HerdrBackend() - with patch.object(tfp.subprocess, "run") as run, \ + with patch.dict(os.environ, {"SOURCE_PANE_AGENT": "pi"}), \ + patch.object(tfp.subprocess, "run") as run, \ patch.object(backend, "_herdr_bin", return_value="herdr"): backend.send_annotate_command("w1:p1", "/tmp/repo/src/main.py") @@ -1095,6 +1338,17 @@ def test_send_annotate_command_runs_slash_command(self): self.assertEqual(calls[1].args[0][:3], ["herdr", "notification", "show"]) + def test_send_annotate_command_runs_cli_for_shell(self): + backend = tfp.HerdrBackend() + with patch.dict(os.environ, {"SOURCE_PANE_AGENT": ""}), \ + patch.object(tfp.subprocess, "run") as run, \ + patch.object(backend, "_herdr_bin", return_value="herdr"): + backend.send_annotate_command("w1:p1", "/tmp/repo/src/main.py") + + command = run.call_args_list[0].args[0] + self.assertEqual(command[:4], ["herdr", "pane", "run", "w1:p1"]) + self.assertEqual(command[4], "plannotator annotate /tmp/repo/src/main.py") + def test_open_in_nvim_split_splits_then_runs(self): backend = tfp.HerdrBackend() split_json = json.dumps({