"docs/guides/dynamo_deploy/metrics.md" did not exist on "5375af2c52d2d5a218e3936d9c23918da5855b17"
conftest.py 32.4 KB
Newer Older
1
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
Neelay Shah's avatar
Neelay Shah committed
2
3
# SPDX-License-Identifier: Apache-2.0

4
import importlib.util
Neelay Shah's avatar
Neelay Shah committed
5
6
import logging
import os
7
import shutil
8
import subprocess
Neelay Shah's avatar
Neelay Shah committed
9
import tempfile
10
from pathlib import Path
11
from typing import Generator, Optional
Neelay Shah's avatar
Neelay Shah committed
12
13

import pytest
14
from filelock import FileLock
Neelay Shah's avatar
Neelay Shah committed
15

16
from tests.utils.constants import TEST_MODELS, DefaultPort
Neelay Shah's avatar
Neelay Shah committed
17
from tests.utils.managed_process import ManagedProcess
18
from tests.utils.port_utils import (
19
    ServicePorts,
20
21
22
23
24
    allocate_port,
    allocate_ports,
    deallocate_port,
    deallocate_ports,
)
25
from tests.utils.test_output import resolve_test_output_path
26
27

_logger = logging.getLogger(__name__)
Neelay Shah's avatar
Neelay Shah committed
28

Alec's avatar
Alec committed
29
30

def pytest_configure(config):
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
    # Defining markers to avoid `<marker> not found in 'markers' configuration option`
    # errors when pyproject.toml is not available in the container (e.g. some CI jobs).
    # IMPORTANT: Keep this marker list in sync with [tool.pytest.ini_options].markers
    # in pyproject.toml. If you add or remove markers there, mirror the change here.
    markers = [
        "pre_merge: marks tests to run before merging",
        "post_merge: marks tests to run after merge",
        "parallel: marks tests that can run in parallel with pytest-xdist",
        "nightly: marks tests to run nightly",
        "weekly: marks tests to run weekly",
        "gpu_0: marks tests that don't require GPU",
        "gpu_1: marks tests to run on GPU",
        "gpu_2: marks tests to run on 2GPUs",
        "gpu_4: marks tests to run on 4GPUs",
        "gpu_8: marks tests to run on 8GPUs",
        "e2e: marks tests as end-to-end tests",
        "integration: marks tests as integration tests",
        "unit: marks tests as unit tests",
        "stress: marks tests as stress tests",
        "performance: marks tests as performance tests",
        "vllm: marks tests as requiring vllm",
        "trtllm: marks tests as requiring trtllm",
        "sglang: marks tests as requiring sglang",
54
        "lmcache: mark tests as requiring lmcache",
55
56
57
        "multimodal: marks tests as multimodal (image/video) tests",
        "slow: marks tests as known to be slow",
        "h100: marks tests to run on H100",
58
        "aiconfigurator: marks e2e tests that cover aiconfigurator functionality",
59
60
61
62
        "router: marks tests for router component",
        "planner: marks tests for planner component",
        "kvbm: marks tests for KV behavior and model determinism",
        "kvbm_v2: marks tests using KVBM V2",
63
        "kvbm_concurrency: marks concurrency stress tests for KVBM (runs separately)",
64
65
66
67
        "model: model id used by a test or parameter",
        "custom_build: marks tests that require custom builds or special setup (e.g., MoE models)",
        "k8s: marks tests as requiring Kubernetes",
        "fault_tolerance: marks tests as fault tolerance tests",
68
        "deploy: marks tests as deployment tests",
69
70
        # Third-party plugin markers
        "timeout: test timeout in seconds (pytest-timeout plugin)",
71
72
73
    ]
    for marker in markers:
        config.addinivalue_line("markers", marker)
Alec's avatar
Alec committed
74
75


76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
def pytest_addoption(parser: pytest.Parser) -> None:
    """Add shared command-line options for all tests.

    Shared options that apply across multiple test suites are defined here.
    Suite-specific options (e.g., deploy, fault-tolerance) are defined in
    their respective subdirectory conftest.py files.
    """
    # -------------------------------------------------------------------------
    # Shared Deployment Options (used by multiple test suites)
    # -------------------------------------------------------------------------
    parser.addoption(
        "--image",
        type=str,
        default=None,
        help="Container image to use for deployment (overrides YAML default)",
    )
    parser.addoption(
        "--namespace",
        type=str,
        default=None,  # No default here - subdirectories provide their own
        help="Kubernetes namespace for deployment",
    )
    parser.addoption(
        "--skip-service-restart",
        action="store_true",
        default=None,  # None = use fixture's default behavior
        help="Skip restarting NATS and etcd services before deployment. "
        "Default: deploy tests skip (for speed), fault-tolerance tests restart (for clean state).",
    )


