Skip to content

Commit d43398c

Browse files
committed
Add tests from #265
1 parent 63ce791 commit d43398c

File tree

1 file changed

+140
-0
lines changed

1 file changed

+140
-0
lines changed
Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
import pytest
2+
from httpx import ASGITransport, AsyncClient
3+
4+
from stac_fastapi.pgstac.db import close_db_connection, connect_to_db
5+
6+
root_path_value = "/stac/v1"
7+
8+
# Append the root path to the base URL, this is key to reproducing the issue where the root path appears twice in some links
9+
base_url = f"http://api.acme.com{root_path_value}"
10+
11+
12+
@pytest.fixture(scope="function")
13+
async def app_with_root_path(database, monkeypatch):
14+
"""
15+
Provides the global stac_fastapi.pgstac.app.app instance, configured with a
16+
specific ROOT_PATH environment variable and connected to the test database.
17+
"""
18+
19+
monkeypatch.setenv("ROOT_PATH", root_path_value)
20+
monkeypatch.setenv("PGUSER", database.user)
21+
monkeypatch.setenv("PGPASSWORD", database.password)
22+
monkeypatch.setenv("PGHOST", database.host)
23+
monkeypatch.setenv("PGPORT", str(database.port))
24+
monkeypatch.setenv("PGDATABASE", database.dbname)
25+
monkeypatch.setenv("ENABLE_TRANSACTIONS_EXTENSIONS", "TRUE")
26+
27+
from stac_fastapi.pgstac.app import app, with_transactions
28+
29+
# Ensure the app's root_path is configured as expected
30+
assert (
31+
app.root_path == root_path_value
32+
), f"app_with_root_path fixture: app.root_path is '{app.root_path}', expected '{root_path_value}'"
33+
34+
await connect_to_db(app, add_write_connection_pool=with_transactions)
35+
yield app
36+
await close_db_connection(app)
37+
38+
39+
@pytest.fixture(scope="function")
40+
async def client_with_root_path(app_with_root_path):
41+
async with AsyncClient(
42+
transport=ASGITransport(app=app_with_root_path),
43+
base_url=base_url,
44+
) as c:
45+
yield c
46+
47+
48+
@pytest.fixture(scope="function")
49+
async def loaded_client(client_with_root_path, load_test_data):
50+
col = load_test_data("test_collection.json")
51+
resp = await client_with_root_path.post(
52+
"/collections",
53+
json=col,
54+
)
55+
assert resp.status_code == 201
56+
item = load_test_data("test_item.json")
57+
resp = await client_with_root_path.post(
58+
f"/collections/{col['id']}/items",
59+
json=item,
60+
)
61+
assert resp.status_code == 201
62+
item = load_test_data("test_item2.json")
63+
resp = await client_with_root_path.post(
64+
f"/collections/{col['id']}/items",
65+
json=item,
66+
)
67+
assert resp.status_code == 201
68+
yield client_with_root_path
69+
70+
71+
async def test_search_links_are_valid(loaded_client):
72+
resp = await loaded_client.get("/search?limit=1")
73+
assert resp.status_code == 200
74+
response_json = resp.json()
75+
assert_links_href(response_json.get("links", []), base_url)
76+
77+
78+
async def test_collection_links_are_valid(loaded_client):
79+
resp = await loaded_client.get("/collections?limit=1")
80+
assert resp.status_code == 200
81+
response_json = resp.json()
82+
assert_links_href(response_json.get("links", []), base_url)
83+
84+
85+
async def test_items_collection_links_are_valid(loaded_client):
86+
resp = await loaded_client.get("/collections/test-collection/items?limit=1")
87+
assert resp.status_code == 200
88+
response_json = resp.json()
89+
assert_links_href(response_json.get("links", []), base_url)
90+
91+
92+
def assert_links_href(links, url_prefix):
93+
"""
94+
Ensure all links start with the expected URL prefix and check that
95+
there is no root_path duplicated in the URL.
96+
97+
Args:
98+
links: List of link dictionaries with 'href' keys
99+
url_prefix: Expected URL prefix (e.g., 'http://test/stac/v1')
100+
"""
101+
from urllib.parse import urlparse
102+
103+
failed_links = []
104+
parsed_prefix = urlparse(url_prefix)
105+
root_path = parsed_prefix.path # e.g., '/stac/v1'
106+
107+
for link in links:
108+
href = link["href"]
109+
rel = link.get("rel", "unknown")
110+
111+
# Check if link starts with the expected prefix
112+
if not href.startswith(url_prefix):
113+
failed_links.append(
114+
{
115+
"rel": rel,
116+
"href": href,
117+
"error": f"does not start with expected prefix '{url_prefix}'",
118+
}
119+
)
120+
continue
121+
122+
# Check for duplicated root path
123+
if root_path and root_path != "/":
124+
remainder = href[len(url_prefix) :]
125+
if remainder.startswith(root_path):
126+
failed_links.append(
127+
{
128+
"rel": rel,
129+
"href": href,
130+
"error": f"contains duplicated root path '{root_path}'",
131+
}
132+
)
133+
134+
# If there are failed links, create a detailed error report
135+
if failed_links:
136+
error_report = "Link validation failed:\n"
137+
for failed_link in failed_links:
138+
error_report += f" - rel: '{failed_link['rel']}', href: '{failed_link['href']}' - {failed_link['error']}\n"
139+
140+
raise AssertionError(error_report)

0 commit comments

Comments
 (0)