aio-01d4-stu/Last3Mon/72/tbench/tests/test_engine.py

import json
import socket
import time

import pytest

from tbench import cli, profiles, store
from tbench.runner import run
from tbench.stats import check, percentile, summarize


def free_port():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


# R1 · nearest-rank fixed by a test
def test_percentile_nearest_rank():
    values = list(range(1, 11))
    assert percentile(values, 0.5) == 5
    assert percentile(values, 0.95) == 10
    assert percentile([3.0], 0.95) == 3.0
    assert percentile([], 0.5) is None


def test_failures_are_counted_not_timed(server, body_file):
    results = list(run(server.base + "?delay=0.1&fail_every=3", body_file.read_text(), n=6))
    s = summarize(results, 6)
    assert s["failures"] == 2 and s["by_kind"]["http_5xx"] == 2
    assert s["ok"] == 4 and len(results) == 6


def test_body_is_sent(server, body_file):
    list(run(server.base, body_file.read_text(), n=1))
    assert json.loads(server.bodies[0])["question"].startswith("VPN")


# R6
def test_first_call_and_warmup(server):
    results = list(run(server.base + "?first_delay=1&delay=0.05", None, n=4, method="GET"))
    assert results[0]["note"] == "first call"
    s = summarize(results, 4)
    assert s["max"] >= 1 and s["p50_excl_first"] < 0.5
    server.count = 0
    results = list(run(server.base + "?first_delay=1&delay=0.05", None, n=4, method="GET", warmup=1))
    measured = [r for r in results if not r["warmup"]]
    assert len(measured) == 4 and max(r["seconds"] for r in measured) < 0.5
    assert all(r["note"] != "first call" for r in measured)


# R7
def test_error_kinds(server):
    assert list(run(f"http://127.0.0.1:{free_port()}/x", None, n=1))[0]["error_kind"] == "connect"
    assert list(run(server.base + "?delay=3", None, n=1, timeout=1))[0]["error_kind"] == "timeout"
    assert list(run(server.base + "?status=500", None, n=1))[0]["error_kind"] == "http_5xx"
    assert list(run(server.base + "?status=404", None, n=1))[0]["error_kind"] == "http_4xx"


# R8 · R3 · R11 · R13
def test_save_masks_keys_compare_warns_history_and_log(server, body_file, tmp_path, capsys):
    out_a, out_b, log = tmp_path / "a.json", tmp_path / "b.json", tmp_path / "runs.jsonl"
    url = server.base + "?delay=0.05"
    cli.execute(cli.settings_from(type("A", (), {"url": url, "body_file": str(body_file), "n": 3,
                "header": ["Authorization: Bearer sk-test", "X-Api-Key: sk-test2"]})()),
                True, str(out_a), str(log))
    other = tmp_path / "other.json"
    other.write_text('{"question": "점심 메뉴"}')
    cli.execute(cli.settings_from(type("A", (), {"url": url, "body_file": str(other), "n": 3})()), True, str(out_b), None)
    assert "sk-test" not in out_a.read_text() and "sk-test" not in log.read_text()
    assert json.loads(out_a.read_text())["headers"]["Authorization"] == "***"
    assert len(log.read_text().splitlines()) == 3
    lines = store.compare(store.load(out_a), store.load(out_b))
    assert any(line.startswith("WARNING: different body_sha256") for line in lines)
    assert not any("WARNING" in line for line in store.compare(store.load(out_a), store.load(out_a)))

    cli.execute(cli.settings_from(type("A", (), {"url": url, "n": 2})()), True, "", None)
    time.sleep(1.1)
    cli.execute(cli.settings_from(type("A", (), {"url": url, "n": 4})()), True, "", None)
    rows = store.history()
    assert [r[2] for r in rows] == [4, 2]


# R4
def test_concurrency_is_faster(server):
    t = time.perf_counter()
    list(run(server.base + "?delay=0.3", None, n=6, concurrency=1))
    one = time.perf_counter() - t
    t = time.perf_counter()
    results = list(run(server.base + "?delay=0.3", None, n=6, concurrency=3))
    three = time.perf_counter() - t
    assert one > 1.7 and three < 1.0
    assert sorted(r["i"] for r in results) == [1, 2, 3, 4, 5, 6]


# R10
def test_limits_and_exit_code(server):
    s = summarize(list(run(server.base + "?delay=0.2", None, n=3)), 3)
    assert check(s, max_p95=0.1) == (False, f"FAIL p95 {s['p95']:.3f} > 0.1")
    assert check(s, max_p95=5, max_failures=0) == (True, "PASS")
    with pytest.raises(SystemExit) as e:
        cli.main([server.base + "?status=500", "-n", "2", "--plain", "--max-failures", "0"])
    assert e.value.code == 1
    with pytest.raises(SystemExit) as e:
        cli.main([server.base, "-n", "2", "--plain", "--max-p95", "5"])
    assert e.value.code == 0


# R9
def test_timings_average(server):
    s = summarize(list(run(server.base + "?timings=1", None, n=2)), 2)
    assert s["timings"] == {"classify_ms": 150.0, "answer_ms": 300.0, "total_ms": 550.0}
    assert summarize(list(run(server.base, None, n=1)), 1)["timings"] == {}


# R12
def test_profile_save_and_run_from_another_folder(server, body_file, tmp_path, monkeypatch, capsys):
    with pytest.raises(SystemExit):
        cli.main(["profile", "save", "vpn", server.base, "--body-file", str(body_file), "-n", "3",
                  "--header", "Authorization: Bearer sk-live"])
    assert "sk-live" not in profiles.path().read_text()
    monkeypatch.chdir(tmp_path)
    with pytest.raises(SystemExit) as e:
        cli.main(["run", "vpn", "--plain"])
    assert e.value.code == 0
    out = capsys.readouterr().out
    assert "done 3/3" in out and server.bodies