42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
import json
|
|
from perfkit import results
|
|
from perfkit.aggregate import summarize, render_md
|
|
|
|
|
|
def rec(**kw):
|
|
r = {"run_id": "r1", "scenario": "s", "target": "t", "role": "admin", "scale": 1,
|
|
"cpu_rate": 1, "repeat": 0, "metrics": {}, "artifacts": [], "incomplete": False, "ts": "x"}
|
|
r.update(kw)
|
|
return r
|
|
|
|
|
|
def test_summarize_median_p95(tmp_path):
|
|
for v in [10, 20, 30, 40, 100]:
|
|
results.append(tmp_path, rec(metrics={"long_task_ms": v}))
|
|
g = summarize(tmp_path)["groups"][0]
|
|
m = g["metrics"]["long_task_ms"]
|
|
assert (m["median"], m["p95"], m["n"], m["min"], m["max"]) == (30, 100, 5, 10, 100)
|
|
|
|
|
|
def test_summarize_groups_by_cpu_rate(tmp_path):
|
|
results.append(tmp_path, rec(cpu_rate=1, metrics={"a": 1}))
|
|
results.append(tmp_path, rec(cpu_rate=6, metrics={"a": 2}))
|
|
assert len(summarize(tmp_path)["groups"]) == 2
|
|
|
|
|
|
def test_incomplete_records_counted_separately(tmp_path):
|
|
results.append(tmp_path, rec(metrics={"a": 1}))
|
|
results.append(tmp_path, rec(metrics={"a": 999}, incomplete=True))
|
|
g = summarize(tmp_path)["groups"][0]
|
|
assert g["incomplete_n"] == 1
|
|
assert g["metrics"]["a"]["max"] == 1 and g["metrics"]["a"]["n"] == 1
|
|
|
|
|
|
def test_append_path_and_md(tmp_path):
|
|
results.append(tmp_path, rec(metrics={"a": 1.5, "flag": True, "s": "x"}))
|
|
assert (tmp_path / "s.jsonl").exists()
|
|
s = summarize(tmp_path)
|
|
assert "flag" not in s["groups"][0]["metrics"]
|
|
md = render_md(s)
|
|
assert "| a |" in md and "median" in md
|