test_router_e2e_with_mockers.py 22.1 KB
Newer Older
1
2
3
4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
import logging
import os
5
from typing import Any, Dict, Optional
6
7
8

import pytest

9
10
11
12
from tests.router.common import (  # utilities
    _test_python_router_bindings,
    _test_router_basic,
    _test_router_decisions,
13
    _test_router_disagg_decisions,
14
15
16
17
18
19
20
    _test_router_indexers_sync,
    _test_router_overload_503,
    _test_router_query_instance_id,
    _test_router_two_routers,
    generate_random_suffix,
    get_runtime,
)
Alec's avatar
Alec committed
21
from tests.utils.constants import ROUTER_MODEL_NAME
22
23
from tests.utils.managed_process import ManagedProcess

24
25
26
27
28
pytestmark = [
    pytest.mark.pre_merge,
    pytest.mark.gpu_0,
    pytest.mark.integration,
]
29

30

31
logger = logging.getLogger(__name__)
32
33


Alec's avatar
Alec committed
34
MODEL_NAME = ROUTER_MODEL_NAME
35
36
NUM_MOCKERS = 2
SPEEDUP_RATIO = 10.0
37
BASE_PORT = 9100  # Base port for all tests (high port to avoid conflicts)
38
NUM_REQUESTS = 100
39
BLOCK_SIZE = 16
40
41


42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def get_unique_ports(
    request, num_ports: int = 1, store_backend: str = "etcd"
) -> list[int]:
    """Generate unique ports for parallel test execution.

    Ports are unique based on:
    - Test function name (each test gets a base offset)
    - Parametrization value (etcd=0, file=50)
    - Port index (for multi-port tests)

    Args:
        request: Pytest request fixture
        num_ports: Number of ports needed (1 for single router, 2 for two routers)
        store_backend: Storage backend parameter ("etcd" or "file")

    Returns:
        List of unique port numbers
    """
    # Get test name without parametrization suffix
    test_name = request.node.name.split("[")[0]

    # Base offsets per test function (ensures each test gets unique range)
    test_offsets = {
        "test_mocker_kv_router": 0,
        "test_mocker_two_kv_router": 100,
        "test_mocker_kv_router_overload_503": 200,
        "test_query_instance_id_returns_worker_and_tokens": 300,
69
        "test_router_disagg_decisions": 400,
70
71
72
73
74
75
76
77
78
79
80
81
    }

    base_offset = test_offsets.get(test_name, 0)

    # Parametrization offset (etcd=0, file=50)
    param_offset = 0 if store_backend == "etcd" else 50

    # Generate ports
    ports = [BASE_PORT + base_offset + param_offset + i for i in range(num_ports)]
    return ports


82
83
84
85
86
87
88
89
90
91
92
93
94
# Shared test payload for all tests
TEST_PAYLOAD: Dict[str, Any] = {
    "model": MODEL_NAME,
    "messages": [
        {
            "role": "user",
            "content": "In a quiet meadow tucked between rolling hills, a plump gray rabbit nibbled on clover beneath the shade of a gnarled oak tree. Its ears twitched at the faint rustle of leaves, but it remained calm, confident in the safety of its burrow just a few hops away. The late afternoon sun warmed its fur, and tiny dust motes danced in the golden light as bees hummed lazily nearby. Though the rabbit lived a simple life, every day was an adventure of scents, shadows, and snacks—an endless search for the tastiest patch of greens and the softest spot to nap.",
        }
    ],
    "stream": True,
    "max_tokens": 10,
}

95

