tests/test_simulator.py

from inferscale import run_simulation


def base_config():
    return {
        "duration_s": 8,
        "request_rate_rps": 1.5,
        "prompt_tokens_mean": 128,
        "prompt_tokens_cv": 0.1,
        "output_tokens_mean": 16,
        "output_tokens_cv": 0.1,
        "seed": 3,
    }


def test_simulation_completes_requests():
    result = run_simulation(base_config())
    assert result["summary"]["requests_generated"] > 0
    assert result["summary"]["requests_unfinished"] == 0
    assert result["latency"]["ttft_ms"]["p95"] > 0
    assert result["resource"]["peak_kv_gb"] >= 0
    assert result["provenance"]["latency_profile_type"] == "analytical-reference"


def test_high_load_increases_tail_latency():
    low = base_config()
    high = base_config()
    low["request_rate_rps"] = 0.5
    high["request_rate_rps"] = 8.0
    a = run_simulation(low)
    b = run_simulation(high)
    assert b["latency"]["ttft_ms"]["p95"] >= a["latency"]["ttft_ms"]["p95"]


def test_static_and_continuous_are_distinct():
    cfg = base_config()
    cfg.update({"request_rate_rps": 4.0, "output_tokens_mean": 32})
    cfg["scheduler"] = "static_fcfs"
    static = run_simulation(cfg)
    cfg["scheduler"] = "continuous_fcfs"
    continuous = run_simulation(cfg)
    assert static["summary"]["goodput_rps"] != continuous["summary"]["goodput_rps"] or static["latency"]["ttft_ms"]["p95"] != continuous["latency"]["ttft_ms"]["p95"]