Neelay Shah's avatar
Neelay Shah committed
107
LOG_FORMAT = "[TEST] %(asctime)s %(levelname)s %(name)s: %(message)s"
108
DATE_FORMAT = "%Y-%m-%dT%H:%M:%S"
Neelay Shah's avatar
Neelay Shah committed
109
110
111
112

logging.basicConfig(
    level=logging.INFO,
    format=LOG_FORMAT,
113
    datefmt=DATE_FORMAT,  # ISO 8601 UTC format
Neelay Shah's avatar
Neelay Shah committed
114
115
)

116

Alec's avatar
Alec committed
117
118
119
120
121
122
123
124
125
126
@pytest.fixture()
def set_ucx_tls_no_mm():
    """Set UCX env defaults for all tests."""
    mp = pytest.MonkeyPatch()
    # CI note:
    # - Affected test: tests/fault_tolerance/cancellation/test_vllm.py::test_request_cancellation_vllm_decode_cancel
    # - Symptom on L40 CI: UCX/NIXL mm transport assertion during worker init
    #   (uct_mem.c:482: mem.memh != UCT_MEM_HANDLE_NULL) when two workers
    #   start on the same node (maybe a shared-memory segment collision/limits).
    # - Mitigation: disable UCX "mm" shared-memory transport globally for tests
127
128
129
130
131
    #
    # Also exclude gdr_copy transport to prevent GDRCopy driver initialization
    # failures (driverInitFileInfo result=11) that can abort the process when
    # the gdrdrv kernel module is not loaded.
    mp.setenv("UCX_TLS", "^mm,gdr_copy")
Alec's avatar
Alec committed
132
133
134
135
    yield
    mp.undo()


136
def download_models(model_list=None, ignore_weights=False):
137
138
139
140
    """Download models - can be called directly or via fixture

    Args:
        model_list: List of model IDs to download. If None, downloads TEST_MODELS.
141
        ignore_weights: If True, skips downloading model weight files. Default is False.
142
143
144
145
146
    """
    if model_list is None:
        model_list = TEST_MODELS

    # Check for HF_TOKEN in environment
147
    hf_token = os.environ.get("HF_TOKEN", "").strip() or None
148
149
150
151
152
153
154
155
156
157
158
    if hf_token:
        logging.info("HF_TOKEN found in environment")
    else:
        logging.warning(
            "HF_TOKEN not found in environment. "
            "Some models may fail to download or you may encounter rate limits. "
            "Get a token from https://huggingface.co/settings/tokens"
        )

    try:
        from huggingface_hub import snapshot_download
159
160
161
162
163
164
165
166
167
168
    except ImportError as exc:
        raise RuntimeError(
            "huggingface_hub is required to pre-download models for tests"
        ) from exc

    failures = []
    for model_id in model_list:
        logging.info(
            f"Pre-downloading {'model (no weights)' if ignore_weights else 'model'}: {model_id}"
        )
169

170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
        try:
            if ignore_weights:
                # Weight file patterns to exclude (based on hub.rs implementation)
                weight_patterns = [
                    "*.bin",
                    "*.safetensors",
                    "*.h5",
                    "*.msgpack",
                    "*.ckpt.index",
                ]

                # Download everything except weight files
                snapshot_download(
                    repo_id=model_id,
                    token=hf_token,
                    ignore_patterns=weight_patterns,
                )
            else:
                # Download the full model snapshot (includes all files)
                snapshot_download(
                    repo_id=model_id,
                    token=hf_token,
                )
            logging.info(f"Successfully pre-downloaded: {model_id}")

        except Exception as exc:
            logging.error(f"Failed to pre-download {model_id}: {exc}")
            failures.append(f"{model_id}: {exc}")

    if failures:
        raise RuntimeError(
            "Failed to pre-download required Hugging Face models:\n"
            + "\n".join(failures)
203
204
205
206
        )


