1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
|
# pylint: disable=line-too-long,useless-suppression
# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------
"""
Unit tests for STAC Collection lifecycle operations (create, update, delete).
Note: These tests are marked with pytest.mark.live_test_only as they modify collections.
"""
import logging
import time
import pytest
from pathlib import Path
from devtools_testutils import recorded_by_proxy, is_live
from testpreparer import PlanetaryComputerProClientTestBase, PlanetaryComputerPreparer
from azure.planetarycomputer.models import (
StacExtensionSpatialExtent,
StacCollectionTemporalExtent,
StacExtensionExtent,
)
# Set up test logger
test_logger = logging.getLogger("test_collection_lifecycle")
test_logger.setLevel(logging.DEBUG)
# Create logs directory if it doesn't exist
log_dir = Path(__file__).parent / "logs"
log_dir.mkdir(exist_ok=True)
# File handler for test logs
log_file = log_dir / "collection_lifecycle_test_results.log"
file_handler = logging.FileHandler(log_file, mode="w")
file_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
file_handler.setFormatter(formatter)
test_logger.addHandler(file_handler)
class TestPlanetaryComputerCollectionLifecycle(PlanetaryComputerProClientTestBase):
"""Test suite for STAC Collection lifecycle operations."""
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_01_begin_create_collection(self, planetarycomputer_endpoint):
"""
Test creating a new STAC collection.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_01_begin_create_collection")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
test_collection_id = "test-collection-lifecycle"
# Check if collection exists and delete it first
try:
existing_collection = client.stac.get_collection(
collection_id=test_collection_id
)
if existing_collection:
test_logger.info(
f"Collection '{test_collection_id}' already exists, deleting first..."
)
delete_poller = client.stac.begin_delete_collection(
collection_id=test_collection_id, polling=True
)
delete_poller.result()
test_logger.info(f"Deleted existing collection '{test_collection_id}'")
except Exception:
test_logger.info(
f"Collection '{test_collection_id}' does not exist, proceeding with creation"
)
# Define collection extents
spatial_extent = StacExtensionSpatialExtent(bounding_box=[[-180, -90, 180, 90]])
temporal_extent = StacCollectionTemporalExtent(
interval=[["2020-01-01T00:00:00Z", "2024-12-31T23:59:59Z"]]
)
extent = StacExtensionExtent(spatial=spatial_extent, temporal=temporal_extent)
# Create collection payload
collection_data = {
"id": test_collection_id,
"description": "Test collection for lifecycle operations",
"extent": extent.as_dict(),
"license": "proprietary",
"links": [],
"stac_version": "1.0.0",
"title": "Test Collection Lifecycle",
"type": "Collection",
}
test_logger.info("Calling: begin_create_collection(body=collection_data)")
create_poller = client.stac.begin_create_collection(
body=collection_data, polling=True
)
result = create_poller.result()
test_logger.info(f"Collection created: {result}")
created_collection = client.stac.get_collection(
collection_id=test_collection_id
)
assert created_collection is not None
assert created_collection.id == test_collection_id
assert created_collection.title == "Test Collection Lifecycle"
test_logger.info("Test PASSED\n")
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_02_create_or_replace_collection(self, planetarycomputer_endpoint):
"""
Test updating a collection using create or replace.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_02_create_or_replace_collection")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
test_collection_id = "test-collection-lifecycle"
# Get existing collection
collection = client.stac.get_collection(collection_id=test_collection_id)
# Update description
collection.description = "Test collection for lifecycle operations - UPDATED"
test_logger.info(
f"Calling: create_or_replace_collection(collection_id='{test_collection_id}', body=collection)"
)
updated_collection = client.stac.create_or_replace_collection(
collection_id=test_collection_id, body=collection
)
test_logger.info(f"Collection updated: {updated_collection}")
updated_collection = client.stac.get_collection(
collection_id=test_collection_id
)
assert (
updated_collection.description
== "Test collection for lifecycle operations - UPDATED"
)
test_logger.info("Test PASSED\n")
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_03_begin_delete_collection(self, planetarycomputer_endpoint):
"""
Test deleting a STAC collection.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_03_begin_delete_collection")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
test_collection_id = "test-collection-lifecycle"
test_logger.info(
f"Calling: begin_delete_collection(collection_id='{test_collection_id}')"
)
delete_poller = client.stac.begin_delete_collection(
collection_id=test_collection_id, polling=True
)
result = delete_poller.result()
test_logger.info(f"Delete operation completed: {result}")
try:
client.stac.get_collection(collection_id=test_collection_id)
assert False, "Collection should have been deleted"
except Exception as e:
test_logger.info(f"Collection successfully deleted (404 expected): {e}")
assert (
"404" in str(e) or "Not Found" in str(e) or "ResourceNotFound" in str(e)
)
test_logger.info("Test PASSED\n")
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_04_create_collection_asset(
self, planetarycomputer_endpoint, planetarycomputer_collection_id
):
"""
Test creating a collection asset.
Note: This test uses the existing test collection.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_04_create_collection_asset")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
# Create a simple text file asset
from io import BytesIO
# Delete the asset if it already exists
try:
test_logger.info(
"Checking if asset 'test-asset' already exists and deleting if found..."
)
client.stac.delete_collection_asset(
collection_id=planetarycomputer_collection_id, asset_id="test-asset"
)
test_logger.info("Deleted existing 'test-asset'")
except Exception as e:
if (
"404" in str(e)
or "Not Found" in str(e)
or "not found" in str(e).lower()
):
test_logger.info(
"Asset 'test-asset' does not exist, proceeding with creation"
)
else:
test_logger.warning(f"Error checking/deleting asset: {e}")
asset_data = {
"key": "test-asset",
"href": "https://example.com/test-asset.txt",
"type": "text/plain",
"roles": ["metadata"],
"title": "Test Asset",
}
file_content = BytesIO(b"Test asset content")
file_tuple = ("test-asset.txt", file_content)
test_logger.info(
f"Calling: create_collection_asset(collection_id='{planetarycomputer_collection_id}', body={{...}})"
)
response = client.stac.create_collection_asset(
collection_id=planetarycomputer_collection_id,
body={"data": asset_data, "file": file_tuple},
)
test_logger.info(f"Response: {response}")
assert response is not None
test_logger.info("Test PASSED\n")
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_05_replace_collection_asset(
self, planetarycomputer_endpoint, planetarycomputer_collection_id
):
"""
Test creating or replacing a collection asset.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_05_replace_collection_asset")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
from io import BytesIO
asset_data = {
"key": "test-asset",
"href": "https://example.com/test-asset-updated.txt",
"type": "text/plain",
"roles": ["metadata"],
"title": "Test Asset - Updated",
}
file_content = BytesIO(b"Test asset content - updated")
file_tuple = ("test-asset.txt", file_content)
test_logger.info(
f"Calling: create_or_replace_collection_asset(collection_id='{planetarycomputer_collection_id}', asset_id='test-asset', body={{...}})"
)
response = client.stac.replace_collection_asset(
collection_id=planetarycomputer_collection_id,
asset_id="test-asset",
body={"data": asset_data, "file": file_tuple},
)
test_logger.info(f"Response: {response}")
assert response is not None
test_logger.info("Test PASSED\n")
@PlanetaryComputerPreparer()
@recorded_by_proxy
def test_06_delete_collection_asset(
self, planetarycomputer_endpoint, planetarycomputer_collection_id
):
"""
Test deleting a collection asset.
First creates an asset specifically for deletion.
"""
test_logger.info("=" * 80)
test_logger.info("TEST: test_06_delete_collection_asset")
test_logger.info("=" * 80)
client = self.create_client(endpoint=planetarycomputer_endpoint)
# Create an asset to be deleted using create_collection_asset first
test_logger.info("Creating asset for deletion: test-asset-to-be-deleted")
# First create the asset
from io import BytesIO
file_content = BytesIO(b"Test asset content for deletion")
# Prepare data and file tuple for multipart form
data = {
"key": "test-asset-to-be-deleted",
"href": "https://example.com/test-asset-to-delete.txt",
"type": "text/plain",
"roles": ["metadata"],
"title": "Test Asset To Be Deleted",
}
file_tuple = ("test-asset-to-delete.txt", file_content)
client.stac.create_collection_asset(
collection_id=planetarycomputer_collection_id,
body={"data": data, "file": file_tuple},
)
test_logger.info("Asset created successfully")
# Now delete it
test_logger.info(
f"Calling: delete_collection_asset(collection_id='{planetarycomputer_collection_id}', asset_id='test-asset-to-be-deleted')"
)
client.stac.delete_collection_asset(
collection_id=planetarycomputer_collection_id,
asset_id="test-asset-to-be-deleted",
)
test_logger.info("Asset deleted successfully")
# Verify deletion by checking collection assets
collection = client.stac.get_collection(
collection_id=planetarycomputer_collection_id
)
if hasattr(collection, "assets") and collection.assets:
assert (
"test-asset-to-be-deleted" not in collection.assets
), "Asset should have been deleted"
test_logger.info("Test PASSED\n")
|
