From 85ddfb4a71cd33e2ebb98bd5b2d3547b01bc48a9 Mon Sep 17 00:00:00 2001 From: Kosmos Date: Thu, 27 Aug 2026 07:46:48 +0000 Subject: [PATCH] Add arrow-key menus and report loading status wizard: when stdin and stdout are both TTYs, menus render inline and are driven by keys: up/down/j/k/Home/End move the bold '>' highlight, Enter accepts, Esc/Ctrl-C/q back out of cancellable menus while Ctrl-C on fixed menus still aborts. Rows redraw with erase+CR across the option block only and clear before returning so prompts stay aligned. Non-TTY stdio keeps the original numbered prompt verbatim. report: each configured source shows a transient "# Generating report: pulling data..." line that is erased on completion, only when stdout is a TTY; piped output remains exactly the report. tests: cover the numbered fallback, row rendering, status suppression and show+clear parity; reformat stray test files to satisfy ruff. --- src/weekly_activity/cli.py | 46 +++++++++++- src/weekly_activity/wizard.py | 127 +++++++++++++++++++++++++++++++++- tests/test_aggregate.py | 15 ++-- tests/test_cli.py | 76 ++++++++++++++++++++ tests/test_config.py | 46 +++++++----- tests/test_wizard.py | 58 ++++++++++++++++ 6 files changed, 340 insertions(+), 28 deletions(-) create mode 100644 tests/test_cli.py create mode 100644 tests/test_wizard.py diff --git a/src/weekly_activity/cli.py b/src/weekly_activity/cli.py index 223764c..8cc3275 100644 --- a/src/weekly_activity/cli.py +++ b/src/weekly_activity/cli.py @@ -8,7 +8,13 @@ from collections.abc import Sequence from datetime import UTC, datetime from pathlib import Path -from weekly_activity.aggregate import build_specs, collect_specs, format_combined +from weekly_activity.aggregate import ( + Collected, + SourceSpec, + build_specs, + collect_specs, + format_combined, +) from weekly_activity.config import ConfigError, default_config_path, load_config from weekly_activity.model import last_week_window from weekly_activity.report import format_report @@ -128,6 +134,42 @@ def resolve_period(args: argparse.Namespace) -> tuple[datetime, datetime]: return last_week_window() +_STATUS_PREFIX = "# Generating report" +_STATUS_ERASE = "\x1b[2K\r" + + +def _report_status_active() -> bool: + """Transient progress lines only make sense on an interactive terminal.""" + return sys.stdout.isatty() + + +def _show_report_status(name: str) -> None: + """Erase-then-overwrite in place; no newline keeps updates on one row.""" + sys.stdout.write(f"{_STATUS_ERASE}{_STATUS_PREFIX}: pulling {name} data...") + sys.stdout.flush() + + +def _hide_report_status() -> None: + """Erase the status row so nothing of it survives into the final output.""" + sys.stdout.write(_STATUS_ERASE) + sys.stdout.flush() + + +def _collect_report_sources( + specs: list[SourceSpec], since: datetime, until: datetime +) -> list[Collected]: + """Run collect_specs per source, showing a TTY-only progress line each.""" + show = _report_status_active() + collected: list[Collected] = [] + for spec in specs: + if show: + _show_report_status(spec.name) + collected.extend(collect_specs([spec], since, until)) + if show: + _hide_report_status() + return collected + + def main() -> None: args = parse_args() command = args.command or "report" @@ -150,7 +192,7 @@ def main() -> None: if not specs: raise SystemExit(f"error: {path} enables no sources; run `weekly-activity config`.") since, until = resolve_period(args) - collected = collect_specs(specs, since, until) + collected = _collect_report_sources(specs, since, until) print(format_combined(collected, since, until)) return diff --git a/src/weekly_activity/wizard.py b/src/weekly_activity/wizard.py index 6b97e72..a2b0b21 100644 --- a/src/weekly_activity/wizard.py +++ b/src/weekly_activity/wizard.py @@ -286,7 +286,17 @@ def _display_name(account: object) -> str: def _menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None: - """Numbered-choice menu: selected index, or None when backed out.""" + """Pick an option: interactive arrows on a TTY, numbered prompt otherwise. + + Returns the selected index, or None when backed out of a cancellable menu. + """ + if sys.stdin.isatty() and sys.stdout.isatty(): + return _arrow_menu(title, options, cancellable=cancellable) + return _numbered_menu(title, options, cancellable=cancellable) + + +def _numbered_menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None: + """Numbered-choice fallback for non-TTY stdin (piped input, tests, CI).""" print(title) for number, option in enumerate(options, 1): print(f" {number}) {option}") @@ -307,6 +317,121 @@ def _menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> i print(f"Please enter a number between 1 and {valid}.") +# --- Arrow-key menu (interactive TTYs only) --- + + +_BOLD = "\x1b[1m" +_RESET = "\x1b[0m" + +_ERASE_LINE = "\x1b[2K" +_CURSOR_DOWN = "\x1b[1B" +_CURSOR_UP_TEMPLATE = "\x1b[{}A" + +# Key classifications shared by the reader and the menu loop. +_KEY_UP = "up" +_KEY_DOWN = "down" +_KEY_HOME = "home" +_KEY_END = "end" +_KEY_ENTER = "enter" +_KEY_ESCAPE = "escape" +_KEY_OTHER = "other" + +_CSI_KEYS = {"[A": _KEY_UP, "[B": _KEY_DOWN, "[H": _KEY_HOME, "[F": _KEY_END} + + +def _read_keypress() -> str: + """Read one keystroke from stdin (cbreak mode); classify it.""" + import select + + char = sys.stdin.read(1) + if char == "\x1b": + # A bare Esc is not followed by more bytes; CSI sequences are. Peek briefly. + readable, _, _ = select.select([sys.stdin], [], [], 0.05) + if not readable: + return _KEY_ESCAPE + tail = sys.stdin.read(2) + return _CSI_KEYS.get(tail, _KEY_OTHER) + if char in ("\r", "\n"): + return _KEY_ENTER + if char == "\x01": # Ctrl-A + return _KEY_HOME + if char == "\x05": # Ctrl-E + return _KEY_END + if char == "k": + return _KEY_UP + if char == "j": + return _KEY_DOWN + return _KEY_OTHER + + +def _option_line(option: str, selected: bool) -> str: + """One rendered row: bold with a '>' marker on the selected line.""" + prefix = "> " if selected else " " + if selected: + return f"{_BOLD}{prefix}{option}{_RESET}" + return f"{prefix}{option}" + + +def _draw_menu_rows(options: Sequence[str], selected: int | None) -> None: + """Redraw or clear the option rows. + + On entry the cursor sits at column 0 of the blank row just below the + block; writes move up and repaint each row so the title above stays + untouched. With ``selected=None`` every row is erased instead, + clearing the menu before trailing prompts run. + """ + out = sys.stdout + out.write(_CURSOR_UP_TEMPLATE.format(len(options))) + for index, option in enumerate(options): + out.write(_ERASE_LINE + "\r") + if selected is not None: + out.write(_option_line(option, index == selected)) + out.write(_CURSOR_DOWN) + out.flush() + + +def _arrow_menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None: + """Inline re-rendered menu driven by raw-mode keypresses.""" + import termios + import tty + + print(title) + selected = 0 + for index, option in enumerate(options): # fresh draw lands one row per option + print(_option_line(option, index == selected)) + fd = sys.stdin.fileno() + saved_state = termios.tcgetattr(fd) + choice: int | None = None + try: + tty.setcbreak(fd) + while True: + key = _read_keypress() + if key == _KEY_ENTER: + choice = selected + break + if key == _KEY_UP: + selected = (selected - 1) % len(options) + elif key == _KEY_DOWN: + selected = (selected + 1) % len(options) + elif key == _KEY_HOME: + selected = 0 + elif key == _KEY_END: + selected = len(options) - 1 + elif cancellable and key in (_KEY_ESCAPE, "q"): + break + else: # Esc/q on fixed menus, or any unrecognised key: no-op. + continue + _draw_menu_rows(options, selected) + except KeyboardInterrupt: + # ISIG stays enabled under cbreak, so Ctrl-C raises SIGINT here. + if not cancellable: + raise + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, saved_state) + _draw_menu_rows(options, None) + return choice + + def _ask(prompt: str, default: str = "") -> str: suffix = f" [{default}]" if default else "" answer = input(f"{prompt}{suffix}: ").strip() diff --git a/tests/test_aggregate.py b/tests/test_aggregate.py index 1b6e732..111be78 100644 --- a/tests/test_aggregate.py +++ b/tests/test_aggregate.py @@ -74,8 +74,9 @@ class BuildSpecsTest(unittest.TestCase): ) specs = build_specs(config) # All launchpad accounts first, then github, then bts; listing order kept. - self.assertEqual([s.username for s in specs], ["lp-a", "lp-b", "gh-a", "gh-b", - "one@x.org", "two@x.org"]) + self.assertEqual( + [s.username for s in specs], ["lp-a", "lp-b", "gh-a", "gh-b", "one@x.org", "two@x.org"] + ) self.assertEqual( [s.name for s in specs], [ @@ -99,8 +100,9 @@ class BuildSpecsTest(unittest.TestCase): def test_launchpad_spec_expands_tilde_credentials_file(self) -> None: config = ActivityConfig(launchpad=[LaunchpadSettings(username="anon")]) - config.launchpad.append(LaunchpadSettings(mode="credentials", username="u", - credentials_file="~/creds.json")) + config.launchpad.append( + LaunchpadSettings(mode="credentials", username="u", credentials_file="~/creds.json") + ) specs = build_specs(config) with mock.patch("weekly_activity.aggregate.LaunchpadSource") as fake_source: for spec in specs: @@ -152,8 +154,9 @@ class CollectAndFormatTest(unittest.TestCase): def test_heading_uses_account_label_not_bare_source_name(self) -> None: since, until = window() - specs = [SourceSpec(name="github/work", username="acme-jane", - make=lambda: StubSource("github"))] + specs = [ + SourceSpec(name="github/work", username="acme-jane", make=lambda: StubSource("github")) + ] text = format_combined(collect_specs(specs, since, until), since, until) self.assertIn("Activity report — github/work / ~acme-jane", text) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..3136816 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,76 @@ +"""Unit tests for the report command's loading-status presentation layer.""" + +from __future__ import annotations + +import io +import sys +import unittest +from datetime import datetime +from unittest import mock + +from weekly_activity.aggregate import SourceSpec, collect_specs +from weekly_activity.cli import _collect_report_sources +from weekly_activity.model import ActivityReport, Section + + +class StubSource: + """Deterministic stand-in satisfying the ActivitySource interface.""" + + def __init__(self, *, fail: bool = False) -> None: + self._fail = fail + self.name = "stub" + + def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport: + if self._fail: + raise RuntimeError("some query failed") + return ActivityReport( + source=username, + username=username, + since=since, + until=until, + sections=[Section(title="Patches", entries=["one commit"])], + ) + + +def spec(name: str, *, fail: bool = False) -> SourceSpec: + source = StubSource(fail=fail) + return SourceSpec(name=name, username="me", make=lambda: source) + + +WINDOW = (datetime(2026, 8, 19), datetime(2026, 8, 26)) + + +class FakeTty(io.StringIO): + """Capture buffer that claims to be a terminal so status lines activate.""" + + def isatty(self) -> bool: + return True + + +class ReportStatusTest(unittest.TestCase): + def test_status_is_silent_without_a_tty(self) -> None: + specs = [spec("alpha"), spec("beta")] + buffer = io.StringIO() + with mock.patch.object(sys, "stdout", buffer): + collected = _collect_report_sources(specs, *WINDOW) + self.assertEqual(buffer.getvalue(), "") + self.assertEqual(collected, collect_specs(specs, *WINDOW)) + + def test_status_names_current_source_then_clears_on_a_tty(self) -> None: + specs = [spec("launchpad"), spec("github/personal"), spec("broken", fail=True)] + buffer = FakeTty() + with mock.patch.object(sys, "stdout", buffer): + collected = _collect_report_sources(specs, *WINDOW) + text = buffer.getvalue() + self.assertIn("# Generating report: pulling launchpad data...", text) + self.assertIn("# Generating report: pulling github/personal data...", text) + self.assertNotIn("# Generating report: pulling broken data...\n", text) + # The final erase leaves nothing visible behind the transient status. + self.assertTrue(text.endswith("\x1b[2K\r")) + self.assertEqual(collected, collect_specs(specs, *WINDOW)) + errors = [item.error for item in collected if item.error is not None] + self.assertTrue(errors and all("RuntimeError" in err for err in errors)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 4622586..944c5ae 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -43,9 +43,7 @@ def sample_v2_config() -> ActivityConfig: ], github=[ GithubSettings(username='gh"user', token="tok\\en"), - GithubSettings( - username="acme-jane", token_env="WORK_GH_TOKEN", name="work" - ), + GithubSettings(username="acme-jane", token_env="WORK_GH_TOKEN", name="work"), ], gitlab=[ GitlabSettings( @@ -121,13 +119,11 @@ class ConfigLoadingTest(unittest.TestCase): def test_unknown_top_level_key_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text('version = 2\nsources = 3\n') + load_config_from_text("version = 2\nsources = 3\n") def test_unknown_account_key_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text( - '[github]\naccounts = [{username = "x", tokken = "t"}]\n' - ) + load_config_from_text('[github]\naccounts = [{username = "x", tokken = "t"}]\n') def test_section_must_hold_only_accounts_table(self) -> None: with self.assertRaises(ConfigError): @@ -141,13 +137,11 @@ class ConfigLoadingTest(unittest.TestCase): with self.assertRaises(ConfigError): load_config_from_text('version = 2\n\n[github]\naccounts = "nope"\n') with self.assertRaises(ConfigError): - load_config_from_text('version = 2\n\n[gitlab]\naccounts = [42]\n') + load_config_from_text("version = 2\n\n[gitlab]\naccounts = [42]\n") def test_non_string_value_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text( - 'version = 2\n\n[[github.accounts]]\nusername = 42\n' - ) + load_config_from_text("version = 2\n\n[[github.accounts]]\nusername = 42\n") def test_bad_launchpad_mode_rejected(self) -> None: with self.assertRaises(ConfigError): @@ -169,8 +163,8 @@ class ConfigLoadingTest(unittest.TestCase): def test_account_errors_carry_position(self) -> None: text = ( "version = 2\n" - "\n[[launchpad.accounts]]\nusername = \"ok\"\n" - "\n[[launchpad.accounts]]\nusername = \"bad\"\nservice = \"oops\"\n" + '\n[[launchpad.accounts]]\nusername = "ok"\n' + '\n[[launchpad.accounts]]\nusername = "bad"\nservice = "oops"\n' ) with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "weekly-activity.toml" @@ -209,13 +203,27 @@ email = "dev@example.org" def test_v1_file_migrates_to_single_unnamed_accounts_in_memory(self) -> None: config = load_config_from_text(self.V1_TEXT) self.assertEqual(config.enabled_sources(), ["launchpad", "github", "gitlab", "bts"]) - self.assertEqual(config.launchpad, [LaunchpadSettings( - username="lp-user", mode="credentials", credentials_file="~/creds.json", - )]) + self.assertEqual( + config.launchpad, + [ + LaunchpadSettings( + username="lp-user", + mode="credentials", + credentials_file="~/creds.json", + ) + ], + ) self.assertEqual(config.github, [GithubSettings(username="octo", token="t0k")]) - self.assertEqual(config.gitlab, [GitlabSettings( - url="https://salsa.debian.org", username="gl", token_env="SALSA_TOKEN", - )]) + self.assertEqual( + config.gitlab, + [ + GitlabSettings( + url="https://salsa.debian.org", + username="gl", + token_env="SALSA_TOKEN", + ) + ], + ) self.assertEqual(config.bts, [BtsSettings(email="dev@example.org")]) def test_saving_migrated_config_persists_v2_shape(self) -> None: diff --git a/tests/test_wizard.py b/tests/test_wizard.py new file mode 100644 index 0000000..4ca709b --- /dev/null +++ b/tests/test_wizard.py @@ -0,0 +1,58 @@ +"""Unit tests for the wizard menus — scripted input, no raw terminal involved.""" + +from __future__ import annotations + +import io +import sys +import unittest +from unittest import mock + +from weekly_activity.wizard import _menu, _option_line + + +class NumberedMenuFallbackTest(unittest.TestCase): + """When stdin/stdout are not a TTY the numbered prompt stays as it was.""" + + def run_numbered(self, answers: list[str], options: list[str], *, cancellable: bool = False): + """Run _menu with forced non-TTY stdio and scripted interactive answers.""" + fake_stdin, fake_stdout = io.StringIO(), io.StringIO() + with ( + mock.patch.object(sys, "stdin", fake_stdin), + mock.patch.object(sys, "stdout", fake_stdout), + mock.patch("builtins.input", side_effect=iter(answers)), + ): + picked = _menu("Pick one", options, cancellable=cancellable) + return picked, fake_stdout.getvalue() + + def test_returns_zero_based_index_of_chosen_option(self) -> None: + picked, output = self.run_numbered(["2"], ["Alpha", "Beta", "Gamma"]) + self.assertEqual(picked, 1) + self.assertIn("Pick one", output) + self.assertIn(" 2) Beta", output) + + def test_empty_answer_backs_out_of_cancellable_menu(self) -> None: + picked, _ = self.run_numbered([""], ["Alpha", "Beta"], cancellable=True) + self.assertIsNone(picked) + + def test_back_number_returns_none(self) -> None: + picked, _ = self.run_numbered(["3"], ["Alpha", "Beta"], cancellable=True) + self.assertIsNone(picked) + + def test_invalid_answers_loop_until_valid_one(self) -> None: + picked, output = self.run_numbered(["9", "nope", "3"], ["Alpha", "Beta", "Gamma"]) + self.assertEqual(picked, 2) + self.assertEqual(output.count("Please enter a number between 1 and 3."), 2) + + +class OptionLineRenderingTest(unittest.TestCase): + """Selected rows carry the '>' marker and bold codes; others stay plain.""" + + def test_selected_line_is_bold_and_marked(self) -> None: + self.assertEqual(_option_line("Exit", True), "\x1b[1m> Exit\x1b[0m") + + def test_unselected_line_is_indented_and_plain(self) -> None: + self.assertEqual(_option_line("Remove a source", False), " Remove a source") + + +if __name__ == "__main__": + unittest.main()