@pytest.fixture(scope="session")
Alec's avatar
Alec committed
207
208
209
210
211
212
213
214
215
216
217
218
def predownload_models(pytestconfig):
    """Fixture wrapper around download_models for models used in collected tests"""
    # Get models from pytest config if available, otherwise fall back to TEST_MODELS
    models = getattr(pytestconfig, "models_to_download", None)
    if models:
        logging.info(
            f"Downloading {len(models)} models needed for collected tests\nModels: {models}"
        )
        download_models(model_list=list(models))
    else:
        # Fallback to original behavior if extraction failed
        download_models()
219
220

    os.environ["HF_HUB_OFFLINE"] = "1"
221
    yield
222
    os.environ.pop("HF_HUB_OFFLINE", None)
223

Neelay Shah's avatar
Neelay Shah committed
224

225
@pytest.fixture(scope="session")
Alec's avatar
Alec committed
226
227
228
229
230
231
232
233
234
235
236
237
def predownload_tokenizers(pytestconfig):
    """Fixture wrapper around download_models for tokenizers used in collected tests"""
    # Get models from pytest config if available, otherwise fall back to TEST_MODELS
    models = getattr(pytestconfig, "models_to_download", None)
    if models:
        logging.info(
            f"Downloading tokenizers for {len(models)} models needed for collected tests\nModels: {models}"
        )
        download_models(model_list=list(models), ignore_weights=True)
    else:
        # Fallback to original behavior if extraction failed
        download_models(ignore_weights=True)
238
239
240
241
242

    # Skip redundant HuggingFace API calls in worker subprocesses since
    # tokenizers are already cached. This avoids flaky timeouts from slow
    # HF API responses (the RepoInfo fetch still happens even for cached models).
    os.environ["HF_HUB_OFFLINE"] = "1"
243
    yield
244
    os.environ.pop("HF_HUB_OFFLINE", None)
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
@pytest.fixture(scope="session")
def build_kv_indexer():
    """Pre-build the standalone KV indexer binary once per session.

    Runs `cargo build` so that `cargo run` in tests starts instantly.
    No-op if the binary is already cached in target/.
    """
    _logger.info("Building dynamo-kv-indexer binary (cached after first build)")
    subprocess.check_call(
        [
            "cargo",
            "build",
            "-p",
            "dynamo-kv-router",
            "--features",
            "indexer-bin",
            "--bin",
            "dynamo-kv-indexer",
        ],
        timeout=600,
    )
    _logger.info("dynamo-kv-indexer binary ready")


271
272
@pytest.fixture(autouse=True)
def logger(request):
273
274
    log_dir = resolve_test_output_path(request.node.name)
    log_path = os.path.join(log_dir, "test.log.txt")
275
    logger = logging.getLogger()
276
277
    shutil.rmtree(log_dir, ignore_errors=True)
    os.makedirs(log_dir, exist_ok=True)
278
279
280
281
282
283
284
285
286
    handler = logging.FileHandler(log_path, mode="w")
    formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    yield
    handler.close()
    logger.removeHandler(handler)


287
288
289
290
291
292
293
294
295
296
297
298
299
300
def _item_has_marker(item, marker_name):
    """Check if a test item has a marker, including module-level pytestmark."""
    if item.get_closest_marker(marker_name):
        return True
    module = getattr(item, "module", None)
    if module is not None:
        marks = getattr(module, "pytestmark", [])
        if not isinstance(marks, list):
            marks = [marks]
        if any(getattr(m, "name", "") == marker_name for m in marks):
            return True
    return False


Alec's avatar
Alec committed
301
@pytest.hookimpl(trylast=True)
302
303
304
305
def pytest_collection_modifyitems(config, items):
    """
    This function is called to modify the list of tests to run.
    """
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
    # Auto-skip tests marked with a framework marker when the framework is not installed
    framework_markers = {
        "trtllm": "tensorrt_llm",
        "vllm": "vllm",
        "sglang": "sglang",
        "kvbm": "kvbm",
        "lmcache": "lmcache",
    }
    for marker_name, module_name in framework_markers.items():
        if importlib.util.find_spec(module_name) is None:
            skip = pytest.mark.skip(reason=f"{module_name} is not installed")
            for item in items:
                if _item_has_marker(item, marker_name):
                    item.add_marker(skip)

