Item type and asset type validation in Data and Subscriptions

planet/cli/data.py

@@ -407,10 +407,6 @@ async def search_run(ctx, search_id, sort, limit, pretty):
             echo_json(item, pretty)
 
 
-# TODO: search-update
-# TODO: search-delete
-
-
 @data.command(epilog=valid_item_string)
 @click.pass_context
 @translate_exceptions
@@ -471,7 +467,7 @@ async def search_delete(ctx, search_id):
         await cl.delete_search(search_id)
 
 
-@data.command()
+@data.command(epilog=valid_item_string)
 @click.pass_context
 @translate_exceptions
 @coro
@@ -514,7 +510,7 @@ async def search_update(ctx,
         echo_json(items, pretty)
 
 
-@data.command()
+@data.command(epilog=valid_item_string)
 @click.pass_context
 @translate_exceptions
 @coro
@@ -576,7 +572,7 @@ async def asset_download(ctx,
             cl.validate_checksum(asset, path)
 
 
-@data.command()
+@data.command(epilog=valid_item_string)
 @click.pass_context
 @translate_exceptions
 @coro
@@ -590,7 +586,7 @@ async def asset_activate(ctx, item_type, item_id, asset_type):
         await cl.activate_asset(asset)
 
 
-@data.command()
+@data.command(epilog=valid_item_string)
 @click.pass_context
 @translate_exceptions
 @coro

planet/cli/subscriptions.py

@@ -1,5 +1,6 @@
 """Subscriptions CLI"""
 from contextlib import asynccontextmanager
+from typing import List, Optional
 
 import click
 
@@ -10,6 +11,32 @@
 from .session import CliSession
 from planet.clients.subscriptions import SubscriptionsClient
 from .. import subscription_request
+from ..specs import get_item_types, validate_item_type, SpecificationException
+
+ALL_ITEM_TYPES = get_item_types()
+valid_item_string = "Valid entries for ITEM_TYPES: " + "|".join(ALL_ITEM_TYPES)
+
+
+def check_item_types(ctx, param, item_types) -> Optional[List[dict]]:
+    '''Validates each item types provided by comparing them to all supported
+    item types.'''
+    try:
+        for item_type in item_types:
+            validate_item_type(item_type)
+        return item_types
+    except SpecificationException as e:
+        raise click.BadParameter(str(e))
+
+
+def check_item_type(ctx, param, item_type) -> Optional[List[dict]]:
+    '''Validates the item type provided by comparing it to all supported
+    item types.'''
+    try:
+        validate_item_type(item_type)
+    except SpecificationException as e:
+        raise click.BadParameter(str(e))
+
+    return item_type
 
 
 @asynccontextmanager
@@ -161,6 +188,7 @@ async def list_subscription_results_cmd(ctx,
 
 
 @subscriptions.command()
+@translate_exceptions
 @click.option('--name',
               required=True,
               type=str,
@@ -192,11 +220,13 @@ def request(name, source, delivery, notifications, tools, pretty):
     echo_json(res, pretty)
 
 
-@subscriptions.command()
+@subscriptions.command(epilog=valid_item_string)
+@translate_exceptions
 @click.option('--item-types',
               required=True,
+              help='Item type for requested item ids.',
               type=types.CommaSeparatedString(),
-              help='One or more comma-separated item types.')
+              callback=check_item_types)
 @click.option('--asset-types',
               required=True,
               type=types.CommaSeparatedString(),

planet/clients/data.py

@@ -533,6 +533,7 @@ async def get_asset(self,
             planet.exceptions.ClientError: If asset type identifier is not
             valid.
         """
+        item_type_id = validate_item_type(item_type_id)
         assets = await self.list_item_assets(item_type_id, item_id)
 
         try:

planet/specs.py

@@ -109,6 +109,14 @@ def validate_supported_bundles(item_type, bundle, all_product_bundles):
     return _validate_field(bundle, supported_bundles, 'bundle')
 
 
+def validate_asset_type(item_type, asset_type):
+    '''Validates an asset type for a given item type.'''
+    item_type = validate_item_type(item_type)
+    supported_assets = get_supported_assets(item_type)
+
+    return _validate_field(asset_type, supported_assets, 'asset_type')
+
+
 def _get_product_bundle_spec():
     with open(DATA_DIR / PRODUCT_BUNDLE_SPEC_NAME) as f:
         data = json.load(f)
@@ -162,3 +170,17 @@ def get_item_types(product_bundle=None):
                 for bundle in get_product_bundles()))
 
     return item_types
+
+
+def get_supported_assets(item_type):
+    '''Get all assets supported by a given item type.'''
+    item_type = validate_item_type(item_type)
+    supported_bundles = get_product_bundles(item_type)
+    spec = _get_product_bundle_spec()
+    supported_assets = [
+        spec['bundles'][bundle]["assets"][item_type]
+        for bundle in supported_bundles
+    ]
+    supported_assets = list(set(list(itertools.chain(*supported_assets))))
+
+    return supported_assets

planet/subscription_request.py

@@ -127,6 +127,18 @@ def catalog_source(
         planet.exceptions.ClientError: If start_time or end_time are not valid
             datetimes
     '''
+    if len(item_types) > 1:
+        raise ClientError(
+            "Subscription can only be successfully created if one item type",
+            "is specified.")
+    try:
+        asset_types = [
+            specs.validate_asset_type(item, asset) for asset in asset_types
+            for item in item_types
+        ]
+    except specs.SpecificationException as exception:
+        raise ClientError(exception)
+
     parameters = {
         "item_types": item_types,
         "asset_types": asset_types,

tests/unit/test_callback_functions.py

@@ -0,0 +1,89 @@
+# Copyright 2022 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
+import pytest
+import click
+from planet.cli import data
+from planet.cli import subscriptions
+
+LOGGER = logging.getLogger(__name__)
+
+
+class MockContext:
+
+    def __init__(self):
+        self.obj = {}
+
+
+def test_item_types_success_data():
+    ctx = MockContext()
+    result = data.check_item_types(ctx, 'item_types', ["PSScene"])
+    assert result == ["PSScene"]
+
+
+def test_item_types_fail_data():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        data.check_item_types(ctx, 'item_types', "bad_item_type")
+
+
+def test_item_type_success_data():
+    ctx = MockContext()
+    item_type = "PSScene"
+    result = data.check_item_type(ctx, 'item_type', item_type)
+    assert result == item_type
+
+
+def test_item_type_fail_data():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        data.check_item_type(ctx, 'item_type', "bad_item_type")
+
+
+def test_item_type_too_many_item_types_data():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        data.check_item_types(ctx, 'item_type', "PSScene,SkySatScene")
+
+
+# Identical tests to above, but for subscriptions CLI
+def test_item_types_success_subscriptions():
+    ctx = MockContext()
+    result = subscriptions.check_item_types(ctx, 'item_types', ["PSScene"])
+    assert result == ["PSScene"]
+
+
+def test_item_types_fail_subscriptions():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        subscriptions.check_item_types(ctx, 'item_types', "bad_item_type")
+
+
+def test_item_type_success_subscriptions():
+    ctx = MockContext()
+    item_type = "PSScene"
+    result = subscriptions.check_item_type(ctx, 'item_type', item_type)
+    assert result == item_type
+
+
+def test_item_type_fail_subscriptions():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        subscriptions.check_item_type(ctx, 'item_type', "bad_item_type")
+
+
+def test_item_type_too_many_item_types_subscriptions():
+    ctx = MockContext()
+    with pytest.raises(click.BadParameter):
+        subscriptions.check_item_types(ctx, 'item_type', "PSScene,SkySatScene")

tests/unit/test_specs.py

@@ -66,6 +66,7 @@
     'MYD09GQ',
     'SkySatScene'
 ]
+TEST_ASSET_TYPE = "basic_udm2"
 
 
 def test_get_type_match():
@@ -162,3 +163,25 @@ def test_validate_supported_bundles_fail():
         specs.validate_supported_bundles(TEST_ITEM_TYPE,
                                          'analytic',
                                          ALL_PRODUCT_BUNDLES)
+
+
+def test_get_supported_assets_success():
+    supported_assets = specs.get_supported_assets(TEST_ITEM_TYPE)
+    assert TEST_ASSET_TYPE in supported_assets
+
+
+def test_get_supported_assets_not_supported_item_type():
+    with pytest.raises(specs.SpecificationException):
+        specs.get_supported_assets('notsupported')
+
+
+def test_validate_asset_type_supported():
+    '''Ensures that a validated asset type for a given item type matches the
+    the given asset type.'''
+    assert TEST_ASSET_TYPE == specs.validate_asset_type(
+        TEST_ITEM_TYPE, TEST_ASSET_TYPE)
+
+
+def test_validate_asset_type_notsupported():
+    with pytest.raises(specs.SpecificationException):
+        specs.validate_asset_type(TEST_ITEM_TYPE, 'notsupported')