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

usage_report: cover _usage_report_parsing.py 39% -> 100%

Browse Source 
36 tests, no production change - the module is byte-identical to HEAD.
Lands separately from the split that follows, because tests written in
the same commit as a split are written against the post-split shape.

This one mattered more than the last: the uncovered 61% was almost
exactly the code a split would extract - the subprocess plumbing and the
native atop_agg path. Splitting that first would have proved nothing.

_atop_agg_binary had no tests at all; all five outcomes are now covered
(fresh cache, no compiler, make raising, make exiting 0 with no binary,
successful build + chmod). The fresh-cache test asserts by installing a
subprocess.run that fails if called. Every subprocess is faked - no test
spawns atop, make or cc.

Also covers _run's two error paths, _iter_atop_lines with and without a
stdout pipe, the native N/W TSV rows including malformed ones,
_parse_name's six paths, _parse_prm, _window_from_epochs' median
interval, _fold_pid_aggregates, and aggregate_atop end to end.

219 stmts / 62 branches, 0 missed. Directory: 143 -> 179 passing. No
suppressions - two mypy exit-return findings were fixed by typing the
fakes' __exit__ as None rather than silencing them.

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:04:13 +02:00
parent 05d8c429d1
commit 40d1073eb7
3 changed files with 672 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -0,0 +1,41 @@
{
"intent": "Take _usage_report_parsing.py from 39% to 100% line and branch coverage, so the split that follows can be verified against tests written before the seam existed.",
"scope": [
"New linux_configuration/tests/test_usage_report_atop_io.py - 17 tests for the subprocess and native-helper machinery",
"New linux_configuration/tests/test_usage_report_parse_records.py - 19 tests for the pure-Python record parsing",
"No production file changed. _usage_report_parsing.py is byte-identical to HEAD.",
"Non-goal: the split itself, which is a separate commit"
],
"changes": [
"Covered the subprocess seam: _run's success path plus its OSError and TimeoutExpired fallbacks, and _iter_atop_lines both with lines and with no stdout pipe. Every subprocess is faked; no test spawns atop, make or cc.",
"Covered all five _atop_agg_binary outcomes, which were entirely untested: fresh cache reused without rebuilding, no compiler, make raising, make exiting 0 without producing a binary, and a successful build that is copied to the cache and chmodded. The fresh-cache test asserts by installing a subprocess.run that fails the test if called.",
"Covered the native TSV path: _apply_native_name, _window_from_native with and without samples, and _aggregate_atop_native reading N and W rows, skipping malformed rows, and handling a helper with no stdout. The pipeline fake needed a stdout object answering .close(), since the real code closes the producer's end.",
"Covered the pure-Python parser: _parse_name's six paths (single token, multi-token rejoin, past-end, empty parens, bare token, unterminated parens), _parse_prm's three, _window_from_epochs including the median-interval calculation, _fold_pid_aggregates for both CPU and RAM folding, and aggregate_atop end to end including its delegation to the native path.",
"No suppression was used. Two mypy exit-return findings were fixed by giving the fake context managers' __exit__ a None return rather than silencing them."
],
"verification": [
{
"command": "pytest linux_configuration/tests/ --cov=_usage_report_parsing --cov-branch --cov-report=term-missing",
"result": "pass",
"evidence": "Before: 219 stmts, 127 miss, 62 branch, 3 BrPart, 39%, missing 53-63, 98-109, 120, 133, 140-141, 157-167, 172-181, 205-228, 233-241, 246-254, 273-305, 323-353, 361-372. After: 219 stmts, 0 miss, 62 branch, 0 BrPart, 100%, nothing missing."
},
{
"command": "pytest linux_configuration/tests/ (whole directory)",
"result": "pass",
"evidence": "179 passed, up from 143 - exactly the 36 tests added, with no collateral change to the 143 that were already there."
},
{
"command": "pre-commit run --files <both new test files>",
"result": "pass",
"evidence": "All applicable hooks pass, including ruff, ruff format, mypy and the two noqa blockers. grep for 'noqa|type: ignore' in both files returns 0."
}
],
"risks": [
"The uncovered regions before this commit were almost exactly the code a split would extract - the subprocess plumbing and the native-helper path. That is the situation the brief warns about: extracting the untested half proves nothing. Writing these tests first is what makes the next commit's split verifiable.",
"These tests fake subprocess.Popen and subprocess.run at the module seam rather than running atop. They therefore verify the parsing and control flow, not that atop's real output format matches the field indices. That risk is covered separately by running the actual program, which the split commit does."
],
"rollback": [
"git rm both test files. No production code depends on them.",
"After rollback, coverage of _usage_report_parsing.py returns to 39% and linux_configuration/tests reports 143 passing tests."
]
}

