diff --git a/runpod/__init__.py b/runpod/__init__.py index 427023537..6add070c1 100644 --- a/runpod/__init__.py +++ b/runpod/__init__.py @@ -54,6 +54,7 @@ "Model": "runpod.apps", "NetworkVolume": "runpod.apps", "Queue": "runpod.apps", + "Task": "runpod.apps", "Secret": "runpod.apps", "Volume": "runpod.apps", "is_local": "runpod.apps", diff --git a/runpod/api/graphql.py b/runpod/api/graphql.py index 367874250..5ab5ddf27 100644 --- a/runpod/api/graphql.py +++ b/runpod/api/graphql.py @@ -87,6 +87,7 @@ async def run_graphql_query_async( variables: Optional[Dict[str, Any]] = None, timeout: float = 60.0, anonymous: bool = False, + headers: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """ Async variant of run_graphql_query, sharing the same url, headers, @@ -110,7 +111,7 @@ async def run_graphql_query_async( async with aiohttp.ClientSession(timeout=client_timeout) as session: async with session.post( _graphql_url(), - headers=_build_headers(effective_api_key), + headers={**_build_headers(effective_api_key), **(headers or {})}, json=_build_payload(query, variables), ) as response: if response.status == HTTP_STATUS_UNAUTHORIZED: diff --git a/runpod/api/mutations/apps.py b/runpod/api/mutations/apps.py index 2e3c73a44..d8a7d71b1 100644 --- a/runpod/api/mutations/apps.py +++ b/runpod/api/mutations/apps.py @@ -88,7 +88,7 @@ MUTATION_FINALIZE_ARTIFACT_UPLOAD = """ mutation FinalizeArtifactUpload($input: FinalizeFlashArtifactUploadInput!) { - finalizeFlashArtifactUpload(input: $input) { id manifest } + finalizeFlashArtifactUpload(input: $input) { id manifest resourceSpec } } """ diff --git a/runpod/apps/__init__.py b/runpod/apps/__init__.py index 6a62fef93..ceb449a49 100644 --- a/runpod/apps/__init__.py +++ b/runpod/apps/__init__.py @@ -15,7 +15,7 @@ from .markers import delete, get, init, patch, post, put from .schedule import schedule from .spec import ResourceKind, ResourceSpec, RouteSpec -from .stubs import Api, Queue +from .stubs import Api, Queue, Task from .model import Model from .secret import Secret from .volume import GlobalVolume, NetworkVolume, Volume @@ -34,6 +34,7 @@ "Model", "NetworkVolume", "Queue", + "Task", "Secret", "Volume", "RemoteExecutionError", diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 9b557800a..0e053dad6 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -19,6 +19,41 @@ _TRANSPORT_RETRIES = 4 +class FlashTaskApiClient: + """named task lifecycle operations with runtime-scoped authentication.""" + + async def _execute(self, query: str, task_input: Dict[str, Any]) -> Dict[str, Any]: + from .utils.network import api_key, runtime_headers + + headers = runtime_headers() + response = await run_graphql_query_async( + query, + variables={"input": task_input}, + api_key=(headers["Authorization"][7:] if headers else api_key()), + headers=headers, + ) + return response["data"] + + async def create(self, task_input: Dict[str, Any]) -> Dict[str, Any]: + data = await self._execute( + """mutation FlashTaskCreate($input: FlashTaskCreateInput!) { + flashTaskCreate(input: $input) { podId flashBuildId resourceName } + }""", + task_input, + ) + return data["flashTaskCreate"] + + async def terminate(self, pod_id: str, token: str) -> None: + data = await self._execute( + """mutation FlashTaskTerminate($input: FlashTaskTerminateInput!) { + flashTaskTerminate(input: $input) + }""", + {"podId": pod_id, "executionToken": token}, + ) + if data.get("flashTaskTerminate") is not True: + raise RuntimeError("task termination was not acknowledged") + + def _rest_endpoint_input(endpoint: Dict[str, Any]) -> Optional[Dict[str, Any]]: """translate only saves whose capabilities rest can preserve atomically.""" fields = { diff --git a/runpod/apps/app.py b/runpod/apps/app.py index c2a6a0e5d..8ad3214c1 100644 --- a/runpod/apps/app.py +++ b/runpod/apps/app.py @@ -259,6 +259,12 @@ async def _resolve(self, spec: ResourceSpec) -> InvocationTarget: """resolve a resource spec to an invocation target.""" spec.validate() if spec.kind is ResourceKind.TASK: + from .context import deployed_identity + from .targets import DeployedTaskTarget + + identity = deployed_identity() + if identity is not None: + return DeployedTaskTarget(self.name, identity[1], spec.name) handle = self._resources.get(spec.name) fn = getattr(handle, "_fn", None) return PodTarget( diff --git a/runpod/apps/context.py b/runpod/apps/context.py index b21d38760..46283fbce 100644 --- a/runpod/apps/context.py +++ b/runpod/apps/context.py @@ -7,7 +7,7 @@ import threading import time from enum import Enum -from typing import Any, Coroutine +from typing import Any, Coroutine, Optional, Tuple log = logging.getLogger(__name__) CLEANUP_TIMEOUT = 30.0 @@ -36,6 +36,27 @@ def current_context() -> Context: return Context.LOCAL +def deployed_identity() -> Optional[Tuple[str, str, str]]: + """app, environment and resource identity for a deployed worker.""" + if not ( + os.getenv("RUNPOD_RUNTIME_KIND") == "task" + or os.getenv("RUNPOD_BUILD_ID") + or os.getenv("FLASH_APP_NAME") + or os.getenv("FLASH_RESOURCE_NAME") + ): + return None + fields = tuple( + os.getenv(key) + for key in ("FLASH_APP_NAME", "FLASH_ENVIRONMENT", "FLASH_RESOURCE_NAME") + ) + build = os.getenv("RUNPOD_BUILD_ID") + if any(not value or value != value.strip() for value in (*fields, build)): + raise RuntimeError( + "deployed flash requires app, environment, resource and build identity" + ) + return fields + + def is_local() -> bool: """true when not running inside a runpod container. diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index a0ef47ca7..4eddb6bf3 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -11,8 +11,6 @@ from pathspec import GitIgnoreSpec -import runpod - from .api import AppsApiClient from .app import App from .build import ( @@ -137,6 +135,17 @@ def build_manifest( ) entry = spec.to_manifest() + if spec.kind is ResourceKind.TASK: + from .images import image_for_spec + + entry["imageName"] = image_for_spec(spec, python_version=python_version) + entry["customImage"] = bool(spec.image) + for mount, volume in zip(entry.get("mounts", []), spec.mounts.values()): + mount["create"] = volume.create + if volume.kind == "network": + mount["size"] = volume.size + if volume.datacenter is not None: + mount["datacenter"] = volume.datacenter fn = ( getattr(handle, "_fn", None) @@ -162,6 +171,26 @@ def build_manifest( return manifest +async def _prepare_task_resources( + app: App, manifest: Dict[str, Any], client: AppsApiClient, events=None +) -> None: + from .registry import resolve_registry_auth + from .volume import VolumeResolver + + resolver = VolumeResolver(client, events=events) + specs = [handle.spec for handle in app.resources.values()] + for resource in manifest["resources"]: + if resource["kind"] != "task": + continue + spec = app.resources[resource["name"]].spec + bindings = await resolver.resolve_mounts(spec.mounts, specs) + for mount, binding in zip(resource.get("mounts", []), bindings): + mount["id"] = binding["id"] + registry_id = await resolve_registry_auth(spec.registry_auth, api=client) + if registry_id: + resource["registryAuth"] = registry_id + + def _load_ignores(path: Path) -> GitIgnoreSpec: if path.is_symlink() or not path.is_file(): return GitIgnoreSpec.from_lines([]) @@ -299,21 +328,32 @@ def _deployed_endpoint_input( from .secret import render_env spec.validate() + resource_env = render_env(spec.env) + reserved = { + "RUNPOD_API_KEY", + "RUNPOD_AI_API_KEY", + "RUNPOD_ENDPOINT_ID", + "RUNPOD_POD_ID", + "RUNPOD_TASK_TOKEN", + "RUNPOD_TASK_PORT", + "FLASH_BUILD_ARTIFACT_PATH", + "RUNPOD_RUNTIME_KIND", + "RUNPOD_DEV_APP", + "RUNPOD_DEV_RESOURCE", + "RUNPOD_DEV_SESSION", + } + if reserved.intersection(resource_env): + raise InvalidResourceError( + "production endpoint env contains reserved runtime fields" + ) template_env = { + **resource_env, + "FLASH_APP_NAME": app.name, + "FLASH_ENVIRONMENT": env_name or app.env, "FLASH_RESOURCE_NAME": spec.name, - # version-triggering: a new build recreates all workers "RUNPOD_BUILD_ID": build_id, - **render_env(spec.env), - "FLASH_ENVIRONMENT": env_name or app.env, } - # cross-resource calls from inside a worker go through the sentinel, - # which authenticates with the account api key; without it any - # .remote() from worker code would fail - api_key = runpod.api_key - if api_key and "RUNPOD_API_KEY" not in template_env: - template_env["RUNPOD_API_KEY"] = api_key - # nested tasks use the same runtime and sdk channels as the endpoint for key in ( "RUNPOD_RUNTIME_TAG", @@ -531,6 +571,7 @@ async def deploy_app( python_version=python_version, excluded_packages=build.excluded, ) + await _prepare_task_resources(app, manifest, client, events) _phase(events, "package") tar_path = package_project(project_root, manifest, env_dir=build.env_dir) tar_size = tar_path.stat().st_size @@ -562,6 +603,20 @@ async def deploy_app( upload["uploadUrl"], str(tar_path), progress=upload_progress ) build = await client.finalize_artifact_upload(app_id, upload["objectKey"], manifest) + definitions = build.get("resourceSpec") + tasks = {r["name"] for r in manifest["resources"] if r["kind"] == "task"} + if ( + not isinstance(definitions, dict) + or not isinstance(definitions.get("tasks"), dict) + or set(definitions["tasks"]) != tasks + or any( + not isinstance(task, dict) or not task + for task in definitions["tasks"].values() + ) + ): + raise RuntimeError( + "backend did not acknowledge the registered task definitions" + ) log.info("uploaded build %s", build["id"]) await client.deploy_build(environment["id"], build["id"]) diff --git a/runpod/apps/handles.py b/runpod/apps/handles.py index b66af1d94..75c190cb6 100644 --- a/runpod/apps/handles.py +++ b/runpod/apps/handles.py @@ -233,6 +233,15 @@ def _guard_discovery(self) -> None: def _is_current_worker(self) -> bool: import os + from .context import deployed_identity + + identity = deployed_identity() + if identity is not None: + return ( + self.spec.kind is not ResourceKind.TASK + and identity[0] == self._app.name + and identity[2] == self.spec.name + ) if dev_app := os.getenv("RUNPOD_DEV_APP"): return ( dev_app == self._app.name diff --git a/runpod/apps/stubs.py b/runpod/apps/stubs.py index f0e5aeef4..5d9865d9f 100644 --- a/runpod/apps/stubs.py +++ b/runpod/apps/stubs.py @@ -123,6 +123,33 @@ def __repr__(self) -> str: return f"" +class Task(_StubBase): + """client for an approved task deployed in a flash environment.""" + + def __init__(self, *, app: str, name: str, env: Optional[str] = None): + super().__init__(app=app, name=name, env=env) + if not app or not name: + raise ValueError("task app and name must be non-empty") + self.remote = Invoker(self._remote_async) + self.spawn = Invoker(self._spawn_async) + + def _target(self): + from .targets import DeployedTaskTarget + + return DeployedTaskTarget(self.app_name, self.env, self.resource_name) + + async def _remote_async(self, *args: Any, **kwargs: Any) -> Any: + target = self._target() + return await target.invoke(target.build_payload(None, None, args, kwargs)) + + async def _spawn_async(self, *args: Any, **kwargs: Any) -> Any: + target = self._target() + return await target.submit(target.build_payload(None, None, args, kwargs)) + + def __repr__(self): + return f"" + + class _StubRouteCaller: __slots__ = ("_stub", "_method") diff --git a/runpod/apps/targets.py b/runpod/apps/targets.py index 28e720513..cf166ec8c 100644 --- a/runpod/apps/targets.py +++ b/runpod/apps/targets.py @@ -39,11 +39,16 @@ def _lb_domain() -> str: def _headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]: + from .utils.network import runtime_headers + headers = { - "Authorization": f"Bearer {_api_key()}", "Content-Type": "application/json", "User-Agent": USER_AGENT, } + runtime = runtime_headers() + headers.update( + runtime if runtime is not None else {"Authorization": f"Bearer {_api_key()}"} + ) if extra: headers.update(extra) return headers @@ -797,63 +802,28 @@ async def request( return result -class PodTarget(InvocationTarget): - """ephemeral pod execution for @app.task: provision a pod, run the - function body, collect the result, terminate. - - tasks run to completion far beyond the queue data plane's sync - window, so the transport is a direct http channel to a single-shot - runner on the pod (see runpod.apps.tasks). - """ - - def __init__( - self, - spec: ResourceSpec, - fn: Callable, - events: Optional[object] = None, - *, - specs: Optional[List[ResourceSpec]] = None, - ): - self.spec = spec - self.fn = fn - self.events = events - self.specs = specs +class _TaskTarget(InvocationTarget): + """shared execution, result collection, and cleanup for task targets.""" - def build_payload( - self, fn: Callable, spec: ResourceSpec, args: tuple, kwargs: dict - ) -> Dict[str, Any]: - from .serialization import ( - get_function_source, - serialize_args, - serialize_kwargs, - ) - - request = FunctionRequest( - function_name=fn.__name__, - function_code=get_function_source(fn), - args=serialize_args(args), - kwargs=serialize_kwargs(kwargs), - dependencies=spec.dependencies, - system_dependencies=spec.system_dependencies, - accelerate_downloads=spec.accelerate_downloads, - ) - return request.to_input() + @abstractmethod + def _execution(self, payload, timeout=None): + raise NotImplementedError async def invoke( self, payload: Dict[str, Any], *, timeout: Optional[float] = None ) -> Any: import time - from .tasks import TaskExecution, finish_cleanup, unwrap_task_response + from .tasks import finish_cleanup, unwrap_task_response - name = self.spec.name - hardware = ",".join(self.spec.cpu or self.spec.gpu or ["any"]) + name = self.resource_name + hardware = self.hardware emit(self.events, "dispatch", name) emit(self.events, "task_status", name, f"provisioning pod on {hardware}") start = time.monotonic() stream = None - execution = TaskExecution(self.spec, specs=self.specs) + execution = self._execution(payload, timeout) try: await execution.start() if execution.pod_id: @@ -899,9 +869,9 @@ async def cleanup(): return result async def submit(self, payload: Dict[str, Any]) -> Any: - from .tasks import TaskExecution, TaskJob, finish_cleanup + from .tasks import TaskJob, finish_cleanup - execution = TaskExecution(self.spec, specs=self.specs) + execution = self._execution(payload) try: await execution.start() await execution.wait_ready() @@ -910,3 +880,82 @@ async def submit(self, payload: Dict[str, Any]) -> Any: await finish_cleanup(execution.terminate()) raise return TaskJob(execution) + + +class PodTarget(_TaskTarget): + """ephemeral pod execution for @app.task: provision a pod, run the + function body, collect the result, terminate. + + tasks run to completion far beyond the queue data plane's sync + window, so the transport is a direct http channel to a single-shot + runner on the pod (see runpod.apps.tasks). + """ + + def __init__( + self, + spec: ResourceSpec, + fn: Callable, + events: Optional[object] = None, + *, + specs: Optional[List[ResourceSpec]] = None, + ): + self.spec = spec + self.fn = fn + self.events = events + self.specs = specs + self.resource_name = spec.name + self.hardware = ",".join(spec.cpu or spec.gpu or ["any"]) + + def _execution(self, payload, timeout=None): + from .tasks import TaskExecution + + return TaskExecution(self.spec, specs=self.specs) + + def build_payload( + self, fn: Callable, spec: ResourceSpec, args: tuple, kwargs: dict + ) -> Dict[str, Any]: + from .serialization import ( + get_function_source, + serialize_args, + serialize_kwargs, + ) + + request = FunctionRequest( + function_name=fn.__name__, + function_code=get_function_source(fn), + args=serialize_args(args), + kwargs=serialize_kwargs(kwargs), + dependencies=spec.dependencies, + system_dependencies=spec.system_dependencies, + accelerate_downloads=spec.accelerate_downloads, + ) + return request.to_input() + + +class DeployedTaskTarget(_TaskTarget): + """invoke an approved deployed task without transmitting source or specs.""" + + def __init__(self, app, env, name, events=None): + self.app_name = app + self.env_name = env + self.resource_name = name + self.hardware = "approved resources" + self.events = events + + def _execution(self, payload, timeout=None): + from .tasks import DeployedTaskExecution + + if ( + set(payload) != {"args", "kwargs"} + or not isinstance(payload["args"], list) + or not isinstance(payload["kwargs"], dict) + or any(not isinstance(key, str) for key in payload["kwargs"]) + ): + raise ValueError("deployed tasks accept only JSON args and kwargs") + json.dumps(payload, allow_nan=False) + return DeployedTaskExecution( + self.app_name, self.env_name, self.resource_name, timeout=timeout + ) + + def build_payload(self, fn, spec, args, kwargs): + return {"args": list(args), "kwargs": kwargs} diff --git a/runpod/apps/tasks.py b/runpod/apps/tasks.py index e6a6419b4..6a60f5f82 100644 --- a/runpod/apps/tasks.py +++ b/runpod/apps/tasks.py @@ -36,6 +36,16 @@ _pending_cleanups = set() +def _validate_timeout(timeout): + if timeout is not None and ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(timeout) + or timeout < 0 + ): + raise ValueError("task timeout must be a finite non-negative number") + + async def finish_cleanup(coro) -> None: """finish cleanup despite repeated cancellation, without hiding its cause.""" original = sys.exc_info()[1] @@ -185,19 +195,11 @@ def _pod_input(spec: ResourceSpec, token: str, task_name: str) -> Dict[str, Any] return pod -class TaskExecution: - """one pod running one function.""" +class _TaskTransport: + """token-authenticated execution transport with cancellation-safe cleanup.""" - def __init__( - self, - spec: ResourceSpec, - api: Optional[AppsApiClient] = None, - *, - specs: Optional[List[ResourceSpec]] = None, - ): - self.spec = spec - self.specs = specs if specs is not None else [spec] - self.api = api or AppsApiClient() + def __init__(self, api): + self.api = api self.token = secrets.token_urlsafe(32) self.pod_id: Optional[str] = None self._deployment: Optional[asyncio.Task] = None @@ -206,55 +208,6 @@ def __init__( def _headers(self) -> Dict[str, str]: return {"Authorization": f"Bearer {self.token}"} - async def start(self) -> None: - pod = _pod_input(self.spec, self.token, self.spec.name) - if self.spec.registry_auth: - from .registry import resolve_registry_auth - - pod["containerRegistryAuthId"] = await resolve_registry_auth( - self.spec.registry_auth, api=self.api - ) - pod = await self._attach_mounts(pod) - self._deployment = asyncio.create_task(self._deploy_and_record(pod)) - await asyncio.shield(self._deployment) - - async def _deploy_and_record(self, pod: Dict[str, Any]) -> None: - result = await self._deploy_pod(pod) - self.pod_id = result["id"] - log.info("task pod %s deployed for %s", self.pod_id, self.spec.name) - - async def _deploy_pod(self, pod: Dict[str, Any]) -> Dict[str, Any]: - from ..error import QueryError - - try: - return await self.api.deploy_task_pod(pod, is_cpu=self.spec.is_cpu) - except QueryError as exc: - if not is_capacity_error(exc): - raise - capacity_error = exc - - datacenters = pod.get("dataCenterIds") or [] - if len(datacenters) < 2: - raise capacity_error - - for datacenter in datacenters: - candidate = dict(pod, dataCenterIds=[datacenter]) - try: - return await self.api.deploy_task_pod( - candidate, is_cpu=self.spec.is_cpu - ) - except QueryError as exc: - if not is_capacity_error(exc): - raise - raise capacity_error - - async def _attach_mounts(self, pod: Dict[str, Any]) -> Dict[str, Any]: - """resolve storage mounts and apply their placement constraints.""" - from .volume import VolumeResolver, attach_pod_mounts - - await attach_pod_mounts(pod, self.spec, VolumeResolver(self.api), self.specs) - return pod - async def wait_ready(self, timeout: float = READY_TIMEOUT) -> None: """poll the runner's /ping through the pod proxy until it answers.""" url = f"{_proxy_url(self.pod_id)}/ping" @@ -312,13 +265,7 @@ async def set_timeout(self, timeout: float) -> None: async def _post(self, path: str, request: Dict[str, Any]) -> None: timeout = request.get("timeout") - if timeout is not None and ( - isinstance(timeout, bool) - or not isinstance(timeout, (int, float)) - or not math.isfinite(timeout) - or timeout < 0 - ): - raise ValueError("task timeout must be a finite non-negative number") + _validate_timeout(timeout) url = f"{_proxy_url(self.pod_id)}{path}" attempts = 6 async with aiohttp.ClientSession() as session: @@ -385,9 +332,7 @@ async def terminate(self) -> None: try: for attempt in range(DELETE_ATTEMPTS): try: - await asyncio.wait_for( - self.api.terminate_pod(self.pod_id), DELETE_TIMEOUT - ) + await asyncio.wait_for(self._terminate_pod(), DELETE_TIMEOUT) break except Exception as exc: status = getattr(exc, "status_code", None) or getattr( @@ -420,6 +365,109 @@ async def terminate(self) -> None: raise self.pod_id = None + async def _terminate_pod(self): + raise NotImplementedError + + +class TaskExecution(_TaskTransport): + """one pod running one function.""" + + def __init__( + self, + spec: ResourceSpec, + api: Optional[AppsApiClient] = None, + *, + specs: Optional[List[ResourceSpec]] = None, + ): + self.spec = spec + self.specs = specs if specs is not None else [spec] + super().__init__(api or AppsApiClient()) + + async def start(self) -> None: + pod = _pod_input(self.spec, self.token, self.spec.name) + if self.spec.registry_auth: + from .registry import resolve_registry_auth + + pod["containerRegistryAuthId"] = await resolve_registry_auth( + self.spec.registry_auth, api=self.api + ) + pod = await self._attach_mounts(pod) + self._deployment = asyncio.create_task(self._deploy_and_record(pod)) + await asyncio.shield(self._deployment) + + async def _deploy_and_record(self, pod: Dict[str, Any]) -> None: + result = await self._deploy_pod(pod) + self.pod_id = result["id"] + log.info("task pod %s deployed for %s", self.pod_id, self.spec.name) + + async def _deploy_pod(self, pod: Dict[str, Any]) -> Dict[str, Any]: + from ..error import QueryError + + try: + return await self.api.deploy_task_pod(pod, is_cpu=self.spec.is_cpu) + except QueryError as exc: + if not is_capacity_error(exc): + raise + capacity_error = exc + + datacenters = pod.get("dataCenterIds") or [] + if len(datacenters) < 2: + raise capacity_error + + for datacenter in datacenters: + candidate = dict(pod, dataCenterIds=[datacenter]) + try: + return await self.api.deploy_task_pod( + candidate, is_cpu=self.spec.is_cpu + ) + except QueryError as exc: + if not is_capacity_error(exc): + raise + raise capacity_error + + async def _attach_mounts(self, pod: Dict[str, Any]) -> Dict[str, Any]: + """resolve storage mounts and apply their placement constraints.""" + from .volume import VolumeResolver, attach_pod_mounts + + await attach_pod_mounts(pod, self.spec, VolumeResolver(self.api), self.specs) + return pod + + async def _terminate_pod(self): + await self.api.terminate_pod(self.pod_id) + + +class DeployedTaskExecution(_TaskTransport): + """one approved named task using the task transport and cleanup lifecycle.""" + + def __init__(self, app, env, name, api=None, *, timeout=None): + from .api import FlashTaskApiClient + + _validate_timeout(timeout) + self.app_name = app + self.env_name = env + self.resource_name = name + self.timeout = timeout + super().__init__(api or FlashTaskApiClient()) + + async def start(self): + self._deployment = asyncio.create_task(self._create_and_record()) + await asyncio.shield(self._deployment) + + async def _create_and_record(self): + task_input = { + "flashAppName": self.app_name, + "flashEnvironmentName": self.env_name, + "resourceName": self.resource_name, + "executionToken": self.token, + } + if self.timeout is not None: + task_input["executionTimeoutSeconds"] = self.timeout + result = await self.api.create(task_input) + self.pod_id = result["podId"] + + async def _terminate_pod(self): + await self.api.terminate(self.pod_id, self.token) + def unwrap_task_response(response: Dict[str, Any]) -> Any: """extract the function result from a runner response.""" @@ -437,7 +485,7 @@ def unwrap_task_response(response: Dict[str, Any]) -> Any: class TaskJob: """handle for a spawned task; owns the pod until the result is read.""" - def __init__(self, execution: TaskExecution): + def __init__(self, execution: _TaskTransport): self._execution = execution self._result: Any = None self._done = False diff --git a/runpod/apps/utils/network.py b/runpod/apps/utils/network.py index d681766ea..d879f8cb6 100644 --- a/runpod/apps/utils/network.py +++ b/runpod/apps/utils/network.py @@ -1,11 +1,33 @@ """shared authentication and lightweight data-plane requests.""" import os -from typing import Any, Dict +from typing import Any, Dict, Optional import aiohttp +def runtime_headers() -> Optional[Dict[str, str]]: + from ..context import deployed_identity + + if deployed_identity() is None: + return None + endpoint = os.getenv("RUNPOD_ENDPOINT_ID") + pod = os.getenv("RUNPOD_POD_ID") + if os.getenv("RUNPOD_RUNTIME_KIND") == "task": + endpoint = None + if not endpoint and not pod: + raise RuntimeError("production flash runtime source identity is missing") + key = os.getenv("RUNPOD_AI_API_KEY" if endpoint else "RUNPOD_API_KEY") + if not key or key != key.strip(): + raise RuntimeError("production flash runtime credential is missing") + if (endpoint or pod) != (endpoint or pod).strip(): + raise RuntimeError("production flash runtime source identity is malformed") + return { + "Authorization": f"Bearer {key}", + "X-RunPod-Endpoint-Id" if endpoint else "X-RunPod-Pod-Id": endpoint or pod, + } + + def api_key() -> str: key = os.getenv("RUNPOD_API_KEY") if not key: diff --git a/runpod/rp_cli/console.py b/runpod/rp_cli/console.py index f165ff56e..fe0e48738 100644 --- a/runpod/rp_cli/console.py +++ b/runpod/rp_cli/console.py @@ -233,11 +233,12 @@ def resources_table(rows: Iterable[tuple]) -> None: w_name = max(len(r[0]) for r in rows) w_hw = max(len(r[2]) for r in rows) for name, kind, hardware, endpoint_id in rows: - tail = ( - "[dim]per-call[/dim]" - if not endpoint_id or endpoint_id == "per-call" - else endpoint_link(endpoint_id) - ) + if endpoint_id == "registered": + tail = "[dim]registered / per-call[/dim]" + elif not endpoint_id or endpoint_id == "per-call": + tail = "[dim]per-call[/dim]" + else: + tail = endpoint_link(endpoint_id) console.print( f" [white]{name:<{w_name}}[/white]" f" {kind_badge(kind)}" diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 5c4b4bfdc..51adab6c5 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -282,14 +282,17 @@ def deploy(target, env, python_version, exclude, build_only): f"[bold white]{result.app_name}/{env or found.env}[/bold white] " f"is live [dim]{t.elapsed:.1f}s[/dim]" ) - if result.endpoints: + if result.resources: ui.console.print() - w = max(len(n) for n in result.endpoints) - for name, endpoint_id in sorted(result.endpoints.items()): - ui.console.print( - f" [white]{name:<{w}}[/white] " - f"{ui.endpoint_link(endpoint_id)}" + ui.resources_table( + ( + name, + handle.spec.kind.value, + ",".join(handle.spec.cpu or handle.spec.gpu or ["any"]), + result.endpoints.get(name, "registered"), ) + for name, handle in sorted(found.resources.items()) + ) ui.console.print() diff --git a/tests/test_apps/test_deploy.py b/tests/test_apps/test_deploy.py index 5445c5b52..46912810e 100644 --- a/tests/test_apps/test_deploy.py +++ b/tests/test_apps/test_deploy.py @@ -191,7 +191,9 @@ def test_excluded_paths_are_logged_without_contents(self, tmp_path, caplog): with tarfile.open(package_project(tmp_path, {})) as tar: assert set(tar.getnames()) == {".runpodignore", "runpod_manifest.json"} for name in (".env", "private.key", "private.pem", "local.txt", ".aws/"): - assert f"excluded source path {name!r} from deployment artifact" in caplog.text + assert ( + f"excluded source path {name!r} from deployment artifact" in caplog.text + ) assert "sensitive-file-contents" not in caplog.text def test_vendored_env_included_under_env(self, tmp_path): @@ -280,7 +282,10 @@ async def test_deploy_app_full_flow(self, tmp_path): "uploadUrl": "https://upload", "objectKey": "key-1", } - api.finalize_artifact_upload.return_value = {"id": "build-1"} + api.finalize_artifact_upload.return_value = { + "id": "build-1", + "resourceSpec": {"tasks": {}}, + } api.deploy_build.return_value = {"id": "env-1"} with _stub_build(tmp_path): @@ -305,7 +310,10 @@ async def test_deploy_reuses_existing_app_and_env(self, tmp_path): "uploadUrl": "https://upload", "objectKey": "key-1", } - api.finalize_artifact_upload.return_value = {"id": "build-2"} + api.finalize_artifact_upload.return_value = { + "id": "build-2", + "resourceSpec": {"tasks": {}}, + } with _stub_build(tmp_path): await deploy_app(app, tmp_path, api=api) @@ -329,7 +337,10 @@ async def test_deploy_environment_is_used_for_nested_calls( "uploadUrl": "https://upload", "objectKey": "key-1", } - api.finalize_artifact_upload.return_value = {"id": "build-1"} + api.finalize_artifact_upload.return_value = { + "id": "build-1", + "resourceSpec": {"tasks": {}}, + } api.save_endpoint.return_value = {"id": "ep-1"} with _stub_build(tmp_path): await deploy_app(app, tmp_path, env_name="prod", api=api) diff --git a/tests/test_apps/test_deployed_tasks.py b/tests/test_apps/test_deployed_tasks.py new file mode 100644 index 000000000..4fa5ce1ae --- /dev/null +++ b/tests/test_apps/test_deployed_tasks.py @@ -0,0 +1,471 @@ +"""production task identity, lifecycle, and deployment contract tests.""" + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +import runpod +from runpod.apps import App, Task +from runpod.apps.api import FlashTaskApiClient +from runpod.apps.app import _clear_registry +from runpod.apps.build import BuildResult +from runpod.apps.deploy import ( + _deployed_endpoint_input, + _prepare_task_resources, + build_manifest, + deploy_app, +) +from runpod.apps.errors import InvalidResourceError +from runpod.apps.targets import DeployedTaskTarget, PodTarget, _headers +from runpod.apps.tasks import DeployedTaskExecution, _pod_input +from runpod.apps.utils.network import runtime_headers +from runpod.apps.volume import GlobalVolume, NetworkVolume + + +@pytest.fixture(autouse=True) +def isolated_identity(monkeypatch): + _clear_registry() + for name in ( + "RUNPOD_RUNTIME_KIND", + "RUNPOD_BUILD_ID", + "RUNPOD_ENDPOINT_ID", + "RUNPOD_POD_ID", + "RUNPOD_AI_API_KEY", + "RUNPOD_API_KEY", + "FLASH_APP_NAME", + "FLASH_ENVIRONMENT", + "FLASH_RESOURCE_NAME", + "RUNPOD_DEV_APP", + "RUNPOD_DEV_SESSION", + ): + monkeypatch.delenv(name, raising=False) + yield + _clear_registry() + + +@pytest.fixture +def production(monkeypatch): + values = { + "RUNPOD_RUNTIME_KIND": "queue", + "RUNPOD_BUILD_ID": "build-1", + "FLASH_APP_NAME": "my-app", + "FLASH_ENVIRONMENT": "prod", + "FLASH_RESOURCE_NAME": "work", + "RUNPOD_ENDPOINT_ID": "ep-1", + "RUNPOD_POD_ID": "worker-pod", + "RUNPOD_AI_API_KEY": "endpoint-key", + "RUNPOD_API_KEY": "must-not-use-account-key", + } + for name, value in values.items(): + monkeypatch.setenv(name, value) + + +def test_endpoint_auth_emits_only_endpoint_source(production): + headers = _headers() + assert headers["Authorization"] == "Bearer endpoint-key" + assert headers["X-RunPod-Endpoint-Id"] == "ep-1" + assert "X-RunPod-Pod-Id" not in headers + + +def test_task_auth_emits_only_pod_source(production, monkeypatch): + monkeypatch.setenv("RUNPOD_RUNTIME_KIND", "task") + monkeypatch.setenv("RUNPOD_API_KEY", "child-key") + assert runtime_headers() == { + "Authorization": "Bearer child-key", + "X-RunPod-Pod-Id": "worker-pod", + } + + +@pytest.mark.parametrize("missing", ["RUNPOD_POD_ID", "RUNPOD_API_KEY"]) +def test_task_auth_never_uses_endpoint_or_global_key(production, monkeypatch, missing): + monkeypatch.setenv("RUNPOD_RUNTIME_KIND", "task") + monkeypatch.setattr(runpod, "api_key", "account-key") + monkeypatch.delenv(missing) + with pytest.raises(RuntimeError): + runtime_headers() + + +def test_endpoint_auth_does_not_fall_back_to_pod_key(production, monkeypatch): + monkeypatch.delenv("RUNPOD_AI_API_KEY") + monkeypatch.setattr(runpod, "api_key", "account-key") + with pytest.raises(RuntimeError, match="credential"): + runtime_headers() + + +@pytest.mark.parametrize( + "missing", + [ + "RUNPOD_AI_API_KEY", + "FLASH_APP_NAME", + "FLASH_ENVIRONMENT", + "FLASH_RESOURCE_NAME", + "RUNPOD_BUILD_ID", + ], +) +def test_missing_production_identity_never_falls_back(production, monkeypatch, missing): + monkeypatch.delenv(missing) + with pytest.raises(RuntimeError): + _headers() + + +def test_dev_auth_retains_account_key(monkeypatch): + monkeypatch.setenv("RUNPOD_DEV_APP", "my-app") + monkeypatch.setenv("RUNPOD_AI_API_KEY", "endpoint-key") + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-1") + monkeypatch.setenv("RUNPOD_API_KEY", "developer-key") + headers = _headers() + assert headers["Authorization"] == "Bearer developer-key" + assert "X-RunPod-Endpoint-Id" not in headers + + +@pytest.mark.parametrize("image", [None, "custom/image:tag"]) +def test_dev_task_payload_preserves_source_runtime(image): + app = App("my-app") + + @app.task(name="work", cpu="cpu3c-1-2", image=image) + def work(): + pass + + payload = _pod_input(work.spec, "execution-token", "work") + env = {entry["key"]: entry["value"] for entry in payload["env"]} + assert "RUNPOD_RUNTIME_KIND" not in env + assert "RUNPOD_BUILD_ID" not in env + assert "FLASH_APP_NAME" not in env + if image: + assert "runpod_sdk_runtime.task.runner" in payload["dockerArgs"] + + +async def test_graphql_create_and_terminate_use_exact_contract(production): + api = FlashTaskApiClient() + create_input = { + "flashAppName": "my-app", + "flashEnvironmentName": "prod", + "resourceName": "work", + "executionToken": "execution-token", + } + execute = AsyncMock( + side_effect=[ + { + "data": { + "flashTaskCreate": { + "podId": "child", + "flashBuildId": "build", + "resourceName": "work", + } + } + }, + {"data": {"flashTaskTerminate": True}}, + ] + ) + with patch("runpod.apps.api.run_graphql_query_async", execute): + assert (await api.create(create_input))["podId"] == "child" + await api.terminate("child", "execution-token") + create, terminate = execute.await_args_list + assert "FlashTaskCreateInput!" in create.args[0] + assert create.kwargs["variables"] == {"input": create_input} + assert create.kwargs["api_key"] == "endpoint-key" + assert create.kwargs["headers"]["X-RunPod-Endpoint-Id"] == "ep-1" + assert terminate.kwargs["variables"] == { + "input": {"podId": "child", "executionToken": "execution-token"} + } + + +async def test_explicit_local_task_uses_account_named_api(monkeypatch): + monkeypatch.setenv("RUNPOD_API_KEY", "local-key") + execute = AsyncMock(return_value={"data": {"flashTaskCreate": {"podId": "child"}}}) + with patch("runpod.apps.api.run_graphql_query_async", execute): + await FlashTaskApiClient().create({"resourceName": "work"}) + assert execute.await_args.kwargs["api_key"] == "local-key" + assert execute.await_args.kwargs["headers"] is None + + +async def test_task_target_selection_is_production_only(production, monkeypatch): + app = App("my-app") + + @app.task(name="work", cpu="cpu3c-1-2") + def work(value): + return value + + target = await app._resolve(work.spec) + assert isinstance(target, DeployedTaskTarget) + assert target.env_name == "prod" + assert target.build_payload(work._fn, work.spec, (1,), {"other": 2}) == { + "args": [1], + "kwargs": {"other": 2}, + } + for name in ("RUNPOD_BUILD_ID", "FLASH_APP_NAME", "FLASH_RESOURCE_NAME"): + monkeypatch.delenv(name) + assert isinstance(await app._resolve(work.spec), PodTarget) + monkeypatch.setenv("RUNPOD_DEV_SESSION", "session") + assert type(await app._resolve(work.spec)) is PodTarget + + +async def test_invalid_production_task_identity_cannot_use_source_target( + production, monkeypatch +): + app = App("my-app") + + @app.task(name="work", cpu="cpu3c-1-2") + def work(): + pass + + monkeypatch.delenv("FLASH_APP_NAME") + with pytest.raises(RuntimeError, match="identity"): + await app._resolve(work.spec) + + +@pytest.mark.parametrize("sync", [False, True]) +def test_task_self_remote_launches_child(production, sync): + app = App("my-app") + + @app.task(name="work", cpu="cpu3c-1-2") + def work(value): + raise AssertionError("task body must not run inline") + + target = DeployedTaskTarget("my-app", "prod", "work") + target.invoke = AsyncMock(return_value="child-result") + with patch.object(app, "_resolve", AsyncMock(return_value=target)): + result = work.remote(1) if sync else asyncio.run(work.remote.aio(1)) + assert result == "child-result" + target.invoke.assert_awaited_once_with({"args": [1], "kwargs": {}}) + + +def test_production_queue_self_optimization_checks_app(production): + other = App("other-app") + + @other.queue(name="work", cpu="cpu3c-1-2") + def work(): + pass + + assert not work._is_current_worker() + assert work._app.name != "my-app" + + +async def test_task_stub_remote_and_spawn_are_json_only(): + stub = Task(app="my-app", env="prod", name="work") + assert runpod.Task is Task + target = stub._target() + target.invoke = AsyncMock(return_value=3) + target.submit = AsyncMock(return_value="job") + with patch.object(stub, "_target", return_value=target): + assert await stub.remote.aio(1, value=2) == 3 + assert await stub.spawn.aio(1, value=2) == "job" + target.invoke.assert_awaited_once_with({"args": [1], "kwargs": {"value": 2}}) + target.submit.assert_awaited_once_with({"args": [1], "kwargs": {"value": 2}}) + + +@pytest.mark.parametrize( + "payload", + [ + {"args": [], "kwargs": {}, "function_code": "bad"}, + {"args": "pickle", "kwargs": {}}, + {"args": [object()], "kwargs": {}}, + {"args": [float("nan")], "kwargs": {}}, + {"args": [], "kwargs": {1: "bad"}}, + ], +) +async def test_deployed_payload_rejects_overrides_and_non_json(payload): + target = DeployedTaskTarget("my-app", "prod", "work") + with pytest.raises((ValueError, TypeError)): + await target.submit(payload) + + +async def test_cancellation_retains_pending_create_and_cleans_success(): + entered = asyncio.Event() + finish = asyncio.Event() + + async def create(task_input): + entered.set() + await finish.wait() + return {"podId": "child"} + + api = AsyncMock() + api.create.side_effect = create + execution = DeployedTaskExecution("my-app", "prod", "work", api) + target = DeployedTaskTarget("my-app", "prod", "work") + with patch.object(target, "_execution", return_value=execution): + invocation = asyncio.create_task(target.submit({"args": [], "kwargs": {}})) + await entered.wait() + invocation.cancel() + await asyncio.sleep(0) + assert not invocation.done() + finish.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(invocation, timeout=5) + api.create.assert_awaited_once() + api.terminate.assert_awaited_once_with("child", execution.token) + assert execution.pod_id is None + + +async def test_create_failure_is_not_retried_or_source_fallback(): + api = AsyncMock() + api.create.side_effect = RuntimeError("approval missing") + execution = DeployedTaskExecution("my-app", "prod", "work", api) + target = DeployedTaskTarget("my-app", "prod", "work") + with patch.object(target, "_execution", return_value=execution): + with pytest.raises(RuntimeError, match="approval missing"): + await target.invoke({"args": [], "kwargs": {}}) + api.create.assert_awaited_once() + api.terminate.assert_not_awaited() + + +async def test_execution_creation_contains_only_named_identity_and_timeout(): + api = AsyncMock() + api.create.return_value = {"podId": "child"} + execution = DeployedTaskExecution("my-app", "prod", "work", api, timeout=12.5) + await execution.start() + api.create.assert_awaited_once_with( + { + "flashAppName": "my-app", + "flashEnvironmentName": "prod", + "resourceName": "work", + "executionToken": execution.token, + "executionTimeoutSeconds": 12.5, + } + ) + assert execution._headers == {"Authorization": f"Bearer {execution.token}"} + await execution.terminate() + + +def test_manifest_pins_task_image_and_full_storage_references(tmp_path, monkeypatch): + monkeypatch.setenv("RUNPOD_RUNTIME_TAG", "release") + app = App("my-app") + + @app.task( + name="work", + gpu="4090", + mounts={ + "/data": NetworkVolume("data", size=75, datacenter="EU-RO-1", create=False), + "/global": GlobalVolume("shared", create=True), + }, + ) + def work(): + pass + + resource = build_manifest(app, tmp_path, python_version="3.12")["resources"][0] + assert resource["imageName"] == "runpod/task-gpu:py3.12-release" + assert resource["mounts"] == [ + { + "kind": "network", + "reference": "data", + "path": "/data", + "create": False, + "size": 75, + "datacenter": "EU-RO-1", + }, + {"kind": "global", "reference": "shared", "path": "/global", "create": True}, + ] + assert resource["module"] + assert resource["qualname"].endswith("work") + + +def test_deployed_endpoint_uses_runtime_identity_without_account_key(monkeypatch): + monkeypatch.setattr(runpod, "api_key", "developer-key") + app = App("my-app") + + @app.queue(name="work", cpu="cpu3c-1-2") + def work(): + pass + + payload = _deployed_endpoint_input(app, work.spec, "env", "build", "3.12") + env = {e["key"]: e["value"] for e in payload["template"]["env"]} + assert env["FLASH_APP_NAME"] == "my-app" + assert "RUNPOD_API_KEY" not in env + assert env["RUNPOD_BUILD_ID"] == "build" + assert env["FLASH_RESOURCE_NAME"] == "work" + assert env["FLASH_ENVIRONMENT"] == "default" + work.spec.env = {"RUNPOD_AI_API_KEY": "override"} + with pytest.raises(InvalidResourceError, match="reserved"): + _deployed_endpoint_input(app, work.spec, "env", "build", "3.12") + + +async def test_task_storage_and_registry_are_resolved_with_deployer_client(tmp_path): + app = App("my-app") + + @app.task( + name="work", + cpu="cpu3c-1-2", + image="private/image:tag", + registry_auth="private", + mounts={"/data": NetworkVolume("data", size=75, create=True)}, + ) + def work(): + pass + + manifest = build_manifest(app, tmp_path) + api = AsyncMock() + api.list_registry_auths.return_value = [{"id": "registry-id", "name": "private"}] + bindings = [ + { + "id": "volume-id", + "kind": "network", + "reference": "data", + "path": "/data", + "dataCenterId": "EU-RO-1", + } + ] + with patch( + "runpod.apps.volume.VolumeResolver.resolve_mounts", + AsyncMock(return_value=bindings), + ): + await _prepare_task_resources(app, manifest, api) + resource = manifest["resources"][0] + assert resource["mounts"][0]["id"] == "volume-id" + assert resource["mounts"][0]["reference"] == "data" + assert resource["mounts"][0]["size"] == 75 + assert resource["registryAuth"] == "registry-id" + assert resource["customImage"] is True + api.list_registry_auths.assert_awaited_once() + + +@pytest.mark.parametrize( + "definitions,approved", + [ + (None, False), + ({}, False), + ({"tasks": {}}, False), + ({"tasks": {"work": None}}, False), + ({"tasks": {"work": {}}}, False), + ({"tasks": {"other": {"name": "other"}}}, False), + ({"tasks": {"work": {"name": "work"}, "extra": {"name": "extra"}}}, False), + ({"tasks": {"work": {"name": "work"}}}, True), + ], +) +async def test_task_only_deployment_requires_definitions_without_launching( + tmp_path, definitions, approved +): + app = App("my-app") + + @app.task(name="work", cpu="cpu3c-1-2") + def work(): + pass + + api = AsyncMock() + api.get_app_by_name.return_value = { + "id": "app", + "flashEnvironments": [{"id": "env", "name": "default"}], + } + api.prepare_artifact_upload.return_value = { + "uploadUrl": "upload", + "objectKey": "key", + } + api.finalize_artifact_upload.return_value = {"id": "build"} + api.finalize_artifact_upload.return_value["resourceSpec"] = definitions + env_dir = tmp_path / "vendor" + env_dir.mkdir() + with patch( + "runpod.apps.deploy.build_environment", + return_value=BuildResult(env_dir=env_dir, requirements=[]), + ): + if approved: + result = await deploy_app(app, tmp_path, api=api) + assert result.resources == ["work"] + assert result.endpoints == {} + api.deploy_build.assert_awaited_once_with("env", "build") + else: + with pytest.raises(RuntimeError, match="acknowledge"): + await deploy_app(app, tmp_path, api=api) + api.deploy_build.assert_not_awaited() + api.save_endpoint.assert_not_awaited() + api.deploy_task_pod.assert_not_awaited() diff --git a/tests/test_apps/test_dispatch.py b/tests/test_apps/test_dispatch.py index 6a2543a83..5d0c27d57 100644 --- a/tests/test_apps/test_dispatch.py +++ b/tests/test_apps/test_dispatch.py @@ -207,6 +207,9 @@ def t(x): def test_worker_executes_own_body(self, monkeypatch): monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-1") monkeypatch.setenv("FLASH_RESOURCE_NAME", "q") + monkeypatch.setenv("FLASH_APP_NAME", "my-app") + monkeypatch.setenv("FLASH_ENVIRONMENT", "production") + monkeypatch.setenv("RUNPOD_BUILD_ID", "build-1") app = App("my-app") @app.queue(name="q") @@ -251,6 +254,9 @@ async def main(): def test_worker_calls_sibling_via_sentinel(self, monkeypatch): monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-1") monkeypatch.setenv("FLASH_RESOURCE_NAME", "other") + monkeypatch.setenv("FLASH_APP_NAME", "my-app") + monkeypatch.setenv("FLASH_ENVIRONMENT", "production") + monkeypatch.setenv("RUNPOD_BUILD_ID", "build-1") app = App("my-app") @app.queue(name="q") diff --git a/tests/test_cli/test_deploy_command.py b/tests/test_cli/test_deploy_command.py index 2fe5942e7..0757ae059 100644 --- a/tests/test_cli/test_deploy_command.py +++ b/tests/test_cli/test_deploy_command.py @@ -62,6 +62,34 @@ def _result(name="demo", endpoints=None): class TestDeploy: + + def test_task_only_deployment_shows_registered_per_call( + self, tmp_path, monkeypatch + ): + source = """ +import runpod +app = runpod.App("tasks") +@app.task(name="batch", cpu="cpu3c-1-2") +def batch(): + return 1 +""" + (tmp_path / "main.py").write_text(source) + monkeypatch.chdir(tmp_path) + deployed = DeployResult( + app_name="tasks", + build_id="build", + environment_id="env", + resources=["batch"], + endpoints={}, + ) + with patch("runpod.apps.deploy.deploy_app", AsyncMock(return_value=deployed)): + result = _runner().invoke(cli, ["flash", "deploy"]) + assert result.exit_code == 0, result.output + assert "batch" in result.output + assert "task" in result.output + assert "cpu3c-1-2" in result.output + assert "registered / per-call" in result.output + def test_deploys_discovered_app(self, tmp_path, monkeypatch): (tmp_path / "main.py").write_text(APP_SOURCE) monkeypatch.chdir(tmp_path)