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

usage_report: split _usage_report_parsing.py 425 -> 243 + 206
Some checks are pending
Pre-commit checks / pre-commit (push) Waiting to run
Details
Python tests / test (push) Waiting to run
Details
Shell tests / shell-tests (push) Waiting to run
Details
Shell tests / arch-tests (push) Waiting to run
Details

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-17 12:12:11 +02:00
parent 40d1073eb7
commit b5dbe1e67a
12 changed files with 676 additions and 524 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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 <today> for the runtime check; pre-commit run --files for lint."
}

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

@ -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 <today> --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 <all changed 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."
]
}

206
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_atop.py Normal file
View File

@ -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<name>\t<cpu>\t<peak>\t<sum_avg>\t<ram_n>\t<pids>` 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<start>\t<end>\t<n>\t<interval>` 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

200
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_parsing.py
View File

@ -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<name>\t<cpu>\t<peak>\t<sum_avg>\t<ram_n>\t<pids>` 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<start>\t<end>\t<n>\t<interval>` 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,

2
linux_configuration/scripts/periodic_background/system-maintenance/bin/_usage_report_render.py
View File

@ -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

111
linux_configuration/tests/test_usage_report_aggregate.py Normal file
View File

@ -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

134
linux_configuration/tests/test_usage_report_atop_build.py Normal file
View File

@ -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

219
linux_configuration/tests/test_usage_report_atop_io.py
View File

@ -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

135
linux_configuration/tests/test_usage_report_atop_pipeline.py Normal file
View File

@ -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

13
linux_configuration/tests/test_usage_report_modes.py
View File

@ -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

98
linux_configuration/tests/test_usage_report_parse_records.py
View File

@ -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

6
linux_configuration/tests/test_usage_report_since.py
View File

@ -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",