359
linux_configuration/tests/test_usage_report_atop_io.py Normal file
View File

@ -0,0 +1,359 @@
"""Tests for usage_report's atop subprocess plumbing and the native atop_agg
helper: command construction, line streaming, binary build/cache decisions and
the TSV rows the C helper emits.
"""
from __future__ import annotations
from pathlib import Path
import subprocess
from typing import TYPE_CHECKING
import _usage_report_parsing as parsing
from _usage_report_types import ProcAgg, _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)
# --------------------------------------------------------------------------- #
# _run
# --------------------------------------------------------------------------- #
def test_run_returns_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
"""A successful command's stdout is returned verbatim."""
monkeypatch.setattr(
parsing.subprocess,
"run",
lambda *_a, **_k: subprocess.CompletedProcess([], 0, stdout="out", stderr=""),
)
assert parsing._run(["true"]) == "out"
def test_run_swallows_os_error(monkeypatch: pytest.MonkeyPatch) -> None:
"""A missing binary yields an empty string rather than raising."""
def boom(*_a: object, **_k: object) -> None:
msg = "no such binary"
raise OSError(msg)
monkeypatch.setattr(parsing.subprocess, "run", boom)
assert parsing._run(["nope"]) == ""
def test_run_swallows_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
"""A command that overruns its timeout also yields an empty string."""
def boom(*_a: object, **_k: object) -> None:
raise subprocess.TimeoutExpired(cmd="slow", timeout=60)
monkeypatch.setattr(parsing.subprocess, "run", boom)
assert parsing._run(["slow"]) == ""
# --------------------------------------------------------------------------- #
# _iter_atop_lines
# --------------------------------------------------------------------------- #
class _FakePopen:
"""Minimal Popen stand-in yielding canned stdout lines."""
def __init__(self, lines: list[str] | None) -> None:
self.stdout = lines
self.stdin = None
def __enter__(self) -> Self:
return self
def __exit__(self, *_exc: object) -> None:
"""Never suppresses an exception."""
def test_iter_atop_lines_strips_newlines(monkeypatch: pytest.MonkeyPatch) -> None:
"""Each streamed line comes back without its trailing newline."""
monkeypatch.setattr(
parsing.subprocess,
"Popen",
lambda *_a, **_k: _FakePopen(["a\n", "b\n"]),
)
assert list(parsing._iter_atop_lines(Path("log"), "PRC")) == ["a", "b"]
def test_iter_atop_lines_handles_no_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
"""A process with no stdout pipe yields nothing instead of raising."""
monkeypatch.setattr(
parsing.subprocess,
"Popen",
lambda *_a, **_k: _FakePopen(None),
)
assert list(parsing._iter_atop_lines(Path("log"), "PRC")) == []
# --------------------------------------------------------------------------- #
# _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
# --------------------------------------------------------------------------- #
# _apply_native_name / _window_from_native
# --------------------------------------------------------------------------- #
def test_apply_native_name_fills_every_field() -> None:
"""An N row populates the ProcAgg, with pid_set sized to the PID count."""
agg: dict[str, ProcAgg] = {}
parsing._apply_native_name(
["N", "firefox", "1200", "8192", "4096", "10", "3"],
agg,
)
entry = agg["firefox"]
assert entry.cpu_ticks == 1200
assert entry.peak_rss_kb == 8192
assert entry.rss_kb_sum == 4096
assert entry.rss_samples == 10
assert entry.pid_count == 3
def test_window_from_native_zero_samples_is_empty() -> None:
"""A W row reporting no epochs yields the default empty window."""
window = parsing._window_from_native(["W", "0", "0", "0", "0"])
assert window.distinct_samples == 0
assert window.start == "n/a"
def test_window_from_native_spans_the_reported_epochs() -> None:
"""A populated W row carries its bounds, count and interval through."""
window = parsing._window_from_native(["W", "1000", "4600", "7", "600"])
assert window.distinct_samples == 7
assert window.interval_s == 600
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

272
linux_configuration/tests/test_usage_report_parse_records.py Normal file
View File