Alec's avatar
Alec committed
321
322
    # Collect models via explicit pytest mark from final filtered items only
    models_to_download = set()
323
    for item in items:
Alec's avatar
Alec committed
324
325
326
327
328
329
330
331
332
333
334
335
        # Only collect from items that are not skipped
        if any(
            getattr(m, "name", "") == "skip" for m in getattr(item, "own_markers", [])
        ):
            continue
        model_mark = item.get_closest_marker("model")
        if model_mark and model_mark.args:
            models_to_download.add(model_mark.args[0])

    # Store models to download in pytest config for fixtures to access
    if models_to_download:
        config.models_to_download = models_to_download
336

337

Neelay Shah's avatar
Neelay Shah committed
338
339
class EtcdServer(ManagedProcess):
    def __init__(self, request, port=2379, timeout=300):
340
341
342
343
344
345
346
347
348
349
350
351
        # Allocate free ports if port is 0
        use_random_port = port == 0
        if use_random_port:
            # Need two ports: client port and peer port for parallel execution
            # Start from 2380 (etcd default 2379 + 1)
            port, peer_port = allocate_ports(2, 2380)
        else:
            peer_port = None

        self.port = port
        self.peer_port = peer_port  # Store for cleanup
        self.use_random_port = use_random_port  # Track if we allocated the port
Neelay Shah's avatar
Neelay Shah committed
352
353
354
355
        port_string = str(port)
        etcd_env = os.environ.copy()
        etcd_env["ALLOW_NONE_AUTHENTICATION"] = "yes"
        data_dir = tempfile.mkdtemp(prefix="etcd_")
356

Neelay Shah's avatar
Neelay Shah committed
357
358
359
360
361
362
363
        command = [
            "etcd",
            "--listen-client-urls",
            f"http://0.0.0.0:{port_string}",
            "--advertise-client-urls",
            f"http://0.0.0.0:{port_string}",
        ]
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384

        # Add peer port configuration only for random ports (parallel execution)
        if peer_port is not None:
            peer_port_string = str(peer_port)
            command.extend(
                [
                    "--listen-peer-urls",
                    f"http://0.0.0.0:{peer_port_string}",
                    "--initial-advertise-peer-urls",
                    f"http://localhost:{peer_port_string}",
                    "--initial-cluster",
                    f"default=http://localhost:{peer_port_string}",
                ]
            )

        command.extend(
            [
                "--data-dir",
                data_dir,
            ]
        )
Neelay Shah's avatar
Neelay Shah committed
385
386
387
388
389
        super().__init__(
            env=etcd_env,
            command=command,
            timeout=timeout,
            display_output=False,
390
            terminate_all_matching_process_names=not use_random_port,  # For distributed tests, do not terminate all matching processes
Neelay Shah's avatar
Neelay Shah committed
391
            health_check_ports=[port],
392
            data_dir=data_dir,
Neelay Shah's avatar
Neelay Shah committed
393
394
395
            log_dir=request.node.name,
        )

