fdc8ecb2a3
Сетевой лог сценариев, bench запросов, pg_stat_statements и EXPLAIN, aggregate исключает трассировочные повторы (O-9). Refs #202 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
from perfkit.bench import path_template, select_requests, record_for, parse_target
|
|
|
|
|
|
def e(method="GET", url="http://localhost:8000/api/v1/form/1/sheet/AHR", role="rf", status=200, **kw):
|
|
d = {"method": method, "url": url, "role": role, "status": status}
|
|
d.update(kw)
|
|
return d
|
|
|
|
|
|
def test_path_template():
|
|
assert path_template("/api/v1/form/123/sheet/AHR/cells") == "/api/v1/form/{id}/sheet/AHR/cells"
|
|
|
|
|
|
def test_path_template_ignores_query_and_digits_inside_names():
|
|
assert path_template("/api/v1/form/5/sheet/AHR_2?year=2026") == "/api/v1/form/{id}/sheet/AHR_2"
|
|
assert path_template("/api/v1/projects/7/report/2026/PLAN") == "/api/v1/projects/{id}/report/{id}/PLAN"
|
|
|
|
|
|
def test_select_requests_skips_non_idempotent():
|
|
got = select_requests([
|
|
e("GET"),
|
|
e("POST", "http://localhost:8000/api/v1/form/1/sheet/AHR/line", post_data='{"x":1}'),
|
|
e("DELETE", "http://localhost:8000/api/v1/form/1/sheet/AHR/line/9"),
|
|
])
|
|
assert [g["method"] for g in got] == ["GET"]
|
|
|
|
|
|
def test_select_requests_dedups_by_role_method_url():
|
|
got = select_requests([e(), e(), e(role="admin"), e(url="http://localhost:8000/api/v1/form/1/sheet/AHR?direction=x")])
|
|
assert len(got) == 3
|
|
|
|
|
|
def test_select_requests_skips_errors_and_auth():
|
|
got = select_requests([e(status=404), e(status=500), e(url="http://localhost:8000/api/v1/auth/login"),
|
|
e(url="http://localhost:8000/assets/app.js")])
|
|
assert got == []
|
|
|
|
|
|
def test_select_requests_patch_only_when_proven_idempotent():
|
|
p = "http://localhost:8000/api/v1/form/1/sheet/AHR/cell"
|
|
assert select_requests([e("PATCH", p, post_data='{"line_id":1,"column":"c","value":5}')]) == []
|
|
got = select_requests([e("PATCH", p, post_data='{"line_id":1,"column":"c","value":5}', idempotent=True)])
|
|
assert len(got) == 1
|
|
|
|
|
|
def test_select_requests_skips_slow_logged():
|
|
got = select_requests([e(duration_ms=120_000)], max_logged_ms=60_000)
|
|
assert got == []
|
|
|
|
|
|
def test_record_for_shape():
|
|
r = record_for(e(), repeat=3, ms=12.5, nbytes=100, run_id="r1", scale=1)
|
|
assert r["scenario"] == "api" and r["target"] == "GET /api/v1/form/{id}/sheet/AHR"
|
|
assert r["role"] == "rf" and r["repeat"] == 3
|
|
assert r["metrics"] == {"ms": 12.5, "bytes": 100}
|
|
assert r["incomplete"] is False and r["cpu_rate"] == 1
|
|
|
|
|
|
def test_parse_target():
|
|
assert parse_target("GET /api/v1/x/{id}") == ("GET", "/api/v1/x/{id}")
|