@ -0,0 +1,272 @@
"""Tests for usage_report's pure-Python atop record parsing: process-name
extraction, PRM folding, epoch windows, per-PID collapsing, and the
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."""
return _Progress(enabled=False, total_stages=1)
# --------------------------------------------------------------------------- #
# _parse_name
# --------------------------------------------------------------------------- #
def test_parse_name_reads_single_token() -> None:
"""The common `(name)` case returns the bare name and the next index."""
assert parsing._parse_name(["x", "(firefox)"], 1) == ("firefox", 2)
def test_parse_name_rejoins_a_name_with_spaces() -> None:
"""A parenthesised name split across tokens is rejoined."""
parts = ["x", "(Draw", "GameThread)", "S"]
assert parsing._parse_name(parts, 1) == ("Draw GameThread", 3)
def test_parse_name_past_end_is_unknown() -> None:
"""An index beyond the record yields the unknown placeholder."""
assert parsing._parse_name(["x"], 5) == ("unknown", 6)
def test_parse_name_empty_parens_is_unknown() -> None:
"""`()` carries no name, so the placeholder is used."""
assert parsing._parse_name(["x", "()"], 1) == ("unknown", 2)
def test_parse_name_unparenthesised_token_passes_through() -> None:
"""A token with no leading paren is taken as the name verbatim."""
assert parsing._parse_name(["x", "bash"], 1) == ("bash", 2)
def test_parse_name_unterminated_parens_consumes_the_rest() -> None:
"""A name whose closing paren never arrives stops at the record end."""
name, nxt = parsing._parse_name(["x", "(never", "closed"], 1)
assert name == "never close"
assert nxt == 3
# --------------------------------------------------------------------------- #
# _parse_prm
# --------------------------------------------------------------------------- #
def _prm(pid: str, name: str, rsize: str) -> list[str]:
"""Build a PRM record: 6 header fields, then pid name state pagesz vsize rsize."""
return [
"PRM",
"host",
"1000",
"d",
"t",
"600",
pid,
name,
"S",
"4096",
"9999",
rsize,
]
def test_parse_prm_records_rss() -> None:
"""A well-formed PRM row lands in the per-PID RSS map."""
ram: dict[int, _PidRam] = {}
parsing._parse_prm(_prm("42", "(firefox)", "8192"), ram)
assert ram[42].name == "firefox"
assert ram[42].peak_kb == 8192
def test_parse_prm_non_numeric_pid_is_ignored() -> None:
"""A row whose PID will not parse is dropped rather than raising."""
ram: dict[int, _PidRam] = {}
parsing._parse_prm(_prm("notapid", "(x)", "1"), ram)
assert ram == {}
def test_parse_prm_truncated_row_is_ignored() -> None:
"""A row that ends before the RSS column is dropped."""
ram: dict[int, _PidRam] = {}
parsing._parse_prm(["PRM", "host", "1000", "d", "t", "600", "42", "(x)"], ram)
assert ram == {}
# --------------------------------------------------------------------------- #
# _window_from_epochs
# --------------------------------------------------------------------------- #
def test_window_from_epochs_empty_is_default() -> None:
"""No samples yields the placeholder window."""
window = parsing._window_from_epochs(set())
assert window.distinct_samples == 0
assert window.start == "n/a"
def test_window_from_epochs_single_sample_has_no_interval() -> None:
"""One sample cannot imply an interval, so it stays zero."""
window = parsing._window_from_epochs({1000})
assert window.distinct_samples == 1
assert window.interval_s == 0
assert window.seconds == 0
def test_window_from_epochs_uses_median_interval() -> None:
"""The interval is the median gap, so one long gap does not skew it."""
window = parsing._window_from_epochs({0, 600, 1200, 6000})
assert window.distinct_samples == 4
assert window.interval_s == 600
assert window.seconds == 6000
assert window.start_epoch == 0
assert window.end_epoch == 6000
# --------------------------------------------------------------------------- #
# _fold_pid_aggregates
# --------------------------------------------------------------------------- #
def test_fold_pid_aggregates_sums_cpu_across_pids() -> None:
"""Two PIDs of one program contribute their deltas to one entry."""
cpu = {1: _PidCpu(), 2: _PidCpu()}
cpu[1].observe("python", 100)
cpu[1].observe("python", 400)
cpu[2].observe("python", 50)
agg = parsing._fold_pid_aggregates(cpu, {})
# PID 1 seen twice contributes last-first = 300; PID 2 seen once gives 50.
assert agg["python"].cpu_ticks == 350
assert agg["python"].pid_count == 2
def test_fold_pid_aggregates_takes_peak_and_counts_rss_samples() -> None:
"""Peak RSS is the max across PIDs and each PID adds one RSS sample."""
ram = {1: _PidRam(), 2: _PidRam()}
ram[1].observe("firefox", 4096)
ram[2].observe("firefox", 8192)
agg = parsing._fold_pid_aggregates({}, ram)
assert agg["firefox"].peak_rss_kb == 8192
assert agg["firefox"].rss_samples == 2
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] = {}
parsing._parse_prc(
["PRC", "host", "1000", "d", "t", "600", "notapid", "(x)", "S", "1", "2", "3"],
cpu,
)
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