396
397
398
399
400
401
402
403
404
405
406
407
408
409
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Release allocated ports when server exits."""
        try:
            # Only deallocate ports that were dynamically allocated (not default ports)
            if self.use_random_port:
                ports_to_release = [self.port]
                if self.peer_port is not None:
                    ports_to_release.append(self.peer_port)
                deallocate_ports(ports_to_release)
        except Exception as e:
            logging.warning(f"Failed to release EtcdServer port: {e}")

        return super().__exit__(exc_type, exc_val, exc_tb)

Neelay Shah's avatar
Neelay Shah committed
410
411

class NatsServer(ManagedProcess):
412
    def __init__(self, request, port=4222, timeout=300, disable_jetstream=False):
413
414
415
416
417
418
419
420
        # Allocate a free port if port is 0
        use_random_port = port == 0
        if use_random_port:
            # Start from 4223 (nats-server default 4222 + 1)
            port = allocate_port(4223)

        self.port = port
        self.use_random_port = use_random_port  # Track if we allocated the port
421
422
        self._request = request  # Store for restart
        self._timeout = timeout
423
424
        self._disable_jetstream = disable_jetstream
        data_dir = tempfile.mkdtemp(prefix="nats_") if not disable_jetstream else None
425
426
427
428
429
430
        command = [
            "nats-server",
            "--trace",
            "-p",
            str(port),
        ]
431
432
        if not disable_jetstream and data_dir:
            command.extend(["-js", "--store_dir", data_dir])
Neelay Shah's avatar
Neelay Shah committed
433
434
435
436
        super().__init__(
            command=command,
            timeout=timeout,
            display_output=False,
437
            terminate_all_matching_process_names=not use_random_port,  # For distributed tests, do not terminate all matching processes
Neelay Shah's avatar
Neelay Shah committed
438
439
            data_dir=data_dir,
            health_check_ports=[port],
440
            health_check_funcs=[self._nats_ready],
Neelay Shah's avatar
Neelay Shah committed
441
442
443
            log_dir=request.node.name,
        )

444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
    def _nats_ready(self, timeout: float = 5) -> bool:
        """Verify NATS server is ready by connecting and optionally checking JetStream."""
        import asyncio

        import nats

        async def check():
            try:
                nc = await nats.connect(
                    f"nats://localhost:{self.port}",
                    connect_timeout=min(timeout, 2),
                )
                try:
                    if not self._disable_jetstream:
                        # Verify JetStream is initialized
                        js = nc.jetstream()
                        await js.account_info()
                    return True
                finally:
                    await nc.close()
            except Exception:
                return False

        # Handle both sync and async contexts
        try:
            asyncio.get_running_loop()  # Check if we're in async context
            # Already in async context - run in a thread to avoid blocking
            import concurrent.futures

            with concurrent.futures.ThreadPoolExecutor() as pool:
                return pool.submit(asyncio.run, check()).result(timeout=timeout)
        except RuntimeError:
            # No running loop - safe to use asyncio.run()
            return asyncio.run(check())

479
480
481
482
483
484
485
486
487
488
489
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Release allocated port when server exits."""
        try:
            # Only deallocate ports that were dynamically allocated (not default ports)
            if self.use_random_port:
                deallocate_port(self.port)
        except Exception as e:
            logging.warning(f"Failed to release NatsServer port: {e}")

        return super().__exit__(exc_type, exc_val, exc_tb)

490
491
492
    def stop(self):
        """Stop the NATS server for restart. Does not release port or clean up fully."""
        _logger.info(f"Stopping NATS server on port {self.port}")
493
        self._stop_started_processes()
494
495
496
497

    def start(self):
        """Restart a stopped NATS server with fresh state."""
        _logger.info(f"Starting NATS server on port {self.port} with fresh state")
498
499
500
501
502
503
504
505
        # Clean up old data directory and create fresh one (only if JetStream enabled)
        if not self._disable_jetstream:
            old_data_dir = self.data_dir  # type: ignore[has-type]
            if old_data_dir is not None:
                shutil.rmtree(old_data_dir, ignore_errors=True)
            self.data_dir = tempfile.mkdtemp(prefix="nats_")

        # Rebuild command
506
507
508
509
510
511
        self.command = [
            "nats-server",
            "--trace",
            "-p",
            str(self.port),
        ]
512
513
        if not self._disable_jetstream and self.data_dir:
            self.command.extend(["-js", "--store_dir", self.data_dir])
514
515

        self._start_process()
516
517
        elapsed = self._check_ports(self._timeout)
        self._check_funcs(self._timeout - elapsed)
518

Neelay Shah's avatar
Neelay Shah committed
519

520
class SharedManagedProcess:
521
522
523
524
525
526
    """Base class for persistent shared processes across pytest-xdist workers.

    Simplified design: first worker starts the process on a dynamic port, it lives forever
    (until the container dies). No ref counting, no teardown. Subsequent workers just
    reuse via port check. This eliminates race conditions and simplifies the logic.
    """
527
528
529
530
531
532

    def __init__(
        self,
        request,
        tmp_path_factory,
        resource_name: str,
533
        start_port: int,
534
535
536
        timeout: int = 300,
    ):
        self.request = request
537
538
        self.start_port = start_port
        self.port: Optional[int] = None  # Set when entering context
539
540
541
542
        self.timeout = timeout
        self.resource_name = resource_name
        self._server: Optional[ManagedProcess] = None

543
        root_tmp = Path(tempfile.gettempdir()) / "pytest_shared_services"
