Coverage for tests/streamwise/test_streamwise_auto_deploy.py: 100%
107 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2Tests for the auto-deploy API endpoints in streamwise.py.
4Covers:
5- POST /api/auto_deploy — returns optimized plan.
6- POST /api/auto_deploy/confirm — deploys the plan.
7- GET /api/auto_deploy/workflows — lists available options.
8- Error cases (missing fields, invalid inputs).
9"""
11from __future__ import annotations
13import sys
15import pytest
17from http import HTTPStatus
18from unittest.mock import patch
20from tests.test_utils import temp_sys_path
21from tests.k8s_mock import K8sMock
23mock_k8s = K8sMock()
25mock_modules = {}
26mock_modules.update(mock_k8s.get_sub_modules())
28import streamwise.http_session_manager # noqa: F401 — registers the streamwise package
30# Permanently inject K8s mocks into sys.modules (not via context manager)
31# so that simulator modules loaded alongside streamwise remain importable
32# after setup completes.
33_original_modules = {}
34for mod_name, mock_mod in mock_modules.items():
35 _original_modules[mod_name] = sys.modules.get(mod_name)
36 sys.modules[mod_name] = mock_mod
38with temp_sys_path("streamwise"):
39 from streamwise import streamwise as sw
42def _get_client(): # type: ignore[no-untyped-def]
43 app = sw.app
44 return app.test_client()
47@pytest.fixture(scope="function", autouse=True)
48def setup_k8s_cluster() -> None:
49 sw.k8s_cluster = "unittest"
50 sw.use_https = False
53# ---------------------------------------------------------------------------
54# GET /api/auto_deploy/workflows
55# ---------------------------------------------------------------------------
57@pytest.mark.asyncio
58async def test_auto_deploy_workflows() -> None:
59 """Should return available workflows and GPU types."""
60 client = _get_client()
61 response = await client.get("/api/auto_deploy/workflows")
62 assert response.status_code == HTTPStatus.OK
63 data = await response.get_json()
64 assert "workflows" in data
65 assert "gpu_types" in data
66 assert "streamcast" in data["workflows"]
67 assert "A100" in data["gpu_types"]
70# ---------------------------------------------------------------------------
71# POST /api/auto_deploy
72# ---------------------------------------------------------------------------
74@pytest.mark.asyncio
75async def test_auto_deploy_success() -> None:
76 """Valid request returns an optimized deployment plan."""
77 fake_json = {
78 "workflow_name": "streamcast",
79 "gpu_budget": {"A100": 8},
80 "metrics": {"total_time_s": 3.5, "ttff_s": 1.0, "cost": 12.0, "gpus_used": {"A100": 3}},
81 "specs": [
82 {"container_name": "gemma", "cpu": 4, "memory_gib": 16,
83 "ephemeral_storage_gib": 10, "gpu": 1, "gpu_type": "A100", "mig_profile": None},
84 {"container_name": "flux", "cpu": 4, "memory_gib": 16,
85 "ephemeral_storage_gib": 10, "gpu": 2, "gpu_type": "A100", "mig_profile": None},
86 ],
87 }
88 # Patch on the actual module object that streamwise.py holds a reference to.
89 with patch.object(sw.allocator_bridge, "run_allocator") as mock_alloc, \
90 patch.object(sw.allocator_bridge, "deployment_plan_to_json", return_value=fake_json):
91 mock_alloc.return_value = "fake_plan"
92 client = _get_client()
93 response = await client.post(
94 "/api/auto_deploy",
95 json={
96 "gpu_budget": {"A100": 8},
97 "workflow": "streamcast",
98 },
99 )
100 assert response.status_code == HTTPStatus.OK
101 data = await response.get_json()
102 assert "specs" in data
103 assert "metrics" in data
104 assert len(data["specs"]) == 2
105 assert data["metrics"]["total_time_s"] == 3.5
108@pytest.mark.asyncio
109async def test_auto_deploy_missing_gpu_budget() -> None:
110 """Missing gpu_budget field returns 400."""
111 client = _get_client()
112 response = await client.post(
113 "/api/auto_deploy",
114 json={"workflow": "streamcast"},
115 )
116 assert response.status_code == HTTPStatus.BAD_REQUEST
119@pytest.mark.asyncio
120async def test_auto_deploy_missing_workflow() -> None:
121 """Missing workflow field returns 400."""
122 client = _get_client()
123 response = await client.post(
124 "/api/auto_deploy",
125 json={"gpu_budget": {"A100": 8}},
126 )
127 assert response.status_code == HTTPStatus.BAD_REQUEST
130@pytest.mark.asyncio
131async def test_auto_deploy_invalid_workflow() -> None:
132 """Invalid workflow name returns 400."""
133 client = _get_client()
134 response = await client.post(
135 "/api/auto_deploy",
136 json={
137 "gpu_budget": {"A100": 8},
138 "workflow": "nonexistent",
139 },
140 )
141 assert response.status_code == HTTPStatus.BAD_REQUEST
142 data = await response.get_json()
143 assert "error" in data
146@pytest.mark.asyncio
147async def test_auto_deploy_insufficient_gpus() -> None:
148 """Too few GPUs returns 400."""
149 client = _get_client()
150 response = await client.post(
151 "/api/auto_deploy",
152 json={
153 "gpu_budget": {"A100": 2},
154 "workflow": "streamcast",
155 },
156 )
157 assert response.status_code == HTTPStatus.BAD_REQUEST
160@pytest.mark.asyncio
161async def test_auto_deploy_no_json_body() -> None:
162 """No JSON body returns 400."""
163 client = _get_client()
164 response = await client.post("/api/auto_deploy")
165 assert response.status_code == HTTPStatus.BAD_REQUEST
168# ---------------------------------------------------------------------------
169# POST /api/auto_deploy/confirm
170# ---------------------------------------------------------------------------
172@pytest.mark.asyncio
173async def test_auto_deploy_confirm_success() -> None:
174 """Valid confirm request deploys containers."""
175 client = _get_client()
176 specs = [
177 {
178 "container_name": "gemma",
179 "cpu": 16,
180 "memory_gib": 192,
181 "ephemeral_storage_gib": 64,
182 "gpu": 2,
183 "gpu_type": "a100",
184 "mig_profile": None,
185 },
186 {
187 "container_name": "flux",
188 "cpu": 12,
189 "memory_gib": 128,
190 "ephemeral_storage_gib": 64,
191 "gpu": 2,
192 "gpu_type": "a100",
193 "mig_profile": None,
194 },
195 ]
196 with patch.object(sw.pod_manager, "add_pod") as mock_add_pod:
197 response = await client.post(
198 "/api/auto_deploy/confirm",
199 json={"specs": specs},
200 )
201 # Should succeed without invoking the real pod_manager.add_pod flow
202 assert response.status_code in (HTTPStatus.OK, HTTPStatus.MULTI_STATUS)
203 data = await response.get_json()
204 assert "deployed" in data
205 assert "message" in data
206 assert mock_add_pod.call_count == len(specs)
209@pytest.mark.asyncio
210async def test_auto_deploy_confirm_missing_specs() -> None:
211 """Missing specs returns 400."""
212 client = _get_client()
213 response = await client.post(
214 "/api/auto_deploy/confirm",
215 json={},
216 )
217 assert response.status_code == HTTPStatus.BAD_REQUEST
220@pytest.mark.asyncio
221async def test_auto_deploy_confirm_tracks_add_pod_status_failures() -> None:
222 """Non-2xx add_pod return statuses are surfaced as deployment errors."""
223 client = _get_client()
224 specs = [
225 {"container_name": "gemma", "gpu": 2, "gpu_type": "a100"},
226 {"container_name": "flux", "gpu": 2, "gpu_type": "a100"},
227 ]
228 with patch.object(
229 sw.pod_manager,
230 "add_pod",
231 side_effect=[
232 (None, HTTPStatus.OK),
233 (None, HTTPStatus.BAD_REQUEST),
234 ],
235 ):
236 response = await client.post("/api/auto_deploy/confirm", json={"specs": specs})
238 assert response.status_code == HTTPStatus.MULTI_STATUS
239 data = await response.get_json()
240 assert data["deployed"] == ["gemma"]
241 assert len(data["errors"]) == 1
242 assert "flux" in data["errors"][0]
243 assert "status=400" in data["errors"][0]
246@pytest.mark.asyncio
247async def test_auto_deploy_confirm_empty_specs() -> None:
248 """Empty specs list returns 400."""
249 client = _get_client()
250 response = await client.post(
251 "/api/auto_deploy/confirm",
252 json={"specs": []},
253 )
254 assert response.status_code == HTTPStatus.BAD_REQUEST