Harden DHMV WCS request headers
This commit is contained in:
@@ -279,7 +279,13 @@ class DhmvAcquisitionService:
|
||||
|
||||
@staticmethod
|
||||
def _fetch(request_url: str, settings: Settings, opener: Callable[..., Any] | None = None) -> tuple[bytes, str]:
|
||||
request = Request(request_url, headers={"User-Agent": "GeoIntel/0.1 bounded-dhmv-acquisition"})
|
||||
request = Request(
|
||||
request_url,
|
||||
headers={
|
||||
"Accept": "*/*",
|
||||
"User-Agent": "GeoIntel/0.1 bounded-dhmv-acquisition",
|
||||
},
|
||||
)
|
||||
max_bytes = settings.dhmv_max_response_mb * 1024 * 1024
|
||||
try:
|
||||
with (opener or urlopen)(request, timeout=settings.dhmv_timeout_seconds) as response:
|
||||
|
||||
@@ -200,6 +200,28 @@ def test_dhmv_multipart_geotiff_is_extracted_and_invalid_response_fails_closed()
|
||||
assert exc_info.value.code == "DHMV_PROVIDER_INVALID_RESPONSE"
|
||||
|
||||
|
||||
def test_dhmv_fetch_sends_explicit_accept_header_required_by_official_wcs() -> None:
|
||||
observed_headers: dict[str, str | None] = {}
|
||||
|
||||
def opener(request, **_kwargs):
|
||||
observed_headers["accept"] = request.get_header("Accept")
|
||||
observed_headers["user_agent"] = request.get_header("User-agent")
|
||||
return FakeResponse(b"II*\x00test", "image/tiff")
|
||||
|
||||
content, content_type = DhmvAcquisitionService._fetch(
|
||||
"https://geo.api.vlaanderen.be/DHMV/wcs?bounded=true",
|
||||
Settings(_env_file=None),
|
||||
opener,
|
||||
)
|
||||
|
||||
assert content == b"II*\x00test"
|
||||
assert content_type == "image/tiff"
|
||||
assert observed_headers == {
|
||||
"accept": "*/*",
|
||||
"user_agent": "GeoIntel/0.1 bounded-dhmv-acquisition",
|
||||
}
|
||||
|
||||
|
||||
def test_dhmv_acquisition_clips_validates_and_persists_via_dataset_service(tmp_path) -> None:
|
||||
project_id = uuid4()
|
||||
area_id = uuid4()
|
||||
|
||||
Reference in New Issue
Block a user