96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def _build_mocker_command(
    endpoint: str,
    store_backend: str,
    num_workers: int,
    mocker_args: Dict[str, Any],
    worker_type: Optional[str] = None,
) -> list[str]:
    """Build the mocker CLI command with all arguments.

    Args:
        endpoint: The dynamo endpoint string
        store_backend: Storage backend ("etcd" or "file")
        num_workers: Number of workers to spawn (uses --num-workers flag)
        mocker_args: Dictionary of mocker arguments
        worker_type: Optional worker type ("prefill" or "decode") for disagg mode

    Returns:
        List of command arguments for subprocess
    """
    command = [
        "python",
        "-m",
        "dynamo.mocker",
        "--model-path",
        MODEL_NAME,
        "--endpoint",
        endpoint,
        "--store-kv",
        store_backend,
        "--num-workers",
        str(num_workers),
    ]

    # Add worker type flag for disaggregated mode
    if worker_type == "prefill":
        command.append("--is-prefill-worker")
    elif worker_type == "decode":
        command.append("--is-decode-worker")

    # Add individual CLI arguments from mocker_args
    if "speedup_ratio" in mocker_args:
        command.extend(["--speedup-ratio", str(mocker_args["speedup_ratio"])])
    if "block_size" in mocker_args:
        command.extend(["--block-size", str(mocker_args["block_size"])])
    if "num_gpu_blocks" in mocker_args:
        command.extend(
            ["--num-gpu-blocks-override", str(mocker_args["num_gpu_blocks"])]
        )
    if "max_num_seqs" in mocker_args:
        command.extend(["--max-num-seqs", str(mocker_args["max_num_seqs"])])
    if "max_num_batched_tokens" in mocker_args:
        command.extend(
            ["--max-num-batched-tokens", str(mocker_args["max_num_batched_tokens"])]
        )
    if "enable_prefix_caching" in mocker_args:
        if mocker_args["enable_prefix_caching"]:
            command.append("--enable-prefix-caching")
        else:
            command.append("--no-enable-prefix-caching")
    if "enable_chunked_prefill" in mocker_args:
        if mocker_args["enable_chunked_prefill"]:
            command.append("--enable-chunked-prefill")
        else:
            command.append("--no-enable-chunked-prefill")
    if "watermark" in mocker_args:
        command.extend(["--watermark", str(mocker_args["watermark"])])
    if "dp_size" in mocker_args:
        command.extend(["--data-parallel-size", str(mocker_args["dp_size"])])

    return command


168
class MockerProcess:
169
    """Manages mocker engine instances with shared tokio runtime via --num-workers."""
170

171
172
173
174
175
    def __init__(
        self,
        request,
        mocker_args: Optional[Dict[str, Any]] = None,
        num_mockers: int = 1,
176
        store_backend: str = "etcd",
177
    ):
178
179
        namespace_suffix = generate_random_suffix()
        self.namespace = f"test-namespace-{namespace_suffix}"
180
181
        self.component_name = "mocker"
        self.endpoint = f"dyn://{self.namespace}.{self.component_name}.generate"
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
        self.num_workers = num_mockers

        mocker_args = mocker_args or {}

        command = _build_mocker_command(
            endpoint=self.endpoint,
            store_backend=store_backend,
            num_workers=num_mockers,
            mocker_args=mocker_args,
        )

        self._process = ManagedProcess(
            command=command,
            timeout=60,
            display_output=True,
            health_check_ports=[],
            health_check_urls=[],
            log_dir=request.node.name,
            terminate_existing=False,
        )
        logger.info(
            f"Created mocker process with {num_mockers} worker(s), endpoint: {self.endpoint}"
        )

    def __enter__(self):
        logger.info(f"Starting mocker process with {self.num_workers} worker(s)")
        self._process.__enter__()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        logger.info("Stopping mocker process")
        self._process.__exit__(exc_type, exc_val, exc_tb)


