Skip to content

Commit

Permalink
Improve test data collections (#401)
Browse files Browse the repository at this point in the history
* Update tests

split test cases that test fultiple functions or faild cases

* Add is byoc, batch test and rename

* Update get_atribute test

* update is_chack tests

* remove unneeded docstring

* Update test_collection_string and add cases

* Correction test_str -> test_repr

* Combine test_define_fails and test_define

* Update tests

typos, remove number of data contains in test
update test_collection_definition_equality add cases

* Update tests

test_collection_equality split into two
added comments for reasons why test_define should fail
added test for change of number of collections (get_available_collections)

* Update test derive

* spelling correction
  • Loading branch information
jgersak committed Feb 16, 2023
1 parent d7d8025 commit 8456baa
Show file tree
Hide file tree
Showing 2 changed files with 124 additions and 53 deletions.
1 change: 0 additions & 1 deletion sentinelhub/data_collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -649,7 +649,6 @@ def contains_orbit_direction(self, orbit_direction: str) -> bool:
:param orbit_direction: An orbit direction
:return: `True` if data collection contains the orbit direction
:return: bool
"""
defined_direction = self.orbit_direction
if defined_direction is None or defined_direction.upper() == OrbitDirection.BOTH:
Expand Down
176 changes: 124 additions & 52 deletions tests/test_data_collections.py
Original file line number Diff line number Diff line change
@@ -1,85 +1,140 @@
"""
Unit tests for data_collections module
"""
from typing import Any
from typing import Any, Dict

import pytest

from sentinelhub import DataCollection
from sentinelhub.data_collections import DataCollectionDefinition


def test_repr() -> None:
definition = DataCollection.SENTINEL1_IW.value
representation = repr(definition)

assert isinstance(representation, str)
assert representation.count("\n") >= 5


def test_derive() -> None:
definition = DataCollectionDefinition(api_id="X", wfs_id="Y")
derived_definition = definition.derive(wfs_id="Z")

assert derived_definition.api_id == "X"
assert derived_definition.wfs_id == "Z"
assert derived_definition.collection_type is None

@pytest.mark.parametrize(
"data_colection_def, derive_attributes, expected_attributes",
[
(DataCollectionDefinition(), {}, {"api_id": None}),
(
DataCollectionDefinition(api_id="X", wfs_id="Y"),
{"wfs_id": "Z"},
{"api_id": "X", "wfs_id": "Z", "collection_type": None},
),
(DataCollection.LANDSAT_MSS_L1.value, {"api_id": None}, {"api_id": None, "wfs_id": "DSS14"}),
],
)
def test_derive(
data_colection_def: DataCollectionDefinition, derive_attributes: Dict[str, Any], expected_attributes: Dict[str, Any]
) -> None:
derived_definition = data_colection_def.derive(**derive_attributes)

for attribute, value in expected_attributes.items():
assert value == getattr(derived_definition, attribute)


@pytest.mark.parametrize(
"definition_input, expected",
[
({}, "DataCollectionDefinition(\n is_timeless: False\n has_cloud_coverage: False\n)"),
(
{"api_id": "X", "_name": "A"},
"DataCollectionDefinition(\n api_id: X\n is_timeless: False\n has_cloud_coverage: False\n)",
),
(
{"api_id": "Y", "is_timeless": True, "has_cloud_coverage": True},
"DataCollectionDefinition(\n api_id: Y\n is_timeless: True\n has_cloud_coverage: True\n)",
),
],
)
def test_collection_repr(definition_input: Dict[str, Any], expected: str) -> None:
assert repr(DataCollectionDefinition(**definition_input)) == expected


@pytest.mark.parametrize(
"test_definition, equal_definition",
[
({"api_id": "X", "_name": "A"}, {"api_id": "X", "_name": "A"}),
({"api_id": "X", "_name": "A"}, {"api_id": "X", "_name": "B"}),
({"api_id": "X", "is_timeless": False}, {"api_id": "X", "is_timeless": False, "_name": "B"}),
({"api_id": "X", "is_timeless": False}, {"api_id": "X"}),
],
)
def test_collection_definitions_equal(test_definition: Dict[str, Any], equal_definition: Dict[str, Any]) -> None:
def1 = DataCollectionDefinition(**test_definition)
def2 = DataCollectionDefinition(**equal_definition)
assert def1 == def2

def test_compare() -> None:
def1 = DataCollectionDefinition(api_id="X", _name="A")
def2 = DataCollectionDefinition(api_id="X", _name="B")

assert def1 == def2
@pytest.mark.parametrize(
"test_definition, equal_definition",
[
({"api_id": "X", "_name": "A"}, {"api_id": "Y", "_name": "A"}),
({"api_id": "X", "is_timeless": True}, {"api_id": "X"}),
({"api_id": "X", "wfs_id": 2132342143454364}, {"api_id": "X"}),
],
)
def test_collection_definitions_not_equal(test_definition: Dict[str, Any], equal_definition: Dict[str, Any]) -> None:
def1 = DataCollectionDefinition(**test_definition)
def2 = DataCollectionDefinition(**equal_definition)
assert def1 != def2


def test_define() -> None:
for _ in range(3):
data_collection = DataCollection.define(
"NEW", api_id="X", sensor_type="Sensor", bands=("B01",), is_timeless=True
)
data_collection = DataCollection.define("NEW", api_id="X", sensor_type="Sensor", bands=("B01",), is_timeless=True)

assert data_collection == DataCollection.NEW
assert DataCollection.NEW.api_id == "X"

# Should fail because DataCollection with same api_id already exists.
with pytest.raises(ValueError):
DataCollection.define("NEW_NEW", api_id="X", sensor_type="Sensor", bands=("B01",), is_timeless=True)

# Should fail because DataCollection with same name already exists.
with pytest.raises(ValueError):
DataCollection.define("NEW", api_id="Y")


def test_define_from() -> None:
bands = ["B01", "XYZ"]
for _ in range(3):
data_collection = DataCollection.define_from(DataCollection.SENTINEL5P, "NEW_5P", api_id="X", bands=bands)
data_collection = DataCollection.define_from(DataCollection.SENTINEL5P, "NEW_5P", api_id="X", bands=bands)

assert data_collection == DataCollection.NEW_5P
assert data_collection.api_id == "X"
assert data_collection.wfs_id == DataCollection.SENTINEL5P.wfs_id
assert data_collection.bands == tuple(bands)


def test_define_byoc_and_batch() -> None:
def test_define_byoc() -> None:
byoc_id = "0000d273-7e89-4f00-971e-9024f89a0000"
byoc = DataCollection.define_byoc(byoc_id, name="MY_BYOC")
batch = DataCollection.define_batch(byoc_id, name="MY_BATCH")

assert byoc == DataCollection.MY_BYOC
assert byoc.api_id.endswith(byoc_id)
assert byoc.collection_id == byoc_id

assert DataCollection.MY_BYOC.is_byoc
assert not DataCollection.SENTINEL5P.is_byoc


def test_define_batch() -> None:
batch_id = "0000d273-7e89-4f00-971e-9024f89a0000"
batch = DataCollection.define_batch(batch_id, name="MY_BATCH")

assert batch == DataCollection.MY_BATCH
assert batch.api_id.endswith(batch_id)
assert batch.collection_id == batch_id

for data_collection in [byoc, batch]:
assert data_collection.api_id.endswith(byoc_id)
assert data_collection.collection_id == byoc_id
assert DataCollection.MY_BATCH.is_batch
assert not DataCollection.SENTINEL2_L2A.is_batch


def test_attributes() -> None:
data_collection = DataCollection.SENTINEL3_OLCI
@pytest.mark.parametrize("data_collection", [DataCollection.SENTINEL3_OLCI, DataCollection.SENTINEL2_L2A])
@pytest.mark.parametrize("attribute", ["api_id", "catalog_id", "wfs_id", "service_url", "bands", "sensor_type"])
def test_attributes(data_collection: DataCollection, attribute: str) -> None:
value = getattr(data_collection, attribute)
assert value is not None
assert value == getattr(data_collection.value, attribute)

for attr_name in ["api_id", "catalog_id", "wfs_id", "service_url", "bands", "sensor_type"]:
value = getattr(data_collection, attr_name)
assert value is not None
assert value == getattr(data_collection.value, attr_name)

def test_attributes_empty_fail() -> None:
data_collection = DataCollection.define("EMPTY")

for attr_name in ["api_id", "catalog_id", "wfs_id", "bands"]:
Expand All @@ -89,25 +144,42 @@ def test_attributes() -> None:
assert data_collection.service_url is None


def test_sentinel1_checks() -> None:
assert DataCollection.SENTINEL1_IW.is_sentinel1
assert not DataCollection.SENTINEL2_L1C.is_sentinel1

assert DataCollection.SENTINEL1_IW_ASC.contains_orbit_direction("ascending")
assert not DataCollection.SENTINEL1_IW_DES.contains_orbit_direction("ascending")

assert DataCollection.SENTINEL2_L2A.contains_orbit_direction("descending")
@pytest.mark.parametrize(
"test_collection, expected",
[
(DataCollection.SENTINEL2_L1C, False),
(DataCollection.SENTINEL1_EW, True),
(DataCollection.LANDSAT_TM_L1, False),
],
)
def test_is_sentinel1(test_collection: DataCollection, expected: bool) -> None:
assert test_collection.is_sentinel1 == expected


@pytest.mark.parametrize(
"collection, direction, expected",
[
("SENTINEL1_IW_ASC", "ascending", True),
("SENTINEL1_IW_ASC", "descending", False),
("SENTINEL1_IW_DES", "ascending", False),
("SENTINEL2_L2A", "descending", True),
("SENTINEL2_L2A", "ascending", True),
],
)
def test_contains_orbit_direction(collection: str, direction: str, expected: bool) -> None:
data_collection = getattr(DataCollection, collection)
assert data_collection.contains_orbit_direction(direction) == expected


def test_get_available_collections() -> None:
number_of_collection = len(DataCollection.get_available_collections())
DataCollection.define("NEW_NEW", api_id="Z")
DataCollection.define_batch("batch_id", name="MY_NEW_BATCH")
DataCollection.define_byoc("byoc_id", name="MY_NEW_BYOC")
collections = DataCollection.get_available_collections()
assert helper_check_collection_list(collections)


def helper_check_collection_list(collection_list: Any) -> bool:
is_list = isinstance(collection_list, list)
contains_collections = all(isinstance(data_collection, DataCollection) for data_collection in collection_list)
return is_list and contains_collections
assert len(collections) == number_of_collection + 3
assert all(isinstance(collection, DataCollection) for collection in collections)


def test_transfer_with_ray(ray: Any) -> None:
Expand Down

0 comments on commit 8456baa

Please sign in to comment.