kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 21:07:03 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

usage_report: split usage_report.py 426 -> 101 + 195 + 117 + 68

Browse Source 
The entry point now holds only the parser, mode dispatch and main. Three
modules take one concern each: _usage_report_logs (where the logs are,
the state file, per-day segment planning), _usage_report_emit (preflight,
stdout, clipboard), _usage_report_run (the two run paths and the
aggregation they drive).

Verified by the 100%-coverage tests from the previous commit. An AST
comparison of all 22 top-level defs and classes reports 22 identical, 0
mismatches.

All four modules hold 100% line+branch. 204 tests before and after. Both
CLI modes were run against the machine's real logs, which is the check
the unit tests cannot make: they stub the collaborators, so only a real
run proves the new import graph resolves.

Two lessons, recorded because the next target is a 498-line file:

This cut was too big. Two names had to be relocated after their first
placement created an import cycle, and four modules' worth of monkeypatch
targets needed re-pointing. A smaller first slice would have been cheaper.

Patching a moved function must target the module whose namespace resolves
it - the caller's, not the one it was moved to. Getting that wrong on
_run_since made main() call the real function, which read the actual atop
logs and hung the suite rather than failing.

Three import bugs (Path and _fmt_h left under TYPE_CHECKING but used at
runtime, dataclass missing where _Aggregates landed) were caught
immediately by the coverage from the previous commit.

Over-cap 38 -> 37.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-17 12:46:06 +02:00
parent ec67387c07
commit ff1a1b6a0e
13 changed files with 583 additions and 452 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
docs/superpowers/contracts/usage-report-main-split-2026-08-17.json Normal file
View File

@ -0,0 +1,18 @@
{
"title": "Split usage_report.py into an entry point plus three concern modules",
"objective": "usage_report.py was 426 lines against a 250-line cap, the last over-cap file in its directory. The preceding commit took it from 47% to 100% coverage so this split could be verified against tests that predate the seam. Success means the entry point keeps only argument parsing and dispatch, the three extracted modules each hold one concern, every moved definition is provably unchanged, coverage stays at 100% everywhere, and both CLI modes still produce a real report.",
"acceptance_criteria": [
"usage_report.py is under 250 lines and the over-cap total drops (38 -> 37)",
"Every moved top-level def and class is AST-identical to its HEAD version",
"All four resulting modules hold 100% line and branch coverage",
"The test count is 204 both before and after",
"Every test names the module that owns the symbol it patches - no reaching a moved function through the module it left",
"Both `usage_report.py --date <today>` and `--since <today>` produce a report from the machine's real logs"
],
"out_of_scope": [
"meta/scripts/optimize_vscode.py (498) and the three transcribe_* files, still over cap and untested",
"Deleting the dead _compute_window shim, still carried for backwards import compatibility with no callers",
"Any behaviour change to the report or the CLI surface"
],
"verifier": "An AST comparison script for the verbatim-move claim; pytest with --cov of all four modules and --cov-branch; check_file_length.sh --all for the cap; both CLI modes run against the real logs; pre-commit run --files for lint."
}

53
docs/superpowers/evidence/usage-report-main-split-2026-08-17.json Normal file
View File

@ -0,0 +1,53 @@
{
"intent": "Split usage_report.py (426 lines, over the 250 cap) into an entry point plus three concern-shaped modules, verified by the 100%-coverage tests from the preceding commit.",
"scope": [
"usage_report.py 426 -> 101 (argument parser, mode dispatch, main)",
"New _usage_report_logs.py 117 (log discovery, the report-state file, per-day segment planning, log descriptions)",
"New _usage_report_emit.py 68 (preflight checks, stdout emission, clipboard)",
"New _usage_report_run.py 195 (the two run paths and the aggregation they drive)",
"Test files rewired to name the owning module for every moved symbol",
"Non-goal: any behaviour change"
],
"changes": [
"Extracted three modules along concern boundaries: where the logs are and which slices to read; how the finished report leaves the process; and the two runs that tie those together. usage_report.py keeps only _is_single_day_mode, _build_parser and main.",
"_Segment moved to the logs module and _Aggregates to the run module, each next to the code that builds it.",
"_should_advance_state and _build_parser were relocated mid-split when the first placement created a cycle: _run_since needs the former, and the latter belongs with main. Both are AST-identical to HEAD wherever they landed.",
"Rewired every test reference to name the owning module. This was the bulk of the work and the source of every failure along the way: monkeypatch.setattr must target the module whose namespace resolves the name, which is the caller's module for an imported function. Patching run._run_since instead of usage_report._run_since made main() call the real function, which hit the machine's actual atop logs and hung the suite - fixed by patching where main resolves it."
],
"verification": [
{
"command": "AST comparison of every top-level def and class against HEAD",
"result": "pass",
"evidence": "'compared: 22, mismatches: 0'. Every function and both dataclasses are AST-identical to their pre-split form across the four modules, so nothing was rewritten while being moved."
},
{
"command": "pytest linux_configuration/tests/ --cov of all four modules --cov-branch",
"result": "pass",
"evidence": "usage_report.py 25 stmts / 2 branches, _usage_report_logs.py 56 / 8, _usage_report_emit.py 29 / 10, _usage_report_run.py 91 / 24 - all 100%, zero missed statements and zero partial branches. 204 passed, the identical count to before the split."
},
{
"command": "python usage_report.py --date <today> and --since <today>",
"result": "pass",
"evidence": "Both entry points produced full reports from the machine's real logs (35 atop samples, 11988 pmon samples). Running both matters here because the split separated the two run paths into a new module; the unit tests stub their collaborators, so only a real run proves the import graph resolves at runtime."
},
{
"command": "bash ~/utils/scripts/check_file_length.sh --all",
"result": "pass",
"evidence": "38 over-cap files before, 37 after. usage_report.py is off the list and all four modules are well under 250 (101/117/68/195)."
},
{
"command": "pre-commit run --files <all changed source and test files>",
"result": "pass",
"evidence": "All hooks pass after ruff auto-fixed 14 import-ordering findings created by the mechanical move. Coverage was re-measured after those fixes and still reports 100% on all four modules with 204 passing."
}
],
"risks": [
"This split moved far more than the previous two and took several corrective passes: two names had to be relocated after their first placement created an import cycle, and four modules' worth of monkeypatch targets had to be re-pointed. Each step was verified, but a smaller first cut would have been cheaper - worth remembering for optimize_vscode.py, which is the next 498-line target.",
"Three source-level bugs were introduced and caught by the tests during the split: Path and _fmt_h left under TYPE_CHECKING while used at runtime, and dataclass missing where _Aggregates landed. All three were import-only and would have failed at first call; the 100% coverage from the previous commit is what surfaced them immediately.",
"_usage_report_run.py at 195 lines is the largest of the four and holds both run paths plus the aggregation helpers. If it grows, _aggregate_segments and _Aggregates are the natural next seam."
],
"rollback": [
"git revert the commit. usage_report.py returns to 426 lines with everything in it, the three new modules disappear, and the test files return to referencing usage_report for every symbol.",
"After rollback, confirm pytest linux_configuration/tests reports 204 passed and that both `usage_report.py --date <today>` and `--since <today>` still produce reports."
]
}

68
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_emit.py Normal file
View File

