97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
import urllib.error
|
|
|
|
import pytest
|
|
|
|
from pipeline.download import satellite_tiles
|
|
|
|
|
|
class _Response:
|
|
headers = {"content-type": "image/jpeg"}
|
|
|
|
def __init__(self, data: bytes = b"jpeg") -> None:
|
|
self._data = data
|
|
|
|
def read(self) -> bytes:
|
|
return self._data
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
|
|
class _Throttle:
|
|
def __init__(self) -> None:
|
|
self.deferred: list[float] = []
|
|
self.waits = 0
|
|
|
|
def wait(self) -> None:
|
|
self.waits += 1
|
|
|
|
def defer(self, delay: float) -> bool:
|
|
self.deferred.append(delay)
|
|
return False
|
|
|
|
|
|
def _http_error(url: str, code: int) -> urllib.error.HTTPError:
|
|
return urllib.error.HTTPError(url, code, "Forbidden", {}, None)
|
|
|
|
|
|
def test_fetch_tile_retries_eox_403_with_shared_cooldown(monkeypatch):
|
|
tile = satellite_tiles.Tile(zoom=9, x=248, y=172)
|
|
calls: list[str] = []
|
|
|
|
def fake_urlopen(req, timeout):
|
|
calls.append(req.full_url)
|
|
if len(calls) == 1:
|
|
raise _http_error(req.full_url, 403)
|
|
return _Response()
|
|
|
|
monkeypatch.setattr(satellite_tiles.urllib.request, "urlopen", fake_urlopen)
|
|
|
|
throttle = _Throttle()
|
|
fetched_tile, data = satellite_tiles._fetch_tile(
|
|
tile,
|
|
satellite_tiles.DEFAULT_TILE_URL,
|
|
timeout=1.0,
|
|
retries=1,
|
|
throttle=throttle,
|
|
retry_cooldown=15.0,
|
|
)
|
|
|
|
assert fetched_tile == tile
|
|
assert data == b"jpeg"
|
|
assert calls == [
|
|
"https://tiles.maps.eox.at/wmts/1.0.0/s2cloudless_3857/default/"
|
|
"GoogleMapsCompatible/9/172/248.jpg",
|
|
"https://tiles.maps.eox.at/wmts/1.0.0/s2cloudless_3857/default/"
|
|
"GoogleMapsCompatible/9/172/248.jpg",
|
|
]
|
|
assert throttle.deferred == [15.0]
|
|
assert throttle.waits == 2
|
|
|
|
|
|
def test_fetch_tile_does_not_retry_non_eox_403(monkeypatch):
|
|
tile = satellite_tiles.Tile(zoom=9, x=248, y=172)
|
|
calls: list[str] = []
|
|
|
|
def fake_urlopen(req, timeout):
|
|
calls.append(req.full_url)
|
|
raise _http_error(req.full_url, 403)
|
|
|
|
monkeypatch.setattr(satellite_tiles.urllib.request, "urlopen", fake_urlopen)
|
|
|
|
throttle = _Throttle()
|
|
with pytest.raises(RuntimeError, match="HTTP Error 403"):
|
|
satellite_tiles._fetch_tile(
|
|
tile,
|
|
"https://example.com/{z}/{x}/{y}.jpg",
|
|
timeout=1.0,
|
|
retries=1,
|
|
throttle=throttle,
|
|
retry_cooldown=15.0,
|
|
)
|
|
|
|
assert calls == ["https://example.com/9/248/172.jpg"]
|
|
assert throttle.deferred == []
|