Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_validate_label(self):
with open(self.label_example_1_uri) as f:
label_example_1_dict = json.load(f)
self.validator.validate_dict(label_example_1_dict, "ITEM")
with TemporaryDirectory() as tmp_dir:
cat_dir = os.path.join(tmp_dir, 'catalog')
catalog = TestCases.test_case_1()
catalog.normalize_and_save(cat_dir, catalog_type=CatalogType.SELF_CONTAINED)
cat_read = Catalog.from_file(os.path.join(cat_dir, 'catalog.json'))
label_item_read = cat_read.get_item("area-2-2-labels", recursive=True)
self.validator.validate_object(label_item_read)
def test_case_2():
return Catalog.from_file(
TestCases.get_path('data-files/catalogs/test-case-2/catalog.json'))
def asset_mapper(key, asset):
if key == changed_asset:
asset.title = 'NEW TITLE'
return asset
with TemporaryDirectory() as tmp_dir:
catalog = TestCases.test_case_2()
new_cat = catalog.map_assets(asset_mapper)
new_cat.normalize_hrefs(os.path.join(tmp_dir, 'cat'))
new_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
result_cat = Catalog.from_file(os.path.join(tmp_dir, 'cat', 'catalog.json'))
found = False
for item in result_cat.get_all_items():
for key, asset in item.assets.items():
if key == changed_asset:
found = True
self.assertEqual(asset.title, 'NEW TITLE')
else:
self.assertNotEqual(asset.title, 'NEW TITLE')
self.assertTrue(found)
if 'geotiff' in asset.media_type:
asset.title = 'NEW TITLE'
changed_assets.append(key)
return ('{}-modified'.format(key), asset)
else:
return asset
with TemporaryDirectory() as tmp_dir:
catalog = TestCases.test_case_2()
new_cat = catalog.map_assets(asset_mapper)
new_cat.normalize_hrefs(os.path.join(tmp_dir, 'cat'))
new_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
result_cat = Catalog.from_file(os.path.join(tmp_dir, 'cat', 'catalog.json'))
found = False
not_found = False
for item in result_cat.get_all_items():
for key, asset in item.assets.items():
if key.replace('-modified', '') in changed_assets:
found = True
self.assertEqual(asset.title, 'NEW TITLE')
else:
not_found = True
self.assertNotEqual(asset.title, 'NEW TITLE')
self.assertTrue(found)
self.assertTrue(not_found)
mod1.title = 'NEW TITLE 1'
mod2 = asset.clone()
mod2.title = 'NEW TITLE 2'
return {'{}-mod-1'.format(key): mod1, '{}-mod-2'.format(key): mod2}
else:
return asset
with TemporaryDirectory() as tmp_dir:
catalog = TestCases.test_case_2()
new_cat = catalog.map_assets(asset_mapper)
new_cat.normalize_hrefs(os.path.join(tmp_dir, 'cat'))
new_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
result_cat = Catalog.from_file(os.path.join(tmp_dir, 'cat', 'catalog.json'))
found1 = False
found2 = False
not_found = False
for item in result_cat.get_all_items():
for key, asset in item.assets.items():
if key.replace('-mod-1', '') in changed_assets:
found1 = True
self.assertEqual(asset.title, 'NEW TITLE 1')
elif key.replace('-mod-2', '') in changed_assets:
found2 = True
self.assertEqual(asset.title, 'NEW TITLE 2')
else:
not_found = True
self.assertNotEqual(asset.title, 'NEW TITLE')
""" Test case to cover issue #88 """
stac_uri = 'tests/data-files/catalogs/test-case-6/catalog.json'
cat = Catalog.from_file(stac_uri)
# Iterate over the items. This was causing failure in
# in the later iterations as per issue #88
for item in cat.get_all_items():
pass
with TemporaryDirectory() as tmp_dir:
new_stac_uri = os.path.join(tmp_dir, 'test-case-6')
cat.normalize_hrefs(new_stac_uri)
cat.save(catalog_type=CatalogType.SELF_CONTAINED)
# Open the local copy and iterate over it.
cat2 = Catalog.from_file(os.path.join(new_stac_uri, 'catalog.json'))
for item in cat2.get_all_items():
# Iterate again over the items. This would fail in #88
pass
def test_reading_iterating_and_writing_works_as_expected(self):
""" Test case to cover issue #88 """
stac_uri = 'tests/data-files/catalogs/test-case-6/catalog.json'
cat = Catalog.from_file(stac_uri)
# Iterate over the items. This was causing failure in
# in the later iterations as per issue #88
for item in cat.get_all_items():
pass
with TemporaryDirectory() as tmp_dir:
new_stac_uri = os.path.join(tmp_dir, 'test-case-6')
cat.normalize_hrefs(new_stac_uri)
cat.save(catalog_type=CatalogType.SELF_CONTAINED)
# Open the local copy and iterate over it.
cat2 = Catalog.from_file(os.path.join(new_stac_uri, 'catalog.json'))
for item in cat2.get_all_items():
# Iterate again over the items. This would fail in #88
def test_create_and_read(self):
with TemporaryDirectory() as tmp_dir:
cat_dir = os.path.join(tmp_dir, 'catalog')
catalog = TestCases.test_case_1()
catalog.normalize_and_save(cat_dir, catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
read_catalog = Catalog.from_file('{}/catalog.json'.format(cat_dir))
collections = catalog.get_children()
self.assertEqual(len(list(collections)), 2)
items = read_catalog.get_all_items()
self.assertEqual(len(list(items)), 8)
def test_read_remote(self):
# TODO: Move this URL to the main stac-spec repo once the example JSON is fixed.
catalog_url = (
'https://raw.githubusercontent.com/lossyrob/stac-spec/0.9.0/pystac-upgrade-fixes'
'/extensions/label/examples/multidataset/catalog.json')
cat = Catalog.from_file(catalog_url)
zanzibar = cat.get_child('zanzibar-collection')
self.assertEqual(len(list(zanzibar.get_items())), 2)