class DisaggMockerProcess:
    """Manages prefill or decode mocker instances for disaggregated serving.

    Uses --num-workers for shared tokio runtime. For disaggregated serving:
    - Prefill workers: worker_type="prefill", endpoint is namespace.prefill.generate
    - Decode workers: worker_type="decode", endpoint is namespace.backend.generate

    Both prefill and decode workers should share the same namespace for proper discovery.
    """

    def __init__(
        self,
        request,
        namespace: str,
        worker_type: str,
        mocker_args: Optional[Dict[str, Any]] = None,
        num_mockers: int = 1,
        store_backend: str = "etcd",
    ):
        if worker_type not in ("prefill", "decode"):
            raise ValueError(
                f"worker_type must be 'prefill' or 'decode', got {worker_type}"
238
            )
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274

        self.namespace = namespace
        self.worker_type = worker_type
        self.num_workers = num_mockers

        # Set component name and endpoint based on worker type
        if worker_type == "prefill":
            self.component_name = "prefill"
            self.endpoint = f"dyn://{self.namespace}.prefill.generate"
        else:
            self.component_name = "backend"
            self.endpoint = f"dyn://{self.namespace}.backend.generate"

        mocker_args = mocker_args or {}

        command = _build_mocker_command(
            endpoint=self.endpoint,
            store_backend=store_backend,
            num_workers=num_mockers,
            mocker_args=mocker_args,
            worker_type=worker_type,
        )

        self._process = ManagedProcess(
            command=command,
            timeout=60,
            display_output=True,
            health_check_ports=[],
            health_check_urls=[],
            log_dir=request.node.name,
            terminate_existing=False,
        )
        logger.info(
            f"Created {worker_type} mocker process with {num_mockers} worker(s), "
            f"endpoint: {self.endpoint}"
        )
275

276
    def __enter__(self):
277
278
279
280
        logger.info(
            f"Starting {self.worker_type} mocker process with {self.num_workers} worker(s)"
        )
        self._process.__enter__()
281
        return self
282

283
    def __exit__(self, exc_type, exc_val, exc_tb):
284
285
        logger.info(f"Stopping {self.worker_type} mocker process")
        self._process.__exit__(exc_type, exc_val, exc_tb)
286
287
288


@pytest.mark.pre_merge
289
290
@pytest.mark.gpu_0
@pytest.mark.integration
291
@pytest.mark.parallel
Alec's avatar
Alec committed
292
@pytest.mark.model(MODEL_NAME)
293
def test_mocker_kv_router(request, runtime_services_session, predownload_tokenizers):
294
295
296
297
298
299
300
301
    """
    Test KV router with multiple mocker engine instances.
    This test doesn't require GPUs and runs quickly for pre-merge validation.
    """

    # runtime_services starts etcd and nats
    logger.info("Starting mocker KV router test")

302
    # Create mocker args dictiona: FixtureRequestry: tuple[NatsServer, EtcdServer]: NoneType
303
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}
304
305

    try:
306
        # Start mocker instances with the new CLI interface
307
        logger.info(f"Starting {NUM_MOCKERS} mocker instances")
308
309
310
        mockers = MockerProcess(
            request, mocker_args=mocker_args, num_mockers=NUM_MOCKERS
        )
311
312
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
        mockers.__enter__()
313

314
315
316
317
        # Get unique port for this test
        frontend_port = get_unique_ports(request, num_ports=1)[0]

        # Run basic router test (starts router internally and waits for workers to be ready)
318
319
320
321
        _test_router_basic(
            engine_workers=mockers,
            block_size=BLOCK_SIZE,
            request=request,
322
            frontend_port=frontend_port,
323
324
            test_payload=TEST_PAYLOAD,
            num_requests=NUM_REQUESTS,
325
326
327
        )

    finally:
328
329
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
330
331


332
@pytest.mark.pre_merge
333
334
@pytest.mark.gpu_0
@pytest.mark.integration
335
@pytest.mark.parallel
Alec's avatar
Alec committed
336
@pytest.mark.model(MODEL_NAME)
337
338
339
@pytest.mark.parametrize("store_backend", ["etcd", "file"])
def test_mocker_two_kv_router(
    request,
340
    runtime_services_session,
341
342
343
344
    predownload_tokenizers,
    file_storage_backend,
    store_backend,
):
345
346
347
    """
    Test with two KV routers and multiple mocker engine instances.
    Alternates requests between the two routers to test load distribution.
348
    Tests with both etcd and file storage backends.
349
350
351
    """

    # runtime_services starts etcd and nats
