Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

993 progress: convert 3 more files to pytest #1528

Merged
merged 12 commits into from
Feb 18, 2025
228 changes: 116 additions & 112 deletions tests/test_stac_io.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
Loading