From e49761f5a1aaad6274887def9cc2898cb3c5bb24 Mon Sep 17 00:00:00 2001 From: asonnenschein Date: Wed, 13 Aug 2025 16:56:43 -0400 Subject: [PATCH] initial commit --- planet/__init__.py | 3 +- planet/cli/cli.py | 3 +- planet/cli/quota.py | 135 +++++++++++++++++ planet/clients/__init__.py | 3 + planet/clients/quota.py | 153 ++++++++++++++++++++ planet/sync/client.py | 3 + planet/sync/quota.py | 104 ++++++++++++++ tests/integration/test_quota_api.py | 189 ++++++++++++++++++++++++ tests/integration/test_quota_cli.py | 215 ++++++++++++++++++++++++++++ tests/unit/test_client.py | 6 + tests/unit/test_quota.py | 74 ++++++++++ 11 files changed, 886 insertions(+), 2 deletions(-) create mode 100644 planet/cli/quota.py create mode 100644 planet/clients/quota.py create mode 100644 planet/sync/quota.py create mode 100644 tests/integration/test_quota_api.py create mode 100644 tests/integration/test_quota_cli.py create mode 100644 tests/unit/test_quota.py diff --git a/planet/__init__.py b/planet/__init__.py index 41a9e62b6..e80e928e8 100644 --- a/planet/__init__.py +++ b/planet/__init__.py @@ -17,7 +17,7 @@ from .__version__ import __version__ # NOQA from .auth import Auth from .auth_builtins import PlanetOAuthScopes -from .clients import DataClient, DestinationsClient, FeaturesClient, MosaicsClient, OrdersClient, SubscriptionsClient # NOQA +from .clients import DataClient, DestinationsClient, FeaturesClient, MosaicsClient, OrdersClient, QuotaClient, SubscriptionsClient # NOQA from .io import collect from .sync import Planet @@ -33,6 +33,7 @@ 'OrdersClient', 'order_request', 'Planet', + 'QuotaClient', 'reporting', 'Session', 'SubscriptionsClient', diff --git a/planet/cli/cli.py b/planet/cli/cli.py index 467b1e5b1..9c6ced11d 100644 --- a/planet/cli/cli.py +++ b/planet/cli/cli.py @@ -22,7 +22,7 @@ import planet from planet.cli import mosaics -from . import auth, cmds, collect, data, destinations, orders, subscriptions, features +from . import auth, cmds, collect, data, destinations, orders, subscriptions, features, quota LOGGER = logging.getLogger(__name__) @@ -130,6 +130,7 @@ def _configure_logging(verbosity): main.add_command(collect.collect) # type: ignore main.add_command(features.features) # type: ignore main.add_command(destinations.destinations) # type: ignore +main.add_command(quota.quota) # type: ignore main.add_command(mosaics.mosaics) # type: ignore if __name__ == "__main__": diff --git a/planet/cli/quota.py b/planet/cli/quota.py new file mode 100644 index 000000000..9e20fd129 --- /dev/null +++ b/planet/cli/quota.py @@ -0,0 +1,135 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +from contextlib import asynccontextmanager +import json +import click +from planet.cli.io import echo_json +from planet.clients.quota import QuotaClient +from .cmds import command +from .options import compact, limit +from .session import CliSession + + +@asynccontextmanager +async def quota_client(ctx): + async with CliSession() as sess: + cl = QuotaClient(sess, base_url=ctx.obj['BASE_URL']) + yield cl + + +@click.group() # type: ignore +@click.pass_context +@click.option('-u', + '--base-url', + default=None, + help='Assign custom base Quota API URL.') +def quota(ctx, base_url): + """Commands for interacting with the Quota API""" + ctx.obj['BASE_URL'] = base_url + + +@quota.group() # type: ignore +def reservations(): + """Commands for managing quota reservations""" + pass + + +@command(reservations, name="create") +@click.argument("request_file", type=click.Path(exists=True)) +async def reservation_create(ctx, request_file, pretty): + """Create a new quota reservation from a JSON request file. + Example: + \b + planet quota reservations create ./reservation_request.json + """ + async with quota_client(ctx) as cl: + with open(request_file) as f: + request = json.load(f) + result = await cl.create_reservation(request) + echo_json(result, pretty) + + +@command(reservations, name="list", extra_args=[limit, compact]) +@click.option("--status", help="Filter reservations by status") +async def reservations_list(ctx, pretty, limit, compact, status): + """List quota reservations. + Example: + \b + planet quota reservations list + planet quota reservations list --status active + """ + async with quota_client(ctx) as cl: + results = cl.list_reservations(status=status, limit=limit) + if compact: + compact_fields = ('id', 'name', 'status', 'created_at') + output = [{ + k: v + for k, v in row.items() if k in compact_fields + } async for row in results] + else: + output = [r async for r in results] + echo_json(output, pretty) + + +@command(reservations, name="get") +@click.argument("reservation_id", required=True) +async def reservation_get(ctx, reservation_id, pretty): + """Get a quota reservation by ID. + Example: + \b + planet quota reservations get 12345678-1234-5678-9012-123456789012 + """ + async with quota_client(ctx) as cl: + result = await cl.get_reservation(reservation_id) + echo_json(result, pretty) + + +@command(reservations, name="cancel") +@click.argument("reservation_id", required=True) +async def reservation_cancel(ctx, reservation_id, pretty): + """Cancel an existing quota reservation. + Example: + \b + planet quota reservations cancel 12345678-1234-5678-9012-123456789012 + """ + async with quota_client(ctx) as cl: + result = await cl.cancel_reservation(reservation_id) + echo_json(result, pretty) + + +@command(quota, name="estimate") +@click.argument("request_file", type=click.Path(exists=True)) +async def quota_estimate(ctx, request_file, pretty): + """Estimate quota requirements for a potential reservation. + Example: + \b + planet quota estimate ./estimation_request.json + """ + async with quota_client(ctx) as cl: + with open(request_file) as f: + request = json.load(f) + result = await cl.estimate_quota(request) + echo_json(result, pretty) + + +@command(quota, name="usage") +async def quota_usage(ctx, pretty): + """Get current quota usage and limits. + Example: + \b + planet quota usage + """ + async with quota_client(ctx) as cl: + result = await cl.get_quota_usage() + echo_json(result, pretty) diff --git a/planet/clients/__init__.py b/planet/clients/__init__.py index 6aae646f6..d831dd1fa 100644 --- a/planet/clients/__init__.py +++ b/planet/clients/__init__.py @@ -17,6 +17,7 @@ from .features import FeaturesClient from .mosaics import MosaicsClient from .orders import OrdersClient +from .quota import QuotaClient from .subscriptions import SubscriptionsClient __all__ = [ @@ -25,6 +26,7 @@ 'FeaturesClient', 'MosaicsClient', 'OrdersClient', + 'QuotaClient', 'SubscriptionsClient' ] @@ -35,5 +37,6 @@ 'features': FeaturesClient, 'mosaics': MosaicsClient, 'orders': OrdersClient, + 'quota': QuotaClient, 'subscriptions': SubscriptionsClient } diff --git a/planet/clients/quota.py b/planet/clients/quota.py new file mode 100644 index 000000000..345194c65 --- /dev/null +++ b/planet/clients/quota.py @@ -0,0 +1,153 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +import logging +from typing import AsyncIterator, Optional +from planet.clients.base import _BaseClient +from planet.constants import PLANET_BASE_URL +from planet.http import Session +from planet.models import Paged + +BASE_URL = f'{PLANET_BASE_URL}/quota/v1' +LOGGER = logging.getLogger(__name__) + + +class Reservations(Paged): + """Asynchronous iterator over reservations from a paged response.""" + NEXT_KEY = '_next' + ITEMS_KEY = 'reservations' + + +class QuotaClient(_BaseClient): + """High-level asynchronous access to Planet's quota API. + The Planet Quota Reservations API allows you to create, estimate, and view + existing quota reservations on the Planet platform for compatible products + including Planetary Variables, Analysis-Ready PlanetScope (ARPS), and select + PlanetScope imagery products. + Example: + ```python + >>> import asyncio + >>> from planet import Session + >>> + >>> async def main(): + ... async with Session() as sess: + ... cl = sess.client('quota') + ... # use client here + ... + >>> asyncio.run(main()) + ``` + """ + + def __init__(self, session: Session, base_url: Optional[str] = None): + """ + Parameters: + session: Open session connected to server. + base_url: The base URL to use. Defaults to production quota API + base url. + """ + super().__init__(session, base_url or BASE_URL) + + def _reservations_url(self): + return f'{self._base_url}/reservations' + + async def create_reservation(self, request: dict) -> dict: + """Create a new quota reservation. + Parameters: + request: Quota reservation request specification. + Returns: + Description of the created reservation. + Raises: + planet.exceptions.APIError: On API error. + """ + url = self._reservations_url() + response = await self._session.request(method='POST', + url=url, + json=request) + return response.json() + + async def estimate_quota(self, request: dict) -> dict: + """Estimate quota requirements for a potential reservation. + Parameters: + request: Quota estimation request specification. + Returns: + Quota estimation details including projected costs and usage. + Raises: + planet.exceptions.APIError: On API error. + """ + url = f'{self._base_url}/estimate' + response = await self._session.request(method='POST', + url=url, + json=request) + return response.json() + + async def list_reservations(self, + status: Optional[str] = None, + limit: int = 100) -> AsyncIterator[dict]: + """Iterate through list of quota reservations. + Parameters: + status: Filter reservations by status (e.g., 'active', 'completed', 'cancelled'). + limit: Maximum number of results to return. When set to 0, no + maximum is applied. + Yields: + Description of a quota reservation. + Raises: + planet.exceptions.APIError: On API error. + """ + url = self._reservations_url() + params = {} + if status: + params['status'] = status + response = await self._session.request(method='GET', + url=url, + params=params) + async for reservation in Reservations(response, + self._session.request, + limit=limit): + yield reservation + + async def get_reservation(self, reservation_id: str) -> dict: + """Get a quota reservation by ID. + Parameters: + reservation_id: Quota reservation identifier. + Returns: + Quota reservation details. + Raises: + planet.exceptions.APIError: On API error. + """ + url = f'{self._reservations_url()}/{reservation_id}' + response = await self._session.request(method='GET', url=url) + return response.json() + + async def cancel_reservation(self, reservation_id: str) -> dict: + """Cancel an existing quota reservation. + Parameters: + reservation_id: Quota reservation identifier. + Returns: + Updated reservation details. + Raises: + planet.exceptions.APIError: On API error. + """ + url = f'{self._reservations_url()}/{reservation_id}/cancel' + response = await self._session.request(method='POST', url=url) + return response.json() + + async def get_quota_usage(self) -> dict: + """Get current quota usage and limits. + Returns: + Current quota usage statistics and limits. + Raises: + planet.exceptions.APIError: On API error. + """ + url = f'{self._base_url}/usage' + response = await self._session.request(method='GET', url=url) + return response.json() diff --git a/planet/sync/client.py b/planet/sync/client.py index 993b35271..a5db765d4 100644 --- a/planet/sync/client.py +++ b/planet/sync/client.py @@ -4,6 +4,7 @@ from .data import DataAPI from .destinations import DestinationsAPI from .orders import OrdersAPI +from .quota import QuotaAPI from .subscriptions import SubscriptionsAPI from planet.http import Session from planet.__version__ import __version__ @@ -24,6 +25,7 @@ class Planet: - `orders`: Orders API. - `subscriptions`: Subscriptions API. - `features`: Features API + - `quota`: Quota API Quick start example: ```python @@ -66,3 +68,4 @@ def __init__(self, self._session, f"{planet_base}/subscriptions/v1/") self.features = FeaturesAPI(self._session, f"{planet_base}/features/v1/ogc/my/") + self.quota = QuotaAPI(self._session, f"{planet_base}/quota/v1") diff --git a/planet/sync/quota.py b/planet/sync/quota.py new file mode 100644 index 000000000..6a9ed7fbb --- /dev/null +++ b/planet/sync/quota.py @@ -0,0 +1,104 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +from typing import Iterator, Optional +from planet.clients.quota import QuotaClient +from planet.http import Session + + +class QuotaAPI: + """Synchronous quota API wrapper. + Provides synchronous access to Planet's quota reservation functionality. + """ + _client: QuotaClient + + def __init__(self, session: Session, base_url: Optional[str] = None): + """ + Parameters: + session: Open session connected to server. + base_url: The base URL to use. Defaults to production quota API + base url. + """ + self._client = QuotaClient(session, base_url) + + def create_reservation(self, request: dict) -> dict: + """Create a new quota reservation. + Parameters: + request: Quota reservation request specification. + Returns: + Description of the created reservation. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._call_sync( + self._client.create_reservation(request)) + + def estimate_quota(self, request: dict) -> dict: + """Estimate quota requirements for a potential reservation. + Parameters: + request: Quota estimation request specification. + Returns: + Quota estimation details including projected costs and usage. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._call_sync(self._client.estimate_quota(request)) + + def list_reservations(self, + status: Optional[str] = None, + limit: int = 100) -> Iterator[dict]: + """Iterate through list of quota reservations. + Parameters: + status: Filter reservations by status (e.g., 'active', 'completed', 'cancelled'). + limit: Maximum number of results to return. When set to 0, no + maximum is applied. + Yields: + Description of a quota reservation. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._aiter_to_iter( + self._client.list_reservations(status=status, limit=limit)) + + def get_reservation(self, reservation_id: str) -> dict: + """Get a quota reservation by ID. + Parameters: + reservation_id: Quota reservation identifier. + Returns: + Quota reservation details. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._call_sync( + self._client.get_reservation(reservation_id)) + + def cancel_reservation(self, reservation_id: str) -> dict: + """Cancel an existing quota reservation. + Parameters: + reservation_id: Quota reservation identifier. + Returns: + Updated reservation details. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._call_sync( + self._client.cancel_reservation(reservation_id)) + + def get_quota_usage(self) -> dict: + """Get current quota usage and limits. + Returns: + Current quota usage statistics and limits. + Raises: + planet.exceptions.APIError: On API error. + """ + return self._client._call_sync(self._client.get_quota_usage()) diff --git a/tests/integration/test_quota_api.py b/tests/integration/test_quota_api.py new file mode 100644 index 000000000..099ad7432 --- /dev/null +++ b/tests/integration/test_quota_api.py @@ -0,0 +1,189 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +from http import HTTPStatus +import json +import httpx +import pytest +import respx +from planet import QuotaClient, Session +from planet.auth import Auth +from planet.sync.quota import QuotaAPI + +pytestmark = pytest.mark.anyio +# Simulated host/path for testing purposes. Not a real subdomain. +TEST_URL = "http://test.planet.com/quota/v1" +TEST_RESERVATION_REQUEST = { + "name": "Test Reservation", + "products": ["PSScene"], + "item_types": ["PSScene4Band"], + "geometry": { + "type": "Polygon", + "coordinates": [[[-122.5, 37.7], [-122.4, 37.7], [-122.4, 37.8], + [-122.5, 37.8], [-122.5, 37.7]]] + } +} +TEST_RESERVATION_RESPONSE = { + "id": "12345678-1234-5678-9012-123456789012", + "name": "Test Reservation", + "status": "active", + "created_at": "2025-01-01T00:00:00Z", + "products": ["PSScene"], + "item_types": ["PSScene4Band"] +} +TEST_ESTIMATE_RESPONSE = { + "estimated_cost": 100.0, + "estimated_usage": { + "area_km2": 25.0, "item_count": 10 + } +} +TEST_RESERVATIONS_LIST = { + "reservations": [TEST_RESERVATION_RESPONSE], "_next": None +} +TEST_QUOTA_USAGE = { + "current_usage": { + "area_km2": 150.0, "item_count": 50 + }, + "quota_limits": { + "area_km2": 1000.0, "item_count": 500 + } +} +# Set up test clients +test_session = Session(auth=Auth.from_key(key="test")) + + +@respx.mock +class TestQuotaClientIntegration: + """Integration tests for QuotaClient with mocked HTTP responses.""" + + async def test_create_reservation(self): + """Test creating a quota reservation.""" + client = QuotaClient(test_session, base_url=TEST_URL) + route = respx.post(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.CREATED, + json=TEST_RESERVATION_RESPONSE) + result = await client.create_reservation(TEST_RESERVATION_REQUEST) + assert result == TEST_RESERVATION_RESPONSE + assert route.call_count == 1 + # Verify request payload + request = route.calls[0].request + assert json.loads(request.content) == TEST_RESERVATION_REQUEST + + async def test_estimate_quota(self): + """Test estimating quota requirements.""" + client = QuotaClient(test_session, base_url=TEST_URL) + route = respx.post(f"{TEST_URL}/estimate") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_ESTIMATE_RESPONSE) + result = await client.estimate_quota(TEST_RESERVATION_REQUEST) + assert result == TEST_ESTIMATE_RESPONSE + assert route.call_count == 1 + + async def test_list_reservations(self): + """Test listing quota reservations.""" + client = QuotaClient(test_session, base_url=TEST_URL) + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + reservations = [] + async for reservation in client.list_reservations(): + reservations.append(reservation) + assert len(reservations) == 1 + assert reservations[0] == TEST_RESERVATION_RESPONSE + assert route.call_count == 1 + + async def test_list_reservations_with_status_filter(self): + """Test listing quota reservations with status filter.""" + client = QuotaClient(test_session, base_url=TEST_URL) + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + reservations = [] + async for reservation in client.list_reservations(status="active"): + reservations.append(reservation) + assert len(reservations) == 1 + assert route.call_count == 1 + # Verify status parameter was sent + request = route.calls[0].request + assert "status=active" in str(request.url) + + async def test_get_reservation(self): + """Test getting a specific quota reservation.""" + client = QuotaClient(test_session, base_url=TEST_URL) + reservation_id = "12345678-1234-5678-9012-123456789012" + route = respx.get(f"{TEST_URL}/reservations/{reservation_id}") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATION_RESPONSE) + result = await client.get_reservation(reservation_id) + assert result == TEST_RESERVATION_RESPONSE + assert route.call_count == 1 + + async def test_cancel_reservation(self): + """Test cancelling a quota reservation.""" + client = QuotaClient(test_session, base_url=TEST_URL) + reservation_id = "12345678-1234-5678-9012-123456789012" + cancelled_response = { + **TEST_RESERVATION_RESPONSE, "status": "cancelled" + } + route = respx.post(f"{TEST_URL}/reservations/{reservation_id}/cancel") + route.return_value = httpx.Response(HTTPStatus.OK, + json=cancelled_response) + result = await client.cancel_reservation(reservation_id) + assert result["status"] == "cancelled" + assert route.call_count == 1 + + async def test_get_quota_usage(self): + """Test getting quota usage statistics.""" + client = QuotaClient(test_session, base_url=TEST_URL) + route = respx.get(f"{TEST_URL}/usage") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_QUOTA_USAGE) + result = await client.get_quota_usage() + assert result == TEST_QUOTA_USAGE + assert route.call_count == 1 + + +@respx.mock +class TestQuotaAPIIntegration: + """Integration tests for synchronous QuotaAPI.""" + + def test_create_reservation_sync(self): + """Test creating a quota reservation using sync API.""" + api = QuotaAPI(test_session, base_url=TEST_URL) + route = respx.post(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.CREATED, + json=TEST_RESERVATION_RESPONSE) + result = api.create_reservation(TEST_RESERVATION_REQUEST) + assert result == TEST_RESERVATION_RESPONSE + assert route.call_count == 1 + + def test_list_reservations_sync(self): + """Test listing quota reservations using sync API.""" + api = QuotaAPI(test_session, base_url=TEST_URL) + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + reservations = list(api.list_reservations()) + assert len(reservations) == 1 + assert reservations[0] == TEST_RESERVATION_RESPONSE + assert route.call_count == 1 + + def test_get_quota_usage_sync(self): + """Test getting quota usage using sync API.""" + api = QuotaAPI(test_session, base_url=TEST_URL) + route = respx.get(f"{TEST_URL}/usage") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_QUOTA_USAGE) + result = api.get_quota_usage() + assert result == TEST_QUOTA_USAGE + assert route.call_count == 1 diff --git a/tests/integration/test_quota_cli.py b/tests/integration/test_quota_cli.py new file mode 100644 index 000000000..6bac033f3 --- /dev/null +++ b/tests/integration/test_quota_cli.py @@ -0,0 +1,215 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +from http import HTTPStatus +import json +import tempfile +import httpx +import pytest +import respx +from click.testing import CliRunner +from planet.cli.quota import quota + +pytestmark = pytest.mark.anyio +# Simulated host/path for testing purposes. Not a real subdomain. +TEST_URL = "http://test.planet.com/quota/v1" +TEST_RESERVATION_REQUEST = { + "name": "Test Reservation", + "products": ["PSScene"], + "item_types": ["PSScene4Band"], + "geometry": { + "type": "Polygon", + "coordinates": [[[-122.5, 37.7], [-122.4, 37.7], [-122.4, 37.8], + [-122.5, 37.8], [-122.5, 37.7]]] + } +} +TEST_RESERVATION_RESPONSE = { + "id": "12345678-1234-5678-9012-123456789012", + "name": "Test Reservation", + "status": "active", + "created_at": "2025-01-01T00:00:00Z" +} +TEST_RESERVATIONS_LIST = { + "reservations": [TEST_RESERVATION_RESPONSE], "_next": None +} +TEST_ESTIMATE_RESPONSE = { + "estimated_cost": 100.0, + "estimated_usage": { + "area_km2": 25.0, "item_count": 10 + } +} +TEST_QUOTA_USAGE = { + "current_usage": { + "area_km2": 150.0, "item_count": 50 + }, + "quota_limits": { + "area_km2": 1000.0, "item_count": 500 + } +} + + +@respx.mock +class TestQuotaCLI: + """Integration tests for quota CLI commands.""" + + def test_quota_reservations_create(self): + """Test creating a reservation via CLI.""" + runner = CliRunner() + # Create temporary file with request JSON + with tempfile.NamedTemporaryFile(mode='w', + suffix='.json', + delete=False) as f: + json.dump(TEST_RESERVATION_REQUEST, f) + temp_file = f.name + try: + route = respx.post(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.CREATED, + json=TEST_RESERVATION_RESPONSE) + result = runner.invoke( + quota, + ['--base-url', TEST_URL, 'reservations', 'create', temp_file]) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output contains reservation ID + output_json = json.loads(result.output) + assert output_json['id'] == TEST_RESERVATION_RESPONSE['id'] + finally: + import os + os.unlink(temp_file) + + def test_quota_reservations_list(self): + """Test listing reservations via CLI.""" + runner = CliRunner() + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + result = runner.invoke( + quota, ['--base-url', TEST_URL, 'reservations', 'list']) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output contains reservation data + output_json = json.loads(result.output) + assert len(output_json) == 1 + assert output_json[0]['id'] == TEST_RESERVATION_RESPONSE['id'] + + def test_quota_reservations_list_with_status(self): + """Test listing reservations with status filter via CLI.""" + runner = CliRunner() + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + result = runner.invoke(quota, + [ + '--base-url', + TEST_URL, + 'reservations', + 'list', + '--status', + 'active' + ]) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify status parameter was sent + request = route.calls[0].request + assert "status=active" in str(request.url) + + def test_quota_reservations_list_compact(self): + """Test listing reservations with compact output via CLI.""" + runner = CliRunner() + route = respx.get(f"{TEST_URL}/reservations") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATIONS_LIST) + result = runner.invoke( + quota, + ['--base-url', TEST_URL, 'reservations', 'list', '--compact']) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify compact output only contains expected fields + output_json = json.loads(result.output) + assert len(output_json) == 1 + compact_fields = {'id', 'name', 'status', 'created_at'} + actual_fields = set(output_json[0].keys()) + assert actual_fields == compact_fields + + def test_quota_reservations_get(self): + """Test getting a specific reservation via CLI.""" + runner = CliRunner() + reservation_id = "12345678-1234-5678-9012-123456789012" + route = respx.get(f"{TEST_URL}/reservations/{reservation_id}") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_RESERVATION_RESPONSE) + result = runner.invoke( + quota, + ['--base-url', TEST_URL, 'reservations', 'get', reservation_id]) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output contains reservation data + output_json = json.loads(result.output) + assert output_json['id'] == reservation_id + + def test_quota_reservations_cancel(self): + """Test cancelling a reservation via CLI.""" + runner = CliRunner() + reservation_id = "12345678-1234-5678-9012-123456789012" + cancelled_response = { + **TEST_RESERVATION_RESPONSE, "status": "cancelled" + } + route = respx.post(f"{TEST_URL}/reservations/{reservation_id}/cancel") + route.return_value = httpx.Response(HTTPStatus.OK, + json=cancelled_response) + result = runner.invoke( + quota, + ['--base-url', TEST_URL, 'reservations', 'cancel', reservation_id]) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output shows cancelled status + output_json = json.loads(result.output) + assert output_json['status'] == 'cancelled' + + def test_quota_estimate(self): + """Test estimating quota via CLI.""" + runner = CliRunner() + # Create temporary file with estimation request JSON + with tempfile.NamedTemporaryFile(mode='w', + suffix='.json', + delete=False) as f: + json.dump(TEST_RESERVATION_REQUEST, f) + temp_file = f.name + try: + route = respx.post(f"{TEST_URL}/estimate") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_ESTIMATE_RESPONSE) + result = runner.invoke( + quota, ['--base-url', TEST_URL, 'estimate', temp_file]) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output contains estimation data + output_json = json.loads(result.output) + assert output_json['estimated_cost'] == 100.0 + finally: + import os + os.unlink(temp_file) + + def test_quota_usage(self): + """Test getting quota usage via CLI.""" + runner = CliRunner() + route = respx.get(f"{TEST_URL}/usage") + route.return_value = httpx.Response(HTTPStatus.OK, + json=TEST_QUOTA_USAGE) + result = runner.invoke(quota, ['--base-url', TEST_URL, 'usage']) + assert result.exit_code == 0 + assert route.call_count == 1 + # Verify output contains usage data + output_json = json.loads(result.output) + assert 'current_usage' in output_json + assert 'quota_limits' in output_json diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 27c4376ae..9cc203798 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -35,6 +35,9 @@ def test_planet_default_initialization(self): assert pl.features is not None assert pl.features._client._base_url == "https://api.planet.com/features/v1/ogc/my" + assert pl.quota is not None + assert pl.quota._client._base_url == "https://api.planet.com/quota/v1" + def test_planet_custom_base_url_initialization(self): """Test that Planet client accepts custom base URL.""" pl = Planet(base_url="https://custom.planet.com") @@ -50,3 +53,6 @@ def test_planet_custom_base_url_initialization(self): assert pl.features is not None assert pl.features._client._base_url == "https://custom.planet.com/features/v1/ogc/my" + + assert pl.quota is not None + assert pl.quota._client._base_url == "https://custom.planet.com/quota/v1" diff --git a/tests/unit/test_quota.py b/tests/unit/test_quota.py new file mode 100644 index 000000000..ecfb457e7 --- /dev/null +++ b/tests/unit/test_quota.py @@ -0,0 +1,74 @@ +# Copyright 2025 Planet Labs PBC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You may obtain a copy of +# the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations under +# the License. +from planet.clients.quota import QuotaClient +from planet.sync.quota import QuotaAPI +from planet.http import Session +from planet.auth import Auth + + +class TestQuotaClient: + """Test cases for the QuotaClient.""" + + def test_quota_client_initialization(self): + """Test that QuotaClient initializes correctly with default and custom URLs.""" + session = Session(auth=Auth.from_key(key="test")) + # Test default URL + client = QuotaClient(session) + assert client._base_url == "https://api.planet.com/quota/v1" + # Test custom URL + custom_url = "https://custom.planet.com/quota/v2/" + client_custom = QuotaClient(session, base_url=custom_url) + assert client_custom._base_url == "https://custom.planet.com/quota/v2" + + def test_quota_client_url_methods(self): + """Test URL construction methods.""" + session = Session(auth=Auth.from_key(key="test")) + client = QuotaClient(session) + assert client._reservations_url( + ) == "https://api.planet.com/quota/v1/reservations" + + +class TestQuotaAPI: + """Test cases for the synchronous QuotaAPI.""" + + def test_quota_api_initialization(self): + """Test that QuotaAPI initializes correctly.""" + session = Session(auth=Auth.from_key(key="test")) + # Test default URL + api = QuotaAPI(session) + assert api._client._base_url == "https://api.planet.com/quota/v1" + # Test custom URL + custom_url = "https://custom.planet.com/quota/v2/" + api_custom = QuotaAPI(session, base_url=custom_url) + assert api_custom._client._base_url == "https://custom.planet.com/quota/v2" + + +class TestPlanetSyncClientQuota: + """Test cases for quota integration in Planet sync client.""" + + def test_planet_quota_initialization(self): + """Test that Planet client includes quota API.""" + from planet.sync import Planet + pl = Planet() + # Test quota is included + assert pl.quota is not None + assert pl.quota._client._base_url == "https://api.planet.com/quota/v1" + + def test_planet_quota_custom_base_url(self): + """Test that Planet client quota uses custom base URL.""" + from planet.sync import Planet + pl = Planet(base_url="https://custom.planet.com") + # Test quota uses custom URL + assert pl.quota is not None + assert pl.quota._client._base_url == "https://custom.planet.com/quota/v1"