352
353
354
    logger.info(
        f"Starting mocker two KV router test with {store_backend} storage backend"
    )
355

356
    # Create mocker args dictionary
357
358
359
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}

    try:
360
        # Start mocker instances with the new CLI interface
361
        logger.info(f"Starting {NUM_MOCKERS} mocker instances")
362
        mockers = MockerProcess(
363
364
365
366
            request,
            mocker_args=mocker_args,
            num_mockers=NUM_MOCKERS,
            store_backend=store_backend,
367
        )
368
369
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
        mockers.__enter__()
370

371
372
373
374
375
        # Get unique ports for this test (2 ports for two routers)
        router_ports = get_unique_ports(
            request, num_ports=2, store_backend=store_backend
        )

376
377
378
379
380
        # Run two-router test (starts KV routers internally and manages their lifecycle)
        _test_router_two_routers(
            engine_workers=mockers,
            block_size=BLOCK_SIZE,
            request=request,
381
            router_ports=router_ports,
382
383
384
            test_payload=TEST_PAYLOAD,
            num_requests=NUM_REQUESTS,
            store_backend=store_backend,
385
386
387
        )

    finally:
388
389
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
390
391


392
@pytest.mark.pre_merge
393
394
@pytest.mark.gpu_0
@pytest.mark.integration
395
@pytest.mark.parallel
Alec's avatar
Alec committed
396
@pytest.mark.model(MODEL_NAME)
397
@pytest.mark.skip(reason="Flaky, temporarily disabled")
Alec's avatar
Alec committed
398
def test_mocker_kv_router_overload_503(
399
    request, runtime_services_session, predownload_tokenizers
Alec's avatar
Alec committed
400
):
401
    """Test that KV router returns 503 when mocker workers are overloaded."""
402
    logger.info("Starting mocker KV router overload test for 503 status")
403
    # Create mocker args dictionary with limited resources
404
405
406
407
408
    mocker_args = {
        "speedup_ratio": 10,
        "block_size": 4,  # Smaller block size
        "num_gpu_blocks": 64,  # Limited GPU blocks to exhaust quickly
    }
409

410
    try:
411
        # Start single mocker instance with limited resources
412
        logger.info("Starting single mocker instance with limited resources")
413
        mockers = MockerProcess(request, mocker_args=mocker_args, num_mockers=1)
414
415
        logger.info(f"Mocker using endpoint: {mockers.endpoint}")
        mockers.__enter__()
416

417
418
419
        # Get unique port for this test
        frontend_port = get_unique_ports(request, num_ports=1)[0]

420
421
422
423
424
425
426
427
428
        # Run overload 503 test
        _test_router_overload_503(
            engine_workers=mockers,
            block_size=4,  # Match the mocker's block size
            request=request,
            frontend_port=frontend_port,
            test_payload=TEST_PAYLOAD,
            busy_threshold=0.2,
        )
429
430

    finally:
431
432
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
433

434
435

@pytest.mark.pre_merge
436
437
@pytest.mark.gpu_0
@pytest.mark.integration
438
@pytest.mark.parallel
Alec's avatar
Alec committed
439
@pytest.mark.model(MODEL_NAME)
440
441
442
def test_kv_push_router_bindings(
    request, runtime_services_session, predownload_tokenizers
):
443
    """Test KvPushRouter Python bindings with mocker engines."""
444
445
446
447
    logger.info("Starting KvPushRouter bindings test")
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}

    try:
448
        # Start mocker instances
449
        logger.info(f"Starting {NUM_MOCKERS} mocker instances")
450
451
452
        mockers = MockerProcess(
            request, mocker_args=mocker_args, num_mockers=NUM_MOCKERS
        )
453
454
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
        mockers.__enter__()
455