544
545
        root_tmp.mkdir(parents=True, exist_ok=True)

546
547
        self.port_file = root_tmp / f"{resource_name}_port"
        self.lock_file = str(self.port_file) + ".lock"
548

549
    def _create_server(self, port: int) -> ManagedProcess:
550
551
552
        """Create the underlying server instance. Must be implemented by subclasses."""
        raise NotImplementedError

553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
    def _is_port_in_use(self, port: int) -> bool:
        """Check if a port is in use (i.e., a process is listening on it)."""
        import socket

        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(1)
            result = sock.connect_ex(("localhost", port))
            sock.close()
            return result == 0  # 0 means connection succeeded (port in use)
        except Exception:
            return False

    def _read_port(self) -> Optional[int]:
        """Read stored port from file."""
        if self.port_file.exists():
569
            try:
570
                return int(self.port_file.read_text().strip())
571
            except (ValueError, IOError):
572
573
574
575
576
577
                return None
        return None

    def _write_port(self, port: int):
        """Write port to file."""
        self.port_file.write_text(str(port))
578
579
580

    def __enter__(self):
        with FileLock(self.lock_file):
581
582
583
584
585
586
587
588
589
            stored_port = self._read_port()

            # Check if a process is already running on the stored port
            if stored_port is not None and self._is_port_in_use(stored_port):
                # Reuse existing process
                self.port = stored_port
                logging.info(
                    f"[{self.resource_name}] Reusing existing process on port {self.port}"
                )
590
            else:
591
592
593
594
595
596
597
598
599
                # Start new process
                if stored_port is not None:
                    logging.warning(
                        f"[{self.resource_name}] Stale port file: port {stored_port} not in use, starting fresh"
                    )
                self.port = allocate_port(self.start_port)
                self._write_port(self.port)
                self._server = self._create_server(self.port)
                self._server.__enter__()
600
                logging.info(
601
                    f"[{self.resource_name}] Started process on port {self.port}"
602
603
604
605
                )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
606
607
608
        # Never tear down - let the process live until the container dies.
        # This avoids race conditions and simplifies the logic.
        pass
609
610
611
612
613


class SharedEtcdServer(SharedManagedProcess):
    """EtcdServer with file-based reference counting for multi-process sharing."""

614
615
    def __init__(self, request, tmp_path_factory, start_port=2380, timeout=300):
        super().__init__(request, tmp_path_factory, "etcd", start_port, timeout)
616
617
618
        # Create a log directory for session-scoped servers
        self._log_dir = tempfile.mkdtemp(prefix=f"pytest_{self.resource_name}_logs_")

619
    def _create_server(self, port: int) -> ManagedProcess:
620
        """Create EtcdServer instance."""
621
        server = EtcdServer(self.request, port=port, timeout=self.timeout)
622
623
624
625
626
627
628
629
        # Override log_dir since request.node.name is empty in session scope
        server.log_dir = self._log_dir
        return server


class SharedNatsServer(SharedManagedProcess):
    """NatsServer with file-based reference counting for multi-process sharing."""

630
631
632
633
634
635
636
637
638
    def __init__(
        self,
        request,
        tmp_path_factory,
        start_port=4223,
        timeout=300,
        disable_jetstream=False,
    ):
        super().__init__(request, tmp_path_factory, "nats", start_port, timeout)
639
640
        # Create a log directory for session-scoped servers
        self._log_dir = tempfile.mkdtemp(prefix=f"pytest_{self.resource_name}_logs_")
641
        self._disable_jetstream = disable_jetstream
642

643
    def _create_server(self, port: int) -> ManagedProcess:
644
        """Create NatsServer instance."""
645
646
647
648
649
650
        server = NatsServer(
            self.request,
            port=port,
            timeout=self.timeout,
            disable_jetstream=self._disable_jetstream,
        )
651
652
653
654
655
        # Override log_dir since request.node.name is empty in session scope
        server.log_dir = self._log_dir
        return server


656
@pytest.fixture
657
def discovery_backend(request):
658
    """
659
    Discovery backend for runtime. Defaults to "etcd".
660

661
662
    To iterate over multiple backends in a test:
        @pytest.mark.parametrize("discovery_backend", ["file", "etcd"], indirect=True)
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
        def test_example(runtime_services):
            ...
    """
    return getattr(request, "param", "etcd")


