import operator import urllib.parse from datetime import datetime, timedelta from typing import Any import pystac import pytest import requests from dateutil.tz import tzutc from pytest_benchmark.fixture import BenchmarkFixture from requests_mock import Mocker from pystac_client import Client from pystac_client.item_search import ItemSearch from .helpers import STAC_URLS, read_data_file SEARCH_URL = f"{STAC_URLS['PLANETARY-COMPUTER']}/search" INTERSECTS_EXAMPLE = { "type": "Polygon", "coordinates": [ [ [-73.21, 43.99], [-73.21, 44.05], [-73.12, 44.05], [-73.12, 43.99], [-73.21, 43.99], ] ], } ITEM_EXAMPLE: dict[str, Any] = {"collections": "io-lulc", "ids": "60U-2020"} class TestItemPerformance: @pytest.fixture(scope="function") def single_href(self) -> str: item_href = ( "https://planetarycomputer.microsoft.com/api/stac/v1/collections/" f"{ITEM_EXAMPLE['collections']}/items/{ITEM_EXAMPLE['ids']}" ) return item_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"] == ITEM_EXAMPLE["ids"] def test_single_item(self, benchmark: BenchmarkFixture, single_href: str) -> None: item = benchmark(pystac.Item.from_file, single_href) assert item.id == ITEM_EXAMPLE["ids"] def test_single_item_search( self, benchmark: BenchmarkFixture, single_href: str ) -> None: search = ItemSearch(url=SEARCH_URL, **ITEM_EXAMPLE) item_collection = benchmark(search.item_collection()) assert len(item_collection.items) == 1 assert item_collection.items[0].id == ITEM_EXAMPLE["ids"] class TestItemSearch: @pytest.fixture(scope="function") def astraea_api(self) -> Client: api_content = read_data_file("astraea_api.json", parse_json=True) return Client.from_dict(api_content) def test_method(self) -> None: # Default method should be POST... search = ItemSearch(url=SEARCH_URL) assert search.method == "POST" # "method" argument should take precedence over presence of "intersects" search = ItemSearch(url=SEARCH_URL, method="GET", intersects=INTERSECTS_EXAMPLE) assert search.method == "GET" def test_method_params(self) -> None: params_in: dict[str, Any] = { "bbox": (-72, 41, -71, 42), "ids": ( "idone", "idtwo", ), "collections": ("collectionone",), "intersects": INTERSECTS_EXAMPLE, } # For POST this is pass through search = ItemSearch(url=SEARCH_URL, **params_in) params = search.get_parameters() assert params == search.get_parameters() # For GET requests, parameters are in query string and must be serialized search = ItemSearch(url=SEARCH_URL, method="GET", **params_in) params = search.get_parameters() 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_results(self) -> None: search = ItemSearch( url=SEARCH_URL, collections="naip", max_items=20, limit=10, ) results = search.items() assert all(isinstance(item, pystac.Item) for item in results) @pytest.mark.vcr def test_ids_results(self) -> None: ids = [ "S2B_MSIL2A_20210610T115639_N0212_R066_T33XXG_20210613T185024.SAFE", "fl_m_2608004_nw_17_060_20191215_20200113", ] search = ItemSearch( url=SEARCH_URL, collections=["naip"], ids=ids, ) results = list(search.items()) assert len(results) == 1 assert all(item.id in ids for item in results) @pytest.mark.vcr def test_datetime_results(self) -> None: min_datetime = datetime(2019, 1, 1, 0, 0, 1, tzinfo=tzutc()) max_datetime = datetime(2019, 1, 1, 0, 0, 10, tzinfo=tzutc()) datetime_ = "2019-01-01T00:00:01Z/2019-01-01T00:00:10Z" search = ItemSearch(url=SEARCH_URL, datetime=datetime_, max_items=20) for item in search.items(): assert item.datetime is not None assert ( min_datetime <= item.datetime <= (max_datetime + timedelta(seconds=1)) ) search = ItemSearch( url=SEARCH_URL, datetime=(min_datetime, max_datetime), max_items=20 ) new_results = search.items() for item in new_results: assert item.datetime is not None assert ( min_datetime <= item.datetime <= (max_datetime + timedelta(seconds=1)) ) @pytest.mark.vcr def test_intersects_results(self) -> None: # GeoJSON-like dict intersects_dict = { "type": "Polygon", "coordinates": [ [ [-73.21, 43.99], [-73.21, 44.05], [-73.12, 44.05], [-73.12, 43.99], [-73.21, 43.99], ] ], } search = ItemSearch( url=SEARCH_URL, intersects=intersects_dict, collections="naip" ) results = list(search.items()) assert len(results) == 42 # Geo-interface object class MockGeoObject: @property def __geo_interface__(self) -> dict[str, Any]: return intersects_dict intersects_obj = MockGeoObject() search = ItemSearch( url=SEARCH_URL, intersects=intersects_obj, collections="naip" ) new_results = search.items() assert all(isinstance(item, pystac.Item) for item in new_results) def test_get_with_query(self, requests_mock: Mocker) -> None: requests_mock.get( ( f"{SEARCH_URL}?query=%7B%22eo%3Acloud_cover%22%3A%7B%22gte%22%3A0%2C%22lte%22%3A10%7D%7D" ), status_code=200, json={"features": [{"foo": "bar"}], "links": []}, ) items = list( ItemSearch( url=SEARCH_URL, method="GET", query={"eo:cloud_cover": {"gte": 0, "lte": 10}}, ).items_as_dicts() ) assert len(items) == 1 @pytest.mark.vcr def test_result_paging(self) -> None: search = ItemSearch( url=SEARCH_URL, bbox=(-73.21, 43.99, -73.12, 44.05), collections="naip", limit=10, max_items=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] assert pages[0].items != pages[1].items @pytest.mark.vcr def test_result_paging_max_items(self) -> None: search = ItemSearch( url=SEARCH_URL, collections="naip", limit=10, max_items=25, ) num_pages = 0 items = list() for page in search.pages_as_dicts(): num_pages += 1 items.extend(page["features"]) assert num_pages == 3 assert len(items) == 25 @pytest.mark.vcr def test_item_collection(self) -> None: search = ItemSearch( url=SEARCH_URL, bbox=(-73.21, 43.99, -73.12, 44.05), collections="naip", limit=10, max_items=20, ) item_collection = search.item_collection() assert isinstance(item_collection, pystac.ItemCollection) assert len(item_collection) == 20 @pytest.mark.vcr @pytest.mark.parametrize( "method, alternative, is_sequence, is_pystac", [ ("get_item_collections", "pages", True, True), ("item_collections", "pages", True, True), ("get_items", "items", True, True), ("get_all_items", "item_collection", False, True), ("get_all_items_as_dict", "item_collection_as_dict", False, False), ], ) def test_deprecations( self, method: str, alternative: str, is_sequence: bool, is_pystac: bool ) -> None: search = ItemSearch( url=SEARCH_URL, bbox=(-73.21, 43.99, -73.12, 44.05), collections="naip", limit=10, max_items=20, ) with pytest.warns(FutureWarning, match=method): result = operator.methodcaller(method)(search) expected = operator.methodcaller(alternative)(search) if is_sequence: result = list(result) expected = list(expected) if is_pystac: result = [x.to_dict() for x in result] expected = [x.to_dict() for x in expected] else: if is_pystac: result = result.to_dict() expected = expected.to_dict() assert result == expected @pytest.mark.vcr def test_items_as_dicts(self) -> None: search = ItemSearch( url=SEARCH_URL, bbox=(-73.21, 43.99, -73.12, 44.05), collections="naip", limit=10, max_items=20, ) assert len(list(search.items_as_dicts())) == 20 class TestItemSearchQuery: @pytest.mark.vcr def test_query_shortcut_syntax(self) -> None: search = ItemSearch( url=SEARCH_URL, collections="naip", bbox=(-73.21, 43.99, -73.12, 44.05), query=["gsd=0.6"], max_items=1, ) items1 = list(search.items()) search = ItemSearch( url=SEARCH_URL, collections="naip", bbox=(-73.21, 43.99, -73.12, 44.05), query={"gsd": {"eq": 0.6}}, max_items=1, ) items2 = list(search.items()) assert len(items1) == 1 assert len(items2) == 1 assert items1[0].id == items2[0].id @pytest.mark.vcr def test_query_json_syntax(self) -> None: # with a list of json strs (format of CLI argument to ItemSearch) search = ItemSearch( url=SEARCH_URL, collections="naip", bbox=(-73.21, 43.99, -73.12, 44.05), query=['{"gsd": { "gte": 0, "lte": 1 }}'], max_items=1, ) item1 = list(search.items())[0] assert item1.properties["gsd"] <= 1 # with a single dict search = ItemSearch( url=SEARCH_URL, collections="naip", bbox=(-73.21, 43.99, -73.12, 44.05), query={"gsd": {"gte": 0, "lte": 1}}, max_items=1, ) item2 = list(search.items())[0] assert item2.properties["gsd"] <= 1 assert item1.id == item2.id def test_query_json_syntax() -> None: item_search = ItemSearch("") assert item_search._format_query( ['{"eo:cloud_cover": { "gte": 0, "lte": 1 }}'] ) == {"eo:cloud_cover": {"gte": 0, "lte": 1}} assert item_search._format_query({"eo:cloud_cover": {"gte": 0, "lte": 1}}) == { "eo:cloud_cover": {"gte": 0, "lte": 1} } assert item_search._format_query(["eo:cloud_cover<=1"]) == { "eo:cloud_cover": {"lte": "1"} } assert item_search._format_query(["eo:cloud_cover<=1", "eo:cloud_cover>0"]) == { "eo:cloud_cover": {"lte": "1", "gt": "0"} } def test_url_with_query_parameter() -> None: # https://github.com/stac-utils/pystac-client/issues/522 search = ItemSearch( url="http://pystac-client.test", query={"eo:cloud_cover": {"lt": 42}} ) url = urllib.parse.urlparse(search.url_with_parameters()) query = urllib.parse.parse_qs(url.query) assert query["query"] == [r'{"eo:cloud_cover":{"lt":42}}'] @pytest.mark.vcr def test_multiple_collections() -> None: search = ItemSearch( url="https://earth-search.aws.element84.com/v1/search", collections=["sentinel-2-l2a", "landsat-c2-l2"], intersects={"type": "Point", "coordinates": [-105.1019, 40.1672]}, datetime="2023-10-08", ) collections = {item.collection_id for item in search.items()} assert collections == {"sentinel-2-l2a", "landsat-c2-l2"} def test_naive_datetime() -> None: search = ItemSearch( url="https://earth-search.aws.element84.com/v1/search", datetime=datetime(2024, 5, 14, 4, 25, 42, tzinfo=None), method="POST", ) assert search.get_parameters()["datetime"] == "2024-05-14T04:25:42Z" @pytest.mark.vcr def test_fields() -> None: search = ItemSearch( url="https://earth-search.aws.element84.com/v1/search", collections=["sentinel-2-c1-l2a"], intersects={"type": "Point", "coordinates": [-105.1019, 40.1672]}, max_items=1, fields=["-geometry", "-assets", "-links"], ) item = next(search.items_as_dicts()) assert "geometry" not in item assert "assets" not in item assert "links" not in item def test_feature() -> None: search = ItemSearch( url="https://earth-search.aws.element84.com/v1/search", intersects={ "type": "Feature", "geometry": {"type": "Point", "coordinates": [-105.1019, 40.1672]}, }, ) assert search.get_parameters()["intersects"] == { "type": "Point", "coordinates": [-105.1019, 40.1672], }