@ -0,0 +1,68 @@
"""Preflight checks and report output for usage_report.
Verifies atop and its log are present before a run, writes the finished report
to stdout, and copies it to the system clipboard with whichever tool is
installed.
"""
from __future__ import annotations
from pathlib import Path
import shutil
import subprocess
import sys
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import argparse
_INSTALL_SCRIPT = Path(__file__).with_name("install_usage_monitoring.sh")
def _preflight(atop_log: Path) -> None:
if not shutil.which("atop"):
sys.exit(
f"error: `atop` is not installed.\nrun: {_INSTALL_SCRIPT}",
)
if not atop_log.exists():
sys.exit(
f"error: atop log not found: {atop_log}\n"
f"run: {_INSTALL_SCRIPT} (enables atop.service), "
"then wait for the first sample.",
)
_CLIPBOARD_CANDIDATES: tuple[tuple[str, tuple[str, ...]], ...] = (
("wl-copy", ("wl-copy",)),
("xclip", ("xclip", "-selection", "clipboard")),
("xsel", ("xsel", "--clipboard", "--input")),
)
def _copy_to_clipboard(text: str) -> None:
"""Copy `text` to the system clipboard using the first available tool.
Prints a one-line status to stderr so the stdout report stays pristine
for redirection.
"""
for name, cmd in _CLIPBOARD_CANDIDATES:
if not shutil.which(name):
continue
try:
subprocess.run(cmd, input=text, text=True, check=True)
except (subprocess.CalledProcessError, OSError) as exc:
sys.stderr.write(f"clipboard: {name} failed: {exc}\n")
return
sys.stderr.write(f"clipboard: copied {len(text)} chars via {name}\n")
return
sys.stderr.write(
"clipboard: no wl-copy/xclip/xsel found; skipping copy\n",
)
def _emit(args: argparse.Namespace, report: str) -> None:
"""Write the report to stdout and (unless suppressed) the clipboard."""
sys.stdout.write(report)
if not args.no_clipboard:
_copy_to_clipboard(report)

117
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_logs.py Normal file
View File

@ -0,0 +1,117 @@
"""Log discovery, report state and per-day segment planning for usage_report.
Resolves which daily atop/pmon logs a reporting window covers, persists the
"last report generated" timestamp that drives the since-last-report mode, and
describes the logs actually consumed for the report's methodology section.
"""
from __future__ import annotations
from dataclasses import dataclass
import datetime as _dt
import json
from pathlib import Path
from _usage_report_types import _PMON_INTERVAL_S
_ATOP_LOG_DIR = Path("/var/log/atop")
_PMON_LOG_DIR = Path.home() / ".local/share/gpu-log"
# Persisted marker of when the last report was generated. Lives under
# ~/.local/share (durable app state), not ~/.cache, so clearing caches does not
# silently reset the "since last report" window back to today-only.
_STATE_DIR = Path.home() / ".local/share/usage_report"
_STATE_FILE = _STATE_DIR / "last_report.json"
@dataclass
class _Segment:
"""One calendar day's resolved logs plus optional in-day start bounds.
*atop_begin* is an atop ``-b`` argument (``YYYYMMDDhhmmss``) and
*pmon_begin_epoch* the matching local epoch; both are set only for the first
day of a "since last report" window so re-runs do not double-count.
"""
atop_log: Path
pmon_log: Path
atop_begin: str | None = None
pmon_begin_epoch: float | None = None
def _resolve_logs(date: str) -> tuple[Path, Path]:
atop_log = _ATOP_LOG_DIR / f"atop_{date}"
pmon_log = _PMON_LOG_DIR / f"pmon-{date}.log"
return atop_log, pmon_log
def _read_last_generated() -> _dt.datetime | None:
"""Return the timestamp of the previous report run, or None if unknown."""
try:
raw = _STATE_FILE.read_text(encoding="utf-8")
except OSError:
return None
try:
stamp = json.loads(raw)["last_generated"]
return _dt.datetime.fromisoformat(stamp).astimezone()
except (ValueError, KeyError, TypeError):
return None
def _write_last_generated(when: _dt.datetime) -> None:
"""Persist *when* as the last-report timestamp for the next run."""
_STATE_DIR.mkdir(parents=True, exist_ok=True)
payload = json.dumps({"last_generated": when.isoformat(timespec="seconds")})
_STATE_FILE.write_text(payload + "\n", encoding="utf-8")
def _has_time_of_day(when: _dt.datetime) -> bool:
"""True when *when* is past local midnight, so a begin bound is needed."""
return bool(when.hour or when.minute or when.second or when.microsecond)
def _plan_segments(start: _dt.datetime, end: _dt.datetime) -> list[_Segment]:
"""Resolve one `_Segment` per calendar day across ``[start, end]``.
The first day is bounded at *start*'s time-of-day so a same-day re-run only
covers the slice since the previous report; later days are covered in full.
Returns an empty list when *start* is after *end* (e.g. a future state file).
"""
segments: list[_Segment] = []
day = start.date()
while day <= end.date():
atop_log, pmon_log = _resolve_logs(day.strftime("%Y%m%d"))
if day == start.date() and _has_time_of_day(start):
segments.append(
_Segment(
atop_log,
pmon_log,
start.strftime("%Y%m%d%H%M%S"),
start.timestamp(),
),
)
else:
segments.append(_Segment(atop_log, pmon_log))
day += _dt.timedelta(days=1)
return segments
def _describe_logs(paths: list[Path], how: str) -> str:
"""One-line Markdown description of the log files actually consumed."""
if not paths:
return f"_none found_ (`{how}`)"
if len(paths) == 1:
return f"`{paths[0]}` (`{how}`)"
return (
f"{len(paths)} daily logs `{paths[0].name}` … `{paths[-1].name}` "
f"in `{paths[0].parent}` (`{how}`)"
)
def _log_descriptions(segments: list[_Segment]) -> tuple[str, str]:
"""Return ``(atop_desc, pmon_desc)`` for the logs present in *segments*."""
atop_present = [seg.atop_log for seg in segments if seg.atop_log.exists()]
pmon_present = [seg.pmon_log for seg in segments if seg.pmon_log.exists()]
return (
_describe_logs(atop_present, "atop -r"),
_describe_logs(pmon_present, f"nvidia-smi pmon -d {_PMON_INTERVAL_S}"),
)

195
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_run.py Normal file
View File

