diff --git a/tests/test_stac_io.py b/tests/test_stac_io.py index 97b6c1fc..2488f148 100644 --- a/tests/test_stac_io.py +++ b/tests/test_stac_io.py @@ -11,121 +11,125 @@ from tests.utils import TestCases -class StacIOTest(unittest.TestCase): - def setUp(self) -> None: - self.stac_io = StacIO.default() - - def test_read_write_collection(self) -> None: - collection = pystac.read_file( - TestCases.get_path("data-files/collections/multi-extent.json") - ) - with tempfile.TemporaryDirectory() as tmp_dir: - dest_href = os.path.join(tmp_dir, "collection.json") - pystac.write_file(collection, dest_href=dest_href) - self.assertTrue(os.path.exists(dest_href), msg="File was not written.") - - def test_read_write_collection_with_file_protocol(self) -> None: - collection = pystac.read_file( - "file://" + TestCases.get_path("data-files/collections/multi-extent.json") - ) - with tempfile.TemporaryDirectory() as tmp_dir: - dest_href = os.path.join(tmp_dir, "collection.json") - pystac.write_file(collection, dest_href="file://" + dest_href) - self.assertTrue(os.path.exists(dest_href), msg="File was not written.") - - def test_read_item(self) -> None: - item = pystac.read_file(TestCases.get_path("data-files/item/sample-item.json")) - with tempfile.TemporaryDirectory() as tmp_dir: - dest_href = os.path.join(tmp_dir, "item.json") - pystac.write_file(item, dest_href=dest_href) - self.assertTrue(os.path.exists(dest_href), msg="File was not written.") - - def test_read_write_catalog(self) -> None: - catalog = pystac.read_file( - TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") - ) - with tempfile.TemporaryDirectory() as tmp_dir: - dest_href = os.path.join(tmp_dir, "catalog.json") - pystac.write_file(catalog, dest_href=dest_href) - self.assertTrue(os.path.exists(dest_href), msg="File was not written.") - - def test_read_item_collection_raises_exception(self) -> None: - with self.assertRaises(pystac.STACTypeError): - _ = pystac.read_file( - TestCases.get_path( - "data-files/item-collection/sample-item-collection.json" - ) - ) - - def test_read_item_dict(self) -> None: - item_dict = self.stac_io.read_json( - TestCases.get_path("data-files/item/sample-item.json") +def test_read_write_collection() -> None: + collection = pystac.read_file( + TestCases.get_path("data-files/collections/multi-extent.json") + ) + with tempfile.TemporaryDirectory() as tmp_dir: + dest_href = os.path.join(tmp_dir, "collection.json") + pystac.write_file(collection, dest_href=dest_href) + assert os.path.exists(dest_href), "File was not written." + + +def test_read_write_collection_with_file_protocol() -> None: + collection = pystac.read_file( + "file://" + TestCases.get_path("data-files/collections/multi-extent.json") + ) + with tempfile.TemporaryDirectory() as tmp_dir: + dest_href = os.path.join(tmp_dir, "collection.json") + pystac.write_file(collection, dest_href="file://" + dest_href) + assert os.path.exists(dest_href), "File was not written." + + +def test_read_item() -> None: + item = pystac.read_file(TestCases.get_path("data-files/item/sample-item.json")) + with tempfile.TemporaryDirectory() as tmp_dir: + dest_href = os.path.join(tmp_dir, "item.json") + pystac.write_file(item, dest_href=dest_href) + assert os.path.exists(dest_href), "File was not written." + + +def test_read_write_catalog() -> None: + catalog = pystac.read_file( + TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") + ) + with tempfile.TemporaryDirectory() as tmp_dir: + dest_href = os.path.join(tmp_dir, "catalog.json") + pystac.write_file(catalog, dest_href=dest_href) + assert os.path.exists(dest_href), "File was not written." + + +def test_read_item_collection_raises_exception() -> None: + with pytest.raises(pystac.STACTypeError): + _ = pystac.read_file( + TestCases.get_path("data-files/item-collection/sample-item-collection.json") ) - item = pystac.read_dict(item_dict) - self.assertIsInstance(item, pystac.Item) - def test_read_collection_dict(self) -> None: - collection_dict = self.stac_io.read_json( - TestCases.get_path("data-files/collections/multi-extent.json") - ) - collection = pystac.read_dict(collection_dict) - self.assertIsInstance(collection, pystac.Collection) - - def test_read_catalog_dict(self) -> None: - catalog_dict = self.stac_io.read_json( - TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") - ) - catalog = pystac.read_dict(catalog_dict) - self.assertIsInstance(catalog, pystac.Catalog) - def test_read_from_stac_object(self) -> None: - catalog = pystac.STACObject.from_file( - TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") - ) - self.assertIsInstance(catalog, pystac.Catalog) - - def test_report_duplicate_keys(self) -> None: - # Directly from dict - class ReportingStacIO(DefaultStacIO, DuplicateKeyReportingMixin): - pass - - stac_io = ReportingStacIO() - test_json = """{ - "key": "value_1", - "key": "value_2" - }""" - - with self.assertRaises(pystac.DuplicateObjectKeyError) as excinfo: - stac_io.json_loads(test_json) - self.assertEqual(str(excinfo.exception), 'Found duplicate object name "key"') - - # From file - with tempfile.TemporaryDirectory() as tmp_dir: - src_href = os.path.join(tmp_dir, "test.json") - with open(src_href, "w") as dst: - dst.write(test_json) - - with self.assertRaises(pystac.DuplicateObjectKeyError) as excinfo: - stac_io.read_json(src_href) - self.assertEqual( - str(excinfo.exception), - f'Found duplicate object name "key" in {src_href}', - ) - - @unittest.mock.patch("pystac.stac_io.urlopen") - def test_headers_stac_io(self, urlopen_mock: unittest.mock.MagicMock) -> None: - stac_io = DefaultStacIO(headers={"Authorization": "api-key fake-api-key-value"}) - - catalog = pystac.Catalog("an-id", "a description").to_dict() - # required until https://github.com/stac-utils/pystac/pull/896 is merged - catalog["links"] = [] - urlopen_mock.return_value.__enter__.return_value.read.return_value = json.dumps( - catalog - ).encode("utf-8") - pystac.Catalog.from_file("https://example.com/catalog.json", stac_io=stac_io) - - request_obj = urlopen_mock.call_args[0][0] - self.assertEqual(request_obj.headers, stac_io.headers) +def test_read_item_dict() -> None: + stac_io = StacIO.default() + item_dict = stac_io.read_json( + TestCases.get_path("data-files/item/sample-item.json") + ) + item = pystac.read_dict(item_dict) + assert isinstance(item, pystac.Item) + + +def test_read_collection_dict() -> None: + stac_io = StacIO.default() + collection_dict = stac_io.read_json( + TestCases.get_path("data-files/collections/multi-extent.json") + ) + collection = pystac.read_dict(collection_dict) + assert isinstance(collection, pystac.Collection) + + +def test_read_catalog_dict() -> None: + stac_io = StacIO.default() + catalog_dict = stac_io.read_json( + TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") + ) + catalog = pystac.read_dict(catalog_dict) + assert isinstance(catalog, pystac.Catalog) + + +def test_read_from_stac_object() -> None: + catalog = pystac.STACObject.from_file( + TestCases.get_path("data-files/catalogs/test-case-1/catalog.json") + ) + assert isinstance(catalog, pystac.Catalog) + + +def test_report_duplicate_keys() -> None: + # Directly from dict + class ReportingStacIO(DefaultStacIO, DuplicateKeyReportingMixin): + pass + + stac_io = ReportingStacIO() + test_json = """{ + "key": "value_1", + "key": "value_2" + }""" + + with pytest.raises(pystac.DuplicateObjectKeyError) as excinfo: + stac_io.json_loads(test_json) + assert str(excinfo.value) == 'Found duplicate object name "key"' + + # From file + with tempfile.TemporaryDirectory() as tmp_dir: + src_href = os.path.join(tmp_dir, "test.json") + with open(src_href, "w") as dst: + dst.write(test_json) + + with pytest.raises(pystac.DuplicateObjectKeyError) as excinfo: + stac_io.read_json(src_href) + assert str(excinfo.value), f'Found duplicate object name "key" in {src_href}' + + +@unittest.mock.patch("pystac.stac_io.urlopen") +def test_headers_stac_io(urlopen_mock: unittest.mock.MagicMock) -> None: + stac_io = DefaultStacIO(headers={"Authorization": "api-key fake-api-key-value"}) + + catalog = pystac.Catalog("an-id", "a description").to_dict() + # required until https://github.com/stac-utils/pystac/pull/896 is merged + catalog["links"] = [] + urlopen_mock.return_value.__enter__.return_value.read.return_value = json.dumps( + catalog + ).encode("utf-8") + pystac.Catalog.from_file("https://example.com/catalog.json", stac_io=stac_io) + + request_obj = urlopen_mock.call_args[0][0] + assert request_obj.headers == stac_io.headers @pytest.mark.vcr() diff --git a/tests/test_summaries.py b/tests/test_summaries.py index a37aabd2..aa5f35e1 100644 --- a/tests/test_summaries.py +++ b/tests/test_summaries.py @@ -1,101 +1,106 @@ import socket -import unittest from typing import Any +import pytest + from pystac.summaries import RangeSummary, Summaries, Summarizer, SummaryStrategy from tests.utils import TestCases -class SummariesTest(unittest.TestCase): - def test_summary(self) -> None: - coll = TestCases.case_5() - summaries = Summarizer().summarize(coll.get_items(recursive=True)) - summaries_dict = summaries.to_dict() - self.assertEqual(len(summaries_dict["eo:bands"]), 4) - self.assertEqual(len(summaries_dict["proj:code"]), 1) - - def test_summary_limit(self) -> None: - coll = TestCases.case_5() - summaries = Summarizer().summarize(coll.get_items(recursive=True)) - summaries.maxcount = 2 - summaries_dict = summaries.to_dict() - self.assertIsNone(summaries_dict.get("eo:bands")) - self.assertEqual(len(summaries_dict["proj:code"]), 1) - - def test_summary_custom_fields_file(self) -> None: - coll = TestCases.case_5() - path = TestCases.get_path("data-files/summaries/fields_no_bands.json") - summaries = Summarizer(path).summarize(coll.get_items(recursive=True)) - summaries_dict = summaries.to_dict() - self.assertIsNone(summaries_dict.get("eo:bands")) - self.assertEqual(len(summaries_dict["proj:code"]), 1) - - def test_summary_custom_fields_dict(self) -> None: - coll = TestCases.case_5() - spec = { - "eo:bands": SummaryStrategy.DONT_SUMMARIZE, - "proj:code": SummaryStrategy.ARRAY, - } - obj = Summarizer(spec) - self.assertTrue("eo:bands" not in obj.summaryfields) - self.assertEqual(obj.summaryfields["proj:code"], SummaryStrategy.ARRAY) - summaries = obj.summarize(coll.get_items(recursive=True)) - summaries_dict = summaries.to_dict() - self.assertIsNone(summaries_dict.get("eo:bands")) - self.assertEqual(len(summaries_dict["proj:code"]), 1) - - def test_summary_wrong_custom_fields_file(self) -> None: - coll = TestCases.case_5() - with self.assertRaises(FileNotFoundError) as context: - Summarizer("wrong/path").summarize(coll.get_items(recursive=True)) - self.assertTrue("No such file or directory" in str(context.exception)) - - def test_can_open_fields_file_even_with_no_nework(self) -> None: - old_socket = socket.socket - try: - - class no_network(socket.socket): - def __init__(self, *args: Any, **kwargs: Any): - raise Exception("Network call blocked") - - socket.socket = no_network # type:ignore - Summarizer() - finally: - # even if this test fails, it should not break the whole test suite - socket.socket = old_socket # type:ignore - - def test_summary_empty(self) -> None: - summaries = Summaries.empty() - self.assertTrue(summaries.is_empty()) - - def test_summary_not_empty(self) -> None: - coll = TestCases.case_5() - summaries = Summarizer().summarize(coll.get_items(recursive=True)) - self.assertFalse(summaries.is_empty()) - - def test_clone_summary(self) -> None: - coll = TestCases.case_5() - summaries = Summarizer().summarize(coll.get_items(recursive=True)) - summaries_dict = summaries.to_dict() - clone = summaries.clone() - self.assertTrue(isinstance(clone, Summaries)) - clone_dict = clone.to_dict() - self.assertDictEqual(clone_dict, summaries_dict) - - -class RangeSummaryTest(unittest.TestCase): - def setUp(self) -> None: - self.maxDiff = None - - def test_repr(self) -> None: - rs = RangeSummary(5, 10) - self.assertEqual("{'minimum': 5, 'maximum': 10}", rs.__repr__()) - - def test_equality(self) -> None: - rs_1 = RangeSummary(5, 10) - rs_2 = RangeSummary(5, 10) - rs_3 = RangeSummary(5, 11) - - self.assertEqual(rs_1, rs_2) - self.assertNotEqual(rs_1, rs_3) - self.assertNotEqual(rs_1, (5, 10)) +def test_summary() -> None: + coll = TestCases.case_5() + summaries = Summarizer().summarize(coll.get_items(recursive=True)) + summaries_dict = summaries.to_dict() + assert len(summaries_dict["eo:bands"]) == 4 + assert len(summaries_dict["proj:code"]) == 1 + + +def test_summary_limit() -> None: + coll = TestCases.case_5() + summaries = Summarizer().summarize(coll.get_items(recursive=True)) + summaries.maxcount = 2 + summaries_dict = summaries.to_dict() + assert summaries_dict.get("eo:bands") is None + assert len(summaries_dict["proj:code"]) == 1 + + +def test_summary_custom_fields_file() -> None: + coll = TestCases.case_5() + path = TestCases.get_path("data-files/summaries/fields_no_bands.json") + summaries = Summarizer(path).summarize(coll.get_items(recursive=True)) + summaries_dict = summaries.to_dict() + assert summaries_dict.get("eo:bands") is None + assert len(summaries_dict["proj:code"]) == 1 + + +def test_summary_custom_fields_dict() -> None: + coll = TestCases.case_5() + spec = { + "eo:bands": SummaryStrategy.DONT_SUMMARIZE, + "proj:code": SummaryStrategy.ARRAY, + } + obj = Summarizer(spec) + assert "eo:bands" not in obj.summaryfields + assert obj.summaryfields["proj:code"] == SummaryStrategy.ARRAY + summaries = obj.summarize(coll.get_items(recursive=True)) + summaries_dict = summaries.to_dict() + assert summaries_dict.get("eo:bands") is None + assert len(summaries_dict["proj:code"]) == 1 + + +def test_summary_wrong_custom_fields_file() -> None: + coll = TestCases.case_5() + with pytest.raises(FileNotFoundError) as context: + Summarizer("wrong/path").summarize(coll.get_items(recursive=True)) + assert "No such file or directory" in str(context.value) + + +def test_can_open_fields_file_even_with_no_nework() -> None: + old_socket = socket.socket + try: + + class no_network(socket.socket): + def __init__(self, *args: Any, **kwargs: Any): + raise Exception("Network call blocked") + + socket.socket = no_network # type:ignore + Summarizer() + finally: + # even if this test fails, it should not break the whole test suite + socket.socket = old_socket # type:ignore + + +def test_summary_empty() -> None: + summaries = Summaries.empty() + assert summaries.is_empty() + + +def test_summary_not_empty() -> None: + coll = TestCases.case_5() + summaries = Summarizer().summarize(coll.get_items(recursive=True)) + assert not summaries.is_empty() + + +def test_clone_summary() -> None: + coll = TestCases.case_5() + summaries = Summarizer().summarize(coll.get_items(recursive=True)) + summaries_dict = summaries.to_dict() + clone = summaries.clone() + assert isinstance(clone, Summaries) + clone_dict = clone.to_dict() + assert clone_dict == summaries_dict + + +def test_RangeSummary_repr() -> None: + rs = RangeSummary(5, 10) + assert "{'minimum': 5, 'maximum': 10}" == rs.__repr__() + + +def test_RangeSummary_equality() -> None: + rs_1 = RangeSummary(5, 10) + rs_2 = RangeSummary(5, 10) + rs_3 = RangeSummary(5, 11) + + assert rs_1 == rs_2 + assert rs_1 != rs_3 + assert rs_1 != (5, 10) diff --git a/tests/test_utils.py b/tests/test_utils.py index 8f2a9f6a..8b53daaa 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,7 +1,6 @@ import json import os import time -import unittest from datetime import datetime, timedelta, timezone import pytest @@ -24,351 +23,318 @@ from tests.utils import TestCases, path_includes_drive_letter -class UtilsTest(unittest.TestCase): - def test_make_relative_href(self) -> None: - # Test cases of (source_href, start_href, expected) - test_cases = [ - ("/a/b/c/d/catalog.json", "/a/b/c/catalog.json", "./d/catalog.json"), - ("/a/b/catalog.json", "/a/b/c/catalog.json", "../catalog.json"), - ("/a/catalog.json", "/a/b/c/catalog.json", "../../catalog.json"), - ("/a/b/c/d/", "/a/b/c/catalog.json", "./d/"), - ("/a/b/c/d/.dotfile", "/a/b/c/d/catalog.json", "./.dotfile"), - ( - "file:///a/b/c/d/catalog.json", - "file:///a/b/c/catalog.json", - "./d/catalog.json", - ), - ] - - for source_href, start_href, expected in test_cases: - actual = make_relative_href(source_href, start_href) - self.assertEqual(actual, expected) - - def test_make_relative_href_url(self) -> None: - test_cases = [ - ( - "http://stacspec.org/a/b/c/d/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "./d/catalog.json", - ), - ( - "http://stacspec.org/a/b/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "../catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "../../catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "http://cogeo.org/a/b/c/catalog.json", - "http://stacspec.org/a/catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "https://stacspec.org/a/b/c/catalog.json", - "http://stacspec.org/a/catalog.json", - ), - ( - "http://stacspec.org/a/", - "https://stacspec.org/a/b/c/catalog.json", - "http://stacspec.org/a/", - ), - ( - "http://stacspec.org/a/b/.dotfile", - "http://stacspec.org/a/b/catalog.json", - "./.dotfile", - ), - ] - - for source_href, start_href, expected in test_cases: - actual = make_relative_href(source_href, start_href) - self.assertEqual(actual, expected) - - def test_make_relative_href_windows(self) -> None: - # Test cases of (source_href, start_href, expected) - test_cases = [ - ( - "C:\\a\\b\\c\\d\\catalog.json", - "C:\\a\\b\\c\\catalog.json", - "./d/catalog.json", - ), - ( - "C:\\a\\b\\catalog.json", - "C:\\a\\b\\c\\catalog.json", - "../catalog.json", - ), - ( - "C:\\a\\catalog.json", - "C:\\a\\b\\c\\catalog.json", - "../../catalog.json", - ), - ("a\\b\\c\\catalog.json", "a\\b\\catalog.json", "./c/catalog.json"), - ("a\\b\\catalog.json", "a\\b\\c\\catalog.json", "../catalog.json"), - ( - "http://stacspec.org/a/b/c/d/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "./d/catalog.json", - ), - ( - "http://stacspec.org/a/b/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "../catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "http://stacspec.org/a/b/c/catalog.json", - "../../catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "http://cogeo.org/a/b/c/catalog.json", - "http://stacspec.org/a/catalog.json", - ), - ( - "http://stacspec.org/a/catalog.json", - "https://stacspec.org/a/b/c/catalog.json", - "http://stacspec.org/a/catalog.json", - ), - ] - - for source_href, start_href, expected in test_cases: - actual = make_relative_href(source_href, start_href) - self.assertEqual(actual, expected) - - def test_make_absolute_href(self) -> None: - # Test cases of (source_href, start_href, expected) - test_cases = [ - ("item.json", "/a/b/c/catalog.json", "/a/b/c/item.json"), - ("./item.json", "/a/b/c/catalog.json", "/a/b/c/item.json"), - ("./z/item.json", "/a/b/c/catalog.json", "/a/b/c/z/item.json"), - ("../item.json", "/a/b/c/catalog.json", "/a/b/item.json"), - ( - "item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/item.json", - ), - ( - "./item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/item.json", - ), - ( - "./z/item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/z/item.json", - ), - ( - "../item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/item.json", - ), - ("http://localhost:8000", None, "http://localhost:8000"), - ("item.json", "file:///a/b/c/catalog.json", "file:///a/b/c/item.json"), - ( - "./z/item.json", - "file:///a/b/c/catalog.json", - "file:///a/b/c/z/item.json", - ), - ("file:///a/b/c/item.json", None, "file:///a/b/c/item.json"), - ] - - for source_href, start_href, expected in test_cases: - actual = make_absolute_href(source_href, start_href) - if expected.startswith("file://"): - _, actual = os.path.splitdrive(actual.replace("file://", "")) - actual = f"file://{actual}" - else: - _, actual = os.path.splitdrive(actual) - self.assertEqual(actual, expected) - - def test_make_absolute_href_on_vsitar(self) -> None: - rel_path = "some/item.json" - cat_path = "/vsitar//tmp/catalog.tar/catalog.json" - expected = "/vsitar//tmp/catalog.tar/some/item.json" - - self.assertEqual(expected, make_absolute_href(rel_path, cat_path)) - - @pytest.mark.skipif(os.name != "nt", reason="Windows only test") - def test_make_absolute_href_windows(self) -> None: - # Test cases of (source_href, start_href, expected) - test_cases = [ - ("item.json", "C:\\a\\b\\c\\catalog.json", "C:/a/b/c/item.json"), - (".\\item.json", "C:\\a\\b\\c\\catalog.json", "C:/a/b/c/item.json"), - ( - ".\\z\\item.json", - "Z:\\a\\b\\c\\catalog.json", - "Z:/a/b/c/z/item.json", - ), - ("..\\item.json", "a:\\a\\b\\c\\catalog.json", "a:/a/b/item.json"), - ( - "item.json", - "HTTPS://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/item.json", - ), - ( - "./item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/item.json", - ), - ( - "./z/item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/c/z/item.json", - ), - ( - "../item.json", - "https://stacspec.org/a/b/c/catalog.json", - "https://stacspec.org/a/b/item.json", - ), - ] - - for source_href, start_href, expected in test_cases: - actual = make_absolute_href(source_href, start_href) - self.assertEqual(actual, expected) - - def test_is_absolute_href(self) -> None: - # Test cases of (href, expected) - test_cases = [ - ("item.json", False), - ("./item.json", False), - ("../item.json", False), - ("http://stacspec.org/item.json", True), - ] - - for href, expected in test_cases: - actual = is_absolute_href(href) - self.assertEqual(actual, expected) - - def test_is_absolute_href_os_aware(self) -> None: - # Test cases of (href, expected) - - is_windows = os.name == "nt" - incl_drive_letter = path_includes_drive_letter() - test_cases = [ - ("/item.json", not incl_drive_letter), - ("/home/someuser/Downloads/item.json", not incl_drive_letter), - ("file:///home/someuser/Downloads/item.json", not incl_drive_letter), - ("d:/item.json", is_windows), - ("c:/files/more_files/item.json", is_windows), - ] - - for href, expected in test_cases: - actual = is_absolute_href(href) - self.assertEqual(actual, expected) - - @pytest.mark.skipif(os.name != "nt", reason="Windows only test") - def test_is_absolute_href_windows(self) -> None: - # Test cases of (href, expected) - - test_cases = [ - ("item.json", False), - (".\\item.json", False), - ("..\\item.json", False), - ("c:\\item.json", True), - ("http://stacspec.org/item.json", True), - ] - - for href, expected in test_cases: - actual = is_absolute_href(href) - self.assertEqual(actual, expected) - - def test_datetime_to_str(self) -> None: - cases = ( - ( - "timezone naive, assume utc", - datetime(2000, 1, 1), - "2000-01-01T00:00:00Z", - ), - ( - "timezone aware, utc", - datetime(2000, 1, 1, tzinfo=timezone.utc), - "2000-01-01T00:00:00Z", - ), - ( - "timezone aware, utc -7", - datetime(2000, 1, 1, tzinfo=timezone(timedelta(hours=-7))), - "2000-01-01T00:00:00-07:00", - ), - ) - - for title, dt, expected in cases: - with self.subTest(title=title): - got = utils.datetime_to_str(dt) - self.assertEqual(expected, got) - - def test_datetime_to_str_with_microseconds_timespec(self) -> None: - cases = ( - ( - "timezone naive, assume utc", - datetime(2000, 1, 1, 0, 0, 0, 0), - "2000-01-01T00:00:00.000000Z", - ), - ( - "timezone aware, utc", - datetime(2000, 1, 1, 0, 0, 0, 0, tzinfo=timezone.utc), - "2000-01-01T00:00:00.000000Z", - ), - ( - "timezone aware, utc -7", - datetime(2000, 1, 1, 0, 0, 0, 0, tzinfo=timezone(timedelta(hours=-7))), - "2000-01-01T00:00:00.000000-07:00", - ), - ) - - for title, dt, expected in cases: - with self.subTest(title=title): - got = utils.datetime_to_str(dt, timespec="microseconds") - self.assertEqual(expected, got) - - def test_str_to_datetime(self) -> None: - def _set_tzinfo(tz_str: str | None) -> None: - if tz_str is None: - if "TZ" in os.environ: - del os.environ["TZ"] - else: - os.environ["TZ"] = tz_str - # time.tzset() only available for Unix/Linux - if hasattr(time, "tzset"): - time.tzset() - - utc_timestamp = "2015-06-27T10:25:31Z" - - prev_tz = os.environ.get("TZ") - - with self.subTest(tz=None): - _set_tzinfo(None) - utc_datetime = str_to_datetime(utc_timestamp) - self.assertIs(utc_datetime.tzinfo, tz.tzutc()) - self.assertIsNot(utc_datetime.tzinfo, tz.tzlocal()) - - with self.subTest(tz="UTC"): - _set_tzinfo("UTC") - utc_datetime = str_to_datetime(utc_timestamp) - self.assertIs(utc_datetime.tzinfo, tz.tzutc()) - self.assertIsNot(utc_datetime.tzinfo, tz.tzlocal()) - - with self.subTest(tz="US/Central"): - _set_tzinfo("US/Central") - utc_datetime = str_to_datetime(utc_timestamp) - self.assertIs(utc_datetime.tzinfo, tz.tzutc()) - self.assertIsNot(utc_datetime.tzinfo, tz.tzlocal()) - - if prev_tz is not None: - _set_tzinfo(prev_tz) - - def test_geojson_bbox(self) -> None: - # Use sample Geojson from https://en.wikipedia.org/wiki/GeoJSON - with open( - TestCases.get_path("data-files/geojson/sample.geojson") - ) as sample_geojson: - all_features = json.load(sample_geojson) - geom_dicts = [f["geometry"] for f in all_features["features"]] - for geom in geom_dicts: - got = utils.geometry_to_bbox(geom) - self.assertNotEqual(got, None) +@pytest.mark.parametrize( + "source_href, start_href, expected", + ( + # relative href + ("/a/b/c/d/catalog.json", "/a/b/c/catalog.json", "./d/catalog.json"), + ("/a/b/catalog.json", "/a/b/c/catalog.json", "../catalog.json"), + ("/a/catalog.json", "/a/b/c/catalog.json", "../../catalog.json"), + ("/a/b/c/d/", "/a/b/c/catalog.json", "./d/"), + ("/a/b/c/d/.dotfile", "/a/b/c/d/catalog.json", "./.dotfile"), + ( + "file:///a/b/c/d/catalog.json", + "file:///a/b/c/catalog.json", + "./d/catalog.json", + ), + # relative href url + ( + "http://stacspec.org/a/b/c/d/catalog.json", + "http://stacspec.org/a/b/c/catalog.json", + "./d/catalog.json", + ), + ( + "http://stacspec.org/a/b/catalog.json", + "http://stacspec.org/a/b/c/catalog.json", + "../catalog.json", + ), + ( + "http://stacspec.org/a/catalog.json", + "http://stacspec.org/a/b/c/catalog.json", + "../../catalog.json", + ), + ( + "http://stacspec.org/a/catalog.json", + "http://cogeo.org/a/b/c/catalog.json", + "http://stacspec.org/a/catalog.json", + ), + ( + "http://stacspec.org/a/catalog.json", + "https://stacspec.org/a/b/c/catalog.json", + "http://stacspec.org/a/catalog.json", + ), + ( + "http://stacspec.org/a/", + "https://stacspec.org/a/b/c/catalog.json", + "http://stacspec.org/a/", + ), + ( + "http://stacspec.org/a/b/.dotfile", + "http://stacspec.org/a/b/catalog.json", + "./.dotfile", + ), + # relative href under windows + ( + "C:\\a\\b\\c\\d\\catalog.json", + "C:\\a\\b\\c\\catalog.json", + "./d/catalog.json", + ), + ( + "C:\\a\\b\\catalog.json", + "C:\\a\\b\\c\\catalog.json", + "../catalog.json", + ), + ( + "C:\\a\\catalog.json", + "C:\\a\\b\\c\\catalog.json", + "../../catalog.json", + ), + ("a\\b\\c\\catalog.json", "a\\b\\catalog.json", "./c/catalog.json"), + ("a\\b\\catalog.json", "a\\b\\c\\catalog.json", "../catalog.json"), + ), +) +def test_make_relative_href(source_href: str, start_href: str, expected: str) -> None: + actual = make_relative_href(source_href, start_href) + assert actual == expected + + +@pytest.mark.parametrize( + "source_href, start_href, expected", + ( + ("item.json", "/a/b/c/catalog.json", "/a/b/c/item.json"), + ("./item.json", "/a/b/c/catalog.json", "/a/b/c/item.json"), + ("./z/item.json", "/a/b/c/catalog.json", "/a/b/c/z/item.json"), + ("../item.json", "/a/b/c/catalog.json", "/a/b/item.json"), + ( + "item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/item.json", + ), + ( + "./item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/item.json", + ), + ( + "./z/item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/z/item.json", + ), + ( + "../item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/item.json", + ), + ("http://localhost:8000", None, "http://localhost:8000"), + ("item.json", "file:///a/b/c/catalog.json", "file:///a/b/c/item.json"), + ( + "./z/item.json", + "file:///a/b/c/catalog.json", + "file:///a/b/c/z/item.json", + ), + ("file:///a/b/c/item.json", None, "file:///a/b/c/item.json"), + ), +) +def test_make_absolute_href(source_href: str, start_href: str, expected: str) -> None: + actual = make_absolute_href(source_href, start_href) + if expected.startswith("file://"): + _, actual = os.path.splitdrive(actual.replace("file://", "")) + actual = f"file://{actual}" + else: + _, actual = os.path.splitdrive(actual) + assert actual == expected + + +def test_make_absolute_href_on_vsitar() -> None: + rel_path = "some/item.json" + cat_path = "/vsitar//tmp/catalog.tar/catalog.json" + expected = "/vsitar//tmp/catalog.tar/some/item.json" + + assert expected == make_absolute_href(rel_path, cat_path) + + +@pytest.mark.skipif(os.name != "nt", reason="Windows only test") +@pytest.mark.parametrize( + "source_href, start_href, expected", + ( + ("item.json", "C:\\a\\b\\c\\catalog.json", "C:/a/b/c/item.json"), + (".\\item.json", "C:\\a\\b\\c\\catalog.json", "C:/a/b/c/item.json"), + ( + ".\\z\\item.json", + "Z:\\a\\b\\c\\catalog.json", + "Z:/a/b/c/z/item.json", + ), + ("..\\item.json", "a:\\a\\b\\c\\catalog.json", "a:/a/b/item.json"), + ( + "item.json", + "HTTPS://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/item.json", + ), + ( + "./item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/item.json", + ), + ( + "./z/item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/c/z/item.json", + ), + ( + "../item.json", + "https://stacspec.org/a/b/c/catalog.json", + "https://stacspec.org/a/b/item.json", + ), + ), +) +def test_make_absolute_href_windows( + source_href: str, start_href: str, expected: str +) -> None: + actual = make_absolute_href(source_href, start_href) + assert actual == expected + + +def test_is_absolute_href() -> None: + # Test cases of (href, expected) + test_cases = [ + ("item.json", False), + ("./item.json", False), + ("../item.json", False), + ("http://stacspec.org/item.json", True), + ] + + for href, expected in test_cases: + actual = is_absolute_href(href) + assert actual == expected + + +def test_is_absolute_href_os_aware() -> None: + # Test cases of (href, expected) + + is_windows = os.name == "nt" + incl_drive_letter = path_includes_drive_letter() + test_cases = [ + ("/item.json", not incl_drive_letter), + ("/home/someuser/Downloads/item.json", not incl_drive_letter), + ("file:///home/someuser/Downloads/item.json", not incl_drive_letter), + ("d:/item.json", is_windows), + ("c:/files/more_files/item.json", is_windows), + ] + + for href, expected in test_cases: + actual = is_absolute_href(href) + assert actual == expected + + +@pytest.mark.skipif(os.name != "nt", reason="Windows only test") +def test_is_absolute_href_windows() -> None: + # Test cases of (href, expected) + + test_cases = [ + ("item.json", False), + (".\\item.json", False), + ("..\\item.json", False), + ("c:\\item.json", True), + ("http://stacspec.org/item.json", True), + ] + + for href, expected in test_cases: + actual = is_absolute_href(href) + assert actual == expected + + +def test_datetime_to_str() -> None: + cases = ( + ( + "timezone naive, assume utc", + datetime(2000, 1, 1), + "2000-01-01T00:00:00Z", + ), + ( + "timezone aware, utc", + datetime(2000, 1, 1, tzinfo=timezone.utc), + "2000-01-01T00:00:00Z", + ), + ( + "timezone aware, utc -7", + datetime(2000, 1, 1, tzinfo=timezone(timedelta(hours=-7))), + "2000-01-01T00:00:00-07:00", + ), + ) + + for title, dt, expected in cases: + got = utils.datetime_to_str(dt) + assert expected == got, f"Failure: {title}" + + +def test_datetime_to_str_with_microseconds_timespec() -> None: + cases = ( + ( + "timezone naive, assume utc", + datetime(2000, 1, 1, 0, 0, 0, 0), + "2000-01-01T00:00:00.000000Z", + ), + ( + "timezone aware, utc", + datetime(2000, 1, 1, 0, 0, 0, 0, tzinfo=timezone.utc), + "2000-01-01T00:00:00.000000Z", + ), + ( + "timezone aware, utc -7", + datetime(2000, 1, 1, 0, 0, 0, 0, tzinfo=timezone(timedelta(hours=-7))), + "2000-01-01T00:00:00.000000-07:00", + ), + ) + + for title, dt, expected in cases: + got = utils.datetime_to_str(dt, timespec="microseconds") + assert expected == got, f"Failure: {title}" + + +def test_str_to_datetime() -> None: + def _set_tzinfo(tz_str: str | None) -> None: + if tz_str is None: + if "TZ" in os.environ: + del os.environ["TZ"] + else: + os.environ["TZ"] = tz_str + # time.tzset() only available for Unix/Linux + if hasattr(time, "tzset"): + time.tzset() + + utc_timestamp = "2015-06-27T10:25:31Z" + + prev_tz = os.environ.get("TZ") + + _set_tzinfo(None) + utc_datetime = str_to_datetime(utc_timestamp) + assert utc_datetime.tzinfo is tz.tzutc() + assert utc_datetime.tzinfo is not tz.tzlocal() + + _set_tzinfo("UTC") + utc_datetime = str_to_datetime(utc_timestamp) + assert utc_datetime.tzinfo is tz.tzutc() + assert utc_datetime.tzinfo is not tz.tzlocal() + + _set_tzinfo("US/Central") + utc_datetime = str_to_datetime(utc_timestamp) + assert utc_datetime.tzinfo is tz.tzutc() + assert utc_datetime.tzinfo is not tz.tzlocal() + + if prev_tz is not None: + _set_tzinfo(prev_tz) + + +def test_geojson_bbox() -> None: + # Use sample Geojson from https://en.wikipedia.org/wiki/GeoJSON + with open( + TestCases.get_path("data-files/geojson/sample.geojson") + ) as sample_geojson: + all_features = json.load(sample_geojson) + geom_dicts = [f["geometry"] for f in all_features["features"]] + for geom in geom_dicts: + got = utils.geometry_to_bbox(geom) + assert got is not None @pytest.mark.parametrize(