Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions src/runpod_flash/core/resources/serverless.py
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,20 @@ def sync_input_fields(self):

return self

@model_validator(mode="after")
def validate_worker_range(self):
"""Ensure worker scaling bounds are valid."""
if (
self.workersMin is not None
and self.workersMax is not None
and self.workersMin > self.workersMax
):
raise ValueError(
f"workersMin ({self.workersMin}) cannot be greater than "
f"workersMax ({self.workersMax})"
)
return self

def _has_cpu_instances(self) -> bool:
"""Check if endpoint has CPU instances configured.

Expand Down
7 changes: 7 additions & 0 deletions tests/unit/resources/test_live_serverless.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,13 @@
class TestLiveServerless:
"""Test LiveServerless (GPU) class behavior."""

def test_live_serverless_workers_min_cannot_exceed_workers_max(self):
with pytest.raises(
ValueError,
match=r"workersMin \(5\) cannot be greater than workersMax \(1\)",
):
LiveServerless(name="broken", workersMin=5, workersMax=1)

def test_live_serverless_gpu_defaults(self):
"""Test LiveServerless uses GPU image and defaults."""
live_serverless = LiveServerless(
Expand Down
7 changes: 7 additions & 0 deletions tests/unit/resources/test_serverless.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,13 @@ async def test_ensure_network_volume_deployed_uses_existing_volume(self):
class TestServerlessResourceValidation:
"""Test field validation and serialization."""

def test_workers_min_cannot_exceed_workers_max(self):
with pytest.raises(
ValueError,
match=r"workersMin \(5\) cannot be greater than workersMax \(1\)",
):
ServerlessResource(name="test", workersMin=5, workersMax=1)

def test_scaler_type_serialization(self):
"""Test ServerlessScalerType enum serialization."""
serverless = ServerlessResource(
Expand Down
Loading