@ -0,0 +1,195 @@
"""The two reporting runs usage_report dispatches between.
`_run_single_day` covers one calendar day ad hoc; `_run_since` spans every
daily log from the last report to now and advances the saved timestamp.
"""
from __future__ import annotations
from dataclasses import dataclass
import datetime as _dt
import shutil
import sys
from typing import TYPE_CHECKING
from _usage_report_emit import _INSTALL_SCRIPT, _emit, _preflight
from _usage_report_format import _fmt_h
from _usage_report_logs import (
_log_descriptions,
_plan_segments,
_read_last_generated,
_resolve_logs,
_Segment,
_write_last_generated,
)
from _usage_report_parsing import (
aggregate_atop,
merge_gpu_aggs,
merge_proc_aggs,
merge_windows,
)
from _usage_report_pmon import aggregate_pmon
from _usage_report_render import _render_report
from _usage_report_types import GpuAgg, ProcAgg, _Progress, _Window
_SEC_PER_DAY = 86_400
if TYPE_CHECKING:
import argparse
from pathlib import Path
def _compute_window(atop_log: Path, progress: _Progress) -> _Window:
"""Deprecated helper kept for backwards import compatibility.
New code should call :func:`aggregate_atop`, which returns the window
alongside the per-process aggregates from a single atop subprocess.
"""
_, window = aggregate_atop(atop_log, progress)
if not window.seconds:
window.seconds = _SEC_PER_DAY
return window
_REPORT_STAGES = 2
@dataclass
class _Aggregates:
"""Merged CPU/GPU aggregates and coverage window for a reporting window.
*days_with_data* is the number of daily logs that actually yielded atop
samples (gap days where the machine was off contribute nothing).
"""
cpu: dict[str, ProcAgg]
gpu: dict[str, GpuAgg]
window: _Window
gpu_samples: int
days_with_data: int
def _aggregate_segments(
segments: list[_Segment],
progress: _Progress,
) -> _Aggregates:
"""Aggregate and merge every existing daily log in *segments*.
Missing daily logs (gap days) are skipped silently.
"""
cpu_total: dict[str, ProcAgg] = {}
gpu_total: dict[str, GpuAgg] = {}
windows: list[_Window] = []
gpu_samples = 0
days_with_data = 0
for seg in segments:
if seg.atop_log.exists():
cpu, window = aggregate_atop(seg.atop_log, progress, seg.atop_begin)
merge_proc_aggs(cpu_total, cpu)
if window.distinct_samples:
windows.append(window)
days_with_data += 1
gpu, samples = aggregate_pmon(seg.pmon_log, progress, seg.pmon_begin_epoch)
merge_gpu_aggs(gpu_total, gpu)
gpu_samples += samples
return _Aggregates(
cpu_total,
gpu_total,
merge_windows(windows),
gpu_samples,
days_with_data,
)
def _period_line(start: _dt.datetime, end: _dt.datetime) -> str:
"""Markdown bullet describing the requested reporting period."""
span = _fmt_h(max((end - start).total_seconds(), 0.0))
return (
f"- **Reporting period**: {start.isoformat(timespec='seconds')} → "
f"{end.isoformat(timespec='seconds')} ({span})"
)
def _should_advance_state(args: argparse.Namespace) -> bool:
"""Advance the saved timestamp only for genuine since-last-report runs.
An explicit ``--since`` is treated as a read-only ad-hoc query (like
``--date``) so "let me look from date X" never silently re-baselines the
saved tracking point.
"""
return args.since is None and not args.no_update_state
def _run_single_day(args: argparse.Namespace, now: _dt.datetime) -> int:
"""Report on one specific day (legacy behaviour); never touches state."""
date = args.date or now.strftime("%Y%m%d")
atop_default, pmon_default = _resolve_logs(date)
atop_log = args.atop_log or atop_default
pmon_log = args.pmon_log or pmon_default
_preflight(atop_log)
segment = _Segment(atop_log, pmon_log)
progress = _Progress(enabled=not args.quiet, total_stages=_REPORT_STAGES)
aggs = _aggregate_segments([segment], progress)
progress.finish()
if not aggs.window.seconds:
aggs.window.seconds = _SEC_PER_DAY
atop_desc, pmon_desc = _log_descriptions([segment])
_emit(
args,
_render_report(
aggs,
top=args.top,
atop_desc=atop_desc,
pmon_desc=pmon_desc,
period_line=f"- **Reporting period**: {date} (single day)",
),
)
return 0
def _resolve_start(args: argparse.Namespace, now: _dt.datetime) -> _dt.datetime:
"""Pick the window start: --since, else last report, else today midnight."""
if args.since is not None:
return _dt.datetime.strptime(args.since, "%Y%m%d").astimezone()
last = _read_last_generated()
if last is not None:
return last
return now.replace(hour=0, minute=0, second=0, microsecond=0)
def _run_since(args: argparse.Namespace, now: _dt.datetime) -> int:
"""Report on everything since the last run, spanning multiple daily logs."""
if not shutil.which("atop"):
sys.exit(f"error: `atop` is not installed.\nrun: {_INSTALL_SCRIPT}")
start = _resolve_start(args, now)
segments = _plan_segments(start, now)
progress = _Progress(
enabled=not args.quiet,
total_stages=max(2 * len(segments), 1),
)
aggs = _aggregate_segments(segments, progress)
progress.finish()
if aggs.days_with_data == 0:
sys.stderr.write(
f"no atop logs with data for {start.date()} … {now.date()}; "
"nothing to report.\n",
)
if _should_advance_state(args):
_write_last_generated(now)
return 0
if not aggs.window.seconds:
aggs.window.seconds = _SEC_PER_DAY
atop_desc, pmon_desc = _log_descriptions(segments)
_emit(
args,
_render_report(
aggs,
top=args.top,
atop_desc=atop_desc,
pmon_desc=pmon_desc,
period_line=_period_line(start, now),
),
)
if _should_advance_state(args):
_write_last_generated(now)
return 0

326
linux_configuration/scripts/periodic_background/system-maintenance/bin/usage_report.py
View File