@pytest.fixture
def request_plane(request):
    """
    Request plane for runtime. Defaults to "nats".

    To iterate over multiple transports in a test:
        @pytest.mark.parametrize("request_plane", ["nats", "tcp"], indirect=True)
        def test_example(runtime_services):
            ...
    """
    return getattr(request, "param", "nats")


682
@pytest.fixture
683
def durable_kv_events(request):
684
    """
685
    Whether to use durable KV events via JetStream. Defaults to False (NATS Core mode).
686

687
    When False (default):
688
    - NATS server starts without JetStream (-js flag omitted) for faster startup
689
    - Workers use local indexer mode (NATS Core / fire-and-forget events)
690

691
692
693
    When True:
    - NATS server starts with JetStream for durable KV event distribution
    - Workers use --durable-kv-events flag to publish to JetStream
694

695
696
    To use JetStream mode:
        @pytest.mark.parametrize("durable_kv_events", [True], indirect=True)
697
698
699
700
701
702
        def test_example(runtime_services_dynamic_ports):
            ...
    """
    return getattr(request, "param", False)


Neelay Shah's avatar
Neelay Shah committed
703
@pytest.fixture()
704
def runtime_services(request, discovery_backend, request_plane):
705
    """
706
    Start runtime services (NATS and/or etcd) based on discovery_backend and request_plane.
707

708
    - If discovery_backend != "etcd", etcd is not started (returns None)
709
    - If request_plane != "nats", NATS is not started (returns None)
710
711

    Returns a tuple of (nats_process, etcd_process) where each has a .port attribute.
712
    """
713
    # Port cleanup is now handled in NatsServer and EtcdServer __exit__ methods
714
    if request_plane == "nats" and discovery_backend == "etcd":
715
716
717
718
719
720
        with NatsServer(request) as nats_process:
            with EtcdServer(request) as etcd_process:
                yield nats_process, etcd_process
    elif request_plane == "nats":
        with NatsServer(request) as nats_process:
            yield nats_process, None
721
    elif discovery_backend == "etcd":
Neelay Shah's avatar
Neelay Shah committed
722
        with EtcdServer(request) as etcd_process:
723
724
725
            yield None, etcd_process
    else:
        yield None, None
726
727


728
@pytest.fixture()
729
730
731
def runtime_services_dynamic_ports(
    request, discovery_backend, request_plane, durable_kv_events
):
732
733
734
735
736
737
    """Provide NATS and Etcd servers with truly dynamic ports per test.

    This fixture actually allocates dynamic ports by passing port=0 to the servers.
    It also sets the NATS_SERVER and ETCD_ENDPOINTS environment variables so that
    Dynamo processes can find the services on the dynamic ports.

738
739
740
741
742
    xdist/parallel safety:
    - Function-scoped: each test gets its own NATS/etcd instances and ports.
    - Each pytest-xdist worker runs tests in a separate process, so env vars do not
      leak across workers.

743
    - If discovery_backend != "etcd", etcd is not started (returns None)
744
745
    - NATS is always started when etcd is used, because KV events require NATS
      regardless of the request_plane (tcp/nats only affects request transport)
746
    - NATS Core mode (no JetStream) is the default; JetStream is enabled when durable_kv_events=True
747
748
749
750
751
752

    Returns a tuple of (nats_process, etcd_process) where each has a .port attribute.
    """
    import os

    # Port cleanup is now handled in NatsServer and EtcdServer __exit__ methods
753
    # Always start NATS when etcd is used - KV events require NATS regardless of request_plane
754
    # When durable_kv_events=False (default), disable JetStream for faster startup
755
    if discovery_backend == "etcd":
756
        with NatsServer(
757
            request, port=0, disable_jetstream=not durable_kv_events
758
        ) as nats_process:
759
            with EtcdServer(request, port=0) as etcd_process:
760
761
762
763
764
                # Save original env vars (may be set by session-scoped fixture)
                orig_nats = os.environ.get("NATS_SERVER")
                orig_etcd = os.environ.get("ETCD_ENDPOINTS")

                # Set environment variables for this test's dynamic ports
765
766
767
768
769
                os.environ["NATS_SERVER"] = f"nats://localhost:{nats_process.port}"
                os.environ["ETCD_ENDPOINTS"] = f"http://localhost:{etcd_process.port}"

                yield nats_process, etcd_process

