test: basic end to end (#1339)

Signed-off-by: Neelay Shah <neelays@nvidia.com>
Co-authored-by: pvijayakrish <pvijayakrish@nvidia.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

container/deps/requirements.test.txt

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+psutil>=5.0.0
 pyright
 pytest
 pytest-asyncio

pyproject.toml

@@ -156,7 +156,16 @@ markers = [
     "pre_merge: marks tests to run before merging",
     "nightly: marks tests to run nightly",
     "weekly: marks tests to run weekly",
-    "gpu: marks tests to run on GPU"
+    "gpu_1: marks tests to run on GPU",
+    "gpu_2: marks tests to run on 2GPUs",
+    "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",
+    "vllm: marks tests as requiring vllm",
+    "tensorrtllm: marks tests as requiring tensorrtllm",
+    "sglang: marks tests as requiring sglang",
+    "slow: marks tests as known to be slow"
 ]
 
 # Linting/formatting

tests/README.md

+# Dynamo Testing Framework
+
+## Overview
+
+This document outlines the testing framework for the Dynamo runtime system, including test discovery, organization, and best practices.
+
+## Directory Structure
+
+```bash
+tests/
+├── serve/              # E2E tests using dynamo serve
+│   ├── conftest.py     # test fixtures as needed for specific test area
+├── run/                # E2E tests using dynamo run
+│   ├── conftest.py     # test fixtures as needed for specific test area
+├── conftest.py         # Shared fixtures and configuration
+└── README.md           # This file
+```
+
+## Test Discovery
+
+Pytest automatically discovers tests based on their naming convention. All test files must follow this pattern:
+
+```bash
+test_<component_or_flow>.py
+```
+
+Where:
+- `component_or_flow`: The component or flow being tested (e.g., planner, kv_router)
+  - For e2e tests, this could be the API or simply "dynamo"
+
+## Running Tests
+
+To run all tests:
+```bash
+pytest
+```
+
+To run only specific tests:
+```bash
+# Run only vLLM tests
+pytest -v -m vllm
+
+# Run only e2e tests
+pytest -v -m e2e
+
+# Run tests for a specific component
+pytest -v -m planner
+
+# Run with print statements visible
+pytest -s
+```
+
+## Test Markers
+
+Markers help control which tests run under different conditions. Add these decorators to your test functions:
+
+### Frequency-based markers
+- `@pytest.mark.nightly` - Tests run nightly
+- `@pytest.mark.weekly` - Tests run weekly
+- `@pytest.mark.pre_merge` - Tests run before merging PRs
+
+### Role-based markers
+- `@pytest.mark.e2e` - End-to-end tests
+- `@pytest.mark.integration` - Integration tests
+- `@pytest.mark.unit` - Unit tests
+- `@pytest.mark.stress` - Stress/load tests
+- `@pytest.mark.benchmark` - Performance benchmark tests
+
+### Component-specific markers
+- `@pytest.mark.vllm` - Framework tests
+- `@pytest.mark.sglang` - Framework tests
+- `@pytest.mark.tensorrtllm` - Framework tests
+- `@pytest.mark.planner` - Planner component tests
+- `@pytest.mark.kv_router` - KV Router component tests
+- etc.
+
+### Execution-related markers
+- `@pytest.mark.slow` - Tests that take a long time to run
+- `@pytest.mark.skip(reason="Example: KV Manager is under development")` - Skip these tests
+- `@pytest.mark.xfail(reason="Expected to fail because...")` - Tests expected to fail
+
+## Environment Setup
+
+Tests are designed to run in the appropriate framework container built
+via ```./container/build.sh --framework X``` and run via
+```./container/run.sh --mount-workspace -it -- pytest```.
+
+
+### Environment Variables
+- `HF_TOKEN` - Your HuggingFace API token to avoid rate limits
+  - Get a token from [HuggingFace Settings](https://huggingface.co/settings/tokens)
+  - Set it before running tests: `export HF_TOKEN=your_token_here`
+
+### Model Download Cache
+
+The tests will automatically use a local cache at `~/.cache/huggingface` to avoid
+repeated downloads of model files. This cache is shared across test runs to improve performance.
+

tests/__init__.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

tests/conftest.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+import tempfile
+
+import pytest
+
+from tests.utils.managed_process import ManagedProcess
+
+# Custom format inspired by your example
+LOG_FORMAT = "[TEST] %(asctime)s %(levelname)s %(name)s: %(message)s"
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format=LOG_FORMAT,
+    datefmt="%Y-%m-%dT%H:%M:%S",  # ISO 8601 UTC format
+)
+
+
+class EtcdServer(ManagedProcess):
+    def __init__(self, request, port=2379, timeout=300):
+        port_string = str(port)
+        etcd_env = os.environ.copy()
+        etcd_env["ALLOW_NONE_AUTHENTICATION"] = "yes"
+        data_dir = tempfile.mkdtemp(prefix="etcd_")
+        command = [
+            "etcd",
+            "--listen-client-urls",
+            f"http://0.0.0.0:{port_string}",
+            "--advertise-client-urls",
+            f"http://0.0.0.0:{port_string}",
+            "--data-dir",
+            data_dir,
+        ]
+        super().__init__(
+            env=etcd_env,
+            command=command,
+            timeout=timeout,
+            display_output=False,
+            health_check_ports=[port],
+            data_dir=tempfile.mkdtemp(prefix="etcd_"),
+            log_dir=request.node.name,
+        )
+
+
+class NatsServer(ManagedProcess):
+    def __init__(self, request, port=4222, timeout=300):
+        data_dir = tempfile.mkdtemp(prefix="nats_")
+        command = ["nats-server", "-js", "--trace", "--store_dir", data_dir]
+        super().__init__(
+            command=command,
+            timeout=timeout,
+            display_output=False,
+            data_dir=data_dir,
+            health_check_ports=[port],
+            log_dir=request.node.name,
+        )
+
+
+@pytest.fixture()
+def runtime_services(request):
+    with NatsServer(request) as nats_process:
+        with EtcdServer(request) as etcd_process:
+            yield nats_process, etcd_process

tests/serve/__init__.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

tests/serve/conftest.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

tests/serve/test_dynamo_serve.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import logging
+import os
+import time
+
+import pytest
+import requests
+
+from tests.utils.deployment_graph import (
+    DeploymentGraph,
+    Payload,
+    completions_response_handler,
+)
+from tests.utils.managed_process import ManagedProcess
+
+text_prompt = "Tell me a short joke about AI."
+
+multimodal_payload = Payload(
+    payload={
+        "model": "llava-hf/llava-1.5-7b-hf",
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "What is in this image?"},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": "http://images.cocodataset.org/test2017/000000155781.jpg"
+                        },
+                    },
+                ],
+            }
+        ],
+        "max_tokens": 300,  # Reduced from 500
+        "stream": False,
+    },
+    expected_log=[],
+    expected_response=["bus"],
+)
+
+text_payload = Payload(
+    payload={
+        "model": "deepseek-ai/DeepSeek-R1-Distill-Llama-8B",
+        "messages": [
+            {
+                "role": "user",
+                "content": text_prompt,  # Shorter prompt
+            }
+        ],
+        "max_tokens": 150,  # Reduced from 500
+        "temperature": 0.1,
+        "seed": 0,
+    },
+    expected_log=[],
+    expected_response=["AI"],
+)
+
+deployment_graphs = {
+    "agg": (
+        DeploymentGraph(
+            module="graphs.agg:Frontend",
+            config="configs/agg.yaml",
+            directory="/workspace/examples/llm",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_1, pytest.mark.vllm],
+        ),
+        text_payload,
+    ),
+    "sglang_agg": (
+        DeploymentGraph(
+            module="graphs.agg:Frontend",
+            config="configs/agg.yaml",
+            directory="/workspace/examples/sglang",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_1, pytest.mark.sglang],
+        ),
+        text_payload,
+    ),
+    "disagg": (
+        DeploymentGraph(
+            module="graphs.disagg:Frontend",
+            config="configs/disagg.yaml",
+            directory="/workspace/examples/llm",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_2, pytest.mark.vllm],
+        ),
+        text_payload,
+    ),
+    "agg_router": (
+        DeploymentGraph(
+            module="graphs.agg_router:Frontend",
+            config="configs/agg_router.yaml",
+            directory="/workspace/examples/llm",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_1, pytest.mark.vllm],
+        ),
+        text_payload,
+    ),
+    "disagg_router": (
+        DeploymentGraph(
+            module="graphs.disagg_router:Frontend",
+            config="configs/disagg_router.yaml",
+            directory="/workspace/examples/llm",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_2, pytest.mark.vllm],
+        ),
+        text_payload,
+    ),
+    "multimodal_agg": (
+        DeploymentGraph(
+            module="graphs.agg:Frontend",
+            config="configs/agg.yaml",
+            directory="/workspace/examples/multimodal",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_2, pytest.mark.vllm],
+        ),
+        multimodal_payload,
+    ),
+    "vllm_v1_agg": (
+        DeploymentGraph(
+            module="graphs.agg:Frontend",
+            config="configs/agg.yaml",
+            directory="/workspace/examples/vllm_v1",
+            endpoint="v1/chat/completions",
+            response_handler=completions_response_handler,
+            marks=[pytest.mark.gpu_1, pytest.mark.vllm],
+        ),
+        text_payload,
+    ),
+}
+
+
+class DynamoServeProcess(ManagedProcess):
+    def __init__(self, graph: DeploymentGraph, request, port=8000, timeout=900):
+        command = ["dynamo", "serve", graph.module]
+
+        if graph.config:
+            command.extend(["-f", os.path.join(graph.directory, graph.config)])
+
+        command.extend(["--Frontend.port", str(port)])
+
+        health_check_urls = [(f"http://localhost:{port}/v1/models", self._check_model)]
+
+        if "multimodal" in graph.directory:
+            health_check_urls = []
+
+        self.port = port
+
+        super().__init__(
+            command=command,
+            timeout=timeout,
+            display_output=True,
+            working_dir=graph.directory,
+            health_check_ports=[port],
+            health_check_urls=health_check_urls,
+            stragglers=["http"],
+            log_dir=request.node.name,
+        )
+
+    def _check_model(self, response):
+        try:
+            data = response.json()
+        except ValueError:
+            return False
+        if data.get("data") and len(data["data"]) > 0:
+            return True
+        return False
+
+
+@pytest.fixture(
+    params=[
+        pytest.param("agg", marks=[pytest.mark.vllm, pytest.mark.gpu_1]),
+        pytest.param("agg_router", marks=[pytest.mark.vllm, pytest.mark.gpu_1]),
+        pytest.param("disagg", marks=[pytest.mark.vllm, pytest.mark.gpu_2]),
+        pytest.param("disagg_router", marks=[pytest.mark.vllm, pytest.mark.gpu_2]),
+        pytest.param("multimodal_agg", marks=[pytest.mark.vllm, pytest.mark.gpu_2]),
+        #        pytest.param("sglang", marks=[pytest.mark.sglang, pytest.mark.gpu_2]),
+    ]
+)
+def deployment_graph_test(request):
+    """
+    Fixture that provides different deployment graph test configurations.
+    """
+    return deployment_graphs[request.param]
+
+
+@pytest.mark.e2e
+@pytest.mark.slow
+def test_serve_deployment(deployment_graph_test, request, runtime_services):
+    """
+    Test dynamo serve deployments with different graph configurations.
+    """
+
+    # runtime_services is used to start nats and etcd
+
+    logger = logging.getLogger(request.node.name)
+    logger.info("Starting test_deployment")
+
+    deployment_graph, payload = deployment_graph_test
+
+    with DynamoServeProcess(deployment_graph, request) as server_process:
+        url = f"http://localhost:{server_process.port}/{deployment_graph.endpoint}"
+        start_time = time.time()
+        retry_delay = 5
+        elapsed = 0.0
+        while time.time() - start_time < deployment_graph.timeout:
+            elapsed = time.time() - start_time
+            try:
+                response = requests.post(
+                    url,
+                    json=payload.payload,
+                    timeout=deployment_graph.timeout - elapsed,
+                )
+            except (requests.RequestException, requests.Timeout) as e:
+                logger.warning("Retrying due to Request failed: %s", e)
+                time.sleep(retry_delay)
+                continue
+            logger.info("Response%r", response)
+            if response.status_code == 500:
+                error = response.json().get("error", "")
+                if "no instances" in error:
+                    logger.warning("Retrying due to no instances available")
+                    time.sleep(retry_delay)
+                    continue
+            if response.status_code == 404:
+                error = response.json().get("error", "")
+                if "Model not found" in error:
+                    logger.warning("Retrying due to model not found")
+                    time.sleep(retry_delay)
+                    continue
+            # Process the response
+            if response.status_code != 200:
+                logger.error(
+                    "Service returned status code %s: %s",
+                    response.status_code,
+                    response.text,
+                )
+                pytest.fail(
+                    "Service returned status code %s: %s"
+                    % (response.status_code, response.text)
+                )
+            else:
+                break
+        else:
+            logger.error(
+                "Service did not return a successful response within %s s",
+                deployment_graph.timeout,
+            )
+            pytest.fail(
+                "Service did not return a successful response within %s s"
+                % deployment_graph.timeout
+            )
+
+        content = deployment_graph.response_handler(response)
+
+        logger.info("Received Content: %s", content)
+
+        # Check for expected responses
+        assert content, "Empty response content"
+
+        for expected in payload.expected_response:
+            assert expected in content, "Expected '%s' not found in response" % expected