456
457
458
        # Get runtime and create endpoint
        runtime = get_runtime()
        namespace = runtime.namespace(mockers.namespace)
459
        component = namespace.component(mockers.component_name)
460
461
        endpoint = component.endpoint("generate")

462
463
464
        # Run Python router bindings test
        _test_python_router_bindings(
            engine_workers=mockers,
465
466
            endpoint=endpoint,
            block_size=BLOCK_SIZE,
467
468
            model_name=MODEL_NAME,
            num_workers=NUM_MOCKERS,
469
        )
470
471

    finally:
472
473
474
475
476
        if "mockers" in locals():
            mockers.__exit__(None, None, None)


@pytest.mark.pre_merge
477
478
@pytest.mark.gpu_0
@pytest.mark.integration
479
@pytest.mark.parallel
Alec's avatar
Alec committed
480
@pytest.mark.model(MODEL_NAME)
481
482
483
@pytest.mark.parametrize("store_backend", ["etcd", "file"])
def test_indexers_sync(
    request,
484
    runtime_services_session,
485
486
487
488
    predownload_tokenizers,
    file_storage_backend,
    store_backend,
):
489
490
491
    """
    Test that two KV routers have synchronized indexer states after processing requests.
    This test verifies that both routers converge to the same internal state.
492
    Tests with both etcd and file storage backends.
493
494
495
    """

    # runtime_services starts etcd and nats
496
    logger.info(f"Starting indexers sync test with {store_backend} storage backend")
497

498
    # Create mocker args dictionary
499
500
501
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}

    try:
502
        # Start mocker instances
503
        logger.info(f"Starting {NUM_MOCKERS} mocker instances")
504
        mockers = MockerProcess(
505
506
507
508
            request,
            mocker_args=mocker_args,
            num_mockers=NUM_MOCKERS,
            store_backend=store_backend,
509
        )
510
511
512
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
        mockers.__enter__()

513
514
515
516
517
518
519
520
521
        # Use the common test implementation (creates its own runtimes for each router)
        # Note: Consumer verification is done inside _test_router_indexers_sync while routers are alive
        _test_router_indexers_sync(
            engine_workers=mockers,
            block_size=BLOCK_SIZE,
            model_name=MODEL_NAME,
            num_workers=NUM_MOCKERS,
            store_backend=store_backend,
        )
522
523
524
525
526
527

        logger.info("Indexers sync test completed successfully")

    finally:
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
528

529
530

@pytest.mark.pre_merge
531
532
@pytest.mark.gpu_0
@pytest.mark.integration
533
@pytest.mark.parallel
Alec's avatar
Alec committed
534
535
@pytest.mark.model(MODEL_NAME)
def test_query_instance_id_returns_worker_and_tokens(
536
    request, runtime_services_session, predownload_tokenizers
Alec's avatar
Alec committed
537
):
538
    """Test query_instance_id annotation with mocker engines."""
539
540
541
542
543
    logger.info("Starting KV router query_instance_id annotation test")
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}
    os.makedirs(request.node.name, exist_ok=True)

    try:
544
        # Start mocker instances
545
        logger.info(f"Starting {NUM_MOCKERS} mocker instances")
546
547
548
        mockers = MockerProcess(
            request, mocker_args=mocker_args, num_mockers=NUM_MOCKERS
        )
549
550
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
        mockers.__enter__()
551

552
553
554
        # Get unique port for this test
        frontend_port = get_unique_ports(request, num_ports=1)[0]

555
556
557
558
559
560
561
562
        # Run query_instance_id annotation test
        _test_router_query_instance_id(
            engine_workers=mockers,
            block_size=BLOCK_SIZE,
            request=request,
            frontend_port=frontend_port,
            test_payload=TEST_PAYLOAD,
        )
563
564

    finally:
565
566
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
567
568
569


