Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def validate_catalog_link_type(href, link_type, should_include_self):
cat_dict = STAC_IO.read_json(href)
cat = STACObject.from_file(href)
for link in cat.get_links():
if not link.rel == 'self':
self.assertEqual(link.link_type, link_type)
rels = set([l['rel'] for l in cat_dict['links']])
self.assertEqual('self' in rels, should_include_self)
for child_link in cat.get_child_links():
child_href = make_absolute_href(child_link.target, href)
validate_catalog_link_type(child_href, link_type,
catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
for item_link in cat.get_item_links():
item_href = make_absolute_href(item_link.target, href)
validate_item_link_type(item_href, link_type,
catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
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)
'eo:gsd': 0.3,
'eo:platform': 'Maxar',
'eo:instrument': 'WorldView3'
}
collection = Collection(id='test',
description='test',
extent=collection_extent,
properties=common_properties,
license='CC-BY-SA-4.0')
collection.add_items([item1, item2])
with TemporaryDirectory() as tmp_dir:
collection.normalize_hrefs(tmp_dir)
collection.save(catalog_type=CatalogType.SELF_CONTAINED)
read_col = Collection.from_file('{}/collection.json'.format(tmp_dir))
items = list(read_col.get_all_items())
self.assertEqual(len(items), 2)
self.assertIsInstance(items[0], EOItem)
self.assertIsInstance(items[1], EOItem)
for link in cat.get_links():
if not link.rel == 'self':
self.assertEqual(link.link_type, link_type)
rels = set([l['rel'] for l in cat_dict['links']])
self.assertEqual('self' in rels, should_include_self)
for child_link in cat.get_child_links():
child_href = make_absolute_href(child_link.target, href)
validate_catalog_link_type(child_href, link_type,
catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
for item_link in cat.get_item_links():
item_href = make_absolute_href(item_link.target, href)
validate_item_link_type(item_href, link_type,
catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
def asset_mapper(key, asset):
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)
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_full_copy_4(self):
with TemporaryDirectory() as tmp_dir:
root_cat = TestCases.test_case_2()
root_cat.normalize_hrefs(os.path.join(tmp_dir, 'catalog-full-copy-4-source'))
root_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
cat2 = root_cat.full_copy()
cat2.normalize_hrefs(os.path.join(tmp_dir, 'catalog-full-copy-4-dest'))
cat2.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
self.check_catalog(root_cat, 'source')
self.check_catalog(cat2, 'dest')
# Check that the relative asset link was saved correctly in the copy.
item = cat2.get_item('cf73ec1a-d790-4b59-b077-e101738571ed', recursive=True)
href = item.assets['cf73ec1a-d790-4b59-b077-e101738571ed'].get_absolute_href()
self.assertTrue(os.path.exists(href))