tests/utils/deployment_graph.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+
+@dataclass
+class DeploymentGraph:
+    """
+    Represents a deployment graph configuration for testing.
+    """
+
+    module: str
+    config: str
+    directory: str
+    endpoint: str
+    response_handler: Callable[[Any], str]
+    timeout: int = 900
+    marks: Optional[List[Any]] = field(default_factory=list)
+
+
+@dataclass
+class Payload:
+    """
+    Represents a test payload with expected response and log patterns.
+    """
+
+    payload: Dict[str, Any]
+    expected_response: List[str]
+    expected_log: List[str]
+
+
+def completions_response_handler(response):
+    """
+    Process chat completions API responses.
+    """
+    if response.status_code != 200:
+        return ""
+    result = response.json()
+    assert "choices" in result, "Missing 'choices' in response"
+    assert len(result["choices"]) > 0, "Empty choices in response"
+    assert "message" in result["choices"][0], "Missing 'message' in first choice"
+    assert "content" in result["choices"][0]["message"], "Missing 'content' in message"
+    return result["choices"][0]["message"]["content"]

tests/utils/managed_process.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+import shutil
+import socket
+import subprocess
+import time
+from dataclasses import dataclass, field
+from typing import Any, List, Optional
+
+import psutil
+import requests
+
+
+@dataclass
+class ManagedProcess:
+    command: List[str]
+    env: Optional[dict] = None
+    health_check_ports: List[int] = field(default_factory=list)
+    health_check_urls: List[Any] = field(default_factory=list)
+    timeout: int = 300
+    working_dir: Optional[str] = None
+    display_output: bool = False
+    data_dir: Optional[str] = None
+    terminate_existing: bool = True
+    stragglers: List[str] = field(default_factory=list)
+    log_dir: str = os.getcwd()
+
+    _logger = logging.getLogger()
+    _command_name = None
+    _log_path = None
+    _tee_proc = None
+    _sed_proc = None
+
+    def __enter__(self):
+        try:
+            self._logger = logging.getLogger(self.__class__.__name__)
+            self._command_name = self.command[0]
+            os.makedirs(self.log_dir, exist_ok=True)
+            log_name = f"{self._command_name}.log.txt"
+            self._log_path = os.path.join(self.log_dir, log_name)
+
+            if self.data_dir:
+                self._remove_directory(self.data_dir)
+
+            self._terminate_existing()
+            self._start_process()
+            elapsed = self._check_ports(self.timeout)
+            self._check_urls(self.timeout - elapsed)
+
+            return self
+
+        except Exception as e:
+            self.__exit__(None, None, None)
+            raise e
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        process_list = [self.proc, self._tee_proc, self._sed_proc]
+        for process in process_list:
+            if process:
+                if process.stdout:
+                    process.stdout.close()
+                if process.stdin:
+                    process.stdin.close()
+                self._terminate_process_tree(process.pid)
+                process.wait()
+        if self.data_dir:
+            self._remove_directory(self.data_dir)
+
+        for ps_process in psutil.process_iter(["name", "cmdline"]):
+            try:
+                if ps_process.name() in self.stragglers:
+                    self._terminate_process_tree(ps_process.pid)
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                # Process may have terminated or become inaccessible during iteration
+                pass
+
+    def _start_process(self):
+        assert self._command_name
+        assert self._log_path
+
+        self._logger.info(
+            "Running command: %s in %s",
+            " ".join(self.command),
+            self.working_dir or os.getcwd(),
+        )
+
+        stdin = subprocess.DEVNULL
+        stdout = subprocess.PIPE
+        stderr = subprocess.STDOUT
+
+        if self.display_output:
+            self.proc = subprocess.Popen(
+                self.command,
+                env=self.env or os.environ.copy(),
+                cwd=self.working_dir,
+                stdin=stdin,
+                stdout=stdout,
+                stderr=stderr,
+            )
+            self._sed_proc = subprocess.Popen(
+                ["sed", "-u", f"s/^/[{self._command_name.upper()}] /"],
+                stdin=self.proc.stdout,
+                stdout=subprocess.PIPE,
+            )
+
+            self._tee_proc = subprocess.Popen(
+                ["tee", self._log_path], stdin=self._sed_proc.stdout
+            )
+
+        else:
+            with open(self._log_path, "w", encoding="utf-8") as f:
+                self.proc = subprocess.Popen(
+                    self.command,
+                    env=self.env or os.environ.copy(),
+                    cwd=self.working_dir,
+                    stdin=stdin,
+                    stdout=stdout,
+                    stderr=stderr,
+                )
+
+                self._sed_proc = subprocess.Popen(
+                    ["sed", "-u", f"s/^/[{self._command_name.upper()}] /"],
+                    stdin=self.proc.stdout,
+                    stdout=f,
+                )
+            self._tee_proc = None
+
+    def _remove_directory(self, path: str) -> None:
+        """Remove a directory."""
+        try:
+            shutil.rmtree(path, ignore_errors=True)
+        except (OSError, IOError) as e:
+            self._logger.warning("Warning: Failed to remove directory %s: %s", path, e)
+
+    def _check_ports(self, timeout):
+        elapsed = 0.0
+        for port in self.health_check_ports:
+            elapsed += self._check_port(port, timeout - elapsed)
+        return elapsed
+
+    def _check_port(self, port, timeout=30, sleep=0.1):
+        """Check if a port is open on localhost."""
+        start_time = time.time()
+        self._logger.info("Checking Port: %s", port)
+        elapsed = 0.0
+        while elapsed < timeout:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                if s.connect_ex(("localhost", port)) == 0:
+                    self._logger.info("SUCCESS: Check Port: %s", port)
+                    return time.time() - start_time
+            time.sleep(sleep)
+            elapsed = time.time() - start_time
+        self._logger.error("FAILED: Check Port: %s", port)
+        raise RuntimeError("FAILED: Check Port: %s" % port)
+
+    def _check_urls(self, timeout):
+        elapsed = 0.0
+        for url in self.health_check_urls:
+            elapsed += self._check_url(url, timeout - elapsed)
+        return elapsed
+
+    def _check_url(self, url, timeout=30, sleep=0.1):
+        if isinstance(url, tuple):
+            response_check = url[1]
+            url = url[0]
+        else:
+            response_check = None
+        start_time = time.time()
+        self._logger.info("Checking URL %s", url)
+        elapsed = 0.0
+        while elapsed < timeout:
+            try:
+                response = requests.get(url, timeout=timeout - elapsed)
+                if response.status_code == 200:
+                    if response_check is None or response_check(response):
+                        self._logger.info("SUCCESS: Check URL: %s", url)
+                        return time.time() - start_time
+            except requests.RequestException as e:
+                self._logger.warning("URL check failed: %s", e)
+            time.sleep(sleep)
+            elapsed = time.time() - start_time
+
+        self._logger.error("FAILED: Check URL: %s", url)
+        raise RuntimeError("FAILED: Check URL: %s" % url)
+
+    def _terminate_existing(self):
+        if self.terminate_existing:
+            self._logger.info("Terminating Existing %s", self._command_name)
+            for proc in psutil.process_iter(["name", "cmdline"]):
+                if proc.name() == self._command_name or proc.name() in self.stragglers:
+                    self._terminate_process_tree(proc.pid)
+
+    def _terminate_process(self, process):
+        try:
+            self._logger.info("Terminating %s", process)
+            process.terminate()
+        except psutil.AccessDenied:
+            self._logger.warning("Access denied for PID %s", process.pid)
+        except psutil.NoSuchProcess:
+            self._logger.warning("PID %s no longer exists", process.pid)
+        except psutil.TimeoutExpired:
+            self._logger.warning(
+                "PID %s did not terminate before timeout, killing", process.pid
+            )
+            process.kill()
+
+    def _terminate_process_tree(self, pid):
+        try:
+            parent = psutil.Process(pid)
+            for child in parent.children(recursive=True):
+                self._terminate_process(child)
+            self._terminate_process(parent)
+        except psutil.NoSuchProcess:
+            # Process already terminated
+            pass
+
+
+def main():
+    with ManagedProcess(
+        command=[
+            "dynamo",
+            "run",
+            "in=http",
+            "out=vllm",
+            "deepseek-ai/DeepSeek-R1-Distill-Llama-8B",
+        ],
+        display_output=True,
+        terminate_existing=True,
+        health_check_ports=[8080],
+        health_check_urls=["http://localhost:8080/v1/models"],
+        timeout=10,
+    ):
+        time.sleep(60)
+        pass
+
+
+if __name__ == "__main__":
+    main()