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 <name> 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.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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()
|
||||
+27
-19
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user