cli test

Author: wiki0831

tests/integration/test_data_cli.py

@@ -1072,59 +1072,154 @@ def test_asset_wait(invoke,
     assert "status: active" in result.output
 
 
-# @respx.mock
-# def test_asset_get(invoke):
-#     item_type = 'PSScene'
-#     item_id = '20221003_002705_38_2461xx'
-#     asset_type_id = 'basic_udm2'
-#     dl_url = f'{TEST_URL}/1?token=IAmAToken'
-
-#     basic_udm2_asset = {
-#         "_links": {
-#             "_self": "SELFURL",
-#             "activate": "ACTIVATEURL",
-#             "type": "https://api.planet.com/data/v1/asset-types/basic_udm2"
-#         },
-#         "_permissions": ["download"],
-#         "md5_digest": None,
-#         "status": 'active',
-#         "location": dl_url,
-#         "type": "basic_udm2"
-#     }
-
-#     page_response = {
-#         "basic_analytic_4b": {
-#             "_links": {
-#                 "_self":
-#                 "SELFURL",
-#                 "activate":
-#                 "ACTIVATEURL",
-#                 "type":
-#                 "https://api.planet.com/data/v1/asset-types/basic_analytic_4b"
-#             },
-#             "_permissions": ["download"],
-#             "md5_digest": None,
-#             "status": "inactive",
-#             "type": "basic_analytic_4b"
-#         },
-#         "basic_udm2": basic_udm2_asset
-#     }
-
-#     mock_resp = httpx.Response(HTTPStatus.OK, json=page_response)
-#     assets_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/assets'
-#     respx.get(assets_url).return_value = mock_resp
-
-#     runner = CliRunner()
-#     result = invoke(['asset-get', item_type, item_id, asset_type_id],
-#                     runner=runner)
-
-#     assert not result.exception
-#     assert json.dumps(basic_udm2_asset) in result.output
-
-# TODO: basic test for "planet data search-list".
-# TODO: basic test for "planet data search-run".
-# TODO: basic test for "planet data item-get".
-# TODO: basic test for "planet data stats".
+@respx.mock
+def test_asset_get_success(invoke,
+                           mock_asset_get_response,
+                           item_type,
+                           item_id,
+                           asset_type):
+    """Test successful asset get command."""
+    mock_asset_get_response()
+    result = invoke(["asset-get", item_type, item_id, asset_type])
+    assert result.exit_code == 0
+    response = json.loads(result.output)
+    assert response["status"] == "active"
+    assert response["type"] == asset_type
+
+
+@respx.mock
+def test_asset_get_not_found(invoke, item_type, item_id):
+    """Test asset get command with non-existent asset."""
+    asset_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/assets'
+    respx.get(asset_url).return_value = httpx.Response(HTTPStatus.NOT_FOUND,
+                                                       json={})
+
+    result = invoke(["asset-get", item_type, item_id, "non_existent_asset"])
+    assert result.exit_code == 1
+
+
+@respx.mock
+def test_asset_list_success(invoke,
+                            mock_asset_get_response,
+                            item_type,
+                            item_id):
+    """Test successful asset list command."""
+    mock_asset_get_response()
+    result = invoke(["asset-list", item_type, item_id])
+    assert result.exit_code == 0
+    assets = json.loads(result.output)
+    assert "basic_udm2" in assets
+    assert "basic_analytic_4b" in assets
+    assert assets["basic_udm2"]["status"] == "active"
+    assert assets["basic_analytic_4b"]["status"] == "inactive"
+
+
+@respx.mock
+def test_asset_list_not_found(invoke, item_type, item_id):
+    """Test asset list command with non-existent item."""
+    asset_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/assets'
+    respx.get(asset_url).return_value = httpx.Response(HTTPStatus.NOT_FOUND,
+                                                       json={})
+
+    result = invoke(["asset-list", item_type, item_id])
+    assert result.exit_code == 1
+
+
+@respx.mock
+def test_item_get_success(invoke, item_type, item_id, search_result):
+    """Test successful item get command."""
+    item_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}'
+    mock_resp = httpx.Response(HTTPStatus.OK, json=search_result)
+    respx.get(item_url).return_value = mock_resp
+
+    result = invoke(["item-get", item_type, item_id])
+    assert result.exit_code == 0
+    assert json.loads(result.output) == search_result
+
+
+@respx.mock
+def test_item_get_not_found(invoke, item_type, item_id):
+    """Test item get command with non-existent item."""
+    item_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}'
+    mock_resp = httpx.Response(HTTPStatus.NOT_FOUND, json={})
+    respx.get(item_url).return_value = mock_resp
+
+    result = invoke(["item-get", item_type, item_id])
+    assert result.exit_code == 1
+
+
+@respx.mock
+def test_item_coverage_success(invoke, item_type, item_id, geom_geojson):
+    """Test successful item coverage command."""
+    coverage_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/coverage'
+    mock_coverage = {"clear_percent": 90.0, "status": "complete"}
+    mock_resp = httpx.Response(HTTPStatus.OK, json=mock_coverage)
+    respx.post(coverage_url).return_value = mock_resp
+
+    result = invoke([
+        "item-coverage",
+        item_type,
+        item_id,
+        "--geom",
+        json.dumps(geom_geojson)
+    ])
+    assert result.exit_code == 0
+    coverage = json.loads(result.output)
+    assert coverage["clear_percent"] == 90.0
+    assert coverage["status"] == "complete"
+
+
+@respx.mock
+def test_item_coverage_with_mode_and_band(invoke,
+                                          item_type,
+                                          item_id,
+                                          geom_geojson):
+    """Test item coverage command with mode and band options."""
+    coverage_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/coverage'
+    mock_coverage = {"cloud_percent": 90.0, "status": "complete"}
+    mock_resp = httpx.Response(HTTPStatus.OK, json=mock_coverage)
+    respx.post(coverage_url).return_value = mock_resp
+
+    result = invoke([
+        "item-coverage",
+        item_type,
+        item_id,
+        "--geom",
+        json.dumps(geom_geojson),
+        "--mode",
+        "UDM2",
+        "--band",
+        "cloud"
+    ])
+    assert result.exit_code == 0
+    coverage = json.loads(result.output)
+    assert coverage["cloud_percent"] == 90.0
+    assert coverage["status"] == "complete"
+
+
+@respx.mock
+def test_item_coverage_invalid_geometry(invoke, item_type, item_id):
+    """Test item coverage command with invalid geometry."""
+    result = invoke(
+        ["item-coverage", item_type, item_id, "--geom", "invalid geom"])
+    assert result.exit_code == 1
+
+
+@respx.mock
+def test_item_coverage_not_found(invoke, item_type, item_id, geom_geojson):
+    """Test item coverage command with non-existent item."""
+    coverage_url = f'{TEST_URL}/item-types/{item_type}/items/{item_id}/coverage'
+    mock_resp = httpx.Response(HTTPStatus.NOT_FOUND, json={})
+    respx.post(coverage_url).return_value = mock_resp
+
+    result = invoke([
+        "item-coverage",
+        item_type,
+        item_id,
+        "--geom",
+        json.dumps(geom_geojson)
+    ])
+    assert result.exit_code == 1
 
 
 @respx.mock