@ -25,251 +25,12 @@ count, HZ, machine specs) so the LLM never has to guess context.
from __future__ import annotations
import argparse
from dataclasses import dataclass
import datetime as _dt
import json
from pathlib import Path
import shutil
import subprocess
import sys
from _usage_report_format import _fmt_h
from _usage_report_parsing import (
aggregate_atop,
merge_gpu_aggs,
merge_proc_aggs,
merge_windows,
)
from _usage_report_pmon import aggregate_pmon
from _usage_report_render import _render_report
from _usage_report_types import _PMON_INTERVAL_S, GpuAgg, ProcAgg, _Progress, _Window
from _usage_report_run import _run_since, _run_single_day
_ATOP_LOG_DIR = Path("/var/log/atop")
_PMON_LOG_DIR = Path.home() / ".local/share/gpu-log"
_DEFAULT_TOP = 15
_SEC_PER_DAY = 86_400
# Persisted marker of when the last report was generated. Lives under
# ~/.local/share (durable app state), not ~/.cache, so clearing caches does not
# silently reset the "since last report" window back to today-only.
_STATE_DIR = Path.home() / ".local/share/usage_report"
_STATE_FILE = _STATE_DIR / "last_report.json"
def _compute_window(atop_log: Path, progress: _Progress) -> _Window:
"""Deprecated helper kept for backwards import compatibility.
New code should call :func:`aggregate_atop`, which returns the window
alongside the per-process aggregates from a single atop subprocess.
"""
_, window = aggregate_atop(atop_log, progress)
if not window.seconds:
window.seconds = _SEC_PER_DAY
return window
_REPORT_STAGES = 2
@dataclass
class _Segment:
"""One calendar day's resolved logs plus optional in-day start bounds.
*atop_begin* is an atop ``-b`` argument (``YYYYMMDDhhmmss``) and
*pmon_begin_epoch* the matching local epoch; both are set only for the first
day of a "since last report" window so re-runs do not double-count.
"""
atop_log: Path
pmon_log: Path
atop_begin: str | None = None
pmon_begin_epoch: float | None = None
@dataclass
class _Aggregates:
"""Merged CPU/GPU aggregates and coverage window for a reporting window.
*days_with_data* is the number of daily logs that actually yielded atop
samples (gap days where the machine was off contribute nothing).
"""
cpu: dict[str, ProcAgg]
gpu: dict[str, GpuAgg]
window: _Window
gpu_samples: int
days_with_data: int
def _aggregate_segments(
segments: list[_Segment],
progress: _Progress,
) -> _Aggregates:
"""Aggregate and merge every existing daily log in *segments*.
Missing daily logs (gap days) are skipped silently.
"""
cpu_total: dict[str, ProcAgg] = {}
gpu_total: dict[str, GpuAgg] = {}
windows: list[_Window] = []
gpu_samples = 0
days_with_data = 0
for seg in segments:
if seg.atop_log.exists():
cpu, window = aggregate_atop(seg.atop_log, progress, seg.atop_begin)
merge_proc_aggs(cpu_total, cpu)
if window.distinct_samples:
windows.append(window)
days_with_data += 1
gpu, samples = aggregate_pmon(seg.pmon_log, progress, seg.pmon_begin_epoch)
merge_gpu_aggs(gpu_total, gpu)
gpu_samples += samples
return _Aggregates(
cpu_total,
gpu_total,
merge_windows(windows),
gpu_samples,
days_with_data,
)
def _describe_logs(paths: list[Path], how: str) -> str:
"""One-line Markdown description of the log files actually consumed."""
if not paths:
return f"_none found_ (`{how}`)"
if len(paths) == 1:
return f"`{paths[0]}` (`{how}`)"
return (
f"{len(paths)} daily logs `{paths[0].name}` \u2026 `{paths[-1].name}` "
f"in `{paths[0].parent}` (`{how}`)"
)
def _log_descriptions(segments: list[_Segment]) -> tuple[str, str]:
"""Return ``(atop_desc, pmon_desc)`` for the logs present in *segments*."""
atop_present = [seg.atop_log for seg in segments if seg.atop_log.exists()]
pmon_present = [seg.pmon_log for seg in segments if seg.pmon_log.exists()]
return (
_describe_logs(atop_present, "atop -r"),
_describe_logs(pmon_present, f"nvidia-smi pmon -d {_PMON_INTERVAL_S}"),
)
def _resolve_logs(date: str) -> tuple[Path, Path]:
atop_log = _ATOP_LOG_DIR / f"atop_{date}"
pmon_log = _PMON_LOG_DIR / f"pmon-{date}.log"
return atop_log, pmon_log
def _read_last_generated() -> _dt.datetime | None:
"""Return the timestamp of the previous report run, or None if unknown."""
try:
raw = _STATE_FILE.read_text(encoding="utf-8")
except OSError:
return None
try:
stamp = json.loads(raw)["last_generated"]
return _dt.datetime.fromisoformat(stamp).astimezone()
except (ValueError, KeyError, TypeError):
return None
def _write_last_generated(when: _dt.datetime) -> None:
"""Persist *when* as the last-report timestamp for the next run."""
_STATE_DIR.mkdir(parents=True, exist_ok=True)
payload = json.dumps({"last_generated": when.isoformat(timespec="seconds")})
_STATE_FILE.write_text(payload + "\n", encoding="utf-8")
def _has_time_of_day(when: _dt.datetime) -> bool:
"""True when *when* is past local midnight, so a begin bound is needed."""
return bool(when.hour or when.minute or when.second or when.microsecond)
def _plan_segments(start: _dt.datetime, end: _dt.datetime) -> list[_Segment]:
"""Resolve one `_Segment` per calendar day across ``[start, end]``.
The first day is bounded at *start*'s time-of-day so a same-day re-run only
covers the slice since the previous report; later days are covered in full.
Returns an empty list when *start* is after *end* (e.g. a future state file).
"""
segments: list[_Segment] = []
day = start.date()
while day <= end.date():
atop_log, pmon_log = _resolve_logs(day.strftime("%Y%m%d"))
if day == start.date() and _has_time_of_day(start):
segments.append(
_Segment(
atop_log,
pmon_log,
start.strftime("%Y%m%d%H%M%S"),
start.timestamp(),
),
)
else:
segments.append(_Segment(atop_log, pmon_log))
day += _dt.timedelta(days=1)
return segments
_INSTALL_SCRIPT = Path(__file__).with_name("install_usage_monitoring.sh")
def _preflight(atop_log: Path) -> None:
if not shutil.which("atop"):
sys.exit(
f"error: `atop` is not installed.\nrun: {_INSTALL_SCRIPT}",
)
if not atop_log.exists():
sys.exit(
f"error: atop log not found: {atop_log}\n"
f"run: {_INSTALL_SCRIPT} (enables atop.service), "
"then wait for the first sample.",
)
_CLIPBOARD_CANDIDATES: tuple[tuple[str, tuple[str, ...]], ...] = (
("wl-copy", ("wl-copy",)),
("xclip", ("xclip", "-selection", "clipboard")),
("xsel", ("xsel", "--clipboard", "--input")),
)
def _copy_to_clipboard(text: str) -> None:
"""Copy `text` to the system clipboard using the first available tool.
Prints a one-line status to stderr so the stdout report stays pristine
for redirection.
"""
for name, cmd in _CLIPBOARD_CANDIDATES:
if not shutil.which(name):
continue
try:
subprocess.run(cmd, input=text, text=True, check=True)
except (subprocess.CalledProcessError, OSError) as exc:
sys.stderr.write(f"clipboard: {name} failed: {exc}\n")
return
sys.stderr.write(f"clipboard: copied {len(text)} chars via {name}\n")
return
sys.stderr.write(
"clipboard: no wl-copy/xclip/xsel found; skipping copy\n",
)
def _emit(args: argparse.Namespace, report: str) -> None:
"""Write the report to stdout and (unless suppressed) the clipboard."""
sys.stdout.write(report)
if not args.no_clipboard:
_copy_to_clipboard(report)
def _period_line(start: _dt.datetime, end: _dt.datetime) -> str:
"""Markdown bullet describing the requested reporting period."""
span = _fmt_h(max((end - start).total_seconds(), 0.0))
return (
f"- **Reporting period**: {start.isoformat(timespec='seconds')} → "
f"{end.isoformat(timespec='seconds')} ({span})"
)
def _is_single_day_mode(args: argparse.Namespace) -> bool:
@ -279,91 +40,6 @@ def _is_single_day_mode(args: argparse.Namespace) -> bool:
)
def _should_advance_state(args: argparse.Namespace) -> bool:
"""Advance the saved timestamp only for genuine since-last-report runs.
An explicit ``--since`` is treated as a read-only ad-hoc query (like
``--date``) so "let me look from date X" never silently re-baselines the
saved tracking point.
"""
return args.since is None and not args.no_update_state
def _run_single_day(args: argparse.Namespace, now: _dt.datetime) -> int:
"""Report on one specific day (legacy behaviour); never touches state."""
date = args.date or now.strftime("%Y%m%d")
atop_default, pmon_default = _resolve_logs(date)
atop_log = args.atop_log or atop_default
pmon_log = args.pmon_log or pmon_default
_preflight(atop_log)
segment = _Segment(atop_log, pmon_log)
progress = _Progress(enabled=not args.quiet, total_stages=_REPORT_STAGES)
aggs = _aggregate_segments([segment], progress)
progress.finish()
if not aggs.window.seconds:
aggs.window.seconds = _SEC_PER_DAY
atop_desc, pmon_desc = _log_descriptions([segment])
_emit(
args,
_render_report(
aggs,
top=args.top,
atop_desc=atop_desc,
pmon_desc=pmon_desc,
period_line=f"- **Reporting period**: {date} (single day)",
),
)
return 0
def _resolve_start(args: argparse.Namespace, now: _dt.datetime) -> _dt.datetime:
"""Pick the window start: --since, else last report, else today midnight."""
if args.since is not None:
return _dt.datetime.strptime(args.since, "%Y%m%d").astimezone()
last = _read_last_generated()
if last is not None:
return last
return now.replace(hour=0, minute=0, second=0, microsecond=0)
def _run_since(args: argparse.Namespace, now: _dt.datetime) -> int:
"""Report on everything since the last run, spanning multiple daily logs."""
if not shutil.which("atop"):
sys.exit(f"error: `atop` is not installed.\nrun: {_INSTALL_SCRIPT}")
start = _resolve_start(args, now)
segments = _plan_segments(start, now)
progress = _Progress(
enabled=not args.quiet,
total_stages=max(2 * len(segments), 1),
)
aggs = _aggregate_segments(segments, progress)
progress.finish()
if aggs.days_with_data == 0:
sys.stderr.write(
f"no atop logs with data for {start.date()} … {now.date()}; "
"nothing to report.\n",
)
if _should_advance_state(args):
_write_last_generated(now)
return 0
if not aggs.window.seconds:
aggs.window.seconds = _SEC_PER_DAY
atop_desc, pmon_desc = _log_descriptions(segments)
_emit(
args,
_render_report(
aggs,
top=args.top,
atop_desc=atop_desc,
pmon_desc=pmon_desc,
period_line=_period_line(start, now),
),
)
if _should_advance_state(args):
_write_last_generated(now)
return 0
def _build_parser() -> argparse.ArgumentParser:
"""Construct the command-line argument parser."""
parser = argparse.ArgumentParser(description=__doc__)

