tests/test_disaggregated.py

from inferscale import run_simulation


def base_config():
    return {
        "model": "Qwen2.5-3B",
        "accelerator": "L4",
        "prefill_accelerator": "L4",
        "decode_accelerator": "L4",
        "quantization": "int8",
        "scheduler": "continuous_slo",
        "topology": "disaggregated_pd",
        "duration_s": 5,
        "request_rate_rps": 1.5,
        "prompt_tokens_mean": 128,
        "prompt_tokens_cv": 0.1,
        "output_tokens_mean": 8,
        "output_tokens_cv": 0.1,
        "seed": 4,
        "slo_ttft_ms": 2000,
        "slo_e2e_ms": 10000,
    }


def test_disaggregated_pipeline_completes_and_reports_transfer():
    result = run_simulation(base_config())
    assert result["provenance"]["topology"] == "disaggregated_pd"
    assert result["summary"]["requests_completed"] > 0
    assert result["resource"]["kv_transfer_gb"] > 0
    assert result["resource"]["accelerator_instances"] == 2
    assert result["resource"]["p95_transfer_ms"] > 0


def test_more_decode_workers_are_modelled():
    cfg = base_config()
    cfg["decode_workers"] = 2
    result = run_simulation(cfg)
    assert result["resource"]["decode_workers"] == 2
    assert result["resource"]["accelerator_instances"] == 3