770
771
772
773
774
775
776
777
778
                # Restore original env vars (or remove if they weren't set)
                if orig_nats is not None:
                    os.environ["NATS_SERVER"] = orig_nats
                else:
                    os.environ.pop("NATS_SERVER", None)
                if orig_etcd is not None:
                    os.environ["ETCD_ENDPOINTS"] = orig_etcd
                else:
                    os.environ.pop("ETCD_ENDPOINTS", None)
779
    elif request_plane == "nats":
780
        with NatsServer(
781
            request, port=0, disable_jetstream=not durable_kv_events
782
783
        ) as nats_process:
            orig_nats = os.environ.get("NATS_SERVER")
784
785
            os.environ["NATS_SERVER"] = f"nats://localhost:{nats_process.port}"
            yield nats_process, None
786
787
788
789
            if orig_nats is not None:
                os.environ["NATS_SERVER"] = orig_nats
            else:
                os.environ.pop("NATS_SERVER", None)
790
791
792
793
    else:
        yield None, None


794
795
796
797
@pytest.fixture(scope="session")
def runtime_services_session(request, tmp_path_factory):
    """Session-scoped fixture that provides shared NATS and etcd instances for all tests.

798
799
800
    Uses file locking to coordinate between pytest-xdist worker processes.
    First worker starts services on dynamic ports, subsequent workers reuse them.
    Services are never torn down (live until container dies) to avoid race conditions.
801

802
803
    This fixture is xdist-safe when tests use unique namespaces (e.g. random suffixes)
    and do not assume exclusive access to global streams/keys.
804

805
806
    For tests that need to restart NATS (e.g. indexer sync), use `runtime_services_dynamic_ports`
    which provides per-test isolated instances.
807
808
809
    """
    with SharedNatsServer(request, tmp_path_factory) as nats:
        with SharedEtcdServer(request, tmp_path_factory) as etcd:
810
811
812
813
            # Set environment variables for Rust/Python runtime to use
            os.environ["NATS_SERVER"] = f"nats://localhost:{nats.port}"
            os.environ["ETCD_ENDPOINTS"] = f"http://localhost:{etcd.port}"

814
815
            yield nats, etcd

816
817
818
819
            # Clean up environment variables
            os.environ.pop("NATS_SERVER", None)
            os.environ.pop("ETCD_ENDPOINTS", None)

820

821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
@pytest.fixture
def file_storage_backend():
    """Fixture that sets up and tears down file storage backend.

    Creates a temporary directory for file-based KV storage and sets
    the DYN_FILE_KV environment variable. Cleans up after the test.
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        old_env = os.environ.get("DYN_FILE_KV")
        os.environ["DYN_FILE_KV"] = tmpdir
        logging.info(f"Set up file storage backend in: {tmpdir}")
        yield tmpdir
        # Cleanup
        if old_env is not None:
            os.environ["DYN_FILE_KV"] = old_env
        else:
            os.environ.pop("DYN_FILE_KV", None)
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871


########################################################
# Shared Port Allocation (Dynamo deployments)
########################################################


@pytest.fixture(scope="function")
def num_system_ports(request) -> int:
    """Number of system ports to allocate for this test.

    Default: 1 port.

    Tests that need multiple system ports (e.g. SYSTEM_PORT1 + SYSTEM_PORT2) must
    explicitly request them via indirect parametrization:
      @pytest.mark.parametrize("num_system_ports", [2], indirect=True)
    """
    return getattr(request, "param", 1)


@pytest.fixture(scope="function")
def dynamo_dynamic_ports(num_system_ports) -> Generator[ServicePorts, None, None]:
    """Allocate per-test ports for Dynamo deployments.

    - frontend_port: OpenAI-compatible HTTP/gRPC ingress (dynamo.frontend)
    - system_ports: List of worker metrics/system ports (configurable count via num_system_ports)
    """
    frontend_port = allocate_port(DefaultPort.FRONTEND.value)
    system_port_list = allocate_ports(num_system_ports, DefaultPort.SYSTEM1.value)
    all_ports = [frontend_port, *system_port_list]
    try:
        yield ServicePorts(frontend_port=frontend_port, system_ports=system_port_list)
    finally:
        deallocate_ports(all_ports)