Add config onboarding wizard and aggregated report

- `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: <source> — 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)
This commit is contained in:
2026-08-26 20:18:41 +00:00
committed by vhaudiquet
parent cbca729693
commit 6d8cc3d817
7 changed files with 921 additions and 12 deletions
+48 -1
View File
@@ -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 <username>
@@ -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 <path>` (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)
+115
View File
@@ -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: <source> …]`` 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])
+74 -11
View File
@@ -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()
+275
View File
@@ -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 (<HOST>_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) + '"'
+186
View File
@@ -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 <HOST>_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)"
+101
View File
@@ -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()
+122
View File
@@ -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()