tests/test_trace_replay.py

from inferscale import run_simulation


def test_trace_replay_preserves_arrivals_and_lengths():
    trace = [
        {"arrival_time": 0.4, "prompt_tokens": 96, "output_tokens": 8},
        {"arrival_time": 0.1, "prompt_tokens": 64, "output_tokens": 6},
        {"arrival_time": 0.9, "prompt_tokens": 128, "output_tokens": 10},
    ]
    result = run_simulation({
        "model": "Qwen2.5-3B",
        "accelerator": "L4",
        "quantization": "int8",
        "arrival_process": "trace",
        "trace_requests": trace,
        "duration_s": 2,
        "slo_ttft_ms": 5000,
        "slo_e2e_ms": 10000,
    })
    rows = result["requests"]
    assert result["summary"]["requests_generated"] == 3
    assert [row["arrival_time"] for row in rows] == [0.1, 0.4, 0.9]
    assert [row["prompt_tokens"] for row in rows] == [64, 96, 128]


def test_trace_replay_rejects_bad_rows():
    try:
        run_simulation({"arrival_process": "trace", "trace_requests": [{"arrival_time": 0.0}]})
    except ValueError as exc:
        assert "Trace row" in str(exc)
    else:
        raise AssertionError("bad trace should fail")