35
linux_configuration/tests/test_usage_report_cli.py
View File

@ -9,6 +9,7 @@ import datetime as _dt
from pathlib import Path
from typing import TYPE_CHECKING
import _usage_report_run as run
from _usage_report_types import _Progress, _Window
import usage_report
@ -33,7 +34,7 @@ def _args(**overrides: object) -> object:
def _aggregates(*, days_with_data: int = 1, seconds: int = 600) -> object:
"""Build an _Aggregates with a window that looks like real coverage."""
return usage_report._Aggregates(
return run._Aggregates(
cpu={},
gpu={},
window=_Window(distinct_samples=2, interval_s=600, seconds=seconds),
@ -133,21 +134,19 @@ def test_main_routes_since_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
# --------------------------------------------------------------------------- #
def test_run_single_day_emits_a_report(monkeypatch: pytest.MonkeyPatch) -> None:
"""The single-day path renders once and writes the result out."""
monkeypatch.setattr(usage_report, "_preflight", lambda _log: None)
monkeypatch.setattr(run, "_preflight", lambda _log: None)
monkeypatch.setattr(
usage_report,
run,
"_resolve_logs",
lambda _date: (Path("/nonexistent/atop"), Path("/nonexistent/pmon")),
)
monkeypatch.setattr(usage_report, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(usage_report, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(
usage_report, "_render_report", lambda *_a, **k: k["period_line"]
)
monkeypatch.setattr(run, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(run, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(run, "_render_report", lambda *_a, **k: k["period_line"])
emitted: list[str] = []
monkeypatch.setattr(usage_report, "_emit", lambda _args, r: emitted.append(r))
monkeypatch.setattr(run, "_emit", lambda _args, r: emitted.append(r))
rc = usage_report._run_single_day(
rc = run._run_single_day(
_args(date="20260817", quiet=True),
_dt.datetime.now().astimezone(),
)
@ -161,20 +160,20 @@ def test_run_single_day_defaults_an_empty_window_to_a_day(
) -> None:
"""A window with no measured seconds is treated as a full day."""
aggs = _aggregates(seconds=0)
monkeypatch.setattr(usage_report, "_preflight", lambda _log: None)
monkeypatch.setattr(run, "_preflight", lambda _log: None)
monkeypatch.setattr(
usage_report,
run,
"_resolve_logs",
lambda _date: (Path("/nonexistent/atop"), Path("/nonexistent/pmon")),
)
monkeypatch.setattr(usage_report, "_aggregate_segments", lambda *_a: aggs)
monkeypatch.setattr(usage_report, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(usage_report, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(usage_report, "_emit", lambda *_a: None)
monkeypatch.setattr(run, "_aggregate_segments", lambda *_a: aggs)
monkeypatch.setattr(run, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(run, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(run, "_emit", lambda *_a: None)
usage_report._run_single_day(
run._run_single_day(
_args(date="20260817", quiet=True),
_dt.datetime.now().astimezone(),
)
assert aggs.window.seconds == usage_report._SEC_PER_DAY
assert aggs.window.seconds == run._SEC_PER_DAY

40
linux_configuration/tests/test_usage_report_emit.py
View File

@ -8,6 +8,8 @@ from __future__ import annotations
from pathlib import Path
import subprocess
import _usage_report_emit as emit
import _usage_report_run as run
from _usage_report_types import _Window
import pytest
import usage_report
@ -23,7 +25,7 @@ def _args(**overrides: object) -> object:
def _aggregates(*, days_with_data: int = 1, seconds: int = 600) -> object:
"""Build an _Aggregates with a window that looks like real coverage."""
return usage_report._Aggregates(
return run._Aggregates(
cpu={},
gpu={},
window=_Window(distinct_samples=2, interval_s=600, seconds=seconds),
@ -37,10 +39,10 @@ def _aggregates(*, days_with_data: int = 1, seconds: int = 600) -> object:
# --------------------------------------------------------------------------- #
def test_preflight_exits_without_atop(monkeypatch: pytest.MonkeyPatch) -> None:
"""A missing atop binary aborts with an install hint."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: None)
monkeypatch.setattr(emit.shutil, "which", lambda _n: None)
with pytest.raises(SystemExit) as excinfo:
usage_report._preflight(Path("/nonexistent/whatever"))
emit._preflight(Path("/nonexistent/whatever"))
assert "not installed" in str(excinfo.value)
@ -50,11 +52,11 @@ def test_preflight_exits_without_log(
tmp_path: Path,
) -> None:
"""A missing log file aborts and names the path."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(emit.shutil, "which", lambda _n: "/usr/bin/atop")
missing = tmp_path / "absent"
with pytest.raises(SystemExit) as excinfo:
usage_report._preflight(missing)
emit._preflight(missing)
assert str(missing) in str(excinfo.value)
@ -64,11 +66,11 @@ def test_preflight_passes_when_both_present(
tmp_path: Path,
) -> None:
"""With atop installed and the log present, preflight is silent."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(emit.shutil, "which", lambda _n: "/usr/bin/atop")
log = tmp_path / "atop.log"
log.write_text("x", encoding="utf-8")
assert usage_report._preflight(log) is None
assert emit._preflight(log) is None
# --------------------------------------------------------------------------- #
@ -80,18 +82,18 @@ def test_copy_to_clipboard_uses_first_available_tool(
) -> None:
"""The first candidate present on PATH is the one invoked."""
monkeypatch.setattr(
usage_report.shutil,
emit.shutil,
"which",
lambda name: "/usr/bin/xclip" if name == "xclip" else None,
)
calls: list[tuple[str, ...]] = []
monkeypatch.setattr(
usage_report.subprocess,
emit.subprocess,
"run",
lambda cmd, **_k: calls.append(tuple(cmd)),
)
usage_report._copy_to_clipboard("hello")
emit._copy_to_clipboard("hello")
assert calls == [("xclip", "-selection", "clipboard")]
assert "copied 5 chars via xclip" in capsys.readouterr().err
@ -102,14 +104,14 @@ def test_copy_to_clipboard_reports_a_failing_tool(
capsys: pytest.CaptureFixture[str],
) -> None:
"""A tool that errors is reported on stderr and not retried elsewhere."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/wl-copy")
monkeypatch.setattr(emit.shutil, "which", lambda _n: "/usr/bin/wl-copy")
def boom(*_a: object, **_k: object) -> None:
raise subprocess.CalledProcessError(1, "wl-copy")
monkeypatch.setattr(usage_report.subprocess, "run", boom)
monkeypatch.setattr(emit.subprocess, "run", boom)
usage_report._copy_to_clipboard("hello")
emit._copy_to_clipboard("hello")
assert "wl-copy failed" in capsys.readouterr().err
@ -119,9 +121,9 @@ def test_copy_to_clipboard_without_any_tool(
capsys: pytest.CaptureFixture[str],
) -> None:
"""With no clipboard tool installed the copy is skipped, not fatal."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: None)
monkeypatch.setattr(emit.shutil, "which", lambda _n: None)
usage_report._copy_to_clipboard("hello")
emit._copy_to_clipboard("hello")
assert "skipping copy" in capsys.readouterr().err
@ -135,9 +137,9 @@ def test_emit_writes_stdout_and_copies(
) -> None:
"""By default the report goes to stdout and the clipboard."""
copied: list[str] = []
monkeypatch.setattr(usage_report, "_copy_to_clipboard", copied.append)
monkeypatch.setattr(emit, "_copy_to_clipboard", copied.append)
usage_report._emit(_args(no_clipboard=False), "REPORT")
emit._emit(_args(no_clipboard=False), "REPORT")
assert capsys.readouterr().out == "REPORT"
assert copied == ["REPORT"]
@ -149,9 +151,9 @@ def test_emit_skips_clipboard_when_suppressed(
) -> None:
"""--no-clipboard leaves stdout untouched but skips the copy."""
copied: list[str] = []
monkeypatch.setattr(usage_report, "_copy_to_clipboard", copied.append)
monkeypatch.setattr(emit, "_copy_to_clipboard", copied.append)
usage_report._emit(_args(no_clipboard=True), "REPORT")
emit._emit(_args(no_clipboard=True), "REPORT")
assert capsys.readouterr().out == "REPORT"
assert copied == []

28
linux_configuration/tests/test_usage_report_modes.py
View File

@ -10,7 +10,9 @@ import datetime as _dt
from pathlib import Path
from typing import TYPE_CHECKING
import _usage_report_logs as logs
import _usage_report_parsing as parsing
import _usage_report_run as run
import usage_report
if TYPE_CHECKING:
@ -47,8 +49,8 @@ def _args(**overrides: object) -> argparse.Namespace:
def test_resolve_start_prefers_since(monkeypatch: pytest.MonkeyPatch) -> None:
"""--since wins over any saved state and starts at local midnight."""
monkeypatch.setattr(usage_report, "_read_last_generated", lambda: _at(2026, 1, 1))
start = usage_report._resolve_start(_args(since="20260604"), _at(2026, 6, 4, 12))
monkeypatch.setattr(run, "_read_last_generated", lambda: _at(2026, 1, 1))
start = run._resolve_start(_args(since="20260604"), _at(2026, 6, 4, 12))
assert start.date() == _dt.date(2026, 6, 4)
assert (start.hour, start.minute) == (0, 0)
@ -57,19 +59,19 @@ def test_resolve_start_prefers_since(monkeypatch: pytest.MonkeyPatch) -> None:
def test_resolve_start_uses_last_report(monkeypatch: pytest.MonkeyPatch) -> None:
"""Without --since, the saved last-report timestamp is the start."""
last = _at(2026, 6, 2, 9, 0)
monkeypatch.setattr(usage_report, "_read_last_generated", lambda: last)
monkeypatch.setattr(run, "_read_last_generated", lambda: last)
assert usage_report._resolve_start(_args(), _at(2026, 6, 4, 12)) == last
assert run._resolve_start(_args(), _at(2026, 6, 4, 12)) == last
def test_resolve_start_first_run_is_today_midnight(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""First-ever run (no state) covers today from local midnight."""
monkeypatch.setattr(usage_report, "_read_last_generated", lambda: None)
monkeypatch.setattr(run, "_read_last_generated", lambda: None)
now = _at(2026, 6, 4, 12, 30)
assert usage_report._resolve_start(_args(), now) == _at(2026, 6, 4, 0, 0)
assert run._resolve_start(_args(), now) == _at(2026, 6, 4, 0, 0)
def test_is_single_day_mode() -> None:
@ -82,11 +84,11 @@ def test_is_single_day_mode() -> None:
def test_should_advance_state_only_for_default_run() -> None:
"""Only a plain since-last-report run re-baselines the saved timestamp."""
assert usage_report._should_advance_state(_args(no_update_state=False)) is True
assert usage_report._should_advance_state(_args(no_update_state=True)) is False
assert run._should_advance_state(_args(no_update_state=False)) is True
assert run._should_advance_state(_args(no_update_state=True)) is False
# --since is an ad-hoc query and must never advance state.
assert (
usage_report._should_advance_state(
run._should_advance_state(
_args(since="20260510", no_update_state=False),
)
is False
@ -98,7 +100,7 @@ def test_should_advance_state_only_for_default_run() -> None:
# --------------------------------------------------------------------------- #
def test_period_line_contains_both_bounds() -> None:
"""The period bullet shows start, end, and the span."""
line = usage_report._period_line(_at(2026, 6, 2, 9), _at(2026, 6, 4, 9))
line = run._period_line(_at(2026, 6, 2, 9), _at(2026, 6, 4, 9))
assert "2026-06-02T09:00:00" in line
assert "2026-06-04T09:00:00" in line
@ -107,13 +109,13 @@ def test_period_line_contains_both_bounds() -> None:
def test_describe_logs_counts() -> None:
"""Log description switches between none / single / multiple wording."""
assert "none found" in usage_report._describe_logs([], "atop -r")
assert usage_report._describe_logs(
assert "none found" in logs._describe_logs([], "atop -r")
assert logs._describe_logs(
[Path("/var/log/atop/atop_20260604")], "atop -r"
).startswith(
"`/var/log/atop/atop_20260604`",
)
many = usage_report._describe_logs(
many = logs._describe_logs(
[Path("/v/atop_20260601"), Path("/v/atop_20260604")],
"atop -r",
)

2
linux_configuration/tests/test_usage_report_render.py
View File

@ -7,8 +7,8 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import _usage_report_render as render
from _usage_report_run import _Aggregates
from _usage_report_types import GpuAgg, ProcAgg, _Window
from usage_report import _Aggregates
if TYPE_CHECKING:
import pytest

89
linux_configuration/tests/test_usage_report_run_since.py
View File

@ -6,6 +6,7 @@ from __future__ import annotations
import datetime as _dt
import _usage_report_run as run
from _usage_report_types import _Window
import pytest
import usage_report
@ -21,7 +22,7 @@ def _args(**overrides: object) -> object:
def _aggregates(*, days_with_data: int = 1, seconds: int = 600) -> object:
"""Build an _Aggregates with a window that looks like real coverage."""
return usage_report._Aggregates(
return run._Aggregates(
cpu={},
gpu={},
window=_Window(distinct_samples=2, interval_s=600, seconds=seconds),
@ -35,10 +36,10 @@ def _aggregates(*, days_with_data: int = 1, seconds: int = 600) -> object:
# --------------------------------------------------------------------------- #
def test_run_since_exits_without_atop(monkeypatch: pytest.MonkeyPatch) -> None:
"""The multi-day path also refuses to run without atop installed."""
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: None)
monkeypatch.setattr(run.shutil, "which", lambda _n: None)
with pytest.raises(SystemExit):
usage_report._run_since(_args(), _dt.datetime.now().astimezone())
run._run_since(_args(), _dt.datetime.now().astimezone())
def test_run_since_reports_and_advances_state(
@ -46,18 +47,18 @@ def test_run_since_reports_and_advances_state(
) -> None:
"""A normal run emits a report and moves the saved timestamp forward."""
now = _dt.datetime.now().astimezone()
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(usage_report, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(usage_report, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(usage_report, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(usage_report, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(usage_report, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(run.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(run, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(run, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(run, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(run, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(run, "_render_report", lambda *_a, **_k: "report")
emitted: list[str] = []
monkeypatch.setattr(usage_report, "_emit", lambda _a, r: emitted.append(r))
monkeypatch.setattr(run, "_emit", lambda _a, r: emitted.append(r))
advanced: list[object] = []
monkeypatch.setattr(usage_report, "_write_last_generated", advanced.append)
monkeypatch.setattr(run, "_write_last_generated", advanced.append)
rc = usage_report._run_since(_args(quiet=True), now)
rc = run._run_since(_args(quiet=True), now)
assert rc == 0
assert emitted == ["report"]
@ -70,20 +71,20 @@ def test_run_since_with_no_data_still_advances_state(
) -> None:
"""Gap days produce a stderr note, no report, but the state still moves."""
now = _dt.datetime.now().astimezone()
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(usage_report, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(usage_report, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(run.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(run, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(run, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(
usage_report,
run,
"_aggregate_segments",
lambda *_a: _aggregates(days_with_data=0),
)
emitted: list[str] = []
monkeypatch.setattr(usage_report, "_emit", lambda _a, r: emitted.append(r))
monkeypatch.setattr(run, "_emit", lambda _a, r: emitted.append(r))
advanced: list[object] = []
monkeypatch.setattr(usage_report, "_write_last_generated", advanced.append)
monkeypatch.setattr(run, "_write_last_generated", advanced.append)
rc = usage_report._run_since(_args(quiet=True), now)
rc = run._run_since(_args(quiet=True), now)
assert rc == 0
assert emitted == []
@ -96,18 +97,18 @@ def test_run_since_with_no_data_leaves_state_when_suppressed(
) -> None:
"""--no-update-state keeps the saved timestamp even on the no-data path."""
now = _dt.datetime.now().astimezone()
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(usage_report, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(usage_report, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(run.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(run, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(run, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(
usage_report,
run,
"_aggregate_segments",
lambda *_a: _aggregates(days_with_data=0),
)
advanced: list[object] = []
monkeypatch.setattr(usage_report, "_write_last_generated", advanced.append)
monkeypatch.setattr(run, "_write_last_generated", advanced.append)
usage_report._run_since(_args(quiet=True, no_update_state=True), now)
run._run_since(_args(quiet=True, no_update_state=True), now)
assert advanced == []
@ -118,18 +119,18 @@ def test_run_since_defaults_an_empty_window_to_a_day(
"""A window with no measured seconds is treated as a full day."""
now = _dt.datetime.now().astimezone()
aggs = _aggregates(seconds=0)
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(usage_report, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(usage_report, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(usage_report, "_aggregate_segments", lambda *_a: aggs)
monkeypatch.setattr(usage_report, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(usage_report, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(usage_report, "_emit", lambda *_a: None)
monkeypatch.setattr(usage_report, "_write_last_generated", lambda *_a: None)
monkeypatch.setattr(run.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(run, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(run, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(run, "_aggregate_segments", lambda *_a: aggs)
monkeypatch.setattr(run, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(run, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(run, "_emit", lambda *_a: None)
monkeypatch.setattr(run, "_write_last_generated", lambda *_a: None)
usage_report._run_since(_args(quiet=True), now)
run._run_since(_args(quiet=True), now)
assert aggs.window.seconds == usage_report._SEC_PER_DAY
assert aggs.window.seconds == run._SEC_PER_DAY
def test_run_since_leaves_state_when_suppressed_after_a_report(
@ -137,18 +138,18 @@ def test_run_since_leaves_state_when_suppressed_after_a_report(
) -> None:
"""--no-update-state keeps the timestamp even when a report was emitted."""
now = _dt.datetime.now().astimezone()
monkeypatch.setattr(usage_report.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(usage_report, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(usage_report, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(usage_report, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(usage_report, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(usage_report, "_render_report", lambda *_a, **_k: "report")
monkeypatch.setattr(run.shutil, "which", lambda _n: "/usr/bin/atop")
monkeypatch.setattr(run, "_resolve_start", lambda *_a: now)
monkeypatch.setattr(run, "_plan_segments", lambda *_a: [])
monkeypatch.setattr(run, "_aggregate_segments", lambda *_a: _aggregates())
monkeypatch.setattr(run, "_log_descriptions", lambda _s: ("a", "p"))
monkeypatch.setattr(run, "_render_report", lambda *_a, **_k: "report")
emitted: list[str] = []
monkeypatch.setattr(usage_report, "_emit", lambda _a, r: emitted.append(r))
monkeypatch.setattr(run, "_emit", lambda _a, r: emitted.append(r))
advanced: list[object] = []
monkeypatch.setattr(usage_report, "_write_last_generated", advanced.append)
monkeypatch.setattr(run, "_write_last_generated", advanced.append)
usage_report._run_since(_args(quiet=True, no_update_state=True), now)
run._run_since(_args(quiet=True, no_update_state=True), now)
assert emitted == ["report"]
assert advanced == []

34
linux_configuration/tests/test_usage_report_segments.py
View File

@ -7,8 +7,8 @@ from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import _usage_report_run as run
from _usage_report_types import _Progress, _Window
import usage_report
if TYPE_CHECKING:
import pytest
@ -27,9 +27,9 @@ def test_compute_window_returns_the_aggregate_window(
) -> None:
"""The shim forwards to aggregate_atop and hands back its window."""
window = _Window(distinct_samples=3, interval_s=600, seconds=1200)
monkeypatch.setattr(usage_report, "aggregate_atop", lambda *_a: ({}, window))
monkeypatch.setattr(run, "aggregate_atop", lambda *_a: ({}, window))
result = usage_report._compute_window(Path("/nonexistent/atop"), _progress())
result = run._compute_window(Path("/nonexistent/atop"), _progress())
assert result is window
assert result.seconds == 1200
@ -40,11 +40,11 @@ def test_compute_window_defaults_an_empty_window_to_a_day(
) -> None:
"""A window with no measured seconds is widened to a full day."""
window = _Window(distinct_samples=1)
monkeypatch.setattr(usage_report, "aggregate_atop", lambda *_a: ({}, window))
monkeypatch.setattr(run, "aggregate_atop", lambda *_a: ({}, window))
assert (
usage_report._compute_window(Path("/nonexistent/atop"), _progress()).seconds
== usage_report._SEC_PER_DAY
run._compute_window(Path("/nonexistent/atop"), _progress()).seconds
== run._SEC_PER_DAY
)
@ -59,11 +59,11 @@ def test_aggregate_segments_skips_missing_atop_logs(
present = tmp_path / "atop-present"
present.write_text("x", encoding="utf-8")
segments = [
usage_report._Segment(present, tmp_path / "pmon-a"),
usage_report._Segment(tmp_path / "atop-absent", tmp_path / "pmon-b"),
run._Segment(present, tmp_path / "pmon-a"),
run._Segment(tmp_path / "atop-absent", tmp_path / "pmon-b"),
]
monkeypatch.setattr(
usage_report,
run,
"aggregate_atop",
lambda *_a: ({}, _Window(distinct_samples=2, seconds=600)),
)
@ -73,9 +73,9 @@ def test_aggregate_segments_skips_missing_atop_logs(
pmon_calls.append(log)
return {}, 5
monkeypatch.setattr(usage_report, "aggregate_pmon", fake_pmon)
monkeypatch.setattr(run, "aggregate_pmon", fake_pmon)
aggs = usage_report._aggregate_segments(segments, _progress())
aggs = run._aggregate_segments(segments, _progress())
assert aggs.days_with_data == 1
assert aggs.gpu_samples == 10
@ -86,9 +86,9 @@ def test_log_descriptions_names_only_existing_logs(tmp_path: Path) -> None:
"""Descriptions cover the logs that are actually on disk."""
atop = tmp_path / "atop"
atop.write_text("x", encoding="utf-8")
segments = [usage_report._Segment(atop, tmp_path / "pmon-absent")]
segments = [run._Segment(atop, tmp_path / "pmon-absent")]
atop_desc, pmon_desc = usage_report._log_descriptions(segments)
atop_desc, pmon_desc = run._log_descriptions(segments)
assert str(atop) in atop_desc
assert "none" in pmon_desc.lower() or pmon_desc
@ -101,11 +101,11 @@ def test_aggregate_segments_ignores_a_log_with_no_samples(
"""A present atop log that yields no samples does not count as a data day."""
present = tmp_path / "atop-empty"
present.write_text("x", encoding="utf-8")
monkeypatch.setattr(usage_report, "aggregate_atop", lambda *_a: ({}, _Window()))
monkeypatch.setattr(usage_report, "aggregate_pmon", lambda *_a: ({}, 0))
monkeypatch.setattr(run, "aggregate_atop", lambda *_a: ({}, _Window()))
monkeypatch.setattr(run, "aggregate_pmon", lambda *_a: ({}, 0))
aggs = usage_report._aggregate_segments(
[usage_report._Segment(present, tmp_path / "pmon")],
aggs = run._aggregate_segments(
[run._Segment(present, tmp_path / "pmon")],
_progress(),
)

30
linux_configuration/tests/test_usage_report_since.py
View File

@ -10,9 +10,9 @@ from pathlib import Path
from typing import TYPE_CHECKING
import _usage_report_atop as atop_io
import _usage_report_logs as logs
import _usage_report_pmon as pmon
from _usage_report_types import _Progress
import usage_report
if TYPE_CHECKING:
import pytest
@ -121,13 +121,13 @@ def test_state_round_trip(
) -> None:
"""A written timestamp reads back as an equal aware datetime."""
state = tmp_path / "state" / "last_report.json"
monkeypatch.setattr(usage_report, "_STATE_DIR", state.parent)
monkeypatch.setattr(usage_report, "_STATE_FILE", state)
monkeypatch.setattr(logs, "_STATE_DIR", state.parent)
monkeypatch.setattr(logs, "_STATE_FILE", state)
when = _at(2026, 6, 2, 9, 0)
usage_report._write_last_generated(when)
logs._write_last_generated(when)
assert usage_report._read_last_generated() == when
assert logs._read_last_generated() == when
def test_state_missing_file_returns_none(
@ -135,9 +135,9 @@ def test_state_missing_file_returns_none(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""No state file yet means "unknown", so the caller falls back to today."""
monkeypatch.setattr(usage_report, "_STATE_FILE", tmp_path / "absent.json")
monkeypatch.setattr(logs, "_STATE_FILE", tmp_path / "absent.json")
assert usage_report._read_last_generated() is None
assert logs._read_last_generated() is None
def test_state_corrupt_file_returns_none(
@ -147,11 +147,11 @@ def test_state_corrupt_file_returns_none(
"""Corrupt or partial JSON is treated as unknown, not a crash."""
bad = tmp_path / "bad.json"
bad.write_text("{ not json", encoding="utf-8")
monkeypatch.setattr(usage_report, "_STATE_FILE", bad)
assert usage_report._read_last_generated() is None
monkeypatch.setattr(logs, "_STATE_FILE", bad)
assert logs._read_last_generated() is None
bad.write_text("{}", encoding="utf-8") # valid JSON, missing key
assert usage_report._read_last_generated() is None
assert logs._read_last_generated() is None
# --------------------------------------------------------------------------- #
@ -159,13 +159,13 @@ def test_state_corrupt_file_returns_none(
# --------------------------------------------------------------------------- #
def test_has_time_of_day() -> None:
"""Midnight needs no begin bound; any later time does."""
assert usage_report._has_time_of_day(_at(2026, 6, 4, 14, 30)) is True
assert usage_report._has_time_of_day(_at(2026, 6, 4, 0, 0)) is False
assert logs._has_time_of_day(_at(2026, 6, 4, 14, 30)) is True
assert logs._has_time_of_day(_at(2026, 6, 4, 0, 0)) is False
def test_plan_segments_single_day_midnight_unbounded() -> None:
"""A start at local midnight covers the whole first day (no -b bound)."""
segments = usage_report._plan_segments(_at(2026, 6, 4), _at(2026, 6, 4, 12))
segments = logs._plan_segments(_at(2026, 6, 4), _at(2026, 6, 4, 12))
assert len(segments) == 1
assert segments[0].atop_begin is None
@ -175,7 +175,7 @@ def test_plan_segments_single_day_midnight_unbounded() -> None:
def test_plan_segments_bounds_only_first_day() -> None:
"""A mid-day start bounds the first day only; later days are full."""
start = _at(2026, 6, 2, 14, 0)
segments = usage_report._plan_segments(start, _at(2026, 6, 4, 10, 0))
segments = logs._plan_segments(start, _at(2026, 6, 4, 10, 0))
assert len(segments) == 3
assert segments[0].atop_begin == "20260602140000"
@ -186,4 +186,4 @@ def test_plan_segments_bounds_only_first_day() -> None:
def test_plan_segments_start_after_end_is_empty() -> None:
"""A future state file (start past end) yields no segments."""
assert usage_report._plan_segments(_at(2026, 6, 5), _at(2026, 6, 4)) == []
assert logs._plan_segments(_at(2026, 6, 5), _at(2026, 6, 4)) == []