71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
import json
|
||
import pytest
|
||
|
||
from perfkit.memtrend import slope_per_cycle, main
|
||
|
||
|
||
def test_slope_linear():
|
||
assert slope_per_cycle([10, 11, 12, 13]) == pytest.approx(1.0)
|
||
|
||
|
||
def test_slope_flat():
|
||
assert slope_per_cycle([5, 5, 5]) == 0.0
|
||
|
||
|
||
def test_slope_noisy_is_least_squares():
|
||
# y = 2x + шум; МНК, а не (last-first)/n
|
||
assert slope_per_cycle([0, 3, 3, 7, 8]) == pytest.approx(2.0)
|
||
|
||
|
||
def test_slope_needs_two_points():
|
||
with pytest.raises(ValueError):
|
||
slope_per_cycle([1.0])
|
||
with pytest.raises(ValueError):
|
||
slope_per_cycle([])
|
||
|
||
|
||
def test_cli_prints_three_slopes(tmp_path, capsys):
|
||
rows = [
|
||
{"scenario": "memory", "repeat": 0, "metrics": {"cycle": c, "heap_used_mb": 100 + 2 * c, "nodes": 1000 + 5 * c, "js_event_listeners": 50}}
|
||
for c in range(4)
|
||
]
|
||
# записи могут идти не по порядку и вперемешку с чужим сценарием
|
||
lines = [json.dumps(r) for r in reversed(rows)] + [json.dumps({"scenario": "other", "metrics": {"cycle": 0}})]
|
||
(tmp_path / "memory.jsonl").write_text("\n".join(lines) + "\n")
|
||
assert main([str(tmp_path)]) == 0
|
||
out = capsys.readouterr().out
|
||
assert "heap_used_mb" in out and "nodes" in out and "js_event_listeners" in out
|
||
assert "2.000" in out and "5.000" in out and "0.000" in out
|
||
|
||
|
||
def _write(tmp_path, heap):
|
||
rows = [{"scenario": "memory", "metrics": {"cycle": c, "heap_used_mb": h, "nodes": 350 if c else 289, "js_event_listeners": 336 if c else 322}} for c, h in enumerate(heap)]
|
||
(tmp_path / "memory.jsonl").write_text("\n".join(json.dumps(r) for r in rows) + "\n")
|
||
|
||
|
||
def test_cli_skips_baseline_step_by_default(tmp_path, capsys):
|
||
# ступенька на цикле 0 (7 -> 11), дальше +1 на цикл, 0 для nodes/listeners
|
||
_write(tmp_path, [7, 11, 12, 13, 14, 15, 16])
|
||
assert main([str(tmp_path)]) == 0
|
||
out = capsys.readouterr().out
|
||
assert "циклы 1-6" in out and "6 точек" in out
|
||
assert "heap_used_mb: 1.000" in out
|
||
assert "nodes: 0.000" in out and "js_event_listeners: 0.000" in out
|
||
|
||
|
||
def test_cli_from_cycle_zero_includes_step(tmp_path, capsys):
|
||
_write(tmp_path, [7, 11, 12, 13, 14, 15, 16])
|
||
assert main([str(tmp_path), "--from-cycle", "0"]) == 0
|
||
out = capsys.readouterr().out
|
||
assert "циклы 0-6" in out and "7 точек" in out
|
||
assert "heap_used_mb: 1.000" not in out.split("вторая половина")[0]
|
||
|
||
|
||
def test_cli_second_half_slope(tmp_path, capsys):
|
||
# рост до цикла 10, потом плато: вторая половина (циклы 11-20) = 0
|
||
_write(tmp_path, [7] + [11 + min(c, 10) for c in range(0, 20)])
|
||
assert main([str(tmp_path)]) == 0
|
||
out = capsys.readouterr().out
|
||
assert "вторая половина (циклы 11-20)" in out
|
||
assert "heap_used_mb: 0.000" in out.split("вторая половина")[1]
|