From b5dbe1e67a06032451431a9ab09ca64ea3ecc523 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 17 Aug 2026 12:12:11 +0200 Subject: [PATCH] usage_report: split _usage_report_parsing.py 425 -> 243 + 206 Lifts the atop subprocess machinery into _usage_report_atop.py: running atop, streaming its parseable output, building/caching the atop_agg C helper, and reading its TSV. What stays is record parsing, epoch windows and per-day merging. Verified by the 100%-coverage tests from the previous commit. An AST comparison of all 16 functions against HEAD reports 16 identical, 0 mismatches. _usage_report_render.py imported _run from the parsing module; it now imports from the module that owns it. That is the same re-export barrel the last split shipped and had to fix in a follow-up - caught in the same commit this time. usage_report.py needed no edit at all. Both modules hold 100% line+branch (94/28 and 130/34, 0 missed). 179 tests before and after; the five new test files collect 8+6+4+14+5 = 37, which is last commit's 36 plus the one re-homed out of test_usage_report_modes.py, where it referenced a constant that had moved out from under it. The two test files added last commit were themselves over cap (373, 272) so they are split five ways here - the DropZone lesson from earlier today, where a lint fix silently added an over-cap file. Ran the program: a real report from today's logs, through the native atop_agg path, which is precisely the code this commit moved. Over-cap 39 -> 38. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY --- ...usage-report-parsing-split-2026-08-17.json | 18 ++ ...usage-report-parsing-split-2026-08-17.json | 58 +++++ .../bin/_usage_report_atop.py | 206 ++++++++++++++++ .../bin/_usage_report_parsing.py | 200 +--------------- .../bin/_usage_report_render.py | 2 +- .../tests/test_usage_report_aggregate.py | 111 +++++++++ .../tests/test_usage_report_atop_build.py | 134 +++++++++++ .../tests/test_usage_report_atop_io.py | 219 +----------------- .../tests/test_usage_report_atop_pipeline.py | 135 +++++++++++ .../tests/test_usage_report_modes.py | 13 -- .../tests/test_usage_report_parse_records.py | 98 -------- .../tests/test_usage_report_since.py | 6 +- 12 files changed, 676 insertions(+), 524 deletions(-) create mode 100644 docs/superpowers/contracts/usage-report-parsing-split-2026-08-17.json create mode 100644 docs/superpowers/evidence/usage-report-parsing-split-2026-08-17.json create mode 100644 linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_atop.py create mode 100644 linux_configuration/tests/test_usage_report_aggregate.py create mode 100644 linux_configuration/tests/test_usage_report_atop_build.py create mode 100644 linux_configuration/tests/test_usage_report_atop_pipeline.py diff --git a/docs/superpowers/contracts/usage-report-parsing-split-2026-08-17.json b/docs/superpowers/contracts/usage-report-parsing-split-2026-08-17.json new file mode 100644 index 00000000..98cfa17e --- /dev/null +++ b/docs/superpowers/contracts/usage-report-parsing-split-2026-08-17.json @@ -0,0 +1,18 @@ +{ + "title": "Split _usage_report_parsing.py at the subprocess seam", + "objective": "_usage_report_parsing.py was 425 lines against a 250-line cap, and the 61% of it that was uncovered was almost exactly the half a split would extract - the atop subprocess plumbing and the native atop_agg helper. The preceding commit took it to 100% coverage so this split could be verified against tests that predate the seam. Success means the file drops under the cap, every moved function is provably unchanged, both resulting modules stay at 100%, the test count is identical, and the program still generates a real report through the native path.", + "acceptance_criteria": [ + "_usage_report_parsing.py is under 250 lines and the over-cap total drops (39 -> 38)", + "Every moved function is AST-identical to its HEAD version", + "Both _usage_report_parsing.py and _usage_report_atop.py hold 100% line and branch coverage", + "The test count is 179 both before and after, and no test file exceeds the cap either", + "No re-export barrel: every import site names the module that owns the symbol, including _usage_report_render.py's _run import", + "usage_report.py generates a report from the machine's real atop logs after the change" + ], + "out_of_scope": [ + "usage_report.py itself, still 426 lines and over cap - it needs its own coverage-first commit", + "meta/scripts/optimize_vscode.py and the three transcribe_* files, also over cap and untested", + "Any behaviour change to the report output or the atop field indices" + ], + "verifier": "An AST comparison script for the verbatim-move claim; pytest with --cov of both modules and --cov-branch for coverage; pytest --collect-only per file for the test count; check_file_length.sh --all for the cap; python usage_report.py --date for the runtime check; pre-commit run --files for lint." +} diff --git a/docs/superpowers/evidence/usage-report-parsing-split-2026-08-17.json b/docs/superpowers/evidence/usage-report-parsing-split-2026-08-17.json new file mode 100644 index 00000000..107731b6 --- /dev/null +++ b/docs/superpowers/evidence/usage-report-parsing-split-2026-08-17.json @@ -0,0 +1,58 @@ +{ + "intent": "Split _usage_report_parsing.py (425 lines, over the 250 cap) by lifting the atop subprocess and native-helper machinery into its own module, verified by the 100%-coverage tests that landed in the preceding commit.", + "scope": [ + "_usage_report_parsing.py 425 -> 243 (record parsing, epoch windows, per-day merging)", + "New _usage_report_atop.py 206 (running atop, streaming its output, building/caching atop_agg, reading its TSV)", + "Import sites updated in usage_report.py's caller chain and _usage_report_render.py", + "Test files re-split to mirror the source and stay under the cap", + "Non-goal: any behaviour change" + ], + "changes": [ + "Moved _run, _atop_read_cmd, _iter_atop_lines, _atop_agg_binary, _apply_native_name, _window_from_native and _aggregate_atop_native into _usage_report_atop.py, with the five constants they own.", + "_usage_report_render.py imported _run from _usage_report_parsing; it now imports from _usage_report_atop. That is the same re-export-barrel trap caught in the previous split - fixing it in the same commit this time rather than a follow-up.", + "usage_report.py needed no edit: it imports aggregate_atop and the three merge_* functions, all of which stayed in the parsing module.", + "Re-homed test_atop_agg_binary_missing_source_falls_back from test_usage_report_modes.py into the new build-focused test file, so it sits with the other four _atop_agg_binary cases rather than referencing a constant that had moved out from under it.", + "Split the two new test files, which were themselves over the cap at 373 and 272 lines, into five: atop_io (subprocess entry points), atop_build (binary build/cache), atop_pipeline (the atop | atop_agg TSV read), parse_records (record parsing) and aggregate (the pure-Python fallback)." + ], + "verification": [ + { + "command": "AST comparison of every moved function against HEAD", + "result": "pass", + "evidence": "All 16 functions report IDENTICAL - _run, _atop_read_cmd, _iter_atop_lines, _parse_name, _parse_prc, _parse_prm, _window_from_epochs, _atop_agg_binary, _apply_native_name, _window_from_native, _aggregate_atop_native, aggregate_atop, _fold_pid_aggregates, merge_proc_aggs, merge_gpu_aggs, merge_windows. 'compared: 16, mismatches: 0'. The comparison normalises through ast.parse(ast.unparse(node)), so it is blind to formatting but catches any logic change." + }, + { + "command": "pytest linux_configuration/tests/ --cov of both modules --cov-branch", + "result": "pass", + "evidence": "_usage_report_atop.py 94 stmts / 28 branches 100%; _usage_report_parsing.py 130 stmts / 34 branches 100%. Zero missed statements and zero partial branches in both, matching the single-file 100% from the previous commit. 179 passed, the same count as before the split." + }, + { + "command": "pytest --collect-only per new test file, summed", + "result": "pass", + "evidence": "atop_io 8 + atop_build 6 + atop_pipeline 4 + parse_records 14 + aggregate 5 = 37. That is the 36 tests added last commit plus the one re-homed from test_usage_report_modes.py, and the directory total of 179 is unchanged - so nothing fell outside the runner's glob." + }, + { + "command": "python usage_report.py --date --top 1 --no-clipboard --no-update-state --quiet", + "result": "pass", + "evidence": "Produced a full report from today's real logs (32 atop samples). This run went through the native atop_agg helper - exactly the code path this commit extracted - so the import rewiring is proven at runtime, not just under mocks." + }, + { + "command": "bash ~/utils/scripts/check_file_length.sh --all", + "result": "pass", + "evidence": "39 over-cap files before, 38 after. _usage_report_parsing.py is off the list and none of the two source or five test files exceeds 250." + }, + { + "command": "pre-commit run --files ", + "result": "pass", + "evidence": "Every hook passes including ruff, ruff format, mypy, bandit and the noqa blockers. Two mypy findings introduced by the mechanical test split were fixed properly - a _parse_prc test that landed in the wrong file was moved to the record-parsing one, and a Self annotation ruff introduced got its import." + } + ], + "risks": [ + "Five test files where there were two makes the mapping from test file to source module less obvious. Each is named after the concern it covers and the module docstrings say which source module they exercise.", + "_usage_report_atop.py is 206 lines - comfortably under the cap, but it holds two distinct concerns (running atop, and building the C helper). If it grows, the build half is the natural next seam; its tests are already isolated in test_usage_report_atop_build.py.", + "The tests fake subprocess at the module seam, so they do not prove atop's real output format still matches the field indices. The end-to-end program run covers that, and it exercised the native path this time rather than the Python fallback." + ], + "rollback": [ + "git revert the commit. _usage_report_parsing.py returns to 425 lines with all sixteen functions, _usage_report_atop.py disappears, and the five test files collapse back into two.", + "After rollback, confirm pytest linux_configuration/tests still reports 179 passed and that usage_report.py still generates a report. Note that reverting this while keeping the previous commit's tests is safe - those tests were written against the pre-split module." + ] +} diff --git a/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_atop.py b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_atop.py new file mode 100644 index 00000000..7886ce6d --- /dev/null +++ b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_atop.py @@ -0,0 +1,206 @@ +"""Subprocess plumbing for usage_report. + +Runs atop, streams its parseable output, and builds/caches the native +`atop_agg` C helper. +""" + +from __future__ import annotations + +import datetime as _dt +from pathlib import Path +import shutil +import subprocess +from typing import TYPE_CHECKING + +from _usage_report_types import ProcAgg, _Window + +if TYPE_CHECKING: + from collections.abc import Iterator + + from _usage_report_types import _Progress + +_ATOP_AGG_CACHE_BIN = Path.home() / ".cache" / "usage_report" / "atop_agg" +_ATOP_AGG_BIN_MODE = 0o755 +# Repo layout: linux_configuration/scripts/system-maintenance/bin/usage_report.py +# -> parents[4] is the repo root which hosts the C/ source tree. +_ATOP_AGG_SRC_DIR = Path(__file__).resolve().parents[4] / "C" / "atop_agg" +_ATOP_AGG_BUILD_TIMEOUT_S = 60 +_NATIVE_TSV_NAME_LEN = 7 +_NATIVE_TSV_WIN_LEN = 5 + + +def _run(cmd: list[str]) -> str: + """Run *cmd* and return stdout (empty string on failure).""" + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + check=False, + timeout=60, + ) + except (OSError, subprocess.TimeoutExpired): + return "" + return proc.stdout + + +def _atop_read_cmd( + log: Path, + labels: str, + begin: str | None, + end: str | None, +) -> list[str]: + """Build an `atop -r` command, optionally bounded by begin/end times. + + *begin*/*end* are atop `-b`/`-e` arguments (`[YYYYMMDD]hhmm[ss]`) used to + restrict replay to a sub-window of the day's log, so a "since last report" + run does not double-count the part of the first day already reported. + """ + cmd = ["atop", "-r", str(log)] + if begin is not None: + cmd += ["-b", begin] + if end is not None: + cmd += ["-e", end] + cmd += ["-P", labels] + return cmd + + +def _iter_atop_lines( + log: Path, + labels: str, + begin: str | None = None, + end: str | None = None, +) -> Iterator[str]: + """Stream `atop -r LOG -P LABELS` stdout line-by-line. + + Uses `Popen` so the report can show progress while atop is still + decoding its binary log, rather than buffering the whole output. + """ + cmd = _atop_read_cmd(log, labels, begin, end) + with subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + ) as proc: + stdout = proc.stdout + if stdout is None: + return + for raw in stdout: + yield raw.rstrip("\n") + + +def _atop_agg_binary() -> Path | None: + """Return a cached `atop_agg` binary path, auto-building if missing/stale. + + Falls back to ``None`` when the C source tree or a system C compiler + is unavailable, in which case callers use the pure-Python parser. + """ + src_c = _ATOP_AGG_SRC_DIR / "atop_agg.c" + if not src_c.exists(): + # Source tree is gone (relocated/extracted): never trust an orphaned + # cached binary whose provenance we can no longer verify against + # source — a stale build can silently carry parsing bugs. Fall back to + # the pure-Python parser instead. + return None + if ( + _ATOP_AGG_CACHE_BIN.exists() + and src_c.stat().st_mtime <= _ATOP_AGG_CACHE_BIN.stat().st_mtime + ): + return _ATOP_AGG_CACHE_BIN + if shutil.which("cc") is None: + return None + _ATOP_AGG_CACHE_BIN.parent.mkdir(parents=True, exist_ok=True) + make_cmd = ["make", "-s", "-C", str(_ATOP_AGG_SRC_DIR), "atop_agg"] + try: + subprocess.run( + make_cmd, + check=True, + capture_output=True, + timeout=_ATOP_AGG_BUILD_TIMEOUT_S, + ) + except (OSError, subprocess.SubprocessError): + return None + built = _ATOP_AGG_SRC_DIR / "atop_agg" + if not built.exists(): + return None + shutil.copy2(built, _ATOP_AGG_CACHE_BIN) + _ATOP_AGG_CACHE_BIN.chmod(_ATOP_AGG_BIN_MODE) + return _ATOP_AGG_CACHE_BIN + + +def _apply_native_name(parts: list[str], agg_map: dict[str, ProcAgg]) -> None: + r"""Fold one `N\t\t\t\t\t\t` row.""" + _, name, cpu_s, peak_s, sum_avg_s, rss_n_s, pids_s = parts + entry = agg_map.setdefault(name, ProcAgg(name=name)) + entry.cpu_ticks = int(cpu_s) + entry.peak_rss_kb = int(peak_s) + entry.rss_kb_sum = int(sum_avg_s) + entry.rss_samples = int(rss_n_s) + # The C helper pre-aggregates by name; pid_set is unused in the native + # path but `len(pid_set)` drives the "PIDs" column in the report. + entry.pid_set = set(range(int(pids_s))) + + +def _window_from_native(parts: list[str]) -> _Window: + r"""Build a `_Window` from a `W\t\t\t\t` row.""" + _, start_s, end_s, n_s, interval_s = parts + n_epochs = int(n_s) + if not n_epochs: + return _Window() + start_epoch = int(start_s) + end_epoch = int(end_s) + start_dt = _dt.datetime.fromtimestamp(start_epoch).astimezone() + end_dt = _dt.datetime.fromtimestamp(end_epoch).astimezone() + return _Window( + start=start_dt.isoformat(timespec="seconds"), + end=end_dt.isoformat(timespec="seconds"), + distinct_samples=n_epochs, + interval_s=int(interval_s), + seconds=end_epoch - start_epoch, + start_epoch=start_epoch, + end_epoch=end_epoch, + ) + + +def _aggregate_atop_native( + log: Path, + progress: _Progress, + binary: Path, + begin: str | None = None, + end: str | None = None, +) -> tuple[dict[str, ProcAgg], _Window]: + """Aggregate via `atop | atop_agg`; return `(by_name, window)`.""" + progress.start_stage("atop: parse PRC+PRM (native)") + agg_map: dict[str, ProcAgg] = {} + window = _Window() + atop_cmd = _atop_read_cmd(log, "PRC,PRM", begin, end) + agg_cmd = [str(binary)] + with ( + subprocess.Popen( + atop_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) as atop, + subprocess.Popen( + agg_cmd, + stdin=atop.stdout, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + ) as agg, + ): + if atop.stdout is not None: + atop.stdout.close() + stdout = agg.stdout + if stdout is None: + return agg_map, window + for raw in stdout: + parts = raw.rstrip("\n").split("\t") + tag = parts[0] + if tag == "N" and len(parts) == _NATIVE_TSV_NAME_LEN: + _apply_native_name(parts, agg_map) + elif tag == "W" and len(parts) == _NATIVE_TSV_WIN_LEN: + window = _window_from_native(parts) + progress.update(1.0) + return agg_map, window diff --git a/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_parsing.py b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_parsing.py index 03dfb2c0..48ccd247 100644 --- a/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_parsing.py +++ b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_parsing.py @@ -1,26 +1,29 @@ -"""atop + pmon log parsing and aggregation helpers for usage_report.""" +"""atop record parsing and per-day aggregate merging for usage_report.""" from __future__ import annotations import contextlib import datetime as _dt -from pathlib import Path -import shutil -import subprocess from typing import TYPE_CHECKING +from _usage_report_atop import ( + _aggregate_atop_native, + _atop_agg_binary, + _iter_atop_lines, +) from _usage_report_types import ( _MIN_SAMPLES_FOR_WINDOW, GpuAgg, ProcAgg, _PidCpu, _PidRam, - _Progress, _Window, ) if TYPE_CHECKING: - from collections.abc import Iterator + from pathlib import Path + + from _usage_report_types import _Progress # atop parseable output layout (atop 2.x, same on Arch/Debian/Ubuntu): # 0 label, 1 host, 2 epoch, 3 YYYY/MM/DD, 4 HH:MM:SS, 5 interval_s, @@ -38,75 +41,6 @@ _PRM_NAME_IDX = 7 _PRM_MIN_LEN = 12 _CPU_RECORD_MIN_LEN = 5 _PAREN_PAIR_MIN = 2 -_ATOP_AGG_CACHE_BIN = Path.home() / ".cache" / "usage_report" / "atop_agg" -_ATOP_AGG_BIN_MODE = 0o755 -# Repo layout: linux_configuration/scripts/system-maintenance/bin/usage_report.py -# -> parents[4] is the repo root which hosts the C/ source tree. -_ATOP_AGG_SRC_DIR = Path(__file__).resolve().parents[4] / "C" / "atop_agg" -_ATOP_AGG_BUILD_TIMEOUT_S = 60 -_NATIVE_TSV_NAME_LEN = 7 -_NATIVE_TSV_WIN_LEN = 5 - - -def _run(cmd: list[str]) -> str: - """Run *cmd* and return stdout (empty string on failure).""" - try: - proc = subprocess.run( - cmd, - capture_output=True, - text=True, - check=False, - timeout=60, - ) - except (OSError, subprocess.TimeoutExpired): - return "" - return proc.stdout - - -def _atop_read_cmd( - log: Path, - labels: str, - begin: str | None, - end: str | None, -) -> list[str]: - """Build an `atop -r` command, optionally bounded by begin/end times. - - *begin*/*end* are atop `-b`/`-e` arguments (`[YYYYMMDD]hhmm[ss]`) used to - restrict replay to a sub-window of the day's log, so a "since last report" - run does not double-count the part of the first day already reported. - """ - cmd = ["atop", "-r", str(log)] - if begin is not None: - cmd += ["-b", begin] - if end is not None: - cmd += ["-e", end] - cmd += ["-P", labels] - return cmd - - -def _iter_atop_lines( - log: Path, - labels: str, - begin: str | None = None, - end: str | None = None, -) -> Iterator[str]: - """Stream `atop -r LOG -P LABELS` stdout line-by-line. - - Uses `Popen` so the report can show progress while atop is still - decoding its binary log, rather than buffering the whole output. - """ - cmd = _atop_read_cmd(log, labels, begin, end) - with subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - ) as proc: - stdout = proc.stdout - if stdout is None: - return - for raw in stdout: - yield raw.rstrip("\n") def _parse_name(parts: list[str], name_idx: int) -> tuple[str, int]: @@ -189,122 +123,6 @@ def _window_from_epochs(epochs: set[int]) -> _Window: ) -def _atop_agg_binary() -> Path | None: - """Return a cached `atop_agg` binary path, auto-building if missing/stale. - - Falls back to ``None`` when the C source tree or a system C compiler - is unavailable, in which case callers use the pure-Python parser. - """ - src_c = _ATOP_AGG_SRC_DIR / "atop_agg.c" - if not src_c.exists(): - # Source tree is gone (relocated/extracted): never trust an orphaned - # cached binary whose provenance we can no longer verify against - # source — a stale build can silently carry parsing bugs. Fall back to - # the pure-Python parser instead. - return None - if ( - _ATOP_AGG_CACHE_BIN.exists() - and src_c.stat().st_mtime <= _ATOP_AGG_CACHE_BIN.stat().st_mtime - ): - return _ATOP_AGG_CACHE_BIN - if shutil.which("cc") is None: - return None - _ATOP_AGG_CACHE_BIN.parent.mkdir(parents=True, exist_ok=True) - make_cmd = ["make", "-s", "-C", str(_ATOP_AGG_SRC_DIR), "atop_agg"] - try: - subprocess.run( - make_cmd, - check=True, - capture_output=True, - timeout=_ATOP_AGG_BUILD_TIMEOUT_S, - ) - except (OSError, subprocess.SubprocessError): - return None - built = _ATOP_AGG_SRC_DIR / "atop_agg" - if not built.exists(): - return None - shutil.copy2(built, _ATOP_AGG_CACHE_BIN) - _ATOP_AGG_CACHE_BIN.chmod(_ATOP_AGG_BIN_MODE) - return _ATOP_AGG_CACHE_BIN - - -def _apply_native_name(parts: list[str], agg_map: dict[str, ProcAgg]) -> None: - r"""Fold one `N\t\t\t\t\t\t` row.""" - _, name, cpu_s, peak_s, sum_avg_s, rss_n_s, pids_s = parts - entry = agg_map.setdefault(name, ProcAgg(name=name)) - entry.cpu_ticks = int(cpu_s) - entry.peak_rss_kb = int(peak_s) - entry.rss_kb_sum = int(sum_avg_s) - entry.rss_samples = int(rss_n_s) - # The C helper pre-aggregates by name; pid_set is unused in the native - # path but `len(pid_set)` drives the "PIDs" column in the report. - entry.pid_set = set(range(int(pids_s))) - - -def _window_from_native(parts: list[str]) -> _Window: - r"""Build a `_Window` from a `W\t\t\t\t` row.""" - _, start_s, end_s, n_s, interval_s = parts - n_epochs = int(n_s) - if not n_epochs: - return _Window() - start_epoch = int(start_s) - end_epoch = int(end_s) - start_dt = _dt.datetime.fromtimestamp(start_epoch).astimezone() - end_dt = _dt.datetime.fromtimestamp(end_epoch).astimezone() - return _Window( - start=start_dt.isoformat(timespec="seconds"), - end=end_dt.isoformat(timespec="seconds"), - distinct_samples=n_epochs, - interval_s=int(interval_s), - seconds=end_epoch - start_epoch, - start_epoch=start_epoch, - end_epoch=end_epoch, - ) - - -def _aggregate_atop_native( - log: Path, - progress: _Progress, - binary: Path, - begin: str | None = None, - end: str | None = None, -) -> tuple[dict[str, ProcAgg], _Window]: - """Aggregate via `atop | atop_agg`; return `(by_name, window)`.""" - progress.start_stage("atop: parse PRC+PRM (native)") - agg_map: dict[str, ProcAgg] = {} - window = _Window() - atop_cmd = _atop_read_cmd(log, "PRC,PRM", begin, end) - agg_cmd = [str(binary)] - with ( - subprocess.Popen( - atop_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - ) as atop, - subprocess.Popen( - agg_cmd, - stdin=atop.stdout, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - ) as agg, - ): - if atop.stdout is not None: - atop.stdout.close() - stdout = agg.stdout - if stdout is None: - return agg_map, window - for raw in stdout: - parts = raw.rstrip("\n").split("\t") - tag = parts[0] - if tag == "N" and len(parts) == _NATIVE_TSV_NAME_LEN: - _apply_native_name(parts, agg_map) - elif tag == "W" and len(parts) == _NATIVE_TSV_WIN_LEN: - window = _window_from_native(parts) - progress.update(1.0) - return agg_map, window - - def aggregate_atop( log: Path, progress: _Progress, diff --git a/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_render.py b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_render.py index 03bc6373..038d5e90 100644 --- a/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_render.py +++ b/linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_render.py @@ -9,8 +9,8 @@ import platform import re from typing import TYPE_CHECKING +from _usage_report_atop import _run from _usage_report_format import _fmt_h -from _usage_report_parsing import _run from _usage_report_tables import _cpu_table, _gpu_table, _ram_table from _usage_report_types import _HZ, _PMON_INTERVAL_S, _Window diff --git a/linux_configuration/tests/test_usage_report_aggregate.py b/linux_configuration/tests/test_usage_report_aggregate.py new file mode 100644 index 00000000..869c8f63 --- /dev/null +++ b/linux_configuration/tests/test_usage_report_aggregate.py @@ -0,0 +1,111 @@ +"""Tests for aggregate_atop, usage_report's pure-Python fallback parser used +when the native atop_agg helper is unavailable. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import _usage_report_parsing as parsing +from _usage_report_types import _Progress + +if TYPE_CHECKING: + import pytest + + +def _progress() -> _Progress: + """A progress reporter that never draws, so stderr stays clean.""" + return _Progress(enabled=False, total_stages=1) + + +# --------------------------------------------------------------------------- # +# aggregate_atop (pure-Python fallback) +# --------------------------------------------------------------------------- # +def test_aggregate_atop_parses_prc_and_prm( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """With no native helper, the Python parser folds PRC and PRM records.""" + log = tmp_path / "atop.log" + log.write_bytes(b"x" * 100) + lines = [ + "# a comment", + "RESET", + "SEP", + "", + "PRC host 1000 d t 600 42 (python) S 100 500 20 a b c", + "PRC host 1600 d t 600 42 (python) S 100 900 40 a b c", + "PRM host 1000 d t 600 42 (python) S 4096 9999 2048 a b c", + "PRC short row", + ] + monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) + monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter(lines)) + + agg, window = parsing.aggregate_atop(log, _progress()) + + # utime+stime deltas: (900+40) - (500+20) = 420. + assert agg["python"].cpu_ticks == 420 + assert agg["python"].peak_rss_kb == 2048 + assert window.distinct_samples == 2 + assert window.interval_s == 600 + + +def test_aggregate_atop_ignores_unparseable_epoch( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A PRC row with a non-numeric epoch still parses, minus the timestamp.""" + log = tmp_path / "atop.log" + log.write_bytes(b"x" * 10) + lines = ["PRC host notanepoch d t 600 42 (python) S 100 500 20 a b c"] + monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) + monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter(lines)) + + agg, window = parsing.aggregate_atop(log, _progress()) + + assert agg["python"].pid_count == 1 + assert window.distinct_samples == 0 + + +def test_aggregate_atop_delegates_to_native_when_available( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """When the C helper builds, the native path is used instead.""" + log = tmp_path / "atop.log" + log.write_bytes(b"x") + monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: Path("/bin/atop_agg")) + called: list[Path] = [] + + def fake_native( + log_arg: Path, + _progress_arg: _Progress, + binary: Path, + _begin: str | None = None, + _end: str | None = None, + ) -> tuple[dict[str, object], object]: + called.append(binary) + return {}, parsing._Window() + + monkeypatch.setattr(parsing, "_aggregate_atop_native", fake_native) + + parsing.aggregate_atop(log, _progress()) + + assert called == [Path("/bin/atop_agg")] + + +def test_aggregate_atop_skips_whitespace_only_rows( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A row of only whitespace splits to nothing and is skipped.""" + log = tmp_path / "atop.log" + log.write_bytes(b"x") + monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) + monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter([" "])) + + agg, window = parsing.aggregate_atop(log, _progress()) + + assert agg == {} + assert window.distinct_samples == 0 diff --git a/linux_configuration/tests/test_usage_report_atop_build.py b/linux_configuration/tests/test_usage_report_atop_build.py new file mode 100644 index 00000000..05cabb67 --- /dev/null +++ b/linux_configuration/tests/test_usage_report_atop_build.py @@ -0,0 +1,134 @@ +"""Tests for usage_report's native atop_agg helper: when the cached binary is +reused, when it is rebuilt, and every way the build can fail into the +pure-Python fallback. +""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING + +import _usage_report_atop as parsing + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + + +def test_atop_agg_binary_uses_fresh_cache( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A cached binary newer than the source is returned without rebuilding.""" + src_dir = tmp_path / "src" + src_dir.mkdir() + src_c = src_dir / "atop_agg.c" + src_c.write_text("int main(void){return 0;}", encoding="utf-8") + cache = tmp_path / "cached" + cache.write_bytes(b"binary") + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) + + def fail(*_a: object, **_k: object) -> None: + msg = "should not rebuild when the cache is fresh" + raise AssertionError(msg) + + monkeypatch.setattr(parsing.subprocess, "run", fail) + + assert parsing._atop_agg_binary() == cache + + +def test_atop_agg_binary_none_without_compiler( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """With a stale cache and no `cc`, the caller falls back to Python.""" + src_dir = tmp_path / "src" + src_dir.mkdir() + (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "absent") + monkeypatch.setattr(parsing.shutil, "which", lambda _name: None) + + assert parsing._atop_agg_binary() is None + + +def test_atop_agg_binary_none_when_build_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A failing `make` falls back rather than propagating the error.""" + src_dir = tmp_path / "src" + src_dir.mkdir() + (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "cache" / "bin") + monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") + + def boom(*_a: object, **_k: object) -> None: + raise subprocess.CalledProcessError(1, "make") + + monkeypatch.setattr(parsing.subprocess, "run", boom) + + assert parsing._atop_agg_binary() is None + + +def test_atop_agg_binary_none_when_make_produces_nothing( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A `make` that exits 0 without writing the binary still falls back.""" + src_dir = tmp_path / "src" + src_dir.mkdir() + (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "cache" / "bin") + monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") + monkeypatch.setattr( + parsing.subprocess, + "run", + lambda *_a, **_k: subprocess.CompletedProcess([], 0), + ) + + assert parsing._atop_agg_binary() is None + + +def test_atop_agg_binary_builds_and_caches( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A successful build is copied into the cache and made executable.""" + src_dir = tmp_path / "src" + src_dir.mkdir() + (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") + cache = tmp_path / "cache" / "atop_agg" + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) + monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") + + def fake_make(*_a: object, **_k: object) -> subprocess.CompletedProcess[bytes]: + (src_dir / "atop_agg").write_bytes(b"ELF") + return subprocess.CompletedProcess([], 0) + + monkeypatch.setattr(parsing.subprocess, "run", fake_make) + + result = parsing._atop_agg_binary() + + assert result == cache + assert cache.read_bytes() == b"ELF" + assert cache.stat().st_mode & 0o777 == parsing._ATOP_AGG_BIN_MODE + + +def test_atop_agg_binary_missing_source_falls_back( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A deleted C source tree yields None (Python fallback) even when a cached + binary exists - never trust an orphaned, unverifiable build.""" + monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", tmp_path / "gone") + cache = tmp_path / "atop_agg" + cache.write_text("stale binary", encoding="utf-8") + monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) + + assert parsing._atop_agg_binary() is None diff --git a/linux_configuration/tests/test_usage_report_atop_io.py b/linux_configuration/tests/test_usage_report_atop_io.py index 26b578e9..d9182a2c 100644 --- a/linux_configuration/tests/test_usage_report_atop_io.py +++ b/linux_configuration/tests/test_usage_report_atop_io.py @@ -9,7 +9,7 @@ from pathlib import Path import subprocess from typing import TYPE_CHECKING -import _usage_report_parsing as parsing +import _usage_report_atop as parsing from _usage_report_types import ProcAgg, _Progress from typing_extensions import Self @@ -101,108 +101,6 @@ def test_iter_atop_lines_handles_no_stdout(monkeypatch: pytest.MonkeyPatch) -> N # --------------------------------------------------------------------------- # # _atop_agg_binary # --------------------------------------------------------------------------- # -def test_atop_agg_binary_uses_fresh_cache( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A cached binary newer than the source is returned without rebuilding.""" - src_dir = tmp_path / "src" - src_dir.mkdir() - src_c = src_dir / "atop_agg.c" - src_c.write_text("int main(void){return 0;}", encoding="utf-8") - cache = tmp_path / "cached" - cache.write_bytes(b"binary") - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) - - def fail(*_a: object, **_k: object) -> None: - msg = "should not rebuild when the cache is fresh" - raise AssertionError(msg) - - monkeypatch.setattr(parsing.subprocess, "run", fail) - - assert parsing._atop_agg_binary() == cache - - -def test_atop_agg_binary_none_without_compiler( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """With a stale cache and no `cc`, the caller falls back to Python.""" - src_dir = tmp_path / "src" - src_dir.mkdir() - (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "absent") - monkeypatch.setattr(parsing.shutil, "which", lambda _name: None) - - assert parsing._atop_agg_binary() is None - - -def test_atop_agg_binary_none_when_build_fails( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A failing `make` falls back rather than propagating the error.""" - src_dir = tmp_path / "src" - src_dir.mkdir() - (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "cache" / "bin") - monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") - - def boom(*_a: object, **_k: object) -> None: - raise subprocess.CalledProcessError(1, "make") - - monkeypatch.setattr(parsing.subprocess, "run", boom) - - assert parsing._atop_agg_binary() is None - - -def test_atop_agg_binary_none_when_make_produces_nothing( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A `make` that exits 0 without writing the binary still falls back.""" - src_dir = tmp_path / "src" - src_dir.mkdir() - (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", tmp_path / "cache" / "bin") - monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") - monkeypatch.setattr( - parsing.subprocess, - "run", - lambda *_a, **_k: subprocess.CompletedProcess([], 0), - ) - - assert parsing._atop_agg_binary() is None - - -def test_atop_agg_binary_builds_and_caches( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A successful build is copied into the cache and made executable.""" - src_dir = tmp_path / "src" - src_dir.mkdir() - (src_dir / "atop_agg.c").write_text("x", encoding="utf-8") - cache = tmp_path / "cache" / "atop_agg" - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", src_dir) - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) - monkeypatch.setattr(parsing.shutil, "which", lambda _name: "/usr/bin/cc") - - def fake_make(*_a: object, **_k: object) -> subprocess.CompletedProcess[bytes]: - (src_dir / "atop_agg").write_bytes(b"ELF") - return subprocess.CompletedProcess([], 0) - - monkeypatch.setattr(parsing.subprocess, "run", fake_make) - - result = parsing._atop_agg_binary() - - assert result == cache - assert cache.read_bytes() == b"ELF" - assert cache.stat().st_mode & 0o777 == parsing._ATOP_AGG_BIN_MODE # --------------------------------------------------------------------------- # @@ -242,118 +140,3 @@ def test_window_from_native_spans_the_reported_epochs() -> None: assert window.seconds == 3600 assert window.start_epoch == 1000 assert window.end_epoch == 4600 - - -# --------------------------------------------------------------------------- # -# _aggregate_atop_native -# --------------------------------------------------------------------------- # -class _ClosableLines(list[str]): - """A line list that also answers .close(), as a real pipe would.""" - - def close(self) -> None: - """Called on the producer's stdout by the pipeline code.""" - - -class _FakePipeline: - """Two-process context manager mimicking the atop | atop_agg pipeline.""" - - def __init__(self, lines: list[str] | None) -> None: - self.stdout = None if lines is None else _ClosableLines(lines) - - def __enter__(self) -> Self: - return self - - def __exit__(self, *_exc: object) -> None: - """Never suppresses an exception.""" - - -def _fake_popen_factory( - rows: list[str] | None, -) -> object: - """Build a Popen replacement: first call is atop, second is atop_agg.""" - calls: list[int] = [] - - def fake_popen(*_a: object, **_k: object) -> _FakePipeline: - calls.append(1) - # The first Popen is atop (its stdout is only closed by the caller); - # the second is atop_agg, whose stdout the loop reads. - return _FakePipeline(["x"] if len(calls) == 1 else rows) - - return fake_popen - - -def test_aggregate_atop_native_reads_n_and_w_rows( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """N rows become aggregates and a W row becomes the window.""" - rows = [ - "N\tfirefox\t900\t4096\t2048\t5\t2\n", - "W\t100\t3700\t6\t600\n", - ] - monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(rows)) - - agg, window = parsing._aggregate_atop_native( - Path("log"), - _progress(), - Path("/bin/atop_agg"), - ) - - assert agg["firefox"].cpu_ticks == 900 - assert window.distinct_samples == 6 - assert window.interval_s == 600 - - -def test_aggregate_atop_native_ignores_malformed_rows( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Rows with the wrong field count are skipped, not fatal.""" - rows = ["N\ttoo\tshort\n", "W\tbad\n", "Z\tunknown\ttag\n"] - monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(rows)) - - agg, window = parsing._aggregate_atop_native( - Path("log"), - _progress(), - Path("/bin/atop_agg"), - ) - - assert agg == {} - assert window.distinct_samples == 0 - - -def test_aggregate_atop_native_handles_no_stdout( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """A helper with no stdout pipe returns empty results rather than raising.""" - monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(None)) - - agg, window = parsing._aggregate_atop_native( - Path("log"), - _progress(), - Path("/bin/atop_agg"), - ) - - assert agg == {} - assert window.distinct_samples == 0 - - -def test_aggregate_atop_native_tolerates_producer_without_stdout( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """When atop exposes no stdout there is nothing to close; the read proceeds.""" - calls: list[int] = [] - - def fake_popen(*_a: object, **_k: object) -> _FakePipeline: - calls.append(1) - # First Popen is atop: give it no stdout, so the close() is skipped. - return _FakePipeline(None if len(calls) == 1 else ["W\t0\t0\t0\t0\n"]) - - monkeypatch.setattr(parsing.subprocess, "Popen", fake_popen) - - agg, window = parsing._aggregate_atop_native( - Path("log"), - _progress(), - Path("/bin/atop_agg"), - ) - - assert agg == {} - assert window.distinct_samples == 0 diff --git a/linux_configuration/tests/test_usage_report_atop_pipeline.py b/linux_configuration/tests/test_usage_report_atop_pipeline.py new file mode 100644 index 00000000..b86f15c4 --- /dev/null +++ b/linux_configuration/tests/test_usage_report_atop_pipeline.py @@ -0,0 +1,135 @@ +"""Tests for the `atop | atop_agg` pipeline: how usage_report reads the native +helper's TSV rows and copes with a producer or consumer that has no stdout. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import _usage_report_atop as parsing +from _usage_report_types import _Progress +from typing_extensions import Self + +if TYPE_CHECKING: + import pytest + + +def _progress() -> _Progress: + """A progress reporter that never draws, so stderr stays clean.""" + return _Progress(enabled=False, total_stages=1) + + +# --------------------------------------------------------------------------- # +# _aggregate_atop_native +# --------------------------------------------------------------------------- # +class _ClosableLines(list[str]): + """A line list that also answers .close(), as a real pipe would.""" + + def close(self) -> None: + """Called on the producer's stdout by the pipeline code.""" + + +class _FakePipeline: + """Two-process context manager mimicking the atop | atop_agg pipeline.""" + + def __init__(self, lines: list[str] | None) -> None: + self.stdout = None if lines is None else _ClosableLines(lines) + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_exc: object) -> None: + """Never suppresses an exception.""" + + +def _fake_popen_factory( + rows: list[str] | None, +) -> object: + """Build a Popen replacement: first call is atop, second is atop_agg.""" + calls: list[int] = [] + + def fake_popen(*_a: object, **_k: object) -> _FakePipeline: + calls.append(1) + # The first Popen is atop (its stdout is only closed by the caller); + # the second is atop_agg, whose stdout the loop reads. + return _FakePipeline(["x"] if len(calls) == 1 else rows) + + return fake_popen + + +def test_aggregate_atop_native_reads_n_and_w_rows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """N rows become aggregates and a W row becomes the window.""" + rows = [ + "N\tfirefox\t900\t4096\t2048\t5\t2\n", + "W\t100\t3700\t6\t600\n", + ] + monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(rows)) + + agg, window = parsing._aggregate_atop_native( + Path("log"), + _progress(), + Path("/bin/atop_agg"), + ) + + assert agg["firefox"].cpu_ticks == 900 + assert window.distinct_samples == 6 + assert window.interval_s == 600 + + +def test_aggregate_atop_native_ignores_malformed_rows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Rows with the wrong field count are skipped, not fatal.""" + rows = ["N\ttoo\tshort\n", "W\tbad\n", "Z\tunknown\ttag\n"] + monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(rows)) + + agg, window = parsing._aggregate_atop_native( + Path("log"), + _progress(), + Path("/bin/atop_agg"), + ) + + assert agg == {} + assert window.distinct_samples == 0 + + +def test_aggregate_atop_native_handles_no_stdout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A helper with no stdout pipe returns empty results rather than raising.""" + monkeypatch.setattr(parsing.subprocess, "Popen", _fake_popen_factory(None)) + + agg, window = parsing._aggregate_atop_native( + Path("log"), + _progress(), + Path("/bin/atop_agg"), + ) + + assert agg == {} + assert window.distinct_samples == 0 + + +def test_aggregate_atop_native_tolerates_producer_without_stdout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When atop exposes no stdout there is nothing to close; the read proceeds.""" + calls: list[int] = [] + + def fake_popen(*_a: object, **_k: object) -> _FakePipeline: + calls.append(1) + # First Popen is atop: give it no stdout, so the close() is skipped. + return _FakePipeline(None if len(calls) == 1 else ["W\t0\t0\t0\t0\n"]) + + monkeypatch.setattr(parsing.subprocess, "Popen", fake_popen) + + agg, window = parsing._aggregate_atop_native( + Path("log"), + _progress(), + Path("/bin/atop_agg"), + ) + + assert agg == {} + assert window.distinct_samples == 0 diff --git a/linux_configuration/tests/test_usage_report_modes.py b/linux_configuration/tests/test_usage_report_modes.py index 54577b9c..e2fbc51b 100644 --- a/linux_configuration/tests/test_usage_report_modes.py +++ b/linux_configuration/tests/test_usage_report_modes.py @@ -165,16 +165,3 @@ def test_parse_prc_too_short_is_ignored() -> None: # --------------------------------------------------------------------------- # # Native helper selection (parsing) -# --------------------------------------------------------------------------- # -def test_atop_agg_binary_missing_source_falls_back( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """A deleted C source tree yields None (Python fallback) even when a cached - binary exists — never trust an orphaned, unverifiable build.""" - monkeypatch.setattr(parsing, "_ATOP_AGG_SRC_DIR", tmp_path / "gone") - cache = tmp_path / "atop_agg" - cache.write_text("stale binary", encoding="utf-8") - monkeypatch.setattr(parsing, "_ATOP_AGG_CACHE_BIN", cache) - - assert parsing._atop_agg_binary() is None diff --git a/linux_configuration/tests/test_usage_report_parse_records.py b/linux_configuration/tests/test_usage_report_parse_records.py index 262819d1..0a3757ec 100644 --- a/linux_configuration/tests/test_usage_report_parse_records.py +++ b/linux_configuration/tests/test_usage_report_parse_records.py @@ -5,15 +5,9 @@ aggregate_atop fallback used when the native helper is unavailable. from __future__ import annotations -from pathlib import Path -from typing import TYPE_CHECKING - import _usage_report_parsing as parsing from _usage_report_types import _PidCpu, _PidRam, _Progress -if TYPE_CHECKING: - import pytest - def _progress() -> _Progress: """A progress reporter that never draws, so stderr stays clean.""" @@ -168,82 +162,6 @@ def test_fold_pid_aggregates_takes_peak_and_counts_rss_samples() -> None: assert agg["firefox"].pid_count == 2 -# --------------------------------------------------------------------------- # -# aggregate_atop (pure-Python fallback) -# --------------------------------------------------------------------------- # -def test_aggregate_atop_parses_prc_and_prm( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """With no native helper, the Python parser folds PRC and PRM records.""" - log = tmp_path / "atop.log" - log.write_bytes(b"x" * 100) - lines = [ - "# a comment", - "RESET", - "SEP", - "", - "PRC host 1000 d t 600 42 (python) S 100 500 20 a b c", - "PRC host 1600 d t 600 42 (python) S 100 900 40 a b c", - "PRM host 1000 d t 600 42 (python) S 4096 9999 2048 a b c", - "PRC short row", - ] - monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) - monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter(lines)) - - agg, window = parsing.aggregate_atop(log, _progress()) - - # utime+stime deltas: (900+40) - (500+20) = 420. - assert agg["python"].cpu_ticks == 420 - assert agg["python"].peak_rss_kb == 2048 - assert window.distinct_samples == 2 - assert window.interval_s == 600 - - -def test_aggregate_atop_ignores_unparseable_epoch( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A PRC row with a non-numeric epoch still parses, minus the timestamp.""" - log = tmp_path / "atop.log" - log.write_bytes(b"x" * 10) - lines = ["PRC host notanepoch d t 600 42 (python) S 100 500 20 a b c"] - monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) - monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter(lines)) - - agg, window = parsing.aggregate_atop(log, _progress()) - - assert agg["python"].pid_count == 1 - assert window.distinct_samples == 0 - - -def test_aggregate_atop_delegates_to_native_when_available( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """When the C helper builds, the native path is used instead.""" - log = tmp_path / "atop.log" - log.write_bytes(b"x") - monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: Path("/bin/atop_agg")) - called: list[Path] = [] - - def fake_native( - log_arg: Path, - _progress_arg: _Progress, - binary: Path, - _begin: str | None = None, - _end: str | None = None, - ) -> tuple[dict[str, object], object]: - called.append(binary) - return {}, parsing._Window() - - monkeypatch.setattr(parsing, "_aggregate_atop_native", fake_native) - - parsing.aggregate_atop(log, _progress()) - - assert called == [Path("/bin/atop_agg")] - - def test_parse_prc_non_numeric_pid_is_ignored() -> None: """A PRC row whose PID will not parse is dropped rather than raising.""" cpu: dict[int, _PidCpu] = {} @@ -254,19 +172,3 @@ def test_parse_prc_non_numeric_pid_is_ignored() -> None: ) assert cpu == {} - - -def test_aggregate_atop_skips_whitespace_only_rows( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """A row of only whitespace splits to nothing and is skipped.""" - log = tmp_path / "atop.log" - log.write_bytes(b"x") - monkeypatch.setattr(parsing, "_atop_agg_binary", lambda: None) - monkeypatch.setattr(parsing, "_iter_atop_lines", lambda *_a, **_k: iter([" "])) - - agg, window = parsing.aggregate_atop(log, _progress()) - - assert agg == {} - assert window.distinct_samples == 0 diff --git a/linux_configuration/tests/test_usage_report_since.py b/linux_configuration/tests/test_usage_report_since.py index 59fee850..a8131e08 100644 --- a/linux_configuration/tests/test_usage_report_since.py +++ b/linux_configuration/tests/test_usage_report_since.py @@ -9,7 +9,7 @@ import datetime as _dt from pathlib import Path from typing import TYPE_CHECKING -import _usage_report_parsing as parsing +import _usage_report_atop as atop_io import _usage_report_pmon as pmon from _usage_report_types import _Progress import usage_report @@ -88,7 +88,7 @@ def test_aggregate_pmon_filters_rows_before_begin(tmp_path: Path) -> None: # --------------------------------------------------------------------------- # def test_atop_read_cmd_unbounded() -> None: """Without bounds the command is a plain replay.""" - cmd = parsing._atop_read_cmd( + cmd = atop_io._atop_read_cmd( Path("/var/log/atop/atop_20260604"), "PRC,PRM", None, None ) @@ -97,7 +97,7 @@ def test_atop_read_cmd_unbounded() -> None: def test_atop_read_cmd_with_begin_and_end() -> None: """Begin/end inject -b/-e before the -P selector.""" - cmd = parsing._atop_read_cmd(Path("/x"), "PRC", "202606041400", "202606042000") + cmd = atop_io._atop_read_cmd(Path("/x"), "PRC", "202606041400", "202606042000") assert cmd == [ "atop",