1
1
import datetime as dt
2
+ import os
2
3
import shutil
3
4
import tempfile
4
5
from pathlib import Path
6
+ from typing import Set
5
7
6
8
from pystac import Asset , Collection , Extent , Item , SpatialExtent , TemporalExtent , CatalogType , Link , RelType
7
9
import pytest
@@ -60,6 +62,14 @@ def test_merge_from_disk_new(tmp_path):
60
62
workspace = DiskWorkspace (root_directory = tmp_path )
61
63
imported_collection = workspace .merge (stac_resource = new_collection , target = target )
62
64
65
+ workspace_dir = (workspace .root_directory / target ).parent
66
+
67
+ assert _paths_relative_to (workspace_dir ) == {
68
+ Path ("collection.json" ),
69
+ Path ("collection.json_items" ) / "asset.tif" ,
70
+ Path ("collection.json_items" ) / "asset.tif.json" ,
71
+ }
72
+
63
73
assert isinstance (imported_collection , Collection )
64
74
asset_workspace_uris = {
65
75
asset_key : asset .extra_fields ["alternate" ]["file" ]
@@ -71,7 +81,6 @@ def test_merge_from_disk_new(tmp_path):
71
81
}
72
82
73
83
# load it again
74
- workspace_dir = (workspace .root_directory / target ).parent
75
84
exported_collection = Collection .from_file (str (workspace_dir / "collection.json" ))
76
85
assert exported_collection .validate_all () == 1
77
86
assert _downloadable_assets (exported_collection ) == 1
@@ -108,6 +117,15 @@ def test_merge_from_disk_into_existing(tmp_path):
108
117
workspace .merge (stac_resource = existing_collection , target = target )
109
118
imported_collection = workspace .merge (stac_resource = new_collection , target = target )
110
119
120
+ workspace_dir = (workspace .root_directory / target ).parent
121
+ assert _paths_relative_to (workspace_dir ) == {
122
+ Path ("collection.json" ),
123
+ Path ("collection.json_items" ) / "asset1.tif" ,
124
+ Path ("collection.json_items" ) / "asset1.tif.json" ,
125
+ Path ("collection.json_items" ) / "asset2.tif" ,
126
+ Path ("collection.json_items" ) / "asset2.tif.json" ,
127
+ }
128
+
111
129
assert isinstance (imported_collection , Collection )
112
130
asset_workspace_uris = {
113
131
asset_key : asset .extra_fields ["alternate" ]["file" ]
@@ -119,7 +137,6 @@ def test_merge_from_disk_into_existing(tmp_path):
119
137
}
120
138
121
139
# load it again
122
- workspace_dir = (workspace .root_directory / target ).parent
123
140
exported_collection = Collection .from_file (str (workspace_dir / "collection.json" ))
124
141
assert exported_collection .validate_all () == 2
125
142
assert _downloadable_assets (exported_collection ) == 2
@@ -230,3 +247,11 @@ def _downloadable_assets(collection: Collection) -> int:
230
247
shutil .copy (asset .get_absolute_href (), temp_file .name ) # "download" the asset without altering its href
231
248
232
249
return len (assets )
250
+
251
+
252
+ def _paths_relative_to (base : Path ) -> Set [Path ]:
253
+ return {
254
+ (Path (dirpath ) / filename ).relative_to (base )
255
+ for dirpath , dirnames , filenames in os .walk (base )
256
+ for filename in filenames
257
+ }
0 commit comments