diff --git a/contributing/ORPHAN-CLEANUP.md b/contributing/ORPHAN-CLEANUP.md index 5c04d67d4f..c720aef344 100644 --- a/contributing/ORPHAN-CLEANUP.md +++ b/contributing/ORPHAN-CLEANUP.md @@ -35,7 +35,7 @@ Processing a row: - No record: ignored. Resources of other servers and projects are never touched. - STEP 5: Owners that were provisioned or terminated around the listing are skipped instead of classified as orphans, since the listing may predate the change. The margin covers clock skew between replicas, the lag between setting timestamps and committing them, and providers listing new resources with a delay. - STEP 6: Handle orphans according to `DSTACK_SERVER_ORPHAN_CLEANUP_MODE`: - - `report` (default): log, send to Sentry, and emit an event targeting the owner instance. If the instance was deleted, the event still targets it by the id and name kept in `ProviderRefModel`, so it shows next to the instance's earlier events. + - `report` (default): log, send to Sentry, and emit an event targeting the owner instance. If the instance was deleted, the event still targets it by the id and name kept in `ProviderRefModel`, so it shows next to the instance's earlier events. Reports are not deduplicated: an orphan that stays is reported again on every cleanup of the backend, at most once per cleanup interval. This is expected and accepted. - `delete`: call `delete_resource_group()` and emit an event. If the backend raises `NotYetTerminated` for a group with an owner resource, the owner's deletion was requested and the event says the resources are being deleted. The remaining internal resources are listed and deleted by next cleanups. - `off`: the pipeline and enrollment are disabled. - STEP 7: Set the outcome: diff --git a/src/dstack/_internal/cli/services/presets/output.py b/src/dstack/_internal/cli/services/presets/output.py index 52df677d61..86f39d54c7 100644 --- a/src/dstack/_internal/cli/services/presets/output.py +++ b/src/dstack/_internal/cli/services/presets/output.py @@ -1,5 +1,4 @@ from collections import defaultdict -from datetime import datetime from typing import Any, Optional, Sequence from rich.table import Table @@ -8,7 +7,7 @@ from dstack._internal.cli.utils.common import add_row_from_dict, console from dstack._internal.core.models.configurations import PresetConfiguration from dstack._internal.core.models.presets import PresetWorkload -from dstack._internal.utils.common import pretty_date, pretty_resources +from dstack._internal.utils.common import parse_datetime, pretty_date, pretty_resources _STATUS_DISPLAY = { "verified": ("verified", "secondary"), @@ -161,9 +160,7 @@ def _add_session(table: Table, session: dict[str, Any], *, verbose: bool = False created_at = session.get("created_at") if isinstance(created_at, str): try: - # `Z` is what pydantic v2 emits for UTC, and `datetime.fromisoformat` only - # accepts it from Python 3.11; dstack supports 3.10. - created = pretty_date(datetime.fromisoformat(created_at.replace("Z", "+00:00"))) + created = pretty_date(parse_datetime(created_at)) except ValueError: created = created_at benchmark = "" diff --git a/src/dstack/_internal/core/backends/runpod/api_client.py b/src/dstack/_internal/core/backends/runpod/api_client.py index c550cb72da..9573e5ef59 100644 --- a/src/dstack/_internal/core/backends/runpod/api_client.py +++ b/src/dstack/_internal/core/backends/runpod/api_client.py @@ -19,6 +19,11 @@ AUTH_ERROR_STATUS_CODES = (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN) +PAGINATION_PAGE_LIMIT = 1000 +"""The maximum page size of paginated v2 endpoints.""" + +PAGINATION_MAX_PAGES = 20 + # Applies to non-JSON error responses, which are not expected to contain anything useful past # the first few lines MAX_ERROR_DETAIL_LEN = 200 @@ -38,16 +43,24 @@ class V2ErrorResponse(TypedDict): class V2Pod(TypedDict): id: str + name: str status: Literal["PROVISIONING", "STARTING", "RUNNING", "EXITED", "ERROR", "TERMINATED"] runtime: Optional["V2PodRuntime"] mounts: "V2Mounts" cluster: NotRequired["V2PodCluster"] + createdAt: str + dataCenterId: Optional[str] class V2PodList(TypedDict): pods: list[V2Pod] +class V2Pagination(TypedDict): + nextCursor: Optional[str] + hasNextPage: bool + + class V2PodCluster(TypedDict): id: str rank: Optional[int] @@ -395,6 +408,12 @@ def get_pod(self, pod_id: str) -> Optional[V2Pod]: raise return response.json() + def list_pods(self) -> list[V2Pod]: + """ + Lists pods except cluster member pods. Deleted pods are not listed. + """ + return self._paginate_v2("pods", response_key="pods") + def terminate_pod(self, pod_id: str) -> bool: try: self._make_v2_request("DELETE", "pods", pod_id) @@ -534,8 +553,35 @@ def _make_graphql_request(self, data: Optional[Dict[str, Any]] = None) -> reques raise BackendInvalidCredentialsError(e.response.text) raise + def _paginate_v2(self, *path_parts: str, response_key: str) -> list[Any]: + """ + Follows `pagination.nextCursor` of a cursor-paginated GET endpoint and returns + the items of all pages, or of the first `PAGINATION_MAX_PAGES` pages. + """ + result = [] + params: dict[str, Any] = {"limit": PAGINATION_PAGE_LIMIT} + for _ in range(PAGINATION_MAX_PAGES): + response_json = self._make_v2_request("GET", *path_parts, params=params).json() + result.extend(response_json[response_key]) + pagination: V2Pagination = response_json["pagination"] + if not pagination["hasNextPage"] or pagination["nextCursor"] is None: + break + params["cursor"] = pagination["nextCursor"] + else: + logger.error( + "Listing %s exceeded the limit of %s pages, the rest is skipped", + "/".join(path_parts), + PAGINATION_MAX_PAGES, + ) + return result + def _make_v2_request( - self, method: str, /, *path_parts: str, json: Optional[Mapping[str, Any]] = None + self, + method: str, + /, + *path_parts: str, + json: Optional[Mapping[str, Any]] = None, + params: Optional[Mapping[str, Any]] = None, ) -> requests.Response: url = "/".join((API_V2_URL, *path_parts)) response = self._session.request( @@ -543,6 +589,7 @@ def _make_v2_request( url=url, timeout=REQUEST_TIMEOUT, json=json, + params=params, ) if HTTPStatus.BAD_REQUEST <= response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR: error = _parse_v2_error_response(response) diff --git a/src/dstack/_internal/core/backends/runpod/compute.py b/src/dstack/_internal/core/backends/runpod/compute.py index 44dc37032e..324a74a9a6 100644 --- a/src/dstack/_internal/core/backends/runpod/compute.py +++ b/src/dstack/_internal/core/backends/runpod/compute.py @@ -1,7 +1,7 @@ import json import re import uuid -from datetime import timedelta +from datetime import datetime, timedelta from typing import Callable, List, Optional from dstack._internal.core.backends.base.authorized_keys import build_authorized_keys @@ -10,12 +10,14 @@ ComputeWithAllOffersCached, ComputeWithGroupProvisioningSupport, ComputeWithMultinodeSupport, + ComputeWithResourceListingSupport, ComputeWithVolumeSupport, generate_unique_instance_name, generate_unique_volume_name, get_docker_commands, get_job_instance_name, get_job_provider_ref, + parse_provider_ref_from_name, ) from dstack._internal.core.backends.base.models import JobConfiguration from dstack._internal.core.backends.base.offers import get_catalog_offers @@ -24,6 +26,11 @@ from dstack._internal.core.consts import DSTACK_RUNNER_SSH_PORT from dstack._internal.core.errors import BackendError, ComputeError from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.backends.orphans import ( + ListedResourceGroup, + OwnerType, + ResourceListing, +) from dstack._internal.core.models.common import CoreModel, RegistryAuth, validate_extra_ignore from dstack._internal.core.models.compute_groups import ComputeGroup, ComputeGroupProvisioningData from dstack._internal.core.models.instances import ( @@ -39,7 +46,7 @@ Volume, VolumeProvisioningData, ) -from dstack._internal.utils.common import get_current_datetime, get_or_error +from dstack._internal.utils.common import get_current_datetime, get_or_error, parse_datetime from dstack._internal.utils.logging import get_logger logger = get_logger(__name__) @@ -69,6 +76,7 @@ class RunpodCompute( ComputeWithVolumeSupport, ComputeWithMultinodeSupport, ComputeWithGroupProvisioningSupport, + ComputeWithResourceListingSupport, Compute, ): _last_cleanup_time = None @@ -136,7 +144,9 @@ def run_job( provider_ref=get_job_provider_ref(job), ) - pod_name = generate_unique_instance_name(instance_config, max_length=MAX_RESOURCE_NAME_LEN) + pod_name = generate_unique_instance_name( + instance_config, max_length=MAX_RESOURCE_NAME_LEN, with_provider_ref=True + ) authorized_keys = build_authorized_keys(project_ssh_public_key, extra_authorized_keys) network_volume_id = None @@ -402,6 +412,36 @@ def update_provisioning_data( provisioning_data.hostname = port.get("ip") provisioning_data.ssh_port = port.get("public") + def list_resource_groups(self) -> ResourceListing: + resource_groups = [] + pods = self.api_client.list_pods() + for pod in pods: + if pod["status"] == "TERMINATED": + continue + provider_ref = parse_provider_ref_from_name(pod["name"]) + if provider_ref is None: + continue + resource_group = ListedResourceGroup( + backend=BackendType.RUNPOD, + owner_type=OwnerType.INSTANCE, + provider_ref=provider_ref, + primary_id=pod["id"], + region=pod["dataCenterId"], + created_at=_parse_pod_created_at(pod), + description=f"pod {pod['id']}", + backend_data=None, + ) + resource_groups.append(resource_group) + return ResourceListing( + resource_groups=resource_groups, + failures=[], + ) + + def delete_resource_group(self, group: ListedResourceGroup) -> None: + if group.primary_id is None: + return + self.api_client.terminate_pod(group.primary_id) + def register_volume(self, volume: Volume) -> VolumeProvisioningData: assert isinstance(volume.configuration, RunpodVolumeConfiguration) volume_id = get_or_error(volume.configuration.volume_id) @@ -524,3 +564,11 @@ def _get_pod_cluster_ip(pod: V2Pod) -> Optional[str]: if cluster is None: return None return cluster.get("ip") + + +def _parse_pod_created_at(pod: V2Pod) -> Optional[datetime]: + try: + return parse_datetime(pod["createdAt"]) + except ValueError: + logger.warning("Failed to parse createdAt %r of pod %s", pod["createdAt"], pod["id"]) + return None diff --git a/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py b/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py index 1a401bb5f0..8406b1a664 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py @@ -552,6 +552,9 @@ async def _handle_orphaned_resource_group( resource_group: ListedResourceGroup, ) -> _ResourceGroupDecision: if settings.ORPHAN_CLEANUP_MODE == OrphanCleanupMode.REPORT: + # Expected and accepted: reports are not deduplicated, so an orphan that stays + # is reported (logged, sent to Sentry, emitted as an event) on every cleanup + # of the backend, i.e. at most once per cleanup interval while the backend is eligible. logger.info( "Found orphaned %s %s in %s/%s", resource_group.owner_type.value, diff --git a/src/dstack/_internal/server/background/scheduled_tasks/__init__.py b/src/dstack/_internal/server/background/scheduled_tasks/__init__.py index 46408d2602..33aa278303 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/__init__.py +++ b/src/dstack/_internal/server/background/scheduled_tasks/__init__.py @@ -59,7 +59,9 @@ def start_scheduled_tasks() -> AsyncIOScheduler: _scheduler.add_job(delete_instance_healthchecks, IntervalTrigger(minutes=5), max_instances=1) if settings.ORPHAN_CLEANUP_MODE != OrphanCleanupMode.OFF: _scheduler.add_job( - enroll_backends_in_orphan_cleanup, IntervalTrigger(minutes=10), max_instances=1 + enroll_backends_in_orphan_cleanup, + IntervalTrigger(seconds=settings.ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS), + max_instances=1, ) if settings.ENABLE_PROMETHEUS_METRICS: _scheduler.add_job( diff --git a/src/dstack/_internal/server/services/logs/fluentbit.py b/src/dstack/_internal/server/services/logs/fluentbit.py index bb97e21f09..f49f56e9a2 100644 --- a/src/dstack/_internal/server/services/logs/fluentbit.py +++ b/src/dstack/_internal/server/services/logs/fluentbit.py @@ -1,4 +1,3 @@ -from datetime import datetime from typing import List, Optional, Protocol from uuid import UUID @@ -19,7 +18,7 @@ LogStorageError, unix_time_ms_to_datetime, ) -from dstack._internal.utils.common import batched +from dstack._internal.utils.common import batched, parse_datetime from dstack._internal.utils.logging import get_logger logger = get_logger(__name__) @@ -113,7 +112,7 @@ def read( if timestamp_str: try: - timestamp = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")) + timestamp = parse_datetime(timestamp_str) except ValueError: continue else: diff --git a/src/dstack/_internal/server/settings.py b/src/dstack/_internal/server/settings.py index 806c45b53b..0b04a51186 100644 --- a/src/dstack/_internal/server/settings.py +++ b/src/dstack/_internal/server/settings.py @@ -240,9 +240,11 @@ class JobNetworkMode(Enum): ORPHAN_CLEANUP_INTERVAL_SECONDS = environ.get_int( "DSTACK_SERVER_ORPHAN_CLEANUP_INTERVAL_SECONDS", default=3600 ) - - DEFAULT_ORPHAN_CLEANUP_MODE = OrphanCleanupMode.REPORT ORPHAN_CLEANUP_MODE = environ.get_enum( "DSTACK_SERVER_ORPHAN_CLEANUP_MODE", OrphanCleanupMode, default=DEFAULT_ORPHAN_CLEANUP_MODE ) +ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS = environ.get_int( + "DSTACK_SERVER_ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS", default=600 +) +"""Internal, e.g. for e2e tests""" diff --git a/src/dstack/_internal/utils/common.py b/src/dstack/_internal/utils/common.py index 33db805f04..9f0f4208ca 100644 --- a/src/dstack/_internal/utils/common.py +++ b/src/dstack/_internal/utils/common.py @@ -85,6 +85,16 @@ def render_datetime_as_api(value: datetime) -> str: return _DATETIME_ADAPTER.dump_python(value, mode="json") +def parse_datetime(value: str) -> datetime: + """ + Parse a datetime the way the API parses one: an ISO 8601 string or a Unix timestamp. + + Unlike `datetime.fromisoformat()` on Python 3.10, accepts the `Z` suffix and fractional + seconds of any precision. Raises `ValueError` if `value` is not a datetime. + """ + return _DATETIME_ADAPTER.validate_strings(value) + + DateFormatter = Callable[[datetime], str] @@ -202,7 +212,7 @@ def parse_since(value: str) -> datetime: except ValueError: pass try: - res = datetime.fromisoformat(value) + res = parse_datetime(value) except ValueError: raise ValueError("Invalid datetime format") else: diff --git a/src/tests/_internal/core/backends/runpod/test_api_client.py b/src/tests/_internal/core/backends/runpod/test_api_client.py index 712aa3c124..315a57cbdd 100644 --- a/src/tests/_internal/core/backends/runpod/test_api_client.py +++ b/src/tests/_internal/core/backends/runpod/test_api_client.py @@ -2,6 +2,7 @@ from dstack._internal.core.backends.runpod.api_client import ( API_V2_URL, + PAGINATION_MAX_PAGES, RunpodApiClient, RunpodV2ClientError, ) @@ -193,3 +194,36 @@ def test_other_errors_are_raised(self, requests_mock): with pytest.raises(RunpodV2ClientError): _client().terminate_pod("pod-1") + + +class TestRunpodApiClientListPods: + def _page(self, pod_ids: list[str], next_cursor: str | None) -> dict: + return { + "pods": [{"id": pod_id} for pod_id in pod_ids], + "pagination": {"nextCursor": next_cursor, "hasNextPage": next_cursor is not None}, + } + + def test_follows_cursor(self, requests_mock): + requests_mock.get( + f"{API_V2_URL}/pods", + [ + {"json": self._page(["pod-1", "pod-2"], "cursor-1")}, + {"json": self._page(["pod-3"], None)}, + ], + ) + + pods = _client().list_pods() + + assert [pod["id"] for pod in pods] == ["pod-1", "pod-2", "pod-3"] + assert [r.qs for r in requests_mock.request_history] == [ + {"limit": ["1000"]}, + {"limit": ["1000"], "cursor": ["cursor-1"]}, + ] + + def test_stops_at_max_pages(self, requests_mock): + requests_mock.get(f"{API_V2_URL}/pods", json=self._page(["pod-1"], "cursor-1")) + + pods = _client().list_pods() + + assert len(pods) == PAGINATION_MAX_PAGES + assert requests_mock.call_count == PAGINATION_MAX_PAGES diff --git a/src/tests/_internal/core/backends/runpod/test_compute.py b/src/tests/_internal/core/backends/runpod/test_compute.py index dd7e982ddb..964b6128df 100644 --- a/src/tests/_internal/core/backends/runpod/test_compute.py +++ b/src/tests/_internal/core/backends/runpod/test_compute.py @@ -1,13 +1,16 @@ +from datetime import datetime, timezone from typing import Optional from unittest.mock import MagicMock, patch import pytest +from dstack._internal.core.backends.base.compute import parse_provider_ref_from_name from dstack._internal.core.backends.runpod.api_client import RunpodV2ClientError from dstack._internal.core.backends.runpod.compute import RunpodCompute from dstack._internal.core.backends.runpod.models import RunpodAPIKeyCreds, RunpodConfig from dstack._internal.core.errors import ComputeError from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.backends.orphans import ListedResourceGroup, OwnerType from dstack._internal.core.models.instances import InstanceOfferWithAvailability from dstack._internal.core.models.resources import ResourcesSpec from dstack._internal.core.models.runs import JobProvisioningData, Requirements @@ -63,14 +66,18 @@ def _run() -> MagicMock: return run -def _run_job(compute: RunpodCompute, offer: InstanceOfferWithAvailability) -> JobProvisioningData: +def _run_job( + compute: RunpodCompute, + offer: InstanceOfferWithAvailability, + job: Optional[MagicMock] = None, +) -> JobProvisioningData: with patch( "dstack._internal.core.backends.runpod.compute.get_docker_commands", return_value=["echo hi"], ): return compute.run_job( run=_run(), - job=_job(), + job=job if job is not None else _job(), instance_offer=offer, project_ssh_public_key=PROJECT_SSH_PUBLIC_KEY, project_ssh_private_key="private-key", @@ -158,6 +165,17 @@ def test_failed_registry_credentials_cleanup_does_not_fail_provisioning(self): assert jpd.instance_id == "pod-1" compute.api_client.terminate_pod.assert_not_called() + def test_encodes_provider_ref_in_pod_name(self): + compute = _compute() + compute.api_client.create_pod.return_value = {"id": "pod-1"} + job = _job() + job.job_submissions = [MagicMock(provider_ref="abcdef123456")] + + _run_job(compute, _gpu_offer(), job=job) + + pod_name = compute.api_client.create_pod.call_args.kwargs["name"] + assert parse_provider_ref_from_name(pod_name) == "abcdef123456" + class TestRunpodComputeRunJobs: def test_sets_internal_ip_from_cluster_ip(self): @@ -256,3 +274,103 @@ def test_cluster_pod_sets_internal_ip(self): assert jpd.hostname == "203.0.113.10" assert jpd.ssh_port == 40022 assert jpd.internal_ip == "10.65.0.3" + + +def _listed_pod( + pod_id: str, + *, + name: str, + status: str = "RUNNING", + data_center_id: Optional[str] = "EU-RO-1", + created_at: str = "2026-10-23T10:00:00Z", +) -> dict: + pod = _pod(pod_id) + pod["status"] = status + pod["name"] = name + pod["dataCenterId"] = data_center_id + pod["createdAt"] = created_at + return pod + + +class TestRunpodComputeListResourceGroups: + def test_lists_pods_with_provider_ref(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod("pod-1", name="dstack-main-test-job-0-abcdef123456wxyz"), + ] + + listing = compute.list_resource_groups() + + assert listing.failures == [] + assert listing.resource_groups == [ + ListedResourceGroup( + backend=BackendType.RUNPOD, + owner_type=OwnerType.INSTANCE, + provider_ref="abcdef123456", + primary_id="pod-1", + region="EU-RO-1", + created_at=datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc), + description="pod pod-1", + backend_data=None, + ) + ] + + def test_skips_pods_without_provider_ref(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod("pod-1", name="dstack-main-test-job-0-abcdefgh"), + _listed_pod("pod-2", name="my-pod"), + ] + + assert compute.list_resource_groups().resource_groups == [] + + def test_skips_terminated_pods(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod( + "pod-1", name="dstack-main-test-job-0-abcdef123456wxyz", status="TERMINATED" + ), + ] + + assert compute.list_resource_groups().resource_groups == [] + + def test_unparsable_created_at(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod( + "pod-1", name="dstack-main-test-job-0-abcdef123456wxyz", created_at="invalid" + ), + ] + + [group] = compute.list_resource_groups().resource_groups + + assert group.created_at is None + + +class TestRunpodComputeDeleteResourceGroup: + def _group(self, primary_id: Optional[str]) -> ListedResourceGroup: + return ListedResourceGroup( + backend=BackendType.RUNPOD, + owner_type=OwnerType.INSTANCE, + provider_ref="abcdef123456", + primary_id=primary_id, + region="EU-RO-1", + created_at=None, + description="pod pod-1", + backend_data=None, + ) + + def test_terminates_pod(self): + compute = _compute() + + compute.delete_resource_group(self._group("pod-1")) + + compute.api_client.terminate_pod.assert_called_once_with("pod-1") + + def test_pod_already_deleted(self): + compute = _compute() + compute.api_client.terminate_pod.return_value = False + + compute.delete_resource_group(self._group("pod-1")) + + compute.api_client.terminate_pod.assert_called_once_with("pod-1") diff --git a/src/tests/_internal/utils/test_common.py b/src/tests/_internal/utils/test_common.py index 91f8bf0768..789b4854e5 100644 --- a/src/tests/_internal/utils/test_common.py +++ b/src/tests/_internal/utils/test_common.py @@ -12,9 +12,12 @@ has_duplicates, local_time, make_proxy_url, + parse_datetime, parse_memory, + parse_since, pretty_date, pretty_resources, + render_datetime_as_api, sizeof_fmt, ) @@ -151,6 +154,55 @@ def test_has_duplicates(iterable, expected): assert has_duplicates(iterable) == expected +class TestParseDatetime: + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("2026-10-23T10:00:00Z", datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc)), + ( + "2026-10-23T10:00:00.12345Z", + datetime(2026, 10, 23, 10, 0, 0, 123450, tzinfo=timezone.utc), + ), + ( + "2026-10-23T12:00:00+02:00", + datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc), + ), + ("2026-10-23T10:00:00", datetime(2026, 10, 23, 10, 0)), + ], + ) + def test_parses(self, value: str, expected: datetime) -> None: + assert parse_datetime(value) == expected + + def test_roundtrips_render_datetime_as_api(self) -> None: + dt = datetime(2026, 10, 23, 10, 0, 0, 123456, tzinfo=timezone.utc) + assert parse_datetime(render_datetime_as_api(dt)) == dt + + @pytest.mark.parametrize("value", ["invalid", "", "2026-13-01T00:00:00Z"]) + def test_raises_value_error(self, value: str) -> None: + with pytest.raises(ValueError): + parse_datetime(value) + + +class TestParseSince: + @pytest.mark.time_machine(datetime(2023, 10, 4, 12, 0, tzinfo=timezone.utc), tick=False) + def test_duration(self) -> None: + assert parse_since("5m") == datetime(2023, 10, 4, 11, 55, tzinfo=timezone.utc) + + @pytest.mark.parametrize( + "value", ["2023-09-24T15:30:00Z", "2023-09-24T15:30:00+00:00", "2023-09-24T17:30:00+02:00"] + ) + def test_datetime(self, value: str) -> None: + assert parse_since(value) == datetime(2023, 9, 24, 15, 30, tzinfo=timezone.utc) + + def test_naive_datetime(self) -> None: + with pytest.raises(ValueError): + parse_since("2023-09-24T15:30:00") + + def test_invalid(self) -> None: + with pytest.raises(ValueError, match="Invalid datetime format"): + parse_since("invalid") + + class TestParseMemory: @pytest.mark.parametrize( "memory,as_units,expected",