Add watch mode for live file monitoring
All checks were successful
CI / lint (push) Successful in -26s
CI / typecheck (push) Successful in -23s
CI / test (3.12) (push) Successful in -25s
CI / test (3.13) (push) Successful in -24s
CI / test (3.14) (push) Successful in -24s
CI / lint (pull_request) Successful in -28s
CI / test (3.14) (pull_request) Successful in -24s
CI / typecheck (pull_request) Successful in -24s
CI / test (3.12) (pull_request) Successful in -25s
CI / test (3.13) (pull_request) Successful in -24s

Add --watch flag that polls a file and redraws the terminal when its
content changes. Useful for monitoring log files, configuration files,
or any file that changes over time.

Features:
- Polls file every 1 second for changes
- Disables pager (pager would block the watch loop)
- File-only: rejects stdin or - input
- Error recovery: retries when file is missing or temporarily unreadable
- State comparison: only redraws when content actually changes
This commit is contained in:
2026-04-04 05:23:28 -07:00
parent d94cf2df30
commit 0f93d3a095
3 changed files with 257 additions and 22 deletions

View File

@@ -5,6 +5,7 @@ from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from typer.testing import CliRunner
from rp import __version__
@@ -171,6 +172,18 @@ class TestStdinInput:
assert result.exit_code == 0
def test_watch_rejects_stdin(self) -> None:
result = runner.invoke(app, ["--watch"], input="hello\n")
assert result.exit_code == 1
assert "--watch only supports file input" in result.output
def test_watch_rejects_dash_input(self) -> None:
result = runner.invoke(app, ["-", "--watch"], input="hello\n")
assert result.exit_code == 1
assert "--watch only supports file input" in result.output
def test_stdin_passes_none_path(self) -> None:
"""When reading from stdin, detect_type should receive path=None."""
with patch("rp.cli.detect_type") as mock_detect_type:
@@ -311,3 +324,96 @@ class TestLineNumbersOption:
result = runner.invoke(app, [str(file), "--no-line-numbers", "--no-pager"])
assert result.exit_code == 0
class TestWatchMode:
"""Tests for the watch mode loop."""
def test_watch_disables_pager(self, tmp_file) -> None:
file = tmp_file("x = 1\n", suffix=".py")
with patch("rp.cli.detect_type", return_value="python"):
with patch("rp.cli.time.sleep", side_effect=KeyboardInterrupt):
with patch("rp.cli.render") as mock_render:
with pytest.raises(KeyboardInterrupt):
main(file=file, pager=True, watch=True)
call_args = mock_render.call_args
assert call_args is not None
options = call_args[0][3]
assert options.pager is False
def test_watch_rerenders_on_file_change(self, tmp_file) -> None:
file = tmp_file("x = 1\n", suffix=".py")
sleep_calls = 0
def fake_sleep(_: float) -> None:
nonlocal sleep_calls
sleep_calls += 1
if sleep_calls == 1:
file.write_text("x = 2\n", encoding="utf-8")
return
raise KeyboardInterrupt
with patch("rp.cli.detect_type", return_value="python"):
with patch("rp.cli.time.sleep", side_effect=fake_sleep):
with patch("rp.cli.render") as mock_render:
with pytest.raises(KeyboardInterrupt):
main(file=file, watch=True)
assert [call.args[0] for call in mock_render.call_args_list] == [
"x = 1\n",
"x = 2\n",
]
def test_watch_retries_after_missing_file_and_recovers(self, tmp_file) -> None:
file = tmp_file("x = 1\n", suffix=".py")
sleep_calls = 0
def fake_sleep(_: float) -> None:
nonlocal sleep_calls
sleep_calls += 1
if sleep_calls == 1:
file.unlink()
return
if sleep_calls == 2:
file.write_text("x = 3\n", encoding="utf-8")
return
raise KeyboardInterrupt
with patch("rp.cli.detect_type", return_value="python"):
with patch("rp.cli.time.sleep", side_effect=fake_sleep):
with patch("rp.cli.Console.print") as mock_print:
with patch("rp.cli.render") as mock_render:
with pytest.raises(KeyboardInterrupt):
main(file=file, watch=True)
assert [call.args[0] for call in mock_render.call_args_list] == [
"x = 1\n",
"x = 3\n",
]
mock_print.assert_any_call(f"[red]Error:[/red] File not found: {file}")
def test_watch_recovers_when_file_is_missing_at_startup(
self, tmp_path: Path
) -> None:
file = tmp_path / "startup.py"
sleep_calls = 0
def fake_sleep(_: float) -> None:
nonlocal sleep_calls
sleep_calls += 1
if sleep_calls == 1:
file.write_text("x = 4\n", encoding="utf-8")
return
raise KeyboardInterrupt
with patch("rp.cli.detect_type", return_value="python"):
with patch("rp.cli.time.sleep", side_effect=fake_sleep):
with patch("rp.cli.Console.print") as mock_print:
with patch("rp.cli.render") as mock_render:
with pytest.raises(KeyboardInterrupt):
main(file=file, watch=True)
assert [call.args[0] for call in mock_render.call_args_list] == ["x = 4\n"]
mock_print.assert_any_call(f"[red]Error:[/red] File not found: {file}")