From 453a03c10a9a818fb456ba5b8bad349d4f783850 Mon Sep 17 00:00:00 2001 From: Valentin Haudiquet Date: Wed, 26 Aug 2026 20:18:41 +0000 Subject: [PATCH] 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()