-
Notifications
You must be signed in to change notification settings - Fork 102
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
- Loading branch information
1 parent
80064f7
commit 8f400e1
Showing
3 changed files
with
80 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,75 @@ | ||
from typing import Iterator | ||
|
||
import pytest | ||
from starlette.testclient import TestClient | ||
|
||
from stac_fastapi.api.app import StacApi | ||
from stac_fastapi.api.models import create_get_request_model, create_post_request_model | ||
from stac_fastapi.extensions.core import FilterExtension | ||
from stac_fastapi.types.config import ApiSettings | ||
from stac_fastapi.types.core import BaseCoreClient | ||
|
||
|
||
class DummyCoreClient(BaseCoreClient): | ||
def all_collections(self, *args, **kwargs): | ||
raise NotImplementedError | ||
|
||
def get_collection(self, *args, **kwargs): | ||
raise NotImplementedError | ||
|
||
def get_item(self, *args, **kwargs): | ||
raise NotImplementedError | ||
|
||
def get_search(self, *args, **kwargs): | ||
raise NotImplementedError | ||
|
||
def post_search(self, *args, **kwargs): | ||
return args[0].model_dump() | ||
|
||
def item_collection(self, *args, **kwargs): | ||
raise NotImplementedError | ||
|
||
|
||
@pytest.fixture | ||
def client() -> Iterator[TestClient]: | ||
settings = ApiSettings() | ||
extensions = [FilterExtension()] | ||
api = StacApi( | ||
settings=settings, | ||
client=DummyCoreClient(), | ||
extensions=extensions, | ||
search_get_request_model=create_get_request_model(extensions), | ||
search_post_request_model=create_post_request_model(extensions), | ||
) | ||
with TestClient(api.app) as client: | ||
yield client | ||
|
||
|
||
def test_search_filter_post_filter_lang_default(client: TestClient): | ||
"""Test search POST endpoint with filter ext.""" | ||
response = client.post( | ||
"/search", | ||
json={ | ||
"collections": ["test"], | ||
"filter": {"op": "=", "args": [{"property": "test_property"}, "test-value"]}, | ||
}, | ||
) | ||
assert response.is_success, response.json() | ||
response_dict = response.json() | ||
assert response_dict["filter_lang"] == "cql2-json" | ||
|
||
|
||
def test_search_filter_post_filter_lang_non_default(client: TestClient): | ||
"""Test search POST endpoint with filter ext.""" | ||
filter_lang_value = "cql2-text" | ||
response = client.post( | ||
"/search", | ||
json={ | ||
"collections": ["test"], | ||
"filter": {"op": "=", "args": [{"property": "test_property"}, "test-value"]}, | ||
"filter-lang": filter_lang_value, | ||
}, | ||
) | ||
assert response.is_success, response.json() | ||
response_dict = response.json() | ||
assert response_dict["filter_lang"] == filter_lang_value |