From 6d8cc3d817819c4fbf0d0ba4c0d8aa0b8a7b2b0f Mon Sep 17 00:00:00 2001 From: Kosmos Date: Wed, 26 Aug 2026 20:18:41 +0000 Subject: [PATCH 1/5] Add config onboarding wizard and aggregated report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `weekly-activity config`: interactive wizard writing ~/.config/weekly-activity.toml (XDG-aware, --config override), one section per source, 0600 permissions, tokens never echoed; re-runs show current settings and pre-fill defaults - `weekly-activity report` (and bare invocation): loads enabled sources, collects each independently, joins blocks under one header; a failing source renders as [skipped: — error] without aborting - config.py: typed schema, tomllib reader with strict validation, hand-formatted TOML writer (stdlib-only, no new dependency) - aggregate.py: source spec building + failure-isolated collection + combined rendering - README: document config/report usage and config schema - tests/: 16 stdlib-unittest cases (config round-trip/escaping/validation, aggregation mapping/isolation) --- README.md | 49 +++++- src/weekly_activity/aggregate.py | 115 +++++++++++++ src/weekly_activity/cli.py | 85 ++++++++-- src/weekly_activity/config.py | 275 +++++++++++++++++++++++++++++++ src/weekly_activity/wizard.py | 186 +++++++++++++++++++++ tests/test_aggregate.py | 101 ++++++++++++ tests/test_config.py | 122 ++++++++++++++ 7 files changed, 921 insertions(+), 12 deletions(-) create mode 100644 src/weekly_activity/aggregate.py create mode 100644 src/weekly_activity/config.py create mode 100644 src/weekly_activity/wizard.py create mode 100644 tests/test_aggregate.py create mode 100644 tests/test_config.py diff --git a/README.md b/README.md index 69468e8..1b812ee 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,17 @@ Aggregate weekly activity across development platforms (Launchpad, GitHub, GitLa ## Usage +```bash +# Interactive setup: pick sources, store identities/tokens once +uv run weekly-activity config + +# Aggregated report across every configured source +uv run weekly-activity report # bare `weekly-activity` does the same +``` + +### Single sources (ad-hoc, no config needed) + + ```bash # Launchpad (uses system keyring for OAuth; --anonymous for public data only) uv run weekly-activity launchpad @@ -54,13 +65,49 @@ Any GitLab instance works via `--url`. Token discovery order: Rolling 7 days ending today (inclusive). Override with `--since` / `--until` (ISO dates, UTC). +## Configuration + +`weekly-activity config` walks through each known source and writes a TOML +file. Tokens are stored as provided (plaintext), but the file is created with +owner-only permissions (`0600`) and secrets are never echoed back. Use +`--config ` (also accepted by `report`) to keep configs elsewhere. + +Default location: `${XDG_CONFIG_HOME:-~/.config}/weekly-activity.toml`. + +```toml +[github] +username = "octocat" +token = "" # empty -> fall back to $GITHUB_TOKEN / `gh auth token` + +[launchpad] +username = "jane" +mode = "credentials" # "anonymous" (public data only) | "credentials" +credentials_file = "" # empty -> system keyring; OAuth runs at report time + +[gitlab] +url = "https://salsa.debian.org" +username = "jane" +token_env = "SALSA_TOKEN" + +[bts] +email = "jane@example.org" +``` + +An absent `[section]` disables that source. Re-running `weekly-activity +config` shows the current settings and offers to redo the wizard with the +existing values as defaults. + + ## Architecture ``` src/weekly_activity/ model.py ActivityReport, Section, last_week_window report.py format_report() — source-agnostic text renderer - cli.py subcommand dispatch (launchpad | github | gitlab | bts) + cli.py subcommand dispatch (launchpad | github | gitlab | bts | report | config) + aggregate.py collect_specs()/format_combined() — one text report across sources + config.py ActivityConfig schema + TOML load/save (tomllib read, hand-written) + wizard.py interactive `config` onboarding sources/ __init__.py ActivitySource protocol launchpad.py LaunchpadSource (launchpadlib) diff --git a/src/weekly_activity/aggregate.py b/src/weekly_activity/aggregate.py new file mode 100644 index 0000000..1333281 --- /dev/null +++ b/src/weekly_activity/aggregate.py @@ -0,0 +1,115 @@ +"""Combined report — build every configured source and render one text output. + +Each enabled source runs independently: a failure (bad token, unreachable +API, unknown user) is recorded as a ``[skipped: …]`` block instead +of aborting the whole report. Per-source warnings surfaced by ``collect`` +are rendered inside the source's block by ``format_report``. +""" + +from __future__ import annotations + +import os +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timedelta + +from weekly_activity.config import ActivityConfig +from weekly_activity.model import ActivityReport +from weekly_activity.report import format_report +from weekly_activity.sources import ActivitySource +from weekly_activity.sources.debian_bts import DebianBtsSource +from weekly_activity.sources.github import GitHubSource +from weekly_activity.sources.gitlab import GitLabSource +from weekly_activity.sources.launchpad import LaunchpadSource + +_RULE = "─" * 60 + + +@dataclass(frozen=True, slots=True) +class SourceSpec: + """One enabled source: display name, identity to query, deferred constructor.""" + + name: str + username: str + make: Callable[[], ActivitySource] + + +@dataclass(frozen=True, slots=True) +class Collected: + spec: SourceSpec + report: ActivityReport | None = None + error: str | None = None + + +def build_specs(config: ActivityConfig) -> list[SourceSpec]: + """Map configured sections onto constructible sources (network deferred).""" + specs: list[SourceSpec] = [] + if (lp := config.launchpad) is not None: + anonymous = lp.mode == "anonymous" + credentials_file = os.path.expanduser(lp.credentials_file) + specs.append( + SourceSpec( + name="launchpad", + username=lp.username, + make=lambda: LaunchpadSource( + service=lp.service, + anonymous=anonymous, + credentials_file=credentials_file or None, + ), + ) + ) + if (gh := config.github) is not None: + token = gh.token + specs.append( + SourceSpec( + name="github", + username=gh.username, + make=lambda: GitHubSource(token=token or None), + ) + ) + if (gl := config.gitlab) is not None: + url, token, token_env = gl.url, gl.token, gl.token_env + specs.append( + SourceSpec( + name="gitlab", + username=gl.username, + make=lambda: GitLabSource( + url=url, token=token or None, token_env=token_env or None + ), + ) + ) + if (bts := config.bts) is not None: + email = bts.email + specs.append(SourceSpec(name="bts", username=email, make=DebianBtsSource)) + return specs + + +def collect_specs(specs: list[SourceSpec], since: datetime, until: datetime) -> list[Collected]: + """Run every spec, converting any per-source failure into an error record.""" + collected: list[Collected] = [] + for spec in specs: + try: + source = spec.make() + report = source.collect(spec.username, since, until) + except Exception as exc: # noqa: BLE001 - one bad source must not kill the rest + collected.append(Collected(spec, error=f"{type(exc).__name__}: {exc}")) + else: + collected.append(Collected(spec, report=report)) + return collected + + +def format_combined(collected: list[Collected], since: datetime, until: datetime) -> str: + """Render all blocks under one header; failed sources become skip notes.""" + end_date = (until - timedelta(days=1)).strftime("%Y-%m-%d") + header = ( + f"Weekly activity report — {len(collected)} source(s) — " + f"{since.strftime('%Y-%m-%d')} .. {end_date} (UTC, inclusive)" + ) + blocks: list[str] = [] + for item in collected: + if item.report is not None: + blocks.append(format_report(item.report)) + else: + detail = item.error or "unknown error" + blocks.append(f"[skipped: {item.spec.name} — {detail}]") + return ("\n\n" + _RULE + "\n\n").join([header, *blocks]) diff --git a/src/weekly_activity/cli.py b/src/weekly_activity/cli.py index 858fea7..223764c 100644 --- a/src/weekly_activity/cli.py +++ b/src/weekly_activity/cli.py @@ -6,13 +6,17 @@ import argparse import sys 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.config import ConfigError, default_config_path, load_config from weekly_activity.model import last_week_window from weekly_activity.report import format_report from weekly_activity.sources.debian_bts import DebianBtsSource from weekly_activity.sources.github import GitHubSource from weekly_activity.sources.gitlab import GitLabSource from weekly_activity.sources.launchpad import LaunchpadSource +from weekly_activity.wizard import run_wizard def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: @@ -26,13 +30,28 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: help="Inclusive end as an ISO date/datetime (UTC). Default: today.", ) + common = argparse.ArgumentParser(add_help=False) + common.add_argument( + "--config", + default=None, + metavar="PATH", + help="Config file location. Default: ~/.config/weekly-activity.toml.", + ) + parser = argparse.ArgumentParser( prog="weekly-activity", - description="Summarize a user's activity for last week (or a custom range).", + description=( + "Summarize dev-platform activity for a week - one source at a time," + " or aggregated across everything configured." + ), + ) + subparsers = parser.add_subparsers( + dest="command", + required=False, + help="a single source, `report` for all configured ones, `config` to set up", ) - subparsers = parser.add_subparsers(dest="source", required=True) - lp = subparsers.add_parser("launchpad", parents=[base], help="Launchpad activity.") + lp = subparsers.add_parser("launchpad", parents=[base, common], help="Launchpad activity.") lp.add_argument("username", help="Launchpad username (the ~name).") lp.add_argument( "--anonymous", @@ -46,13 +65,13 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: default="production", ) - gh = subparsers.add_parser("github", parents=[base], help="GitHub activity.") + gh = subparsers.add_parser("github", parents=[base, common], help="GitHub activity.") gh.add_argument("username", help="GitHub username.") gh.add_argument("--token", default=None, help="GitHub token (or set GITHUB_TOKEN env var).") gl = subparsers.add_parser( "gitlab", - parents=[base], + parents=[base, common], help="GitLab activity (gitlab.com or any self-hosted instance, e.g. Debian Salsa).", ) gl.add_argument("username", help="GitLab username.") @@ -69,9 +88,22 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: help="Env var holding the API token (see README for discovery order).", ) - bts = subparsers.add_parser("bts", parents=[base], help="Debian Bug Tracking System activity.") + bts = subparsers.add_parser( + "bts", parents=[base, common], help="Debian Bug Tracking System activity." + ) bts.add_argument("username", help="Email address used on Debian bugs (submitter/owner).") + subparsers.add_parser( + "report", + parents=[base, common], + help="Aggregate activity from every source in the config file.", + ) + subparsers.add_parser( + "config", + parents=[common], + help="Interactive wizard that writes the config file.", + ) + return parser.parse_args(argv) @@ -98,22 +130,46 @@ def resolve_period(args: argparse.Namespace) -> tuple[datetime, datetime]: def main() -> None: args = parse_args() + command = args.command or "report" + + if command == "config": + run_wizard(_config_path(args)) + return + + if command == "report": + path = _config_path(args) + try: + config = load_config(path) + except FileNotFoundError as exc: + raise SystemExit( + f"error: no configuration found at {path}; run `weekly-activity config` first." + ) from exc + except ConfigError as exc: + raise SystemExit(f"error: {exc}") from exc + specs = build_specs(config) + 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) + print(format_combined(collected, since, until)) + return + since, until = resolve_period(args) - if args.source == "launchpad": + if command == "launchpad": source = LaunchpadSource( service=args.service, anonymous=args.anonymous, credentials_file=args.credentials_file, ) - elif args.source == "github": + elif command == "github": source = GitHubSource(token=args.token) - elif args.source == "gitlab": + elif command == "gitlab": source = GitLabSource(url=args.url, token=args.token, token_env=args.token_env) - elif args.source == "bts": + elif command == "bts": source = DebianBtsSource() else: - raise SystemExit(f"error: unknown source {args.source!r}") + raise SystemExit(f"error: unknown source {command!r}") try: report = source.collect(args.username, since, until) @@ -126,5 +182,12 @@ def main() -> None: print(format_report(report)) +def _config_path(args: argparse.Namespace) -> Path: + """Resolve this run's config location (--config override or XDG default).""" + if args.config: + return Path(args.config).expanduser() + return default_config_path() + + if __name__ == "__main__": main() diff --git a/src/weekly_activity/config.py b/src/weekly_activity/config.py new file mode 100644 index 0000000..ce32c3f --- /dev/null +++ b/src/weekly_activity/config.py @@ -0,0 +1,275 @@ +"""Config file: schema, discovery, loading, and persistence. + +Stored as TOML at ``${XDG_CONFIG_HOME:-~/.config}/weekly-activity.toml`` +(override with ``--config``). One section per source; an absent section +means the source is not enabled. + +Tokens are stored as provided (plaintext) in the user's own local config; +the file is created with ``0600`` permissions so they are not world-readable. +""" + +from __future__ import annotations + +import os +import tomllib +from dataclasses import dataclass +from pathlib import Path + +CONFIG_FILE_NAME = "weekly-activity.toml" + +_LAUNCHPAD_MODES = frozenset({"anonymous", "credentials"}) +_LAUNCHPAD_SERVICES = frozenset({"production", "staging"}) + + +class ConfigError(Exception): + """The config file exists but is malformed or has unknown entries.""" + + +@dataclass(slots=True) +class GithubSettings: + username: str + token: str = "" # empty -> fall back to GITHUB_TOKEN / `gh auth token` + + +@dataclass(slots=True) +class LaunchpadSettings: + username: str + mode: str = "anonymous" # "anonymous" | "credentials" + credentials_file: str = "" # used in "credentials" mode; empty -> keyring + service: str = "production" + + +@dataclass(slots=True) +class GitlabSettings: + username: str + url: str = "https://gitlab.com" + token: str = "" # empty -> env discovery (_TOKEN, GITLAB_TOKEN) + token_env: str = "" # optional named var, takes precedence in discovery + + +@dataclass(slots=True) +class BtsSettings: + email: str + + +@dataclass(slots=True) +class ActivityConfig: + """All configured sources; ``None`` section = source disabled.""" + + launchpad: LaunchpadSettings | None = None + github: GithubSettings | None = None + gitlab: GitlabSettings | None = None + bts: BtsSettings | None = None + + def enabled_sources(self) -> list[str]: + """Names of configured source sections, in stable report order.""" + ordered: list[tuple[str, bool]] = [ + ("launchpad", self.launchpad is not None), + ("github", self.github is not None), + ("gitlab", self.gitlab is not None), + ("bts", self.bts is not None), + ] + return [name for name, on in ordered if on] + + +def default_config_path() -> Path: + xdg = os.environ.get("XDG_CONFIG_HOME") + base = Path(xdg) if xdg else Path.home() / ".config" + return base / CONFIG_FILE_NAME + + +# --- Loading --- + + +def load_config(path: Path) -> ActivityConfig: + """Parse the TOML config at ``path``. + + Raises ``FileNotFoundError`` when the file does not exist and + ``ConfigError`` when it cannot be parsed or contains unknown entries. + """ + try: + text = path.read_text(encoding="utf-8") + except FileNotFoundError: + raise + except OSError as exc: + raise ConfigError(f"cannot read {path}: {exc}") from exc + try: + data = tomllib.loads(text) + except tomllib.TOMLDecodeError as exc: + raise ConfigError(f"{path}: invalid TOML: {exc}") from exc + + where = str(path) + _reject_unknown_top_level(data, where) + return ActivityConfig( + launchpad=_parse_launchpad(data.get("launchpad"), where) if "launchpad" in data else None, + github=_parse_github(data.get("github"), where) if "github" in data else None, + gitlab=_parse_gitlab(data.get("gitlab"), where) if "gitlab" in data else None, + bts=_parse_bts(data.get("bts"), where) if "bts" in data else None, + ) + + +def _reject_unknown_top_level(data: dict[str, object], where: str) -> None: + unknown = sorted(set(data) - {"launchpad", "github", "gitlab", "bts"}) + if unknown: + raise ConfigError(f"{where}: unknown config section(s): {', '.join(unknown)}") + + +def _check_keys(table: dict[str, object], names: set[str], section: str) -> None: + unknown = sorted(set(table) - names) + if unknown: + raise ConfigError(f"[{section}]: unknown key(s): {', '.join(unknown)}") + + +def _required_str(table: dict[str, object], key: str, section: str) -> str: + value = table[key] + if not isinstance(value, str): + raise ConfigError(f"[{section}] {key} must be a string") + return value + + +def _optional_str(table: dict[str, object], key: str, section: str) -> str: + value = table[key] + if not isinstance(value, str): + raise ConfigError(f"[{section}] {key} must be a string") + return value + + +def _one_of(value: str, allowed: frozenset[str], section: str, key: str) -> str: + if value not in allowed: + choices = ", ".join(sorted(allowed)) + raise ConfigError(f"[{section}] {key} must be one of: {choices}") + return value + + +def _parse_github(raw: object, where: str) -> GithubSettings: + del where + table = _as_table(raw, "github") + _check_keys(table, {"username", "token"}, "github") + return GithubSettings( + username=_required_str(table, "username", "github"), + token=_optional_str(table, "token", "github") if "token" in table else "", + ) + + +def _parse_launchpad(raw: object, where: str) -> LaunchpadSettings: + del where + table = _as_table(raw, "launchpad") + _check_keys(table, {"username", "mode", "credentials_file", "service"}, "launchpad") + mode = _one_of( + _optional_str(table, "mode", "launchpad") if "mode" in table else "anonymous", + _LAUNCHPAD_MODES, + "launchpad", + "mode", + ) + service = _one_of( + _optional_str(table, "service", "launchpad") if "service" in table else "production", + _LAUNCHPAD_SERVICES, + "launchpad", + "service", + ) + return LaunchpadSettings( + username=_required_str(table, "username", "launchpad"), + mode=mode, + credentials_file=( + _optional_str(table, "credentials_file", "launchpad") + if "credentials_file" in table + else "" + ), + service=service, + ) + + +def _parse_gitlab(raw: object, where: str) -> GitlabSettings: + del where + table = _as_table(raw, "gitlab") + _check_keys(table, {"username", "url", "token", "token_env"}, "gitlab") + return GitlabSettings( + username=_required_str(table, "username", "gitlab"), + url=(_optional_str(table, "url", "gitlab") if "url" in table else "https://gitlab.com"), + token=_optional_str(table, "token", "gitlab") if "token" in table else "", + token_env=_optional_str(table, "token_env", "gitlab") if "token_env" in table else "", + ) + + +def _parse_bts(raw: object, where: str) -> BtsSettings: + del where + table = _as_table(raw, "bts") + _check_keys(table, {"email"}, "bts") + return BtsSettings(email=_required_str(table, "email", "bts")) + + +def _as_table(raw: object, section: str) -> dict[str, object]: + if not isinstance(raw, dict): + raise ConfigError(f"[{section}] must be a TOML table") + return raw + + +# --- Saving (hand-formatted: the schema is flat and string-only) --- + + +def save_config(config: ActivityConfig, path: Path) -> None: + """Serialize ``config`` to TOML at ``path`` with owner-only permissions.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(format_toml(config)) + os.chmod(path, 0o600) # enforce even when overwriting a pre-existing file + + +def format_toml(config: ActivityConfig) -> str: + lines = [ + "# weekly-activity configuration.", + "# Generated by `weekly-activity config`; edit by hand if you prefer.", + "# Tokens are stored in plaintext and readable only by your user (0600).", + "# Run `weekly-activity report` to aggregate every enabled source below.", + ] + if config.launchpad is not None: + lp = config.launchpad + lines += ["", "[launchpad]", f"username = {_toml(lp.username)}"] + if lp.mode != "anonymous": + lines.append(f"mode = {_toml(lp.mode)}") + if lp.credentials_file: + lines.append(f"credentials_file = {_toml(lp.credentials_file)}") + if lp.service != "production": + lines.append(f"service = {_toml(lp.service)}") + if config.github is not None: + gh = config.github + lines += ["", "[github]", f"username = {_toml(gh.username)}"] + if gh.token: + lines.append(f"token = {_toml(gh.token)}") + if config.gitlab is not None: + gl = config.gitlab + lines += ["", "[gitlab]", f"username = {_toml(gl.username)}"] + if gl.url != "https://gitlab.com": + lines.append(f"url = {_toml(gl.url)}") + if gl.token: + lines.append(f"token = {_toml(gl.token)}") + if gl.token_env: + lines.append(f"token_env = {_toml(gl.token_env)}") + if config.bts is not None: + lines += ["", "[bts]", f"email = {_toml(config.bts.email)}"] + return "\n".join(lines) + "\n" + + +_TOML_ESCAPES = { + "\\": "\\\\", + '"': '\\"', + "\b": "\\b", + "\t": "\\t", + "\n": "\\n", + "\f": "\\f", + "\r": "\\r", +} + + +def _toml(value: str) -> str: + """Render a TOML basic string, escaping quotes and control characters.""" + chars: list[str] = [] + for ch in value: + if ch in _TOML_ESCAPES: + chars.append(_TOML_ESCAPES[ch]) + elif ord(ch) < 0x20 or ord(ch) == 0x7F: + chars.append(f"\\u{ord(ch):04X}") + else: + chars.append(ch) + return '"' + "".join(chars) + '"' diff --git a/src/weekly_activity/wizard.py b/src/weekly_activity/wizard.py new file mode 100644 index 0000000..2ac353f --- /dev/null +++ b/src/weekly_activity/wizard.py @@ -0,0 +1,186 @@ +"""Interactive onboarding wizard behind `weekly-activity config`. + +Asks which sources to enable and the identity/auth details for each, then +writes the TOML config. Never echoes token values back to the console; does +not run any OAuth flow — for Launchpad credentials mode it only records the +choice (launchpadlib's keyring flow runs at report time). +""" + +from __future__ import annotations + +import getpass +import sys +from pathlib import Path + +from weekly_activity.config import ( + ActivityConfig, + BtsSettings, + ConfigError, + GithubSettings, + GitlabSettings, + LaunchpadSettings, + load_config, + save_config, +) + + +def run_wizard(config_path: Path) -> None: + """Walk through source setup and write ``config_path``.""" + existing = _load_existing(config_path) + if existing.enabled_sources(): + print(f"Found an existing configuration at {config_path}:") + print(_describe(existing, indent=" ")) + if not _confirm("Re-run setup?"): + print("Keeping the current configuration.") + return + + print() + print("Welcome! This wizard sets up the sources aggregated by") + print("`weekly-activity report`. Answer with Enter to accept [defaults],") + print("or Ctrl-C to abort without writing anything.") + + config = ActivityConfig( + launchpad=_configure_launchpad(existing.launchpad), + github=_configure_github(existing.github), + gitlab=_configure_gitlab(existing.gitlab), + bts=_configure_bts(existing.bts), + ) + + enabled = config.enabled_sources() + if not enabled: + raise SystemExit("warning: no source selected; nothing was written.") + + save_config(config, config_path) + print() + print(f"Configuration written to {config_path}.") + print(f"Enabled sources: {', '.join(enabled)}.") + print(_describe(config, indent="")) + print("Tokens are never printed back. Run `weekly-activity report` to") + print("aggregate activity across all of them in one output.") + + +def _load_existing(path: Path) -> ActivityConfig: + try: + return load_config(path) + except FileNotFoundError: + return ActivityConfig() + except ConfigError as exc: + print(f"warning: ignoring unusable existing config: {exc}", file=sys.stderr) + print(f"warning: {path} will be replaced by a fresh configuration.", file=sys.stderr) + return ActivityConfig() + + +# --- Per-source prompts --- + + +def _configure_launchpad(prev: LaunchpadSettings | None) -> LaunchpadSettings | None: + if not _confirm("Add Launchpad?", default=False): + return None + print("Launchpad identity is the ~name shown on your profile page.") + username = _ask_required("Launchpad username", prev.username if prev else "") + anonymous = _confirm( + "Anonymous mode (public data only, no OAuth)?", + default=prev.mode == "anonymous" if prev else True, + ) + settings = LaunchpadSettings( + username=username, mode="anonymous" if anonymous else "credentials" + ) + if not anonymous: + print( + "OAuth is obtained at report time: launchpadlib opens a browser and\n" + "stores the token in your system keyring, or reads it from a file." + ) + settings.credentials_file = _ask( + "Credentials file (empty to use the system keyring)", + prev.credentials_file if prev else "", + ) + return settings + + +def _configure_github(prev: GithubSettings | None) -> GithubSettings | None: + if not _confirm("Add GitHub?", default=False): + return None + username = _ask_required("GitHub username", prev.username if prev else "") + print("Leave the token empty to fall back to $GITHUB_TOKEN or `gh auth token`.") + token = _ask_secret("GitHub token (input hidden)", prev.token if prev else "") + return GithubSettings(username=username, token=token) + + +def _configure_gitlab(prev: GitlabSettings | None) -> GitlabSettings | None: + if not _confirm("Add GitLab?", default=False): + return None + defaults = prev or GitlabSettings(username="") + username = _ask_required("GitLab username", defaults.username) + url = _ask("Instance URL", defaults.url or "https://gitlab.com") + print("Leave the token empty to discover it from the environment instead.") + token = _ask_secret("API token (input hidden)", defaults.token) + token_env = _ask("Env var holding the token (optional, e.g. SALSA_TOKEN)", defaults.token_env) + return GitlabSettings(username=username, url=url, token=token, token_env=token_env) + + +def _configure_bts(prev: BtsSettings | None) -> BtsSettings | None: + if not _confirm("Add Debian BTS?", default=False): + return None + print("The BTS identifies people by the email used on bugs (submitter/owner).") + email = _ask_required("Email address on Debian bugs", prev.email if prev else "") + return BtsSettings(email=email) + + +# --- Prompt helpers --- + + +def _ask(prompt: str, default: str = "") -> str: + suffix = f" [{default}]" if default else "" + answer = input(f"{prompt}{suffix}: ").strip() + return answer or default + + +def _ask_required(prompt: str, default: str = "") -> str: + while True: + answer = _ask(prompt, default) + if answer: + return answer + print("A value is required here.") + + +def _ask_secret(prompt: str, default: str = "") -> str: + """Read a secret without echo; keep the stored one when input is empty.""" + try: + entered = getpass.getpass(f"{prompt}: ") + except (EOFError, KeyboardInterrupt): + raise SystemExit(130) from None + return entered.strip() or default + + +def _confirm(prompt: str, *, default: bool = False) -> bool: + hint = "[Y/n]" if default else "[y/N]" + while True: + answer = input(f"{prompt} {hint}: ").strip().lower() + if not answer: + return default + if answer in {"y", "yes"}: + return True + if answer in {"n", "no"}: + return False + + +def _describe(config: ActivityConfig, indent: str = " ") -> str: + """Human-readable summary that masks every token value.""" + lines: list[str] = [] + if (lp := config.launchpad) is not None: + auth = ( + "anonymous (public data only)" + if lp.mode == "anonymous" + else f"OAuth via {lp.credentials_file or 'system keyring'}" + ) + lines.append(f"{indent}launchpad: {lp.username} — {auth}") + if (gh := config.github) is not None: + token_desc = "stored" if gh.token else "not stored (uses GITHUB_TOKEN / gh)" + lines.append(f"{indent}github: {gh.username} — token {token_desc}") + if (gl := config.gitlab) is not None: + fallback = gl.token_env or "uses _TOKEN / GITLAB_TOKEN env" + token_desc = "stored" if gl.token else f"not stored ({fallback})" + lines.append(f"{indent}gitlab: {gl.username} @ {gl.url} — token {token_desc}") + if (bts := config.bts) is not None: + lines.append(f"{indent}bts: {bts.email}") + return "\n".join(lines) if lines else f"{indent}(no sources configured)" diff --git a/tests/test_aggregate.py b/tests/test_aggregate.py new file mode 100644 index 0000000..509f882 --- /dev/null +++ b/tests/test_aggregate.py @@ -0,0 +1,101 @@ +"""Unit tests for weekly_activity.aggregate — stubbed sources, no network.""" + +from __future__ import annotations + +import os +import unittest +from datetime import datetime +from unittest import mock + +from weekly_activity.aggregate import ( + SourceSpec, + build_specs, + collect_specs, + format_combined, +) +from weekly_activity.config import ( + ActivityConfig, + BtsSettings, + GithubSettings, + GitlabSettings, + LaunchpadSettings, +) +from weekly_activity.model import ActivityReport, Section + + +class StubSource: + def __init__(self, name: str = "stub", *, fail: bool = False) -> None: + self.name = name + self._fail = fail + + def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport: + if self._fail: + raise RuntimeError("kaput") + return ActivityReport( + source=self.name, + username=username, + since=since, + until=until, + sections=[Section("Things", ["did a thing"])], + warnings=["some query failed"], + ) + + +def window() -> tuple[datetime, datetime]: + return datetime(2026, 8, 19), datetime(2026, 8, 26) + + +class BuildSpecsTest(unittest.TestCase): + def test_maps_every_section_in_order(self) -> None: + config = ActivityConfig( + launchpad=LaunchpadSettings(username="lp"), + github=GithubSettings(username="gh"), + gitlab=GitlabSettings(username="gl"), + bts=BtsSettings(email="bts@mail"), + ) + specs = build_specs(config) + self.assertEqual([s.name for s in specs], ["launchpad", "github", "gitlab", "bts"]) + self.assertEqual([s.username for s in specs], ["lp", "gh", "gl", "bts@mail"]) + + def test_disabled_sections_are_skipped(self) -> None: + config = ActivityConfig(github=GithubSettings(username="only-one")) + self.assertEqual([s.name for s in build_specs(config)], ["github"]) + + def test_launchpad_spec_expands_tilde_credentials_file(self) -> None: + settings = LaunchpadSettings( + username="u", mode="credentials", credentials_file="~/creds.json" + ) + spec = build_specs(ActivityConfig(launchpad=settings))[0] + with mock.patch("weekly_activity.aggregate.LaunchpadSource") as fake_source: + spec.make() + kwargs = fake_source.call_args.kwargs + self.assertEqual(kwargs["credentials_file"], os.path.expanduser("~") + "/creds.json") + self.assertFalse(kwargs["anonymous"]) + + +class CollectAndFormatTest(unittest.TestCase): + def test_failure_isolated_per_source(self) -> None: + since, until = window() + specs = [ + SourceSpec(name="alpha", username="a", make=lambda: StubSource("alpha")), + SourceSpec(name="beta", username="b", make=lambda: StubSource("beta", fail=True)), + ] + collected = collect_specs(specs, since, until) + self.assertIsNotNone(collected[0].report) + self.assertIn("RuntimeError: kaput", collected[1].error or "") + + text = format_combined(collected, since, until) + self.assertIn("Weekly activity report — 2 source(s)", text) + self.assertIn("Activity report — alpha / ~a", text) + self.assertIn("[skipped: beta — RuntimeError: kaput]", text) + + def test_warning_surfaced_inside_block(self) -> None: + since, until = window() + specs = [SourceSpec(name="gamma", username="g", make=lambda: StubSource("gamma"))] + text = format_combined(collect_specs(specs, since, until), since, until) + self.assertIn("Warnings (partial data — some queries failed):", text) + self.assertIn("some query failed", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..6de67df --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,122 @@ +"""Unit tests for weekly_activity.config — pure logic, no network.""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path + +from weekly_activity.config import ( + ActivityConfig, + BtsSettings, + ConfigError, + GithubSettings, + GitlabSettings, + LaunchpadSettings, + default_config_path, + format_toml, + load_config, + save_config, +) + + +class ConfigRoundTripTest(unittest.TestCase): + def test_round_trip_preserves_all_sections(self) -> None: + config = ActivityConfig( + launchpad=LaunchpadSettings( + username="lp-user", + mode="credentials", + credentials_file="/tmp/lp-creds.json", + service="staging", + ), + github=GithubSettings(username='gh"user', token="tok\\en"), + gitlab=GitlabSettings( + username="gl-user", + url="https://salsa.debian.org", + token_env="SALSA_TOKEN", + ), + bts=BtsSettings(email="dev@example.org"), + ) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "weekly-activity.toml" + save_config(config, path) + self.assertEqual(load_config(path), config) + + def test_save_creates_owner_only_file(self) -> None: + config = ActivityConfig(bts=BtsSettings(email="x@example.org")) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "nested" / "cfg.toml" + save_config(config, path) + self.assertEqual(os.stat(path).st_mode & 0o777, 0o600) + + def test_empty_optionals_round_trip_to_defaults(self) -> None: + config = ActivityConfig( + github=GithubSettings(username="octocat"), + bts=BtsSettings(email="a@b.c"), + ) + text = format_toml(config) + reparsed = load_config_from_text(text) + github = reparsed.github + assert github is not None + self.assertEqual(github.username, "octocat") + self.assertEqual(github.token, "") + self.assertIsNone(reparsed.launchpad) + + def test_escapes_quotes_backslashes_and_newlines(self) -> None: + config = ActivityConfig(github=GithubSettings(username='a"b\\c\nd')) + reparsed = load_config_from_text(format_toml(config)) + github = reparsed.github + assert github is not None + self.assertEqual(github.username, 'a"b\\c\nd') + + +class ConfigLoadingTest(unittest.TestCase): + def test_missing_file_raises_filenotfound(self) -> None: + with self.assertRaises(FileNotFoundError): + load_config(Path("/nonexistent/weekly-activity.toml")) + + def test_unknown_section_rejected(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text("[gitea]\nusername = 'x'\n") + + def test_unknown_key_rejected(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text('[github]\nusername = "x"\ntokken = "t"\n') + + def test_non_string_value_rejected(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text("[github]\nusername = 42\n") + + def test_bad_launchpad_mode_rejected(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text('[launchpad]\nusername = "u"\nmode = "oauth2"\n') + + def test_enabled_sources_stable_order(self) -> None: + config = load_config_from_text( + '[bts]\nemail = "e"\n\n[github]\nusername = "g"\n' + ) + self.assertEqual(config.enabled_sources(), ["github", "bts"]) + + def test_default_path_uses_xdg_or_home(self) -> None: + old = os.environ.get("XDG_CONFIG_HOME") + try: + os.environ["XDG_CONFIG_HOME"] = "/custom/cfg" + self.assertEqual(default_config_path(), Path("/custom/cfg/weekly-activity.toml")) + del os.environ["XDG_CONFIG_HOME"] + self.assertTrue(str(default_config_path()).endswith("/.config/weekly-activity.toml")) + finally: + if old is not None: + os.environ["XDG_CONFIG_HOME"] = old + + +def load_config_from_text(text: str) -> ActivityConfig: + """Helper: parse TOML through the public loader without touching disk.""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "weekly-activity.toml" + path.write_text(text, encoding="utf-8") + return load_config(path) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0 From 5f4487f3ade49ff0abecb119cad9ed5400ca73a5 Mon Sep 17 00:00:00 2001 From: Kosmos Date: Wed, 26 Aug 2026 22:53:17 +0000 Subject: [PATCH 2/5] feat(config): menu-driven multi-account wizard, v2 schema, config-time Launchpad OAuth Address review of PR #1 (Valentin): replace the linear config walk-through and single-section schema. - weekly-activity config becomes a menu loop: Current sources (masked) -> Add a new source (pick kind, repeatable) / Remove a source / Exit; the file is written only when something changed - new versioned config layout (version = 2): [[kind.accounts]] arrays so several accounts or instances per provider coexist; optional display labels; strict unknown-key/type validation kept (errors carry account position, e.g. [github.accounts[1]]) - legacy v1 files (no version key, one section per source) migrate transparently in memory on load; saving from the wizard persists v2; the file itself is never rewritten by load_config - non-anonymous Launchpad accounts now authenticate at config time: constructing LaunchpadSource runs launchpadlib's browser OAuth and the token lands in the system keyring before any report; an explicit credentials_file stays supported as an additive opt-out - report aggregates every configured account in stable order with failure isolation intact; headings add the account label when it disambiguates (source_label shared between aggregate and wizard) - GitHub accounts gain optional token_env resolved lazily against the environment before falling back to GITHUB_TOKEN / `gh auth token` --- README.md | 49 +++-- src/weekly_activity/aggregate.py | 120 ++++++++--- src/weekly_activity/config.py | 257 ++++++++++++++++------- src/weekly_activity/wizard.py | 338 ++++++++++++++++++++++--------- tests/test_aggregate.py | 94 +++++++-- tests/test_config.py | 204 +++++++++++++++---- 6 files changed, 801 insertions(+), 261 deletions(-) diff --git a/README.md b/README.md index 1b812ee..6951bda 100644 --- a/README.md +++ b/README.md @@ -67,35 +67,54 @@ Rolling 7 days ending today (inclusive). Override with `--since` / `--until` (IS ## Configuration -`weekly-activity config` walks through each known source and writes a TOML -file. Tokens are stored as provided (plaintext), but the file is created with -owner-only permissions (`0600`) and secrets are never echoed back. Use -`--config ` (also accepted by `report`) to keep configs elsewhere. +`weekly-activity config` opens a small menu-driven editor: it lists your +current sources (secrets masked), then lets you **Add a new source** +(choose the kind, repeat as often as you like), **Remove a source**, or +**Exit** (writes the file only when something changed). Several accounts +of the same provider can live side by side. Tokens are stored as provided +(plaintext), but the file is created with owner-only permissions (`0600`) +and secrets are never echoed back. Use `--config ` (also accepted by +`report`) to keep configs elsewhere. Default location: `${XDG_CONFIG_HOME:-~/.config}/weekly-activity.toml`. ```toml -[github] -username = "octocat" -token = "" # empty -> fall back to $GITHUB_TOKEN / `gh auth token` +version = 2 -[launchpad] +[[launchpad.accounts]] username = "jane" -mode = "credentials" # "anonymous" (public data only) | "credentials" -credentials_file = "" # empty -> system keyring; OAuth runs at report time +mode = "credentials" # "anonymous" (public data only) | "credentials" +# credentials_file = "~/lp-creds.json" # omit -> system keyring -[gitlab] +[[github.accounts]] +name = "personal" # optional display label +username = "octocat" +token = "" # empty -> $token_env / GITHUB_TOKEN / `gh auth token` + +[[github.accounts]] +name = "work" +username = "acme-jane" +token_env = "WORK_GH_TOKEN" + +[[gitlab.accounts]] +name = "salsa" url = "https://salsa.debian.org" username = "jane" token_env = "SALSA_TOKEN" -[bts] +[[bts.accounts]] email = "jane@example.org" ``` -An absent `[section]` disables that source. Re-running `weekly-activity -config` shows the current settings and offers to redo the wizard with the -existing values as defaults. +An absent section disables that source; an empty one is not written. +When a non-anonymous Launchpad account is added without a credentials +file, the wizard runs launchpadlib's OAuth right there — a browser opens +once and the token is kept in your system keyring, so `report` never has +to authenticate. + +Legacy configs without the `version` key (one single-account section per +source) keep working: they are migrated transparently when loaded and +rewritten in the v2 shape the next time the wizard saves. ## Architecture diff --git a/src/weekly_activity/aggregate.py b/src/weekly_activity/aggregate.py index 1333281..0886344 100644 --- a/src/weekly_activity/aggregate.py +++ b/src/weekly_activity/aggregate.py @@ -1,16 +1,16 @@ -"""Combined report — build every configured source and render one text output. +"""Combined report — build every configured account and render one text output. -Each enabled source runs independently: a failure (bad token, unreachable -API, unknown user) is recorded as a ``[skipped: …]`` block instead -of aborting the whole report. Per-source warnings surfaced by ``collect`` -are rendered inside the source's block by ``format_report``. +Each configured account runs independently: a failure (bad token, +unreachable API, unknown user) is recorded as a ``[skipped: …]`` +block instead of aborting the whole report. Per-source warnings surfaced by +``collect`` are rendered inside the source's block by ``format_report``. """ from __future__ import annotations import os from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timedelta from weekly_activity.config import ActivityConfig @@ -27,7 +27,12 @@ _RULE = "─" * 60 @dataclass(frozen=True, slots=True) class SourceSpec: - """One enabled source: display name, identity to query, deferred constructor.""" + """One configured account: display name, identity to query, deferred constructor. + + ``name`` carries the report-heading label: the source kind plus the + account's ``name`` (or identity) whenever more than one account of that + kind exists and disambiguation helps. + """ name: str username: str @@ -41,49 +46,101 @@ class Collected: error: str | None = None +def source_label(kind: str, name: str, identity: str, ambiguous: bool) -> str: + """Heading label for an account: the source kind, disambiguated when needed. + + An explicit ``name`` label always shows; otherwise sibling accounts of + the same kind force the identity (username/email) into the label. Shared + with the wizard so menus and report headings never drift apart. + """ + if name: + return f"{kind}/{name}" + if ambiguous and identity: + return f"{kind}/{identity}" + return kind + + def build_specs(config: ActivityConfig) -> list[SourceSpec]: - """Map configured sections onto constructible sources (network deferred).""" + """Map every configured account onto a constructible source (network deferred). + + Accounts appear in listing order within each source kind, kinds in + stable order: launchpad, github, gitlab, bts. + """ + ambiguous = { + "launchpad": len(config.launchpad) > 1, + "github": len(config.github) > 1, + "gitlab": len(config.gitlab) > 1, + "bts": len(config.bts) > 1, + } specs: list[SourceSpec] = [] - if (lp := config.launchpad) is not None: - anonymous = lp.mode == "anonymous" - credentials_file = os.path.expanduser(lp.credentials_file) + for lp in config.launchpad: specs.append( SourceSpec( - name="launchpad", + name=source_label("launchpad", lp.name, lp.username, ambiguous["launchpad"]), username=lp.username, - make=lambda: LaunchpadSource( - service=lp.service, - anonymous=anonymous, - credentials_file=credentials_file or None, - ), + make=_make_launchpad(lp.service, lp.mode == "anonymous", lp.credentials_file), ) ) - if (gh := config.github) is not None: - token = gh.token + for gh in config.github: specs.append( SourceSpec( - name="github", + name=source_label("github", gh.name, gh.username, ambiguous["github"]), username=gh.username, - make=lambda: GitHubSource(token=token or None), + make=_make_github(gh.token, gh.token_env), ) ) - if (gl := config.gitlab) is not None: - url, token, token_env = gl.url, gl.token, gl.token_env + for gl in config.gitlab: specs.append( SourceSpec( - name="gitlab", + name=source_label("gitlab", gl.name, gl.username, ambiguous["gitlab"]), username=gl.username, - make=lambda: GitLabSource( - url=url, token=token or None, token_env=token_env or None - ), + make=_make_gitlab(gl.url, gl.token, gl.token_env), + ) + ) + for bts_acct in config.bts: + email = bts_acct.email + specs.append( + SourceSpec( + name=source_label("bts", bts_acct.name, email, ambiguous["bts"]), + username=email, + make=DebianBtsSource, ) ) - if (bts := config.bts) is not None: - email = bts.email - specs.append(SourceSpec(name="bts", username=email, make=DebianBtsSource)) return specs +def _make_launchpad( + service: str, anonymous: bool, credentials_file: str +) -> Callable[[], ActivitySource]: + expanded = os.path.expanduser(credentials_file) + + def make() -> ActivitySource: + return LaunchpadSource( + service=service, + anonymous=anonymous, + credentials_file=expanded or None, + ) + + return make + + +def _make_github(token: str, token_env: str) -> Callable[[], ActivitySource]: + def make() -> ActivitySource: + # No stored token: explicit env var if named, else GitHubSource's own + # GITHUB_TOKEN / `gh auth token` discovery. + resolved = token or (os.environ.get(token_env, "") if token_env else "") + return GitHubSource(token=resolved or None) + + return make + + +def _make_gitlab(url: str, token: str, token_env: str) -> Callable[[], ActivitySource]: + def make() -> ActivitySource: + return GitLabSource(url=url, token=token or None, token_env=token_env or None) + + return make + + def collect_specs(specs: list[SourceSpec], since: datetime, until: datetime) -> list[Collected]: """Run every spec, converting any per-source failure into an error record.""" collected: list[Collected] = [] @@ -94,7 +151,8 @@ def collect_specs(specs: list[SourceSpec], since: datetime, until: datetime) -> except Exception as exc: # noqa: BLE001 - one bad source must not kill the rest collected.append(Collected(spec, error=f"{type(exc).__name__}: {exc}")) else: - collected.append(Collected(spec, report=report)) + # Headings must show the account label, not just the source kind. + collected.append(Collected(spec, report=replace(report, source=spec.name))) return collected diff --git a/src/weekly_activity/config.py b/src/weekly_activity/config.py index ce32c3f..e78d890 100644 --- a/src/weekly_activity/config.py +++ b/src/weekly_activity/config.py @@ -1,8 +1,17 @@ -"""Config file: schema, discovery, loading, and persistence. +"""Config file: schema, discovery, loading, migration, and persistence. Stored as TOML at ``${XDG_CONFIG_HOME:-~/.config}/weekly-activity.toml`` -(override with ``--config``). One section per source; an absent section -means the source is not enabled. +(override with ``--config``). + +Version 2 layout (``version = 2``): one section per source kind, each +holding an array of accounts (``[[github.accounts]]``, …) so several +accounts or instances of the same kind can be configured side by side. +An absent section means that source is not enabled. + +Legacy v1 files (no ``version`` key, one single-account section per +source) are transparently migrated to the account-list shape when loaded, +in memory only — the file on disk is never rewritten by ``load_config``; +the next save from the wizard persists the migrated v2 shape. Tokens are stored as provided (plaintext) in the user's own local config; the file is created with ``0600`` permissions so they are not world-readable. @@ -12,14 +21,20 @@ from __future__ import annotations import os import tomllib -from dataclasses import dataclass +from collections.abc import Callable +from dataclasses import dataclass, field from pathlib import Path +from typing import TypeVar CONFIG_FILE_NAME = "weekly-activity.toml" +CONFIG_VERSION = 2 +_SOURCE_KEYS = ("launchpad", "github", "gitlab", "bts") _LAUNCHPAD_MODES = frozenset({"anonymous", "credentials"}) _LAUNCHPAD_SERVICES = frozenset({"production", "staging"}) +_T = TypeVar("_T") + class ConfigError(Exception): """The config file exists but is malformed or has unknown entries.""" @@ -27,49 +42,67 @@ class ConfigError(Exception): @dataclass(slots=True) class GithubSettings: + """One GitHub account.""" + username: str - token: str = "" # empty -> fall back to GITHUB_TOKEN / `gh auth token` + token: str = "" # empty -> $token_env / GITHUB_TOKEN / `gh auth token` + token_env: str = "" # optional named var, takes precedence in discovery + name: str = "" # optional display label @dataclass(slots=True) class LaunchpadSettings: + """One Launchpad account. + + In credentials mode an empty ``credentials_file`` means launchpadlib + uses its system-keyring credential store; config-time OAuth in the + wizard populates that keyring so ``report`` never needs to authenticate. + """ + username: str mode: str = "anonymous" # "anonymous" | "credentials" credentials_file: str = "" # used in "credentials" mode; empty -> keyring service: str = "production" + name: str = "" @dataclass(slots=True) class GitlabSettings: + """One GitLab instance/account.""" + username: str url: str = "https://gitlab.com" token: str = "" # empty -> env discovery (_TOKEN, GITLAB_TOKEN) token_env: str = "" # optional named var, takes precedence in discovery + name: str = "" @dataclass(slots=True) class BtsSettings: + """One Debian BTS identity (an email address).""" + email: str + name: str = "" @dataclass(slots=True) class ActivityConfig: - """All configured sources; ``None`` section = source disabled.""" + """All configured accounts; an empty list = source disabled.""" - launchpad: LaunchpadSettings | None = None - github: GithubSettings | None = None - gitlab: GitlabSettings | None = None - bts: BtsSettings | None = None + launchpad: list[LaunchpadSettings] = field(default_factory=list) + github: list[GithubSettings] = field(default_factory=list) + gitlab: list[GitlabSettings] = field(default_factory=list) + bts: list[BtsSettings] = field(default_factory=list) def enabled_sources(self) -> list[str]: """Names of configured source sections, in stable report order.""" - ordered: list[tuple[str, bool]] = [ - ("launchpad", self.launchpad is not None), - ("github", self.github is not None), - ("gitlab", self.gitlab is not None), - ("bts", self.bts is not None), - ] - return [name for name, on in ordered if on] + ordered = ( + ("launchpad", self.launchpad), + ("github", self.github), + ("gitlab", self.gitlab), + ("bts", self.bts), + ) + return [name for name, accounts in ordered if accounts] def default_config_path() -> Path: @@ -84,6 +117,11 @@ def default_config_path() -> Path: def load_config(path: Path) -> ActivityConfig: """Parse the TOML config at ``path``. + Files written at ``version = 2`` carry arrays of accounts per source. + Legacy files without a ``version`` key are migrated to that shape in + memory (one unnamed account per configured section); nothing is + rewritten on disk here — saving from the wizard persists v2. + Raises ``FileNotFoundError`` when the file does not exist and ``ConfigError`` when it cannot be parsed or contains unknown entries. """ @@ -100,20 +138,75 @@ def load_config(path: Path) -> ActivityConfig: where = str(path) _reject_unknown_top_level(data, where) + + if "version" in data: + _check_version(data["version"], where) + return ActivityConfig( + launchpad=_account_list(data.get("launchpad"), "launchpad", _parse_launchpad), + github=_account_list(data.get("github"), "github", _parse_github), + gitlab=_account_list(data.get("gitlab"), "gitlab", _parse_gitlab), + bts=_account_list(data.get("bts"), "bts", _parse_bts), + ) + + if _looks_like_v2_without_version(data): + raise ConfigError( + f"{where}: account-array config without a version header; " + f'add "version = {CONFIG_VERSION}" at the top' + ) + # Legacy v1 layout: migrate one single-table section to one unnamed account. return ActivityConfig( - launchpad=_parse_launchpad(data.get("launchpad"), where) if "launchpad" in data else None, - github=_parse_github(data.get("github"), where) if "github" in data else None, - gitlab=_parse_gitlab(data.get("gitlab"), where) if "gitlab" in data else None, - bts=_parse_bts(data.get("bts"), where) if "bts" in data else None, + launchpad=_legacy_accounts(data.get("launchpad"), "launchpad", _parse_launchpad), + github=_legacy_accounts(data.get("github"), "github", _parse_github), + gitlab=_legacy_accounts(data.get("gitlab"), "gitlab", _parse_gitlab), + bts=_legacy_accounts(data.get("bts"), "bts", _parse_bts), ) +def _check_version(version: object, where: str) -> None: + if isinstance(version, bool) or not isinstance(version, int): + raise ConfigError(f"{where}: version must be an integer") + if version != CONFIG_VERSION: + raise ConfigError( + f"{where}: unsupported config version {version} " + f"(this release understands version {CONFIG_VERSION})" + ) + + def _reject_unknown_top_level(data: dict[str, object], where: str) -> None: - unknown = sorted(set(data) - {"launchpad", "github", "gitlab", "bts"}) + unknown = sorted(set(data) - {"version", *_SOURCE_KEYS}) if unknown: raise ConfigError(f"{where}: unknown config section(s): {', '.join(unknown)}") +def _looks_like_v2_without_version(data: dict[str, object]) -> bool: + for key in _SOURCE_KEYS: + section = data.get(key) + if isinstance(section, dict) and "accounts" in section: + return True + return False + + +def _account_list(raw: object, section: str, parse: Callable[[object, str], _T]) -> list[_T]: + """Parse ``[[
.accounts]]`` tables via the given account parser.""" + if raw is None: + return [] + table = _as_table(raw, section) + _check_keys(table, {"accounts"}, section) + accounts = table["accounts"] + if not isinstance(accounts, list): + raise ConfigError( + f"[{section}] accounts must be an array of tables ([[{section}.accounts]])" + ) + return [parse(item, f"{section}.accounts[{index}]") for index, item in enumerate(accounts)] + + +def _legacy_accounts(raw: object, section: str, parse: Callable[[object, str], _T]) -> list[_T]: + """Migrate one v1 single-table section into a single-account list.""" + if raw is None: + return [] + return [parse(raw, section)] + + def _check_keys(table: dict[str, object], names: set[str], section: str) -> None: unknown = sorted(set(table) - names) if unknown: @@ -134,6 +227,10 @@ def _optional_str(table: dict[str, object], key: str, section: str) -> str: return value +def _optional_default_str(table: dict[str, object], key: str, section: str) -> str: + return _optional_str(table, key, section) if key in table else "" + + def _one_of(value: str, allowed: frozenset[str], section: str, key: str) -> str: if value not in allowed: choices = ", ".join(sorted(allowed)) @@ -141,61 +238,60 @@ def _one_of(value: str, allowed: frozenset[str], section: str, key: str) -> str: return value -def _parse_github(raw: object, where: str) -> GithubSettings: - del where - table = _as_table(raw, "github") - _check_keys(table, {"username", "token"}, "github") +def _parse_github(raw: object, section: str) -> GithubSettings: + table = _as_table(raw, section) + _check_keys(table, {"name", "username", "token", "token_env"}, section) return GithubSettings( - username=_required_str(table, "username", "github"), - token=_optional_str(table, "token", "github") if "token" in table else "", + username=_required_str(table, "username", section), + token=_optional_default_str(table, "token", section), + token_env=_optional_default_str(table, "token_env", section), + name=_optional_default_str(table, "name", section), ) -def _parse_launchpad(raw: object, where: str) -> LaunchpadSettings: - del where - table = _as_table(raw, "launchpad") - _check_keys(table, {"username", "mode", "credentials_file", "service"}, "launchpad") +def _parse_launchpad(raw: object, section: str) -> LaunchpadSettings: + table = _as_table(raw, section) + _check_keys(table, {"name", "username", "mode", "credentials_file", "service"}, section) mode = _one_of( - _optional_str(table, "mode", "launchpad") if "mode" in table else "anonymous", + _optional_default_str(table, "mode", section) or "anonymous", _LAUNCHPAD_MODES, - "launchpad", + section, "mode", ) service = _one_of( - _optional_str(table, "service", "launchpad") if "service" in table else "production", + _optional_default_str(table, "service", section) or "production", _LAUNCHPAD_SERVICES, - "launchpad", + section, "service", ) return LaunchpadSettings( - username=_required_str(table, "username", "launchpad"), + username=_required_str(table, "username", section), mode=mode, - credentials_file=( - _optional_str(table, "credentials_file", "launchpad") - if "credentials_file" in table - else "" - ), + credentials_file=_optional_default_str(table, "credentials_file", section), service=service, + name=_optional_default_str(table, "name", section), ) -def _parse_gitlab(raw: object, where: str) -> GitlabSettings: - del where - table = _as_table(raw, "gitlab") - _check_keys(table, {"username", "url", "token", "token_env"}, "gitlab") +def _parse_gitlab(raw: object, section: str) -> GitlabSettings: + table = _as_table(raw, section) + _check_keys(table, {"name", "username", "url", "token", "token_env"}, section) return GitlabSettings( - username=_required_str(table, "username", "gitlab"), - url=(_optional_str(table, "url", "gitlab") if "url" in table else "https://gitlab.com"), - token=_optional_str(table, "token", "gitlab") if "token" in table else "", - token_env=_optional_str(table, "token_env", "gitlab") if "token_env" in table else "", + username=_required_str(table, "username", section), + url=_optional_default_str(table, "url", section) or "https://gitlab.com", + token=_optional_default_str(table, "token", section), + token_env=_optional_default_str(table, "token_env", section), + name=_optional_default_str(table, "name", section), ) -def _parse_bts(raw: object, where: str) -> BtsSettings: - del where - table = _as_table(raw, "bts") - _check_keys(table, {"email"}, "bts") - return BtsSettings(email=_required_str(table, "email", "bts")) +def _parse_bts(raw: object, section: str) -> BtsSettings: + table = _as_table(raw, section) + _check_keys(table, {"name", "email"}, section) + return BtsSettings( + email=_required_str(table, "email", section), + name=_optional_default_str(table, "name", section), + ) def _as_table(raw: object, section: str) -> dict[str, object]: @@ -222,32 +318,47 @@ def format_toml(config: ActivityConfig) -> str: "# Generated by `weekly-activity config`; edit by hand if you prefer.", "# Tokens are stored in plaintext and readable only by your user (0600).", "# Run `weekly-activity report` to aggregate every enabled source below.", + "", + f"version = {CONFIG_VERSION}", ] - if config.launchpad is not None: - lp = config.launchpad - lines += ["", "[launchpad]", f"username = {_toml(lp.username)}"] + + def emit(section: str, entry_lines: list[str]) -> None: + lines.append("") + lines.append(f"[[{section}.accounts]]") + lines.extend(entry_lines) + + for lp in config.launchpad: + entry = [f"name = {_toml(lp.name)}"] if lp.name else [] + entry.append(f"username = {_toml(lp.username)}") if lp.mode != "anonymous": - lines.append(f"mode = {_toml(lp.mode)}") + entry.append(f"mode = {_toml(lp.mode)}") if lp.credentials_file: - lines.append(f"credentials_file = {_toml(lp.credentials_file)}") + entry.append(f"credentials_file = {_toml(lp.credentials_file)}") if lp.service != "production": - lines.append(f"service = {_toml(lp.service)}") - if config.github is not None: - gh = config.github - lines += ["", "[github]", f"username = {_toml(gh.username)}"] + entry.append(f"service = {_toml(lp.service)}") + emit("launchpad", entry) + for gh in config.github: + entry = [f"name = {_toml(gh.name)}"] if gh.name else [] + entry.append(f"username = {_toml(gh.username)}") + if gh.token_env: + entry.append(f"token_env = {_toml(gh.token_env)}") if gh.token: - lines.append(f"token = {_toml(gh.token)}") - if config.gitlab is not None: - gl = config.gitlab - lines += ["", "[gitlab]", f"username = {_toml(gl.username)}"] + entry.append(f"token = {_toml(gh.token)}") + emit("github", entry) + for gl in config.gitlab: + entry = [f"name = {_toml(gl.name)}"] if gl.name else [] + entry.append(f"username = {_toml(gl.username)}") if gl.url != "https://gitlab.com": - lines.append(f"url = {_toml(gl.url)}") + entry.append(f"url = {_toml(gl.url)}") if gl.token: - lines.append(f"token = {_toml(gl.token)}") + entry.append(f"token = {_toml(gl.token)}") if gl.token_env: - lines.append(f"token_env = {_toml(gl.token_env)}") - if config.bts is not None: - lines += ["", "[bts]", f"email = {_toml(config.bts.email)}"] + entry.append(f"token_env = {_toml(gl.token_env)}") + emit("gitlab", entry) + for bts in config.bts: + entry = [f"name = {_toml(bts.name)}"] if bts.name else [] + entry.append(f"email = {_toml(bts.email)}") + emit("bts", entry) return "\n".join(lines) + "\n" diff --git a/src/weekly_activity/wizard.py b/src/weekly_activity/wizard.py index 2ac353f..6b97e72 100644 --- a/src/weekly_activity/wizard.py +++ b/src/weekly_activity/wizard.py @@ -1,17 +1,25 @@ """Interactive onboarding wizard behind `weekly-activity config`. -Asks which sources to enable and the identity/auth details for each, then -writes the TOML config. Never echoes token values back to the console; does -not run any OAuth flow — for Launchpad credentials mode it only records the -choice (launchpadlib's keyring flow runs at report time). +A small menu-driven editor for the account-list config: it shows the +current sources (secrets masked), then offers Add a new source / +Remove a source / Exit, supporting several accounts per provider. + +Adding a non-anonymous Launchpad account runs launchpadlib's interactive +OAuth flow right here (the browser opens once and the token lands in the +system keyring) so `weekly-activity report` never has to authenticate; +supplying a credentials file skips that flow. Token values are never +echoed back. """ from __future__ import annotations import getpass +import os import sys +from collections.abc import Callable, Sequence from pathlib import Path +from weekly_activity.aggregate import source_label from weekly_activity.config import ( ActivityConfig, BtsSettings, @@ -22,38 +30,76 @@ from weekly_activity.config import ( load_config, save_config, ) +from weekly_activity.sources.launchpad import LaunchpadSource + +_APPLICATION_NAME = "weekly-activity" + +# Stable processing order, shared by menus and summaries: (config attr, menu text). +_KINDS: tuple[tuple[str, str], ...] = ( + ("launchpad", "Launchpad account"), + ("github", "GitHub account"), + ("gitlab", "GitLab instance/account"), + ("bts", "Debian BTS email"), +) def run_wizard(config_path: Path) -> None: - """Walk through source setup and write ``config_path``.""" - existing = _load_existing(config_path) - if existing.enabled_sources(): - print(f"Found an existing configuration at {config_path}:") - print(_describe(existing, indent=" ")) - if not _confirm("Re-run setup?"): - print("Keeping the current configuration.") - return + """Edit the source configuration interactively; write when changed.""" + try: + config = _load_existing(config_path) + original = _snapshot(config) + print() + print("Welcome! This wizard manages the sources aggregated by") + print("`weekly-activity report`. Enter a menu number, accept [defaults]") + print("with Enter, and pick Exit to save; Ctrl-C aborts without writing.") + try: + _menu_loop(config) + except EOFError: + print() + print("End of input; applying current selections.") + except KeyboardInterrupt: + raise SystemExit(130) from None # parity with secret-prompt interruption + _finish(config, original, config_path) - print() - print("Welcome! This wizard sets up the sources aggregated by") - print("`weekly-activity report`. Answer with Enter to accept [defaults],") - print("or Ctrl-C to abort without writing anything.") - config = ActivityConfig( - launchpad=_configure_launchpad(existing.launchpad), - github=_configure_github(existing.github), - gitlab=_configure_gitlab(existing.gitlab), - bts=_configure_bts(existing.bts), +def _snapshot(config: ActivityConfig) -> ActivityConfig: + """Copy the lists so edits can be compared against the starting point.""" + return ActivityConfig( + launchpad=list(config.launchpad), + github=list(config.github), + gitlab=list(config.gitlab), + bts=list(config.bts), ) - enabled = config.enabled_sources() - if not enabled: - raise SystemExit("warning: no source selected; nothing was written.") +def _menu_loop(config: ActivityConfig) -> None: + while True: + print() + print("Current sources:") + print(_describe(config)) + action = _menu("What do you want to do?", ["Add a new source", "Remove a source", "Exit"]) + if action == 0: + _add_source_flow(config) + elif action == 1: + _remove_source_flow(config) + else: + if not config.enabled_sources() and not _confirm( + "No sources are configured; write an empty configuration anyway?", default=False + ): + continue + return + + +def _finish(config: ActivityConfig, original: ActivityConfig, config_path: Path) -> None: + if config == original: + print() + print("No changes made; configuration left untouched.") + return save_config(config, config_path) + enabled = ", ".join(config.enabled_sources()) print() print(f"Configuration written to {config_path}.") - print(f"Enabled sources: {', '.join(enabled)}.") + print(f"Enabled sources: {enabled}.") print(_describe(config, indent="")) print("Tokens are never printed back. Run `weekly-activity report` to") print("aggregate activity across all of them in one output.") @@ -70,65 +116,197 @@ def _load_existing(path: Path) -> ActivityConfig: return ActivityConfig() -# --- Per-source prompts --- +# --- Add / remove flows --- -def _configure_launchpad(prev: LaunchpadSettings | None) -> LaunchpadSettings | None: - if not _confirm("Add Launchpad?", default=False): - return None - print("Launchpad identity is the ~name shown on your profile page.") - username = _ask_required("Launchpad username", prev.username if prev else "") - anonymous = _confirm( - "Anonymous mode (public data only, no OAuth)?", - default=prev.mode == "anonymous" if prev else True, +def _add_source_flow(config: ActivityConfig) -> None: + chosen = _menu( + "Which source do you want to add?", [text for _, text in _KINDS], cancellable=True ) - settings = LaunchpadSettings( - username=username, mode="anonymous" if anonymous else "credentials" - ) - if not anonymous: - print( - "OAuth is obtained at report time: launchpadlib opens a browser and\n" - "stores the token in your system keyring, or reads it from a file." - ) - settings.credentials_file = _ask( - "Credentials file (empty to use the system keyring)", - prev.credentials_file if prev else "", - ) - return settings + if chosen is None: + return + kind = _KINDS[chosen][0] + account = _ADDERS[kind]() + if account is None: + print("Cancelled; nothing was added.") + return + accounts = getattr(config, kind) + accounts.append(account) + label = source_label(kind, _display_name(account), _identity(account), len(accounts) > 1) + print(f"Added: {label} ({_identity(account)}).") -def _configure_github(prev: GithubSettings | None) -> GithubSettings | None: - if not _confirm("Add GitHub?", default=False): - return None - username = _ask_required("GitHub username", prev.username if prev else "") - print("Leave the token empty to fall back to $GITHUB_TOKEN or `gh auth token`.") - token = _ask_secret("GitHub token (input hidden)", prev.token if prev else "") - return GithubSettings(username=username, token=token) +def _remove_source_flow(config: ActivityConfig) -> None: + entries: list[tuple[str, int]] = [] + options: list[str] = [] + for kind, _ in _KINDS: + accounts = getattr(config, kind) + for index, account in enumerate(accounts): + entries.append((kind, index)) + options.append(_account_line(kind, account, len(accounts) > 1)) + if not entries: + print("There are no sources to remove yet.") + return + picked = _menu("Which account do you want to remove?", options, cancellable=True) + if picked is None or not _confirm(f"Remove {options[picked]}?", default=False): + return + kind, index = entries[picked] + del getattr(config, kind)[index] + print(f"Removed: {options[picked]}.") -def _configure_gitlab(prev: GitlabSettings | None) -> GitlabSettings | None: - if not _confirm("Add GitLab?", default=False): - return None - defaults = prev or GitlabSettings(username="") - username = _ask_required("GitLab username", defaults.username) - url = _ask("Instance URL", defaults.url or "https://gitlab.com") - print("Leave the token empty to discover it from the environment instead.") - token = _ask_secret("API token (input hidden)", defaults.token) - token_env = _ask("Env var holding the token (optional, e.g. SALSA_TOKEN)", defaults.token_env) - return GitlabSettings(username=username, url=url, token=token, token_env=token_env) +def _add_github() -> GithubSettings | None: + print("GitHub identity is the profile @handle.") + username = _ask_required("GitHub username") + name = _ask("Display name (optional label)") + print("Leave both token fields empty to fall back to $GITHUB_TOKEN or `gh auth token`.") + token_env = _ask("Env var holding the token (optional, e.g. GITHUB_TOKEN)") + token = _ask_secret("GitHub token (input hidden)") + return GithubSettings(username=username, token=token, token_env=token_env, name=name) -def _configure_bts(prev: BtsSettings | None) -> BtsSettings | None: - if not _confirm("Add Debian BTS?", default=False): - return None +def _add_gitlab() -> GitlabSettings | None: + print("GitLab needs the instance URL plus your username there.") + username = _ask_required("GitLab username") + url = _ask("Instance URL", "https://gitlab.com") + name = _ask("Display name (optional label)") + print("Leave both token fields empty to discover the token from the environment.") + token = _ask_secret("API token (input hidden)") + token_env = _ask("Env var holding the token (optional, e.g. SALSA_TOKEN)") + return GitlabSettings(username=username, url=url, token=token, token_env=token_env, name=name) + + +def _add_bts() -> BtsSettings | None: print("The BTS identifies people by the email used on bugs (submitter/owner).") - email = _ask_required("Email address on Debian bugs", prev.email if prev else "") - return BtsSettings(email=email) + email = _ask_required("Email address on Debian bugs") + name = _ask("Display name (optional label)") + return BtsSettings(email=email, name=name) + + +def _add_launchpad() -> LaunchpadSettings | None: + print("Launchpad identity is the ~name shown on your profile page.") + username = _ask_required("Launchpad username") + name = _ask("Display name (optional label)") + anonymous = _confirm("Anonymous mode (public data only, no OAuth)?", default=True) + if anonymous: + return LaunchpadSettings(username=username, mode="anonymous", name=name) + + credentials_file = _ask("Credentials file path (empty to authenticate right now)") + if credentials_file: + expanded = os.path.expanduser(credentials_file) + if not os.path.isfile(expanded): + print(f"note: {expanded} does not exist yet; it will be read at report time.") + print(f"OAuth credentials will be read from {credentials_file} at report time.") + return LaunchpadSettings( + username=username, mode="credentials", credentials_file=credentials_file, name=name + ) + + # Real config-time OAuth: construct LaunchpadSource (this runs + # launchpadlib's browser flow) and keep only its side effect — the token + # in the system keyring keyed to the application name. + while True: + print("A browser window will open to authorize weekly-activity with Launchpad.") + try: + LaunchpadSource( + service="production", + anonymous=False, + application_name=_APPLICATION_NAME, + credentials_file=None, + ) + except Exception as exc: # noqa: BLE001 - network/user errors here are recoverable + print(f"warning: Launchpad authentication failed: {exc}") + if not _confirm("Retry authentication?", default=False): + return None + else: + break + print("Launchpad authorization complete: the token is stored in your system keyring;") + print("`weekly-activity report` will reuse it automatically.") + return LaunchpadSettings(username=username, mode="credentials", name=name) + + +_ADDERS: dict[str, Callable[[], object | None]] = { + "launchpad": _add_launchpad, + "github": _add_github, + "gitlab": _add_gitlab, + "bts": _add_bts, +} + + +# --- Display helpers (secrets masked) --- + + +def _describe(config: ActivityConfig, indent: str = " ") -> str: + """Human-readable summary masking every token value.""" + lines: list[str] = [] + for kind, _ in _KINDS: + accounts = getattr(config, kind) + for account in accounts: + lines.append(indent + _account_line(kind, account, len(accounts) > 1)) + return "\n".join(lines) if lines else f"{indent}(no sources configured)" + + +def _account_line(kind: str, account: object, ambiguous: bool) -> str: + identity = _identity(account) + prefix = f"{source_label(kind, _display_name(account), identity, ambiguous)}: " + if kind == "launchpad" and isinstance(account, LaunchpadSettings): + auth = ( + "anonymous (public data only)" + if account.mode == "anonymous" + else f"OAuth via {account.credentials_file or 'system keyring'}" + ) + return f"{prefix}{identity} — {auth}" + if kind == "github" and isinstance(account, GithubSettings): + fallback = account.token_env or "GITHUB_TOKEN / gh" + token_desc = "stored" if account.token else f"not stored (uses {fallback})" + return f"{prefix}{identity} — token {token_desc}" + if kind == "gitlab" and isinstance(account, GitlabSettings): + fallback = account.token_env or "_TOKEN / GITLAB_TOKEN env" + token_desc = "stored" if account.token else f"not stored ({fallback})" + return f"{prefix}{identity} @ {account.url} — token {token_desc}" + if kind == "bts" and isinstance(account, BtsSettings): + return f"{prefix}{identity}" + return f"{prefix}{identity}" # pragma: no cover - exhaustive over known kinds + + +def _identity(account: object) -> str: + if isinstance(account, BtsSettings): + return account.email + if isinstance(account, (GithubSettings, GitlabSettings, LaunchpadSettings)): + return account.username + return "" # pragma: no cover - exhaustive over known kinds + + +def _display_name(account: object) -> str: + if isinstance(account, (GithubSettings, GitlabSettings, LaunchpadSettings, BtsSettings)): + return account.name + return "" # pragma: no cover - exhaustive over known kinds # --- Prompt helpers --- +def _menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None: + """Numbered-choice menu: selected index, or None when backed out.""" + print(title) + for number, option in enumerate(options, 1): + print(f" {number}) {option}") + back_number = len(options) + 1 if cancellable else 0 + if cancellable: + print(f" {back_number}) Back") + while True: + answer = input("Enter choice: ").strip() + if not answer and cancellable: + return None + if answer.isdigit(): + number = int(answer) + if 1 <= number <= len(options): + return number - 1 + if cancellable and number == back_number: + return None + valid = back_number or len(options) + print(f"Please enter a number between 1 and {valid}.") + + def _ask(prompt: str, default: str = "") -> str: suffix = f" [{default}]" if default else "" answer = input(f"{prompt}{suffix}: ").strip() @@ -162,25 +340,3 @@ def _confirm(prompt: str, *, default: bool = False) -> bool: return True if answer in {"n", "no"}: return False - - -def _describe(config: ActivityConfig, indent: str = " ") -> str: - """Human-readable summary that masks every token value.""" - lines: list[str] = [] - if (lp := config.launchpad) is not None: - auth = ( - "anonymous (public data only)" - if lp.mode == "anonymous" - else f"OAuth via {lp.credentials_file or 'system keyring'}" - ) - lines.append(f"{indent}launchpad: {lp.username} — {auth}") - if (gh := config.github) is not None: - token_desc = "stored" if gh.token else "not stored (uses GITHUB_TOKEN / gh)" - lines.append(f"{indent}github: {gh.username} — token {token_desc}") - if (gl := config.gitlab) is not None: - fallback = gl.token_env or "uses _TOKEN / GITLAB_TOKEN env" - token_desc = "stored" if gl.token else f"not stored ({fallback})" - lines.append(f"{indent}gitlab: {gl.username} @ {gl.url} — token {token_desc}") - if (bts := config.bts) is not None: - lines.append(f"{indent}bts: {bts.email}") - return "\n".join(lines) if lines else f"{indent}(no sources configured)" diff --git a/tests/test_aggregate.py b/tests/test_aggregate.py index 509f882..1b6e732 100644 --- a/tests/test_aggregate.py +++ b/tests/test_aggregate.py @@ -48,29 +48,90 @@ def window() -> tuple[datetime, datetime]: class BuildSpecsTest(unittest.TestCase): def test_maps_every_section_in_order(self) -> None: config = ActivityConfig( - launchpad=LaunchpadSettings(username="lp"), - github=GithubSettings(username="gh"), - gitlab=GitlabSettings(username="gl"), - bts=BtsSettings(email="bts@mail"), + launchpad=[LaunchpadSettings(username="lp")], + github=[GithubSettings(username="gh")], + gitlab=[GitlabSettings(username="gl")], + bts=[BtsSettings(email="bts@mail")], ) specs = build_specs(config) self.assertEqual([s.name for s in specs], ["launchpad", "github", "gitlab", "bts"]) self.assertEqual([s.username for s in specs], ["lp", "gh", "gl", "bts@mail"]) + def test_multiple_accounts_per_kind_in_listing_order(self) -> None: + config = ActivityConfig( + launchpad=[ + LaunchpadSettings(username="lp-a"), + LaunchpadSettings(username="lp-b", mode="credentials"), + ], + github=[ + GithubSettings(username="gh-a", name="work"), + GithubSettings(username="gh-b"), + ], + bts=[ + BtsSettings(email="one@x.org"), + BtsSettings(email="two@x.org", name="two"), + ], + ) + 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.name for s in specs], + [ + "launchpad/lp-a", # siblings force identity into the label + "launchpad/lp-b", + "github/work", # explicit label wins even among siblings + "github/gh-b", + "bts/one@x.org", + "bts/two", + ], + ) + + def test_single_unnamed_account_keeps_plain_kind_label(self) -> None: + config = ActivityConfig(github=[GithubSettings(username="only-one")]) + specs = build_specs(config) + self.assertEqual(specs[0].name, "github") + def test_disabled_sections_are_skipped(self) -> None: - config = ActivityConfig(github=GithubSettings(username="only-one")) - self.assertEqual([s.name for s in build_specs(config)], ["github"]) + config = ActivityConfig(bts=[BtsSettings(email="only@one")]) + self.assertEqual([s.name for s in build_specs(config)], ["bts"]) def test_launchpad_spec_expands_tilde_credentials_file(self) -> None: - settings = LaunchpadSettings( - username="u", mode="credentials", credentials_file="~/creds.json" - ) - spec = build_specs(ActivityConfig(launchpad=settings))[0] + config = ActivityConfig(launchpad=[LaunchpadSettings(username="anon")]) + 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: + spec.make() + anon_kwargs, cred_kwargs = (call.kwargs for call in fake_source.call_args_list) + self.assertTrue(anon_kwargs["anonymous"]) + self.assertIsNone(anon_kwargs["credentials_file"]) + self.assertFalse(cred_kwargs["anonymous"]) + self.assertEqual(cred_kwargs["credentials_file"], os.path.expanduser("~") + "/creds.json") + # The two accounts must not share one late-bound factory. + self.assertNotEqual(anon_kwargs["credentials_file"], cred_kwargs["credentials_file"]) + + def test_github_token_env_resolved_lazily(self) -> None: + config = ActivityConfig(github=[GithubSettings(username="u", token_env="MY_GH_TOKEN")]) + spec = build_specs(config)[0] + with ( + mock.patch("weekly_activity.aggregate.GitHubSource") as fake_source, + mock.patch.dict(os.environ, {"MY_GH_TOKEN": "env-token"}), + ): spec.make() - kwargs = fake_source.call_args.kwargs - self.assertEqual(kwargs["credentials_file"], os.path.expanduser("~") + "/creds.json") - self.assertFalse(kwargs["anonymous"]) + self.assertEqual(fake_source.call_args.kwargs["token"], "env-token") + + def test_github_stored_token_takes_precedence_over_env(self) -> None: + config = ActivityConfig(github=[GithubSettings(username="u", token="tok")]) + spec = build_specs(config)[0] + with ( + mock.patch("weekly_activity.aggregate.GitHubSource") as fake_source, + mock.patch.dict(os.environ, {"MY_GH_TOKEN": "env-token"}), + ): + spec.make() + self.assertEqual(fake_source.call_args.kwargs["token"], "tok") class CollectAndFormatTest(unittest.TestCase): @@ -89,6 +150,13 @@ class CollectAndFormatTest(unittest.TestCase): self.assertIn("Activity report — alpha / ~a", text) self.assertIn("[skipped: beta — RuntimeError: kaput]", text) + 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"))] + text = format_combined(collect_specs(specs, since, until), since, until) + self.assertIn("Activity report — github/work / ~acme-jane", text) + def test_warning_surfaced_inside_block(self) -> None: since, until = window() specs = [SourceSpec(name="gamma", username="g", make=lambda: StubSource("gamma"))] diff --git a/tests/test_config.py b/tests/test_config.py index 6de67df..4622586 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -21,54 +21,93 @@ from weekly_activity.config import ( ) -class ConfigRoundTripTest(unittest.TestCase): - def test_round_trip_preserves_all_sections(self) -> None: - config = ActivityConfig( - launchpad=LaunchpadSettings( +def load_config_from_text(text: str) -> ActivityConfig: + """Helper: parse TOML through the public loader without touching disk.""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "weekly-activity.toml" + path.write_text(text, encoding="utf-8") + return load_config(path) + + +def sample_v2_config() -> ActivityConfig: + return ActivityConfig( + launchpad=[ + LaunchpadSettings( username="lp-user", mode="credentials", credentials_file="/tmp/lp-creds.json", service="staging", + name="work", ), - github=GithubSettings(username='gh"user', token="tok\\en"), - gitlab=GitlabSettings( + LaunchpadSettings(username="anon-lp"), + ], + github=[ + GithubSettings(username='gh"user', token="tok\\en"), + GithubSettings( + username="acme-jane", token_env="WORK_GH_TOKEN", name="work" + ), + ], + gitlab=[ + GitlabSettings( username="gl-user", url="https://salsa.debian.org", token_env="SALSA_TOKEN", - ), - bts=BtsSettings(email="dev@example.org"), - ) + name="salsa", + ) + ], + bts=[BtsSettings(email="dev@example.org")], + ) + + +class ConfigRoundTripTest(unittest.TestCase): + def test_round_trip_preserves_all_accounts(self) -> None: + config = sample_v2_config() with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "weekly-activity.toml" save_config(config, path) self.assertEqual(load_config(path), config) + def test_save_writes_v2_version_header(self) -> None: + text = format_toml(ActivityConfig(github=[GithubSettings(username="g")])) + self.assertIn("version = 2", text) + self.assertIn("[[github.accounts]]", text) + def test_save_creates_owner_only_file(self) -> None: - config = ActivityConfig(bts=BtsSettings(email="x@example.org")) + config = ActivityConfig(bts=[BtsSettings(email="x@example.org")]) with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "nested" / "cfg.toml" save_config(config, path) self.assertEqual(os.stat(path).st_mode & 0o777, 0o600) + def test_empty_config_emits_header_only(self) -> None: + text = format_toml(ActivityConfig()) + self.assertEqual(load_config_from_text(text), ActivityConfig()) + self.assertNotIn("[[", text) + def test_empty_optionals_round_trip_to_defaults(self) -> None: config = ActivityConfig( - github=GithubSettings(username="octocat"), - bts=BtsSettings(email="a@b.c"), + github=[GithubSettings(username="octocat")], + bts=[BtsSettings(email="a@b.c")], ) - text = format_toml(config) - reparsed = load_config_from_text(text) - github = reparsed.github - assert github is not None - self.assertEqual(github.username, "octocat") - self.assertEqual(github.token, "") - self.assertIsNone(reparsed.launchpad) + reparsed = load_config_from_text(format_toml(config)) + self.assertEqual(reparsed.github, [GithubSettings(username="octocat")]) + self.assertEqual(reparsed.bts, [BtsSettings(email="a@b.c")]) + self.assertEqual(reparsed.launchpad, []) def test_escapes_quotes_backslashes_and_newlines(self) -> None: - config = ActivityConfig(github=GithubSettings(username='a"b\\c\nd')) + config = ActivityConfig(github=[GithubSettings(username='a"b\\c\nd')]) reparsed = load_config_from_text(format_toml(config)) - github = reparsed.github - assert github is not None - self.assertEqual(github.username, 'a"b\\c\nd') + self.assertEqual(reparsed.github[0].username, 'a"b\\c\nd') + + def test_load_does_not_rewrite_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "weekly-activity.toml" + original = "[github]\nusername = 'octo'\n" + path.write_text(original, encoding="utf-8") + before = path.read_bytes() + config = load_config(path) + self.assertEqual(config.github, [GithubSettings(username="octo")]) + self.assertEqual(path.read_bytes(), before) class ConfigLoadingTest(unittest.TestCase): @@ -78,26 +117,123 @@ class ConfigLoadingTest(unittest.TestCase): def test_unknown_section_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text("[gitea]\nusername = 'x'\n") + load_config_from_text('version = 2\n\n[[gitea.accounts]]\nusername = "x"\n') - def test_unknown_key_rejected(self) -> None: + def test_unknown_top_level_key_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text('[github]\nusername = "x"\ntokken = "t"\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' + ) + + def test_section_must_hold_only_accounts_table(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text('[version]\nwrong = "shape"\n') # wrong type below + with self.assertRaises(ConfigError): + load_config_from_text( + 'version = 2\n\n[github]\nusername = "legacy"\n' # v2 without .accounts + ) + + def test_accounts_must_be_list_of_tables(self) -> None: + 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') def test_non_string_value_rejected(self) -> None: with self.assertRaises(ConfigError): - load_config_from_text("[github]\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): - load_config_from_text('[launchpad]\nusername = "u"\nmode = "oauth2"\n') + load_config_from_text( + 'version = 2\n\n[[launchpad.accounts]]\nusername = "u"\nmode = "oauth2"\n' + ) + + def test_bad_launchpad_service_rejected(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text( + 'version = 2\n\n[[launchpad.accounts]]\nusername = "u"\nservice = "qa"\n' + ) + + def test_v2_section_without_accounts_table_rejected(self) -> None: + # A version-tagged file must use the [[kind.accounts]] shape. + with self.assertRaises(ConfigError): + load_config_from_text('version = 2\n\n[github]\nusername = "legacy"\n') + + 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" + ) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "weekly-activity.toml" + path.write_text(text, encoding="utf-8") + with self.assertRaisesRegex(ConfigError, r"launchpad\.accounts\[1\]"): + load_config(path) def test_enabled_sources_stable_order(self) -> None: - config = load_config_from_text( - '[bts]\nemail = "e"\n\n[github]\nusername = "g"\n' + config = ActivityConfig( + bts=[BtsSettings(email="e")], + github=[GithubSettings(username="g")], ) self.assertEqual(config.enabled_sources(), ["github", "bts"]) + +class ConfigMigrationTest(unittest.TestCase): + V1_TEXT = """\ +[launchpad] +username = "lp-user" +mode = "credentials" +credentials_file = "~/creds.json" + +[github] +username = "octo" +token = "t0k" + +[gitlab] +url = "https://salsa.debian.org" +username = "gl" +token_env = "SALSA_TOKEN" + +[bts] +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.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.bts, [BtsSettings(email="dev@example.org")]) + + def test_saving_migrated_config_persists_v2_shape(self) -> None: + config = load_config_from_text(self.V1_TEXT) + text = format_toml(config) + self.assertIn("version = 2", text) + self.assertEqual(load_config_from_text(text), config) + + def test_v1_unknown_keys_still_strict(self) -> None: + with self.assertRaises(ConfigError): + load_config_from_text('[github]\nusername = "u"\ntokken = "t"\n') + + def test_v2_shaped_file_without_version_rejected(self) -> None: + with self.assertRaisesRegex(ConfigError, "version"): + load_config_from_text('[[github.accounts]]\nusername = "u"\n') + + +class ConfigPathTest(unittest.TestCase): def test_default_path_uses_xdg_or_home(self) -> None: old = os.environ.get("XDG_CONFIG_HOME") try: @@ -110,13 +246,5 @@ class ConfigLoadingTest(unittest.TestCase): os.environ["XDG_CONFIG_HOME"] = old -def load_config_from_text(text: str) -> ActivityConfig: - """Helper: parse TOML through the public loader without touching disk.""" - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "weekly-activity.toml" - path.write_text(text, encoding="utf-8") - return load_config(path) - - if __name__ == "__main__": unittest.main() -- 2.54.0 From 85ddfb4a71cd33e2ebb98bd5b2d3547b01bc48a9 Mon Sep 17 00:00:00 2001 From: Kosmos Date: Thu, 27 Aug 2026 07:46:48 +0000 Subject: [PATCH 3/5] 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() -- 2.54.0 From 12ca06ba0d9903131d6781ad38469a0efa1acd14 Mon Sep 17 00:00:00 2001 From: Kosmos Date: Thu, 27 Aug 2026 08:07:33 +0000 Subject: [PATCH 4/5] fix(wizard): read keypresses from the raw fd so arrow keys work sys.stdin.read(1) on the buffered TextIOWrapper pulls the whole escape sequence into its Python-level buffer, so the follow-up select() sees an empty fd and every arrow/Home/End press degrades to bare Esc; the unread '[B' tail then leaks into the next keypress. Read raw bytes via os.read() and select() on the file descriptor instead, keeping the same key tokens and classifications. Add pty-driven regression tests over a real cbreak terminal covering CSI sequences, single-byte keys, bare Esc, and tail-leak. --- src/weekly_activity/wizard.py | 23 +++--- tests/test_wizard.py | 148 ++++++++++++++++++++++++++++++++++ 2 files changed, 161 insertions(+), 10 deletions(-) diff --git a/src/weekly_activity/wizard.py b/src/weekly_activity/wizard.py index a2b0b21..c9ca2e1 100644 --- a/src/weekly_activity/wizard.py +++ b/src/weekly_activity/wizard.py @@ -343,23 +343,26 @@ def _read_keypress() -> str: """Read one keystroke from stdin (cbreak mode); classify it.""" import select - char = sys.stdin.read(1) - if char == "\x1b": + fd = sys.stdin.fileno() + # Read raw bytes from the fd: a buffered text read would pull the whole + # escape sequence into its own buffer, leaving nothing for select to see. + first = os.read(fd, 1) + if first == b"\x1b": # A bare Esc is not followed by more bytes; CSI sequences are. Peek briefly. - readable, _, _ = select.select([sys.stdin], [], [], 0.05) + readable, _, _ = select.select([fd], [], [], 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"): + tail = os.read(fd, 2) + return _CSI_KEYS.get(tail.decode("ascii", "replace"), _KEY_OTHER) + if first in (b"\r", b"\n"): return _KEY_ENTER - if char == "\x01": # Ctrl-A + if first == b"\x01": # Ctrl-A return _KEY_HOME - if char == "\x05": # Ctrl-E + if first == b"\x05": # Ctrl-E return _KEY_END - if char == "k": + if first == b"k": return _KEY_UP - if char == "j": + if first == b"j": return _KEY_DOWN return _KEY_OTHER diff --git a/tests/test_wizard.py b/tests/test_wizard.py index 4ca709b..5bb0b56 100644 --- a/tests/test_wizard.py +++ b/tests/test_wizard.py @@ -2,8 +2,16 @@ from __future__ import annotations +from collections.abc import Callable import io +import os +from pathlib import Path +import pty +import re +import select +import subprocess import sys +import time import unittest from unittest import mock @@ -54,5 +62,145 @@ class OptionLineRenderingTest(unittest.TestCase): self.assertEqual(_option_line("Remove a source", False), " Remove a source") +_CHILD_SCRIPT = """ +import os +import sys +import termios +import tty + +sys.path.insert(0, os.environ["WIZARD_SRC"]) +from weekly_activity.wizard import _read_keypress # noqa: E402 + +fd = sys.stdin.fileno() +saved = termios.tcgetattr(fd) +tty.setcbreak(fd) +try: + for _ in range(int(sys.argv[1])): + print("READY", flush=True) + print(f"KEY={_read_keypress()}", flush=True) +finally: + termios.tcsetattr(fd, termios.TCSADRAIN, saved) +""" + + +_KEY_LINE = re.compile(rb"KEY=([a-z]+)\r\n") + + +class ArrowKeyKeypressTest(unittest.TestCase): + """CSI sequences must classify over a real pty instead of surfacing as bare Esc. + + Drives the genuine ``_read_keypress()`` on a cbreak-mode terminal — the path the + arrow-key menu uses, unreachable through scripted ``io.StringIO`` stdin. + """ + + TIMEOUT_S = 20.0 + + def _await( + self, + master: int, + child: subprocess.Popen[bytes], + box: dict[str, bytes], + condition: Callable[[], bool], + stage: str, + ) -> None: + """Block until ``condition`` holds on the pty stream, else fail loudly.""" + deadline = time.monotonic() + self.TIMEOUT_S + while not condition(): + remaining = deadline - time.monotonic() + if remaining <= 0: + break + readable, _, _ = select.select([master], [], [], min(remaining, 0.2)) + if readable: + try: + chunk = os.read(master, 4096) + except OSError: # slave closed: the child exited early + break + if chunk: + box["stream"] += chunk + else: + return + if child.poll() is None: + child.kill() + child.wait(timeout=5) + err = child.stderr.read().decode("utf-8", "replace") if child.stderr else "" + self.fail(f"{stage}: stalled; tail={box['stream'][-300:]!r} stderr={err!r}") + + def _child_tokens(self, sends: list[bytes]) -> list[str]: + """Send each payload after its READY prompt; return printed key tokens.""" + src_dir = str(Path(__file__).resolve().parents[1] / "src") + master, slave = pty.openpty() + child = subprocess.Popen( + [sys.executable, "-c", _CHILD_SCRIPT, str(len(sends))], + stdin=slave, + stdout=slave, + stderr=subprocess.PIPE, + env={**os.environ, "WIZARD_SRC": src_dir}, + ) + os.close(slave) + box: dict[str, bytes] = {"stream": b""} + + try: + tokens: list[str] = [] + for index, payload in enumerate(sends): + self._await( + master, + child, + box, + lambda i=index: box["stream"].count(b"READY") > i, + f"READY prompt {index} before {payload!r}", + ) + seen = len(_KEY_LINE.findall(box["stream"])) + os.write(master, payload) + self._await( + master, + child, + box, + lambda n=seen: len(_KEY_LINE.findall(box["stream"])) > n, + f"token line after {payload!r}", + ) + tokens.append(_KEY_LINE.findall(box["stream"])[-1].decode()) + + self.assertEqual( + box["stream"].count(b"READY"), + len(sends), + f"unexpected output tail {box['stream'][-300:]!r}", + ) + self.assertEqual(len(_KEY_LINE.findall(box["stream"])), len(sends)) + try: + child.wait(timeout=self.TIMEOUT_S) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + self.fail("child ignored termination after completing every keystroke") + err = child.stderr.read().decode("utf-8", "replace") if child.stderr else "" + self.assertEqual(child.returncode, 0, err) + return tokens + finally: + if child.poll() is None: + child.kill() + child.wait() + if child.stderr is not None: + child.stderr.close() + os.close(master) + + def test_arrow_home_end_sequences_map_to_selection_keys(self) -> None: + tokens = self._child_tokens([b"\x1b[A", b"\x1b[B", b"\x1b[H", b"\x1b[F"]) + self.assertEqual(tokens, ["up", "down", "home", "end"]) + + def test_single_byte_keystrokes_still_classify(self) -> None: + # The reader switched to raw fd bytes; letters/Enter must classify as before. + tokens = self._child_tokens([b"\r", b"k", b"j", b"x"]) + self.assertEqual(tokens, ["enter", "up", "down", "other"]) + + def test_bare_escape_is_classified_as_escape(self) -> None: + tokens = self._child_tokens([b"\x1b"]) + self.assertEqual(tokens, ["escape"]) + + def test_escape_sequence_tail_does_not_leak_into_next_keypress(self) -> None: + # The original report: an arrow press poisoned the following reads. + tokens = self._child_tokens([b"\x1b[B", b"j", b"\r"]) + self.assertEqual(tokens, ["down", "down", "enter"]) + + if __name__ == "__main__": unittest.main() -- 2.54.0 From daf0a203a0b1db1b60a43c5b2d1e692d7274f0ed Mon Sep 17 00:00:00 2001 From: Kosmos Date: Thu, 27 Aug 2026 08:36:56 +0000 Subject: [PATCH 5/5] refactor(wizard): replace hand-rolled arrow-key menu with questionary The reviewer asked for a known-good interactive-menu library instead of the custom termios/select key reader; arrow keys had failed in their terminal. The TTY path now delegates to questionary.select (pointer ">", bold highlight via pointer/selected styles), keeping choice values mapped to list indices. Cancellable menus graft an eager Escape binding onto the prompt's own key-binding registry so Esc backs out while Ctrl-C still aborts fixed menus and backs out of cancellable ones. The non-TTY numbered fallback is untouched, and hand-rolled internals (_arrow_menu/_read_keypress/_option_line/_draw_menu_rows) are gone. --- pyproject.toml | 1 + src/weekly_activity/wizard.py | 138 ++++----------- tests/test_wizard.py | 315 ++++++++++++++++++++++------------ uv.lock | 35 ++++ 4 files changed, 277 insertions(+), 212 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e2ce52b..4ff69a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "keyring>=25", "httpx>=0.27", "python-debianbts>=4.1.1", + "questionary>=2.1.1", ] [project.scripts] diff --git a/src/weekly_activity/wizard.py b/src/weekly_activity/wizard.py index c9ca2e1..5608fa3 100644 --- a/src/weekly_activity/wizard.py +++ b/src/weekly_activity/wizard.py @@ -18,6 +18,7 @@ import os import sys from collections.abc import Callable, Sequence from pathlib import Path +from typing import TYPE_CHECKING, cast from weekly_activity.aggregate import source_label from weekly_activity.config import ( @@ -32,6 +33,10 @@ from weekly_activity.config import ( ) from weekly_activity.sources.launchpad import LaunchpadSource +if TYPE_CHECKING: + from prompt_toolkit.key_binding import KeyBindings, KeyPressEvent + + _APPLICATION_NAME = "weekly-activity" # Stable processing order, shared by menus and summaries: (config attr, menu text). @@ -291,7 +296,7 @@ def _menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> i 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 _tty_menu(title, options, cancellable=cancellable) return _numbered_menu(title, options, cancellable=cancellable) @@ -320,119 +325,42 @@ def _numbered_menu(title: str, options: Sequence[str], *, cancellable: bool = Fa # --- 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} +# Sentinel handed back through the prompt when Esc backs out; it can never +# collide with a real choice because every choice's value is its list index. +_CANCELLED = object() -def _read_keypress() -> str: - """Read one keystroke from stdin (cbreak mode); classify it.""" - import select +def _tty_menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None: + """Arrow-key menu rendered by questionary instead of hand-rolled ANSI. - fd = sys.stdin.fileno() - # Read raw bytes from the fd: a buffered text read would pull the whole - # escape sequence into its own buffer, leaving nothing for select to see. - first = os.read(fd, 1) - if first == b"\x1b": - # A bare Esc is not followed by more bytes; CSI sequences are. Peek briefly. - readable, _, _ = select.select([fd], [], [], 0.05) - if not readable: - return _KEY_ESCAPE - tail = os.read(fd, 2) - return _CSI_KEYS.get(tail.decode("ascii", "replace"), _KEY_OTHER) - if first in (b"\r", b"\n"): - return _KEY_ENTER - if first == b"\x01": # Ctrl-A - return _KEY_HOME - if first == b"\x05": # Ctrl-E - return _KEY_END - if first == b"k": - return _KEY_UP - if first == b"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. + Up/Down (plus j/k) move the bold ``>``-marked highlight and Enter + accepts; Esc or Ctrl-C backs out of a cancellable menu while Ctrl-C + alone aborts fixed ones. questionary is imported lazily so + non-interactive commands never pay for prompt-toolkit. """ - 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() + import questionary + choices = [questionary.Choice(title=text, value=index) for index, text in enumerate(options)] + style = questionary.Style([("pointer", "bold"), ("selected", "bold")]) + question = questionary.select(title, choices, pointer=">", style=style) + if cancellable: + # questionary wires arrows, Enter and Ctrl-C itself but leaves Esc as + # a no-op; graft back-out onto this prompt's own key-binding registry. + registry = cast("KeyBindings", question.application.key_bindings) -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 + @registry.add("escape", eager=True) + def _back_out(event: KeyPressEvent) -> None: + event.app.exit(result=_CANCELLED) - 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. + picked = question.unsafe_ask() + except (KeyboardInterrupt, EOFError): if not cancellable: - raise - finally: - termios.tcsetattr(fd, termios.TCSADRAIN, saved_state) - _draw_menu_rows(options, None) - return choice + raise # Ctrl-C aborts fixed menus, like every other wizard input + return None + if not isinstance(picked, int): # the Esc sentinel, or any surprise value + return None + return picked def _ask(prompt: str, default: str = "") -> str: diff --git a/tests/test_wizard.py b/tests/test_wizard.py index 5bb0b56..fbfe9e9 100644 --- a/tests/test_wizard.py +++ b/tests/test_wizard.py @@ -2,20 +2,21 @@ from __future__ import annotations -from collections.abc import Callable import io import os -from pathlib import Path import pty import re import select import subprocess import sys import time +import types import unittest +from pathlib import Path +from typing import Any, cast from unittest import mock -from weekly_activity.wizard import _menu, _option_line +from weekly_activity.wizard import _CANCELLED, _menu class NumberedMenuFallbackTest(unittest.TestCase): @@ -52,58 +53,211 @@ class NumberedMenuFallbackTest(unittest.TestCase): 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") +# --- Interactive (TTY) path ----------------------------------------------- -_CHILD_SCRIPT = """ +class _Tty(io.StringIO): + """A stream claiming to be a terminal so _menu routes to questionary.""" + + def isatty(self) -> bool: + return True + + +class _FakeKeyBindings: + """Registry mimicking prompt_toolkit's enough to capture added handlers.""" + + def __init__(self) -> None: + self.added: list[tuple[tuple[Any, ...], Any]] = [] + + def add(self, *keys: Any, eager: bool = False): + def register(handler): + self.added.append((keys, handler)) + return handler + + return register + + +class _FakeApplication: + def __init__(self) -> None: + self.key_bindings = _FakeKeyBindings() + self.exit_calls: list[dict[str, Any]] = [] + + def exit(self, result: Any = None, exception: BaseException | None = None, style=None) -> None: + self.exit_calls.append({"result": result, "exception": exception}) + + +class _FakeQuestion: + def __init__(self, outcome: Any) -> None: + self.application = _FakeApplication() + self.outcome = outcome + + def unsafe_ask(self) -> Any: + if isinstance(self.outcome, BaseException): + raise self.outcome + return self.outcome + + +class _FakeQuestionary: + """Stand-in exposing exactly the surface wizard._tty_menu touches.""" + + def __init__(self, outcome: Any) -> None: + self.outcome = outcome + self.select_outcome = _FakeQuestion(outcome) + self.seen: dict[str, Any] = {} + + def Choice(self, **kwargs: Any) -> Any: + return types.SimpleNamespace(**kwargs) + + @staticmethod + def Style(styles: list[tuple[str, str]]) -> list[tuple[str, str]]: + return styles + + def select(self, message: str, choices: list[Any], **kwargs: Any) -> _FakeQuestion: + self.seen.update(message=message, choices=list(choices), **kwargs) + return self.select_outcome + + +class QuestionaryDelegationTest(unittest.TestCase): + """On terminals _menu must hand selection to questionary, not re-implement it.""" + + def ask_tty( + self, + options: list[str], + outcome: Any, + *, + cancellable: bool, + ) -> tuple[int | None, _FakeQuestion, dict[str, Any]]: + """Run the TTY branch against an injected fake questionary module.""" + fake = _FakeQuestionary(outcome) + + with ( + mock.patch.object(sys, "stdin", _Tty()), + mock.patch.object(sys, "stdout", _Tty()), + mock.patch.dict(sys.modules, {"questionary": cast("Any", fake)}), + ): + picked = _menu("Pick one", options, cancellable=cancellable) + return picked, fake.select_outcome, fake.seen + + def test_non_cancellable_menu_delegates_and_maps_choice_value_to_index(self) -> None: + picked, question, seen = self.ask_tty( + ["Alpha", "Beta", "Gamma"], + outcome=2, + cancellable=False, + ) + self.assertEqual(picked, 2) + self.assertEqual(seen["message"], "Pick one") + self.assertEqual(seen["pointer"], ">") + self.assertEqual( + [(choice.title, choice.value) for choice in seen["choices"]], + [("Alpha", 0), ("Beta", 1), ("Gamma", 2)], + ) + # A fixed menu grafts nothing onto questionary's own bindings. + self.assertEqual(question.application.key_bindings.added, []) + + def test_cancellable_menu_keeps_fixed_menus_behaviour_on_ctrl_c(self) -> None: + with self.assertRaises(KeyboardInterrupt): + self.ask_tty(["Alpha", "Beta"], outcome=KeyboardInterrupt(), cancellable=False) + + def test_cancellable_menu_maps_ctrl_c_to_none(self) -> None: + picked, _, _ = self.ask_tty( + ["Alpha", "Beta"], + outcome=KeyboardInterrupt(), + cancellable=True, + ) + self.assertIsNone(picked) + + def test_cancellable_menu_grafts_escape_onto_the_prompt_bindings(self) -> None: + picked, question, _ = self.ask_tty( + ["Alpha", "Beta"], + outcome=_CANCELLED, + cancellable=True, + ) + self.assertIsNone(picked) # a non-int answer means "backed out" + keys, handler = question.application.key_bindings.added[-1] + self.assertIn("escape", keys) + handler(types.SimpleNamespace(app=question.application)) + self.assertIs(question.application.exit_calls[0]["result"], _CANCELLED) + + +# --- Real-terminal proof over a pty ---------------------------------------- + + +_MENU_CHILD = """ import os import sys -import termios -import tty sys.path.insert(0, os.environ["WIZARD_SRC"]) -from weekly_activity.wizard import _read_keypress # noqa: E402 +from weekly_activity.wizard import _tty_menu # noqa: E402 -fd = sys.stdin.fileno() -saved = termios.tcgetattr(fd) -tty.setcbreak(fd) -try: - for _ in range(int(sys.argv[1])): - print("READY", flush=True) - print(f"KEY={_read_keypress()}", flush=True) -finally: - termios.tcsetattr(fd, termios.TCSADRAIN, saved) +picked = _tty_menu("Pick one", ["Alpha", "Beta", "Gamma"], cancellable=True) +print(f"PICKED={picked}", flush=True) """ - -_KEY_LINE = re.compile(rb"KEY=([a-z]+)\r\n") +_PICKED_LINE = re.compile(rb"PICKED=(\d+|None)") -class ArrowKeyKeypressTest(unittest.TestCase): - """CSI sequences must classify over a real pty instead of surfacing as bare Esc. +class ArrowMenuPtyTest(unittest.TestCase): + """Drive the genuine questionary menu across a real pty with raw keypresses. - Drives the genuine ``_read_keypress()`` on a cbreak-mode terminal — the path the - arrow-key menu uses, unreachable through scripted ``io.StringIO`` stdin. + This is the regression case the reviewer hit: arrow presses must move the + highlight of a live menu, which scripted ``io.StringIO`` stdin cannot reach. """ - TIMEOUT_S = 20.0 + TIMEOUT_S = 30.0 + + @classmethod + def setUpClass(cls) -> None: + cls.child_env = { + **os.environ, + "WIZARD_SRC": str(Path(__file__).resolve().parents[1] / "src"), + } + + def drive_menu(self, sends: list[bytes]) -> tuple[str, int]: + """Render the menu on a pty, send keystrokes, return screen + exit code.""" + master, slave = pty.openpty() + child = subprocess.Popen( + [sys.executable, "-c", _MENU_CHILD], + stdin=slave, + stdout=slave, + stderr=subprocess.PIPE, + env=self.child_env, + ) + os.close(slave) + screen = bytearray() + + try: + self._await(screen, master, child, lambda: b"Gamma" in screen, "menu render") + for payload in sends: + os.write(master, payload) + time.sleep(0.2) # let the vt100 parser see keystrokes separately + self._await( + screen, + master, + child, + lambda: _PICKED_LINE.search(bytes(screen)) is not None, + "final PICKED line", + ) + child.wait(timeout=self.TIMEOUT_S) + err = child.stderr.read().decode("utf-8", "replace") if child.stderr else "" + self.assertEqual(child.returncode, 0, f"menu child failed: {err}") + return bytes(screen).decode("utf-8", "replace"), child.returncode + finally: + if child.poll() is None: + child.kill() + child.wait() + if child.stderr is not None: + child.stderr.close() + os.close(master) def _await( self, + screen: bytearray, master: int, child: subprocess.Popen[bytes], - box: dict[str, bytes], - condition: Callable[[], bool], + condition, stage: str, ) -> None: - """Block until ``condition`` holds on the pty stream, else fail loudly.""" + """Read the pty until ``condition`` holds on accumulated output.""" deadline = time.monotonic() + self.TIMEOUT_S while not condition(): remaining = deadline - time.monotonic() @@ -116,90 +270,37 @@ class ArrowKeyKeypressTest(unittest.TestCase): except OSError: # slave closed: the child exited early break if chunk: - box["stream"] += chunk + screen.extend(chunk) else: return if child.poll() is None: child.kill() child.wait(timeout=5) err = child.stderr.read().decode("utf-8", "replace") if child.stderr else "" - self.fail(f"{stage}: stalled; tail={box['stream'][-300:]!r} stderr={err!r}") + self.fail(f"{stage}: stalled; tail={bytes(screen)[-300:]!r} stderr={err!r}") - def _child_tokens(self, sends: list[bytes]) -> list[str]: - """Send each payload after its READY prompt; return printed key tokens.""" - src_dir = str(Path(__file__).resolve().parents[1] / "src") - master, slave = pty.openpty() - child = subprocess.Popen( - [sys.executable, "-c", _CHILD_SCRIPT, str(len(sends))], - stdin=slave, - stdout=slave, - stderr=subprocess.PIPE, - env={**os.environ, "WIZARD_SRC": src_dir}, - ) - os.close(slave) - box: dict[str, bytes] = {"stream": b""} + def _picked(self, screen: str) -> str: + found = _PICKED_LINE.search(screen.encode()) + if found is None: + self.fail(f"no PICKED line in {screen[-300:]!r}") + return found.group(1).decode() - try: - tokens: list[str] = [] - for index, payload in enumerate(sends): - self._await( - master, - child, - box, - lambda i=index: box["stream"].count(b"READY") > i, - f"READY prompt {index} before {payload!r}", - ) - seen = len(_KEY_LINE.findall(box["stream"])) - os.write(master, payload) - self._await( - master, - child, - box, - lambda n=seen: len(_KEY_LINE.findall(box["stream"])) > n, - f"token line after {payload!r}", - ) - tokens.append(_KEY_LINE.findall(box["stream"])[-1].decode()) + def test_down_arrow_moves_highlight_and_enter_accepts(self) -> None: + # Start on "Alpha"; one Down must land on "Beta" — the reported bug. + screen, _ = self.drive_menu([b"\x1b[B", b"\r"]) + self.assertEqual(self._picked(screen), "1") - self.assertEqual( - box["stream"].count(b"READY"), - len(sends), - f"unexpected output tail {box['stream'][-300:]!r}", - ) - self.assertEqual(len(_KEY_LINE.findall(box["stream"])), len(sends)) - try: - child.wait(timeout=self.TIMEOUT_S) - except subprocess.TimeoutExpired: - child.kill() - child.wait() - self.fail("child ignored termination after completing every keystroke") - err = child.stderr.read().decode("utf-8", "replace") if child.stderr else "" - self.assertEqual(child.returncode, 0, err) - return tokens - finally: - if child.poll() is None: - child.kill() - child.wait() - if child.stderr is not None: - child.stderr.close() - os.close(master) + def test_repeated_down_arrows_reach_third_option(self) -> None: + screen, _ = self.drive_menu([b"\x1b[B", b"\x1b[B", b"\r"]) + self.assertEqual(self._picked(screen), "2") - def test_arrow_home_end_sequences_map_to_selection_keys(self) -> None: - tokens = self._child_tokens([b"\x1b[A", b"\x1b[B", b"\x1b[H", b"\x1b[F"]) - self.assertEqual(tokens, ["up", "down", "home", "end"]) + def test_enter_without_movement_accepts_first_option(self) -> None: + screen, _ = self.drive_menu([b"\r"]) + self.assertEqual(self._picked(screen), "0") - def test_single_byte_keystrokes_still_classify(self) -> None: - # The reader switched to raw fd bytes; letters/Enter must classify as before. - tokens = self._child_tokens([b"\r", b"k", b"j", b"x"]) - self.assertEqual(tokens, ["enter", "up", "down", "other"]) - - def test_bare_escape_is_classified_as_escape(self) -> None: - tokens = self._child_tokens([b"\x1b"]) - self.assertEqual(tokens, ["escape"]) - - def test_escape_sequence_tail_does_not_leak_into_next_keypress(self) -> None: - # The original report: an arrow press poisoned the following reads. - tokens = self._child_tokens([b"\x1b[B", b"j", b"\r"]) - self.assertEqual(tokens, ["down", "down", "enter"]) + def test_escape_backs_out_of_a_cancellable_menu(self) -> None: + screen, _ = self.drive_menu([b"\x1b"]) + self.assertEqual(self._picked(screen), "None") if __name__ == "__main__": diff --git a/uv.lock b/uv.lock index d92f677..fd75631 100644 --- a/uv.lock +++ b/uv.lock @@ -381,6 +381,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/16/12b82f791c7f50ddec566873d5bdd245baa1491bac11d15ffb98aecc8f8b/pefile-2024.8.26-py3-none-any.whl", hash = "sha256:76f8b485dcd3b1bb8166f1128d395fa3d87af26360c2358fb75b80019b957c6f", size = 74766, upload-time = "2024-08-26T21:01:02.632Z" }, ] +[[package]] +name = "prompt-toolkit" +version = "3.0.53" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wcwidth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7d/ea/39b988c938f75cb75d7045b5c69f8bfed47ee2152c8837fb403de29d6fb8/prompt_toolkit-3.0.53.tar.gz", hash = "sha256:9ec8a0ad96d5c56148b3f914aa79c1564c3fde5d2e6b876e7bc327e353cf8fa6", size = 435492, upload-time = "2026-07-26T20:56:14.758Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/6f/84908cad2d6aa5144abcf7b42709fe4fdb459bc640ec7ac5786e7693dabc/prompt_toolkit-3.0.53-py3-none-any.whl", hash = "sha256:01c0891d7f9237d5e339f7d3e42cdae80b7534abb1c7c0e3352efba6231492f2", size = 392288, upload-time = "2026-07-26T20:56:12.512Z" }, +] + [[package]] name = "pycparser" version = "3.0" @@ -457,6 +469,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/de/3d/8161f7711c017e01ac9f008dfddd9410dff3674334c233bde66e7ba65bbf/pywin32_ctypes-0.2.3-py3-none-any.whl", hash = "sha256:8a1513379d709975552d202d942d9837758905c8d01eb82b8bcc30918929e7b8", size = 30756, upload-time = "2024-08-14T10:15:33.187Z" }, ] +[[package]] +name = "questionary" +version = "2.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "prompt-toolkit" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f6/45/eafb0bba0f9988f6a2520f9ca2df2c82ddfa8d67c95d6625452e97b204a5/questionary-2.1.1.tar.gz", hash = "sha256:3d7e980292bb0107abaa79c68dd3eee3c561b83a0f89ae482860b181c8bd412d", size = 25845, upload-time = "2025-08-28T19:00:20.851Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3c/26/1062c7ec1b053db9e499b4d2d5bc231743201b74051c973dadeac80a8f43/questionary-2.1.1-py3-none-any.whl", hash = "sha256:a51af13f345f1cdea62347589fbb6df3b290306ab8930713bfae4d475a7d4a59", size = 36753, upload-time = "2025-08-28T19:00:19.56Z" }, +] + [[package]] name = "ruff" version = "0.16.4" @@ -550,6 +574,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/f3/00b61e720165f73c6ef3c7a6ea732314c8ca1ba95b9ed1545fd788d9af9b/wadllib-2.1.0-py3-none-any.whl", hash = "sha256:41b58db0bb5fb21e188c7452281ee7b364b2690a4f6c388afea0ceaeafed1132", size = 61997, upload-time = "2026-07-01T10:48:01.37Z" }, ] +[[package]] +name = "wcwidth" +version = "0.8.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/34/74/c6428f875774288bec1396f5bfcbc2d925700a4dad61727fd5f2b12f249d/wcwidth-0.8.2.tar.gz", hash = "sha256:91fbef97204b96a3d4d421609b80340b760cf33e26da123ff243d76b1fda8dda", size = 1466253, upload-time = "2026-06-29T18:11:11.601Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/96/42/3e5985a0a7e57de470b320c6d6a1a67c844f6737a587f3d44dd13d1819e7/wcwidth-0.8.2-py3-none-any.whl", hash = "sha256:d63947694a0539a1d51e01eda7caf800c291020e6cdd7e28ad7b14dd33ad4f85", size = 323166, upload-time = "2026-06-29T18:11:09.888Z" }, +] + [[package]] name = "weekly-activity" version = "0.3.0" @@ -559,6 +592,7 @@ dependencies = [ { name = "keyring" }, { name = "launchpadlib" }, { name = "python-debianbts" }, + { name = "questionary" }, ] [package.dev-dependencies] @@ -576,6 +610,7 @@ requires-dist = [ { name = "keyring", specifier = ">=25" }, { name = "launchpadlib", specifier = ">=2.1.0" }, { name = "python-debianbts", specifier = ">=4.1.1" }, + { name = "questionary", specifier = ">=2.1.1" }, ] [package.metadata.requires-dev] -- 2.54.0