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
|
import tempfile
from typing import Any
import pytest
import pystac
from pystac import HIERARCHICAL_LINKS, Catalog, CatalogType, Collection
from pystac.utils import is_absolute_href, make_absolute_href
from pystac.validation import validate_dict
from tests.utils import TestCases
CTYPES = [
CatalogType.ABSOLUTE_PUBLISHED,
CatalogType.RELATIVE_PUBLISHED,
CatalogType.SELF_CONTAINED,
]
class TestSTACWriting:
"""Tests writing STACs, using JSON Schema validation,
and ensure that links are correctly set to relative or absolute.
"""
def validate_catalog(self, catalog: Catalog) -> int:
catalog.validate()
validated_count = 1
for child in catalog.get_children():
validated_count += self.validate_catalog(child)
for item in catalog.get_items():
item.validate()
validated_count += 1
return validated_count
def validate_file(self, path: str, object_type: str) -> list[Any]:
d = pystac.StacIO.default().read_json(path)
return validate_dict(d, pystac.STACObjectType(object_type))
def validate_link_types(self, root_href: str, catalog_type: CatalogType) -> None:
def validate_item_link_type(
href: str, link_type: str, should_include_self: bool
) -> None:
item_dict = pystac.StacIO.default().read_json(href)
item = pystac.Item.from_file(href)
rel_links = [
*HIERARCHICAL_LINKS,
*pystac.EXTENSION_HOOKS.get_extended_object_links(item),
]
for link in item.get_links():
if not link.rel == "self":
if link_type == "RELATIVE" and link.rel in rel_links:
assert not is_absolute_href(link.href)
else:
assert is_absolute_href(link.href)
rels = {link["rel"] for link in item_dict["links"]}
assert ("self" in rels) == should_include_self
def validate_catalog_link_type(
href: str, link_type: str, should_include_self: bool
) -> None:
cat_dict = pystac.StacIO.default().read_json(href)
cat = pystac.read_file(href)
assert isinstance(cat, Catalog)
rels = {link["rel"] for link in cat_dict["links"]}
assert ("self" in rels) == should_include_self
for child_link in cat.get_child_links():
child_href = make_absolute_href(child_link.href, 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.href, href)
validate_item_link_type(
item_href, link_type, catalog_type == CatalogType.ABSOLUTE_PUBLISHED
)
link_type = "RELATIVE"
if catalog_type == CatalogType.ABSOLUTE_PUBLISHED:
link_type = "ABSOLUTE"
root_should_include_href = catalog_type in [
CatalogType.ABSOLUTE_PUBLISHED,
CatalogType.RELATIVE_PUBLISHED,
]
validate_catalog_link_type(root_href, link_type, root_should_include_href)
def do_test(self, catalog: Catalog, catalog_type: CatalogType) -> None:
with tempfile.TemporaryDirectory() as tmp_dir:
catalog.normalize_hrefs(tmp_dir)
self.validate_catalog(catalog)
catalog.save(catalog_type=catalog_type)
root_href = catalog.self_href
self.validate_link_types(root_href, catalog_type)
for parent, _, items in catalog.walk():
if issubclass(type(parent), Collection):
stac_object_type = pystac.STACObjectType.COLLECTION
else:
stac_object_type = pystac.STACObjectType.CATALOG
self.validate_file(parent.self_href, stac_object_type)
for item in items:
self.validate_file(item.self_href, pystac.STACObjectType.ITEM)
@pytest.mark.vcr()
@pytest.mark.parametrize("catalog", TestCases.all_test_catalogs())
@pytest.mark.parametrize("catalog_type", CTYPES)
def test_testcases(self, catalog: Catalog, catalog_type: CatalogType) -> None:
catalog = catalog.full_copy()
self.do_test(catalog, catalog_type)
|