File: test_writing.py

package info (click to toggle)
pystac 1.13.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 19,904 kB
  • sloc: python: 24,370; makefile: 124; sh: 7
file content (121 lines) | stat: -rw-r--r-- 4,504 bytes parent folder | download
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)