import json from datetime import datetime from math import ceil from typing import Any import pystac import pytest import requests from dateutil.tz import tzutc from pytest_benchmark.fixture import BenchmarkFixture from pystac_client.client import Client from pystac_client.collection_search import ( CollectionSearch, bboxes_overlap, collection_matches, temporal_intervals_overlap, ) from pystac_client.warnings import PystacClientWarning from .helpers import STAC_URLS, read_data_file COLLECTION_SEARCH_URL = f"{STAC_URLS['SPACEBEL']}/collections" COLLECTION_EXAMPLE: dict[str, Any] = { "id": "F-TCC", "q": '"cloudless mosaics"', "datetime": "2021-09-15T00:00:00Z", } class TestCollectionPerformance: @pytest.fixture(scope="function") def single_href(self) -> str: collection_href = COLLECTION_SEARCH_URL + f"/{COLLECTION_EXAMPLE['id']}" return collection_href def test_requests(self, benchmark: BenchmarkFixture, single_href: str) -> None: response = benchmark(requests.get, single_href) assert response.status_code == 200 assert response.json()["id"] == COLLECTION_EXAMPLE["id"] def test_single_collection( self, benchmark: BenchmarkFixture, single_href: str ) -> None: collection = benchmark(pystac.Collection.from_file, single_href) assert collection.id == COLLECTION_EXAMPLE["id"] def test_single_collection_search( self, benchmark: BenchmarkFixture, single_href: str ) -> None: search = CollectionSearch(url=COLLECTION_SEARCH_URL, **COLLECTION_EXAMPLE) collection_list = benchmark(search.collection_list()) assert len(collection_list) == 1 assert collection_list[0].id == COLLECTION_EXAMPLE["id"] class TestCollectionSearch: @pytest.fixture(scope="function") def fedeo_api(self) -> Client: api_content = read_data_file("fedeo_clearinghouse.json", parse_json=True) return Client.from_dict(api_content) def test_method(self) -> None: # Default method should be POST... search = CollectionSearch(url=COLLECTION_SEARCH_URL) assert search.method == "GET" def test_method_params(self) -> None: params_in: dict[str, Any] = { "bbox": (-72, 41, -71, 42), "datetime": "2021-01-01T00:00:00Z/2021-12-31T23:59:59Z", "q": "vegetation AND biomass", } # For POST this is pass through search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, **params_in, ) params = search.get_parameters() assert params == search.get_parameters() # For GET requests, parameters are in query string and must be serialized search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, **params_in, ) params = search.get_parameters() assert search.method == "GET" assert all(key in params for key in params_in) assert all(isinstance(params[key], str) for key in params_in) @pytest.mark.vcr def test_q_results(self) -> None: search_string = "monthly cloudless mosaics" search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, q=f'"{search_string}"', max_collections=20, limit=10, ) results = search.collection_list() assert all(isinstance(collection, pystac.Collection) for collection in results) assert all(search_string in collection.description for collection in results) assert search.matched() == 1 @pytest.mark.vcr def test_datetime_results(self) -> None: min_datetime = datetime(2024, 1, 1, 0, 0, 1, tzinfo=tzutc()) max_datetime = datetime(2024, 1, 1, 0, 0, 10, tzinfo=tzutc()) datetime_ = "/".join( ts.isoformat().replace("+00:00", "Z") for ts in [min_datetime, max_datetime] ) search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, datetime=datetime_, max_collections=20, ) for collection in search.collections(): for temporal_intervals in collection.extent.temporal.intervals: assert temporal_intervals_overlap( (temporal_intervals[0], temporal_intervals[1]), (min_datetime, max_datetime), ) search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, datetime=(min_datetime, max_datetime), max_collections=20, ) new_results = search.collections() for collection in new_results: for temporal_intervals in collection.extent.temporal.intervals: assert temporal_intervals_overlap( (temporal_intervals[0], temporal_intervals[1]), (min_datetime, max_datetime), ) @pytest.mark.vcr def test_bbox_results(self) -> None: bbox = (-159.893, 21.843, -159.176, 22.266) search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, bbox=bbox, limit=10, max_collections=10, ) for collection in search.collections(): for _bbox in collection.extent.spatial.bboxes: assert bboxes_overlap(bbox, tuple(float(coord) for coord in _bbox)) @pytest.mark.vcr def test_result_paging(self) -> None: search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, q="sentinel", limit=10, max_collections=20, ) # Check that the current page changes on the ItemSearch instance when a new page # is requested pages = list(search.pages()) assert pages[0] != pages[1] @pytest.mark.vcr def test_matched(self) -> None: q = "sentinel" search = CollectionSearch( url=f"{STAC_URLS['EARTH-SEARCH']}/collections", collection_search_extension_enabled=False, collection_search_free_text_enabled=False, q=q, limit=4, ) assert search.matched() == 5 @pytest.mark.vcr def test_enabled_but_client_side_q(self) -> None: q = "sentinel" limit = 5 search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=False, limit=limit, max_collections=limit, q=q, ) collection_list = search.collection_list() assert len(collection_list) <= limit collection_list_dict = search.collection_list_as_dict() assert len(collection_list_dict["collections"]) == len(collection_list) for collection in search.collections(): text_fields = [] if collection.description: text_fields.append(collection.description) if collection.title: text_fields.append(collection.title) if collection.keywords: text_fields.extend(collection.keywords) assert any(q in text_field.lower() for text_field in text_fields), ( f"{collection.id} failed check" ) @pytest.mark.vcr def test_client_side_q(self) -> None: q = "sentinel" limit = 10 search = CollectionSearch( url=f"{STAC_URLS['EARTH-SEARCH']}/collections", collection_search_extension_enabled=False, collection_search_free_text_enabled=False, limit=limit, max_collections=limit, q=q, ) collection_list = search.collection_list() assert len(collection_list) <= limit for collection in search.collections(): text_fields = [] if collection.description: text_fields.append(collection.description) if collection.title: text_fields.append(collection.title) if collection.keywords: text_fields.extend(collection.keywords) assert any(q in text_field.lower() for text_field in text_fields), ( f"{collection.id} failed check" ) @pytest.mark.vcr def test_client_side_bbox(self) -> None: bbox = (60, 0, 70, 10) limit = 10 search = CollectionSearch( url=f"{STAC_URLS['EARTH-SEARCH']}/collections", collection_search_extension_enabled=False, collection_search_free_text_enabled=False, limit=limit, max_collections=limit, bbox=bbox, ) collection_list = search.collection_list() assert len(collection_list) <= limit for collection in search.collections(): assert any( bboxes_overlap(bbox, tuple(float(coord) for coord in collection_bbox)) for collection_bbox in collection.extent.spatial.bboxes ), f"{collection.id} failed check" @pytest.mark.vcr def test_client_side_datetime(self) -> None: _datetime_interval = ("2024-09-15T00:00:00+00:00", "2024-09-16T00:00:00+00:00") limit = 10 search = CollectionSearch( url=f"{STAC_URLS['EARTH-SEARCH']}/collections", collection_search_extension_enabled=False, collection_search_free_text_enabled=False, limit=limit, max_collections=limit, datetime="/".join(_datetime_interval), ) collection_list = search.collection_list() assert len(collection_list) <= limit temporal_interval = ( datetime.fromisoformat(_datetime_interval[0]), datetime.fromisoformat(_datetime_interval[1]), ) for collection in search.collections(): assert any( temporal_intervals_overlap( temporal_interval, (collection_temporal_interval[0], collection_temporal_interval[1]), ) for collection_temporal_interval in collection.extent.temporal.intervals ), f"{collection.id} failed check" def test_client_side_extra_args(self) -> None: with pytest.raises(ValueError): CollectionSearch( url=f"{STAC_URLS['EARTH-SEARCH']}/collections", collection_search_extension_enabled=False, collection_search_free_text_enabled=False, filter="title LIKE '%nope%'", ) @pytest.mark.vcr def test_result_paging_max_collections(self) -> None: max_collections = 15 limit = 10 search = CollectionSearch( url=COLLECTION_SEARCH_URL, collection_search_extension_enabled=True, collection_search_free_text_enabled=True, bbox=(-92.379, 46.662, -91.788, 46.919), limit=limit, max_collections=max_collections, ) num_pages = 0 collections = list() for page in search.pages_as_dicts(): num_pages += 1 collections.extend(page["collections"]) assert num_pages == ceil(max_collections / limit) assert len(collections) == max_collections def test_bboxes_overlap() -> None: assert bboxes_overlap( (0, 0, 2, 2), (1, 1, 3, 3), ) assert not bboxes_overlap( (0, 0, 2, 2), (3, 3, 4, 4), ) def test_temporal_intervals_overlap() -> None: assert temporal_intervals_overlap( (datetime(2024, 9, 1, tzinfo=tzutc()), datetime(2024, 9, 2, tzinfo=tzutc())), ( datetime(2024, 9, 1, 12, tzinfo=tzutc()), datetime(2024, 9, 2, 12, tzinfo=tzutc()), ), ) assert temporal_intervals_overlap( (datetime(2024, 9, 1, tzinfo=tzutc()), None), ( datetime(2024, 9, 1, 12, tzinfo=tzutc()), datetime(2024, 9, 2, 12, tzinfo=tzutc()), ), ) assert temporal_intervals_overlap( (None, None), ( datetime(2024, 9, 1, 12, tzinfo=tzutc()), datetime(2024, 9, 2, 12, tzinfo=tzutc()), ), ) assert not temporal_intervals_overlap( (datetime(2023, 9, 1, tzinfo=tzutc()), datetime(2023, 9, 2, tzinfo=tzutc())), ( datetime(2024, 9, 1, 12, tzinfo=tzutc()), datetime(2024, 9, 2, 12, tzinfo=tzutc()), ), ) assert not temporal_intervals_overlap( (datetime(2023, 9, 1, tzinfo=tzutc()), datetime(2023, 9, 2, tzinfo=tzutc())), ( datetime(2024, 9, 1, 12, tzinfo=tzutc()), None, ), ) def test_invalid_collection() -> None: # https://github.com/stac-utils/pystac-client/issues/786 data = json.loads(read_data_file("invalid-collection.json")) with pytest.warns(PystacClientWarning): collection_matches(data)