298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""Tests for the ODMS client, against a real HTTP server on localhost.
|
|
|
|
No mocking library and no monkeypatched sockets: the client's whole job is to
|
|
speak HTTP correctly, and a fake that intercepts above the socket cannot catch a
|
|
malformed query string, a mishandled 429, or a retry that silently repeats a
|
|
POST. ``http.server`` is in the standard library and costs milliseconds.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from odms import NotFound, OdmsClient, RateLimited, Unauthorized # noqa: E402
|
|
from odms.errors import HTTPError, ServiceUnavailable # noqa: E402
|
|
|
|
|
|
class Recorder:
|
|
"""The scripted responses, and what the client actually asked for."""
|
|
|
|
def __init__(self):
|
|
self.requests: list[dict] = []
|
|
self.responses: list[tuple] = []
|
|
|
|
def reply(self, status: int, body="", headers=None):
|
|
"""Queue one response. The last queued one repeats once used up."""
|
|
if isinstance(body, (dict, list)):
|
|
body = json.dumps(body)
|
|
headers = {"Content-Type": "application/json", **(headers or {})}
|
|
self.responses.append((status, body, headers or {}))
|
|
return self
|
|
|
|
def next_response(self):
|
|
if len(self.responses) > 1:
|
|
return self.responses.pop(0)
|
|
return self.responses[0] if self.responses else (200, "{}", {})
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def _server():
|
|
"""One server for the module. Binding a socket and starting a thread costs
|
|
most of a second here, and 22 of them is a suite nobody runs."""
|
|
recorder = Recorder()
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_args):
|
|
pass # the test output is not a web server log
|
|
|
|
def _handle(self):
|
|
parsed = urlparse(self.path)
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
recorder.requests.append({
|
|
"method": self.command,
|
|
"path": parsed.path,
|
|
"query": parse_qs(parsed.query),
|
|
"body": self.rfile.read(length).decode() if length else "",
|
|
"headers": dict(self.headers),
|
|
})
|
|
status, body, headers = recorder.next_response()
|
|
payload = body.encode()
|
|
self.send_response(status)
|
|
for key, value in headers.items():
|
|
self.send_header(key, value)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
do_GET = do_POST = _handle
|
|
|
|
class FastServer(HTTPServer):
|
|
def server_bind(self):
|
|
# HTTPServer.server_bind resolves its own FQDN for the Server header,
|
|
# which costs ~half a second per bind on Windows. Nothing under test
|
|
# reads it.
|
|
super(HTTPServer, self).server_bind()
|
|
self.server_name = "127.0.0.1"
|
|
self.server_port = self.socket.getsockname()[1]
|
|
|
|
httpd = FastServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=httpd.serve_forever, kwargs={"poll_interval": 0.02})
|
|
thread.daemon = True
|
|
thread.start()
|
|
recorder.url = f"http://127.0.0.1:{httpd.server_port}"
|
|
try:
|
|
yield recorder
|
|
finally:
|
|
httpd.shutdown()
|
|
httpd.server_close()
|
|
|
|
|
|
@pytest.fixture
|
|
def server(_server):
|
|
_server.requests.clear()
|
|
_server.responses.clear()
|
|
return _server
|
|
|
|
|
|
@pytest.fixture
|
|
def client(server):
|
|
# Retries with no backoff: the retry *policy* is what is under test, and
|
|
# sleeping through it would only make the suite slow.
|
|
with OdmsClient(server.url, token="t0ken", max_retries=2) as odms:
|
|
odms._transport.backoff_s = 0.0
|
|
# urllib is the floor every consumer gets; httpx, if installed here, is
|
|
# an optimisation the tests should not accidentally become dependent on.
|
|
odms._transport._session = None
|
|
yield odms
|
|
|
|
|
|
# --- request shape ----------------------------------------------------------
|
|
|
|
|
|
def test_the_token_is_sent_as_a_bearer_header(server, client):
|
|
server.reply(200, {"results": []})
|
|
client.satellites()
|
|
assert server.requests[0]["headers"]["Authorization"] == "Bearer t0ken"
|
|
|
|
|
|
def test_unset_filters_are_omitted_rather_than_sent_as_none(server, client):
|
|
"""A `?source=None` filter matches no source and returns nothing, silently."""
|
|
server.reply(200, {"count": 0, "results": []})
|
|
client.query(internal_id="iss", source=None, name=None)
|
|
|
|
query = server.requests[0]["query"]
|
|
assert query == {"internal_id": ["iss"], "format": ["json"]}
|
|
|
|
|
|
def test_datetimes_are_sent_as_iso_8601(server, client):
|
|
server.reply(200, {"count": 0, "results": []})
|
|
client.at("iss", datetime(2026, 4, 1, tzinfo=timezone.utc))
|
|
|
|
assert server.requests[0]["query"]["datetime"] == ["2026-04-01T00:00:00+00:00"]
|
|
|
|
|
|
def test_booleans_are_sent_the_way_django_reads_them(server, client):
|
|
"""Python's str(True) is "True"; str(False) is "False" -- but a lowercased
|
|
"false" is truthy to a naive parser, which is the bug this pins."""
|
|
server.reply(201, {"id": 1})
|
|
client.push_message("CCSDS_OPM_VERS = 3.0", key="iss", is_public=False)
|
|
|
|
assert json.loads(server.requests[0]["body"])["is_public"] is False
|
|
|
|
|
|
def test_an_unknown_filter_raises_instead_of_being_dropped(client):
|
|
"""A typo'd filter that quietly returns the whole catalogue is worse than
|
|
an exception -- the caller gets plausible, wrong data."""
|
|
with pytest.raises(TypeError, match="norrad"):
|
|
client.query(norrad="25544")
|
|
|
|
|
|
def test_an_unknown_format_raises_before_the_request(server, client):
|
|
with pytest.raises(ValueError, match="format"):
|
|
client.query(format="parquet")
|
|
assert server.requests == []
|
|
|
|
|
|
def test_transform_requires_exactly_one_source(client):
|
|
with pytest.raises(TypeError):
|
|
client.transform("ITRF")
|
|
with pytest.raises(TypeError):
|
|
client.transform("ITRF", key="iss", message_id=4)
|
|
|
|
|
|
# --- responses --------------------------------------------------------------
|
|
|
|
|
|
def test_rendered_formats_come_back_as_text_not_json(server, client):
|
|
lines = "ISS (ZARYA)\n1 25544U ...\n2 25544 ..."
|
|
server.reply(200, lines, {"Content-Type": "text/plain"})
|
|
|
|
assert client.download(internal_id="iss", format="tle") == lines
|
|
|
|
|
|
def test_latest_returns_none_when_the_satellite_has_no_elements(server, client):
|
|
server.reply(200, {"count": 0, "results": []})
|
|
assert client.latest("iss") is None
|
|
|
|
|
|
def test_decay_latest_returns_none_when_there_is_no_forecast(server, client):
|
|
"""No forecast is a normal state: an object with no fittable history is
|
|
skipped deliberately rather than predicted from a coefficient nobody can
|
|
justify. Callers must not have to catch an exception for the normal case."""
|
|
server.reply(404, {"detail": "no forecast"})
|
|
assert client.decay_latest("iss") is None
|
|
|
|
|
|
def test_find_returns_none_where_satellite_raises(server, client):
|
|
server.reply(404, {"detail": "not found"})
|
|
assert client.find("nope") is None
|
|
with pytest.raises(NotFound):
|
|
client.satellite("nope")
|
|
|
|
|
|
# --- errors -----------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("status,expected", [
|
|
(401, Unauthorized),
|
|
(403, Unauthorized),
|
|
(404, NotFound),
|
|
(400, HTTPError),
|
|
])
|
|
def test_statuses_map_onto_actionable_exceptions(server, client, status, expected):
|
|
server.reply(status, {"detail": "nope"})
|
|
with pytest.raises(expected):
|
|
client.satellite("iss")
|
|
|
|
|
|
def test_the_error_detail_is_lifted_out_of_a_json_body(server, client):
|
|
server.reply(400, {"detail": "end must be >= start"})
|
|
with pytest.raises(HTTPError) as caught:
|
|
client.satellite("iss")
|
|
assert caught.value.detail == "end must be >= start"
|
|
assert caught.value.status == 400
|
|
|
|
|
|
def test_a_non_json_error_body_still_produces_a_usable_message(server, client):
|
|
server.reply(502, "<html>Bad Gateway</html>", {"Content-Type": "text/html"})
|
|
with pytest.raises(ServiceUnavailable) as caught:
|
|
client.satellite("iss")
|
|
assert "Bad Gateway" in str(caught.value)
|
|
|
|
|
|
# --- retries ----------------------------------------------------------------
|
|
|
|
|
|
def test_a_429_is_retried_and_then_succeeds(server, client):
|
|
server.reply(429, "slow down", {"Retry-After": "0"})
|
|
server.reply(200, {"results": [{"internal_id": "iss"}]})
|
|
|
|
assert client.satellites() == [{"internal_id": "iss"}]
|
|
assert len(server.requests) == 2
|
|
|
|
|
|
def test_a_persistent_429_raises_with_the_servers_advice(server, client):
|
|
server.reply(429, "slow down", {"Retry-After": "0"})
|
|
|
|
with pytest.raises(RateLimited) as caught:
|
|
client.satellites()
|
|
assert caught.value.retry_after == 0
|
|
assert len(server.requests) == 3 # the original plus max_retries
|
|
|
|
|
|
def test_a_500_is_not_retried(server, client):
|
|
"""It means ODMS took the request and broke on it. Repeating it breaks it
|
|
again, and turns one alert into three."""
|
|
server.reply(500, "boom")
|
|
|
|
with pytest.raises(HTTPError):
|
|
client.satellites()
|
|
assert len(server.requests) == 1
|
|
|
|
|
|
def test_a_503_is_retried_even_for_a_post(server, client):
|
|
"""503 proves the request was not acted on, so repeating it cannot
|
|
duplicate anything -- unlike a transport failure."""
|
|
server.reply(503, "busy")
|
|
server.reply(201, {"id": 7})
|
|
|
|
assert client.push_message("CCSDS_OPM_VERS = 3.0", key="iss") == {"id": 7}
|
|
assert len(server.requests) == 2
|
|
|
|
|
|
def test_a_transport_failure_never_repeats_a_post(server):
|
|
"""A push that died in flight may already have stored the message. Retrying
|
|
would store a second copy, and nothing downstream would notice."""
|
|
from odms.errors import TransportError
|
|
|
|
# Nothing is listening on this port, so every attempt fails in transport.
|
|
with OdmsClient("http://127.0.0.1:9", max_retries=3, timeout_s=0.5) as odms:
|
|
odms._transport.backoff_s = 0.0
|
|
odms._transport._session = None
|
|
attempts = []
|
|
original = odms._transport._send_urllib
|
|
|
|
def counting(*args, **kwargs):
|
|
attempts.append(1)
|
|
return original(*args, **kwargs)
|
|
|
|
odms._transport._send_urllib = counting
|
|
|
|
with pytest.raises(TransportError):
|
|
odms.push_message("CCSDS_OPM_VERS = 3.0", key="iss")
|
|
assert len(attempts) == 1
|
|
|
|
attempts.clear()
|
|
with pytest.raises(TransportError):
|
|
odms.satellites()
|
|
assert len(attempts) == 4 # a read may be repeated: 1 + max_retries
|