diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 9b557800..c518e8b0 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -164,6 +164,11 @@ class AppsApiClient: def __init__(self, api_key: Optional[str] = None): self._api_key = api_key + async def gpu_pool_members(self) -> Optional[Dict[str, Set[str]]]: + from .gpu import live_pool_members_async + + return await live_pool_members_async(self._api_key) + async def _execute( self, query: str, diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index a0ef47ca..bddd6361 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -7,7 +7,7 @@ import tempfile from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple from pathspec import GitIgnoreSpec @@ -285,6 +285,7 @@ def _deployed_endpoint_input( python_version: str, *, env_name: Optional[str] = None, + pool_members: Optional[Dict[str, Set[str]]] = None, ) -> Dict[str, Any]: """saveEndpoint payload for one deployed queue/api resource. @@ -372,7 +373,7 @@ def _deployed_endpoint_input( else: from .gpu import gpu_ids_value - payload["gpuIds"] = gpu_ids_value(spec.gpu) + payload["gpuIds"] = gpu_ids_value(spec.gpu, pool_members=pool_members) payload["gpuCount"] = spec.gpu_count if spec.min_cuda_version: payload["minCudaVersion"] = spec.min_cuda_version @@ -406,12 +407,16 @@ async def reconcile_endpoints( if h.spec.kind in (ResourceKind.QUEUE, ResourceKind.API) } + from .gpu import needs_gpu_catalog from .registry import resolve_registry_auth from .volume import VolumeResolver, attach_endpoint_volumes resolver = VolumeResolver(client) endpoints: Dict[str, str] = {} for name, handle in sorted(provisionable.items()): + members = None + if needs_gpu_catalog(handle.spec.gpu): + members = await client.gpu_pool_members() payload = _deployed_endpoint_input( app, handle.spec, @@ -419,6 +424,7 @@ async def reconcile_endpoints( build_id, python_version, env_name=environment["name"], + pool_members=members or {}, ) await attach_endpoint_volumes(payload, handle.spec, resolver, app) auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client) diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index c4e95ad1..be7b951a 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -8,11 +8,12 @@ import logging import os as _os -from typing import Dict, List, Optional, Union +from typing import Dict, List, Optional, Set, Union from .api import AppsApiClient from .app import App from .errors import AppError +from .gpu import needs_gpu_catalog from .handles import ApiHandle, FunctionHandle from .shim import runtime_launcher from .spec import ResourceKind, ResourceSpec @@ -107,7 +108,13 @@ def _cpu_locations(instance_ids: List[str]) -> str: GENERATION_ENV = "RUNPOD_DEV_GENERATION" -def _endpoint_input(app: App, spec: ResourceSpec, generation: int = 1) -> Dict: +def _endpoint_input( + app: App, + spec: ResourceSpec, + generation: int = 1, + *, + pool_members: Optional[Dict[str, Set[str]]] = None, +) -> Dict: """build a live endpoint payload with an atomically bound container config.""" spec.validate() @@ -179,7 +186,7 @@ def _endpoint_input(app: App, spec: ResourceSpec, generation: int = 1) -> Dict: else: from .gpu import gpu_ids_value - payload["gpuIds"] = gpu_ids_value(spec.gpu) + payload["gpuIds"] = gpu_ids_value(spec.gpu, pool_members=pool_members) payload["gpuCount"] = spec.gpu_count if spec.min_cuda_version: payload["minCudaVersion"] = spec.min_cuda_version @@ -274,7 +281,13 @@ async def start(self) -> None: if found: self._endpoints[name] = found["id"] self._resource_names[name] = spec.name - payload = _endpoint_input(app, spec, self.generation) + spec.validate() + members = None + if needs_gpu_catalog(spec.gpu): + members = await self.api.gpu_pool_members() + payload = _endpoint_input( + app, spec, self.generation, pool_members=members or {} + ) await self._attach_volumes(payload, spec, app) hardware = ",".join(spec.cpu or spec.gpu or ["any"]) @@ -351,7 +364,13 @@ async def refresh(self, apps: List[App]) -> None: # update survivors (config + generation bump) and create additions for name, (app, handle) in desired.items(): - payload = _endpoint_input(app, handle.spec, self.generation) + handle.spec.validate() + members = None + if needs_gpu_catalog(handle.spec.gpu): + members = await self.api.gpu_pool_members() + payload = _endpoint_input( + app, handle.spec, self.generation, pool_members=members or {} + ) await self._attach_volumes(payload, handle.spec, app) existing_id = self._endpoints.get(name) diff --git a/runpod/apps/gpu.py b/runpod/apps/gpu.py index 79048b22..4b7ba72f 100644 --- a/runpod/apps/gpu.py +++ b/runpod/apps/gpu.py @@ -5,8 +5,13 @@ configured. CpuInstanceType values are backend cpu flavor ids. """ +import logging +import time +from dataclasses import dataclass from enum import Enum -from typing import List, Optional, Union +from typing import Any, Dict, List, Optional, Set, Tuple, Union + +log = logging.getLogger(__name__) class GpuGroup(Enum): @@ -208,24 +213,168 @@ def resolve_gpu_string(value: str) -> List[str]: ) -def gpu_ids_value(gpu: Optional[List[str]]) -> str: +CATALOG_PATH = "/v2/catalog/gpus" +CATALOG_RETRY_SECONDS = 60.0 + + +@dataclass +class _CatalogCache: + """the live catalog's pool membership, or when fetching it last failed.""" + + members: Optional[Dict[str, Set[str]]] = None + failed_at: Optional[float] = None + + def may_fetch(self) -> bool: + """true until a fetch succeeds, except within a minute of a failure.""" + if self.members is not None: + return False + return ( + self.failed_at is None + or time.monotonic() - self.failed_at >= CATALOG_RETRY_SECONDS + ) + + def fail(self, exc: Exception) -> None: + self.failed_at = time.monotonic() + log.warning( + "gpu catalog unavailable (%s); device pins use the static pool " + "table, which can miss newer devices in a pool", + exc, + ) + + +_catalog_caches: Dict[Tuple[str, str], _CatalogCache] = {} + + +def _clear_catalog_cache() -> None: + _catalog_caches.clear() + + +def _catalog_context(api_key: Optional[str] = None) -> Tuple[str, str]: + from ..api.rest import _build_url, _resolve_api_key + from ..error import AuthenticationError + + try: + key = _resolve_api_key(api_key) + except AuthenticationError: + key = "" + return _build_url(CATALOG_PATH), key + + +def _parse_catalog(data: Any) -> Dict[str, Set[str]]: + """pool id -> device ids from a catalog response.""" + gpus = data.get("gpus", []) if isinstance(data, dict) else data + members: Dict[str, Set[str]] = {} + for gpu in gpus: + if gpu.get("pool") and gpu.get("id"): + members.setdefault(gpu["pool"], set()).add(gpu["id"]) + if not members: + raise ValueError("catalog lists no pools") + return members + + +def live_pool_members() -> Optional[Dict[str, Set[str]]]: + """pool id -> device ids, from the platform's gpu catalog. + + the static POOLS_TO_TYPES table lags the platform (pools gain mig + slices and new skus), so exact-device pinning reads the live + membership. returns None when the catalog is unreachable. a result + is kept for the process; a failure is retried after a minute. + """ + from ..api.ctl_commands import get_gpus + + context = _catalog_context() + cache = _catalog_caches.setdefault(context, _CatalogCache()) + if cache.may_fetch(): + try: + cache.members = _parse_catalog(get_gpus(api_key=context[1])) + except Exception as exc: # noqa: BLE001 - offline, auth, or malformed + cache.fail(exc) + return cache.members + + +async def live_pool_members_async( + api_key: Optional[str] = None, +) -> Optional[Dict[str, Set[str]]]: + """fetch pool membership without blocking, cached by catalog url and credentials.""" + from ..api.rest import run_rest_request_async + + context = _catalog_context(api_key) + cache = _catalog_caches.setdefault(context, _CatalogCache()) + if cache.may_fetch(): + try: + data = await run_rest_request_async( + "GET", CATALOG_PATH, api_key=context[1], timeout=10 + ) + cache.members = _parse_catalog(data) + except Exception as exc: + cache.fail(exc) + return cache.members + + +def _pool_members(pool: str, live: Optional[Dict[str, Set[str]]]) -> Set[str]: + if live is not None and pool in live: + return live[pool] + try: + return {t.value for t in POOLS_TO_TYPES.get(GpuGroup(pool), [])} + except ValueError: + return set() + + +def _device_pool(value: str) -> Optional[str]: + """the pool id of a known device name, or None.""" + try: + pool = pool_for_gpu_type(GpuType(value)) + except ValueError: + return None + return pool.value if pool is not None else None + + +def needs_gpu_catalog(gpu: Optional[List[str]]) -> bool: + if not gpu or any(str(g).lower() == "any" for g in gpu): + return False + return any( + (pool := _device_pool(str(g))) is not None and pool not in gpu for g in gpu + ) + + +def gpu_ids_value( + gpu: Optional[List[str]], *, pool_members: Optional[Dict[str, Set[str]]] = None +) -> str: """the gpuIds string for an endpoint payload. "any gpu" (no selection, or the ANY sentinel) means every pool id: - the api has no wildcard and rejects "any". device names map back - to their pool (endpoints select by pool, not device). + the api has no wildcard and rejects "any". endpoints select by + pool, so device names map to their pool, and every other device in + that pool is excluded with a "-" token so workers only + land on the requested devices. pools named directly allow all of + their devices. + + supplying pool_members keeps serialization offline; an empty mapping + uses the static pool table. omitted membership is fetched synchronously. """ if not gpu or any(str(g).lower() == "any" for g in gpu): return ",".join(g.value for g in GpuGroup.all()) pools: List[str] = [] + named_pools: Set[str] = set() + devices: Set[str] = set() for entry in gpu: value = str(entry) - try: - pool = pool_for_gpu_type(GpuType(value)) - value = pool.value if pool is not None else value - except ValueError: - # not a known GpuType: pass the raw pool/id string through - pass + pool = _device_pool(value) + if pool is None: + # a pool id allows every device in it; other ids pass through + named_pools.add(value) + else: + devices.add(value) + value = pool if value not in pools: pools.append(value) - return ",".join(pools) + device_pools = {_device_pool(d) for d in devices} - named_pools + if pool_members is None and device_pools: + pool_members = live_pool_members() + exclusions = sorted( + member + for pool in device_pools + for member in _pool_members(pool, pool_members) + if member not in devices + ) + return ",".join(pools + [f"-{m}" for m in exclusions]) diff --git a/tests/test_apps/conftest.py b/tests/test_apps/conftest.py new file mode 100644 index 00000000..d95469fb --- /dev/null +++ b/tests/test_apps/conftest.py @@ -0,0 +1,15 @@ +"""shared fixtures for apps tests.""" + +from unittest.mock import AsyncMock + +import pytest + +from runpod.apps import gpu as _gpu + + +@pytest.fixture(autouse=True) +def offline_gpu_catalog(monkeypatch): + """payload builders read the live gpu catalog; keep tests offline and + deterministic on the static pool table unless a test overrides it.""" + monkeypatch.setattr(_gpu, "live_pool_members", lambda: None) + monkeypatch.setattr(_gpu, "live_pool_members_async", AsyncMock(return_value=None)) diff --git a/tests/test_apps/test_app.py b/tests/test_apps/test_app.py index e23614be..882b9570 100644 --- a/tests/test_apps/test_app.py +++ b/tests/test_apps/test_app.py @@ -1,11 +1,14 @@ """tests for the App registry and decorators.""" +from unittest.mock import patch + import pytest import runpod from runpod.apps import App, ApiHandle, FunctionHandle, ResourceKind from runpod.apps.app import _REGISTRY, _clear_registry from runpod.apps.errors import InvalidResourceError +from runpod.apps.gpu import live_pool_members as _real_live_pool_members from runpod.apps.markers import get, init, post @@ -382,6 +385,77 @@ def test_unknown_raises_at_decoration(self): with pytest.raises(InvalidResourceError, match="unknown gpu 'B300'"): normalize_gpu("B300") + def test_gpu_ids_excludes_other_live_pool_members(self, monkeypatch): + from runpod.apps import gpu as gpu_mod + + live = { + "ADA_24": {"NVIDIA GeForce RTX 4090"}, + "AMPERE_24": { + "NVIDIA GeForce RTX 3090", + "NVIDIA L4", + "NVIDIA RTX A5000", + "NVIDIA RTX PRO 6000 Blackwell Server Edition MIG 1g.24gb", + }, + } + monkeypatch.setattr(gpu_mod, "live_pool_members", lambda: live) + value = gpu_mod.gpu_ids_value( + ["NVIDIA GeForce RTX 4090", "NVIDIA L4", "NVIDIA RTX A5000"] + ) + assert value == ( + "ADA_24,AMPERE_24,-NVIDIA GeForce RTX 3090," + "-NVIDIA RTX PRO 6000 Blackwell Server Edition MIG 1g.24gb" + ) + + def test_gpu_ids_named_pool_allows_all_members(self, monkeypatch): + from runpod.apps import gpu as gpu_mod + + monkeypatch.setattr( + gpu_mod, + "live_pool_members", + lambda: {"AMPERE_24": {"NVIDIA L4", "NVIDIA GeForce RTX 3090"}}, + ) + assert gpu_mod.gpu_ids_value(["NVIDIA L4", "AMPERE_24"]) == "AMPERE_24" + + def test_gpu_ids_falls_back_to_static_pools(self, monkeypatch): + from runpod.apps import gpu as gpu_mod + + monkeypatch.setattr(gpu_mod, "live_pool_members", lambda: None) + assert gpu_mod.gpu_ids_value(["NVIDIA L4"]) == ( + "AMPERE_24,-NVIDIA GeForce RTX 3090,-NVIDIA RTX A5000" + ) + + def test_live_pool_members_parses_catalog_and_fails_soft(self, monkeypatch): + from unittest.mock import MagicMock + + from runpod.apps import gpu as gpu_mod + + monkeypatch.setattr(runpod, "api_key", "k") + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = { + "gpus": [ + {"id": "NVIDIA L4", "pool": "AMPERE_24"}, + {"id": "NVIDIA A100-SXM4-40GB", "pool": None}, + ] + } + fetch = _real_live_pool_members + gpu_mod._clear_catalog_cache() + with patch("requests.request", return_value=resp) as get: + assert fetch() == {"AMPERE_24": {"NVIDIA L4"}} + assert get.call_args.kwargs["headers"]["Authorization"] == "Bearer k" + gpu_mod._clear_catalog_cache() + with patch("requests.request", side_effect=OSError("offline")) as get: + assert fetch() is None + # a failure is not retried on every lookup + assert fetch() is None + assert get.call_count == 1 + # it is retried once the window passes + cache = next(iter(gpu_mod._catalog_caches.values())) + cache.failed_at -= gpu_mod.CATALOG_RETRY_SECONDS + with patch("requests.request", return_value=resp): + assert fetch() == {"AMPERE_24": {"NVIDIA L4"}} + gpu_mod._clear_catalog_cache() + def test_gpu_ids_maps_devices_to_pools(self): from runpod.apps.gpu import gpu_ids_value diff --git a/tests/test_apps/test_gpu_catalog.py b/tests/test_apps/test_gpu_catalog.py new file mode 100644 index 00000000..a4baf448 --- /dev/null +++ b/tests/test_apps/test_gpu_catalog.py @@ -0,0 +1,246 @@ +"""gpu catalog fetching across async provisioning and synchronous serialization.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from aiohttp import web + +import runpod +from runpod.apps import App, gpu +from runpod.apps.api import AppsApiClient +from runpod.apps.app import _clear_registry +from runpod.apps.deploy import reconcile_endpoints +from runpod.apps.dev import DevSession +from runpod.apps.gpu import live_pool_members as _real_live_pool_members +from runpod.apps.gpu import live_pool_members_async as _real_live_pool_members_async + + +MIG = "NVIDIA RTX PRO 6000 Blackwell Server Edition MIG 1g.24gb" +CATALOG = { + "gpus": [ + {"id": name, "pool": "AMPERE_24"} + for name in ("NVIDIA L4", "NVIDIA GeForce RTX 3090", "NVIDIA RTX A5000", MIG) + ] +} + + +@pytest.fixture(autouse=True) +def clean_catalog(monkeypatch, offline_gpu_catalog): + _clear_registry() + gpu._clear_catalog_cache() + monkeypatch.delenv("RUNPOD_API_KEY", raising=False) + monkeypatch.setattr(runpod, "api_key", None) + monkeypatch.setattr(gpu, "live_pool_members_async", _real_live_pool_members_async) + monkeypatch.setattr( + gpu, "live_pool_members", Mock(side_effect=AssertionError("synchronous fetch")) + ) + yield + gpu._clear_catalog_cache() + _clear_registry() + + +@pytest.fixture +async def catalog(monkeypatch): + state = SimpleNamespace( + requests=[], + started=asyncio.Event(), + release=asyncio.Event(), + status=200, + data=CATALOG, + ) + state.release.set() + + async def handle(request): + state.requests.append((request.path, request.headers.get("Authorization"))) + state.started.set() + await state.release.wait() + return web.json_response(state.data, status=state.status) + + app = web.Application() + app.router.add_get("/v2/catalog/gpus", handle) + app.router.add_get("/custom/v2/catalog/gpus", handle) + runner = web.AppRunner(app, shutdown_timeout=0.1) + await runner.setup() + try: + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + state.url = f"http://127.0.0.1:{site._server.sockets[0].getsockname()[1]}" + other = web.TCPSite(runner, "127.0.0.1", 0) + await other.start() + other_port = other._server.sockets[0].getsockname()[1] + state.other_url = f"http://127.0.0.1:{other_port}" + monkeypatch.setenv("RUNPOD_API_BASE_URL", state.url) + yield state + finally: + state.release.set() + await runner.cleanup() + + +async def provision(operation, client, gpu_selection="L4", cpu=None): + app = App("catalog-test") + + @app.queue(name="work", gpu=gpu_selection, cpu=cpu) + def work(value): + return value + + client.save_endpoint = AsyncMock(return_value={"id": "test-endpoint"}) + client.list_my_endpoints = AsyncMock(return_value=[]) + if operation == "reconcile": + await reconcile_endpoints( + client, app, {"id": "env", "name": "test"}, "build", "3.12" + ) + else: + session = DevSession([app], api=client) + session._attach_volumes = AsyncMock() + if operation == "start": + await session.start() + else: + await session.refresh([app]) + return client.save_endpoint.call_args.args[0] + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +@pytest.mark.parametrize("global_key", [None, "unrelated-global-key"]) +async def test_provisioning_uses_explicit_credentials( + catalog, monkeypatch, operation, global_key +): + monkeypatch.setattr(runpod, "api_key", global_key) + if global_key: + monkeypatch.setenv("RUNPOD_API_KEY", "unrelated-env-key") + client = AppsApiClient(api_key="explicit-key") + payload = await provision(operation, client) + assert f"-{MIG}" in payload["gpuIds"] + assert catalog.requests == [("/v2/catalog/gpus", "Bearer explicit-key")] + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +async def test_catalog_wait_keeps_event_loop_responsive(catalog, operation): + catalog.release.clear() + client = AppsApiClient(api_key="explicit-key") + task = asyncio.create_task(provision(operation, client)) + try: + await asyncio.wait_for(catalog.started.wait(), timeout=1) + heartbeat = asyncio.Event() + asyncio.get_running_loop().call_later(0.01, heartbeat.set) + await asyncio.wait_for(heartbeat.wait(), timeout=1) + assert not task.done() + catalog.release.set() + payload = await asyncio.wait_for(task, timeout=1) + assert f"-{MIG}" in payload["gpuIds"] + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +async def test_catalog_wait_respects_provisioning_deadline(catalog, operation): + catalog.release.clear() + client = AppsApiClient(api_key="explicit-key") + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(provision(operation, client), timeout=0.1) + assert catalog.started.is_set() + client.save_endpoint.assert_not_called() + cache = next(iter(gpu._catalog_caches.values())) + assert cache.failed_at is None + assert cache.members is None + catalog.release.set() + assert MIG in (await client.gpu_pool_members())["AMPERE_24"] + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +async def test_failed_catalog_uses_static_payload_without_sync_fetch( + catalog, operation +): + catalog.status = 500 + payload = await provision(operation, AppsApiClient(api_key="explicit-key")) + assert payload["gpuIds"] == ( + "AMPERE_24,-NVIDIA GeForce RTX 3090,-NVIDIA RTX A5000" + ) + assert len(catalog.requests) == 1 + + +@pytest.mark.parametrize("malformed", [False, True]) +async def test_failed_catalog_retries_after_cooldown(catalog, malformed, caplog): + if malformed: + catalog.data = {"gpus": []} + else: + catalog.status = 500 + client = AppsApiClient(api_key="explicit-key") + assert await client.gpu_pool_members() is None + assert await client.gpu_pool_members() is None + assert len(catalog.requests) == 1 + assert "device pins use the static pool table" in caplog.text + assert "explicit-key" not in caplog.text + cache = next(iter(gpu._catalog_caches.values())) + cache.failed_at -= gpu.CATALOG_RETRY_SECONDS + catalog.status = 200 + catalog.data = CATALOG + assert MIG in (await client.gpu_pool_members())["AMPERE_24"] + assert len(catalog.requests) == 2 + another_client = AppsApiClient(api_key="explicit-key") + assert MIG in (await another_client.gpu_pool_members())["AMPERE_24"] + assert len(catalog.requests) == 2 + + +async def test_catalog_uses_sdk_authentication_errors(catalog, caplog): + catalog.status = 401 + assert await AppsApiClient(api_key="explicit-key").gpu_pool_members() is None + assert "Unauthorized request, please check your API key." in caplog.text + assert "explicit-key" not in caplog.text + + +def test_sync_catalog_reuses_gpu_catalog_command(monkeypatch): + from runpod.api import ctl_commands + + monkeypatch.setattr(runpod, "api_key", "sdk-key") + monkeypatch.setenv("RUNPOD_API_KEY", "different-env-key") + fetch = Mock(return_value=CATALOG["gpus"]) + monkeypatch.setattr(ctl_commands, "get_gpus", fetch) + assert MIG in _real_live_pool_members()["AMPERE_24"] + assert MIG in _real_live_pool_members()["AMPERE_24"] + fetch.assert_called_once_with(api_key="sdk-key") + + +async def test_cache_is_scoped_to_credentials_and_catalog_url(catalog, monkeypatch): + first = AppsApiClient(api_key="first-key") + assert MIG in (await first.gpu_pool_members())["AMPERE_24"] + catalog.data = {"gpus": [{"id": "NVIDIA L4", "pool": "AMPERE_24"}]} + second = AppsApiClient(api_key="second-key") + assert await second.gpu_pool_members() == {"AMPERE_24": {"NVIDIA L4"}} + assert MIG in (await first.gpu_pool_members())["AMPERE_24"] + assert len(catalog.requests) == 2 + monkeypatch.setenv("RUNPOD_API_BASE_URL", catalog.other_url) + assert await first.gpu_pool_members() == {"AMPERE_24": {"NVIDIA L4"}} + assert len(catalog.requests) == 3 + monkeypatch.setenv("RUNPOD_API_BASE_URL", f"{catalog.url}/custom") + assert await first.gpu_pool_members() == {"AMPERE_24": {"NVIDIA L4"}} + assert catalog.requests[-1] == ("/custom/v2/catalog/gpus", "Bearer first-key") + assert len(catalog.requests) == 4 + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +@pytest.mark.parametrize("selection", [None, "any", "AMPERE_24", ["L4", "AMPERE_24"]]) +async def test_unpinned_gpus_do_not_fetch_catalog(operation, selection): + client = AppsApiClient(api_key="explicit-key") + client.gpu_pool_members = AsyncMock(side_effect=AssertionError("unneeded catalog")) + await provision(operation, client, gpu_selection=selection) + client.gpu_pool_members.assert_not_called() + + +@pytest.mark.parametrize("operation", ["start", "refresh", "reconcile"]) +async def test_cpu_resources_do_not_fetch_catalog(operation): + client = AppsApiClient(api_key="explicit-key") + client.gpu_pool_members = AsyncMock(side_effect=AssertionError("unneeded catalog")) + await provision(operation, client, gpu_selection=None, cpu="cpu3c-1-2") + client.gpu_pool_members.assert_not_called() + + +def test_supplied_membership_serializes_without_network(): + assert f"-{MIG}" in gpu.gpu_ids_value( + ["NVIDIA L4"], pool_members=gpu._parse_catalog(CATALOG) + ) + assert gpu.gpu_ids_value(["NVIDIA L4"], pool_members={}) == ( + "AMPERE_24,-NVIDIA GeForce RTX 3090,-NVIDIA RTX A5000" + )