@pytest.mark.pre_merge
570
571
@pytest.mark.gpu_0
@pytest.mark.integration
572
@pytest.mark.parallel
573
@pytest.mark.model(MODEL_NAME)
574
def test_router_decisions(request, runtime_services_session, predownload_tokenizers):
575
    """Validate KV cache prefix reuse and dp_rank routing by sending progressive requests with overlapping prefixes."""
576
577
578
579

    # runtime_services starts etcd and nats
    logger.info("Starting test router prefix reuse and KV events synchronization")

Yan Ru Pei's avatar
Yan Ru Pei committed
580
581
582
583
584
585
    # Create mocker args dictionary with dp_size=4
    mocker_args = {
        "speedup_ratio": SPEEDUP_RATIO,
        "block_size": BLOCK_SIZE,
        "dp_size": 4,
    }
586
587

    try:
Yan Ru Pei's avatar
Yan Ru Pei committed
588
589
        logger.info(
            "Starting 2 mocker instances with dp_size=4 each (8 total dp ranks)"
590
        )
Yan Ru Pei's avatar
Yan Ru Pei committed
591
        mockers = MockerProcess(request, mocker_args=mocker_args, num_mockers=2)
592
        logger.info(f"All mockers using endpoint: {mockers.endpoint}")
593

594
595
596
597
598
599
600
601
602
603
        # Initialize mockers
        mockers.__enter__()

        # Get runtime and create endpoint
        runtime = get_runtime()
        # Use the namespace from the mockers
        namespace = runtime.namespace(mockers.namespace)
        component = namespace.component("mocker")
        endpoint = component.endpoint("generate")

604
605
        _test_router_decisions(
            mockers, endpoint, MODEL_NAME, request, test_dp_rank=True
606
607
608
609
610
        )

    finally:
        if "mockers" in locals():
            mockers.__exit__(None, None, None)
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678


@pytest.mark.pre_merge
@pytest.mark.parallel
@pytest.mark.model(MODEL_NAME)
def test_router_disagg_decisions(
    request, runtime_services_session, predownload_tokenizers
):
    """Validate KV cache prefix reuse in disaggregated prefill-decode setup.

    Tests that progressive requests with overlapping prefixes are routed to the
    same prefill worker due to KV cache reuse.
    """
    logger.info("Starting disaggregated router prefix reuse test")

    # Generate shared namespace for prefill and decode workers
    namespace_suffix = generate_random_suffix()
    shared_namespace = f"test-namespace-{namespace_suffix}"

    # Create mocker args
    mocker_args = {"speedup_ratio": SPEEDUP_RATIO, "block_size": BLOCK_SIZE}

    prefill_workers = None
    decode_workers = None

    try:
        # Start prefill workers (4 instances)
        logger.info("Starting 4 prefill mocker instances")
        prefill_workers = DisaggMockerProcess(
            request,
            namespace=shared_namespace,
            worker_type="prefill",
            mocker_args=mocker_args,
            num_mockers=4,
        )
        prefill_workers.__enter__()
        logger.info(f"Prefill workers using endpoint: {prefill_workers.endpoint}")

        # Start decode workers (4 instances)
        logger.info("Starting 4 decode mocker instances")
        decode_workers = DisaggMockerProcess(
            request,
            namespace=shared_namespace,
            worker_type="decode",
            mocker_args=mocker_args,
            num_mockers=4,
        )
        decode_workers.__enter__()
        logger.info(f"Decode workers using endpoint: {decode_workers.endpoint}")

        # Get unique port for this test
        frontend_port = get_unique_ports(request, num_ports=1)[0]

        # Run disagg routing test
        _test_router_disagg_decisions(
            prefill_workers=prefill_workers,
            decode_workers=decode_workers,
            block_size=BLOCK_SIZE,
            request=request,
            frontend_port=frontend_port,
            test_payload=TEST_PAYLOAD,
        )

    finally:
        if decode_workers is not None:
            decode_workers.__exit__(None, None, None)
        if prefill_workers is not None:
            prefill_workers.__exit__(None, None, None)