"""The commercial schedule services. None of these has been run against its live service, because each wants a paid key. What is tested is everything that can be: that a source with no key is skipped rather than tried, that each reader takes the documented shape of its answers and finds the leg in it, that the leg chosen is the one in the air at the moment being asked about, and that a service which has changed since costs a route rather than a scan. The recorded shapes below are written from each service's published documentation. If one of them stops matching reality, this is where it will show, and the failure will be a route that is not found rather than one that is wrong. """ import json import pytest from bandsaunter import schedules NOON = 1_788_600_000.0 # a moment to ask about def at(offset_hours: float) -> str: """A time, written the way these services write it.""" from datetime import datetime, timezone return datetime.fromtimestamp(NOON + offset_hours * 3600, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") # --------------------------------------------------------------------------- # Without keys, which is how nearly everyone runs # --------------------------------------------------------------------------- def test_every_source_says_what_it_needs(): for kind in schedules.SOURCES: source = kind() assert source.name and source.needs and source.signup assert all(n.startswith("BANDSAUNTER_") for n in source.needs) def test_a_source_with_no_key_is_not_available(monkeypatch): for kind in schedules.SOURCES: for name in kind.needs: monkeypatch.delenv(name, raising=False) assert kind().available() is False assert kind().missing() == list(kind.needs) def test_a_source_with_no_key_is_never_asked(monkeypatch): """Not asked, rather than asked and refused: there is nothing to ask with, and a request would only be a way of finding that out slowly.""" for kind in schedules.SOURCES: for name in kind.needs: monkeypatch.delenv(name, raising=False) def refuse(self, url, headers=None): raise AssertionError("asked a service with no key") monkeypatch.setattr(schedules.Schedule, "fetch", refuse) assert schedules.route_for("SWA930", NOON) is None def test_half_a_key_is_no_key(monkeypatch): """Cirium wants two; one of them is not enough.""" monkeypatch.setenv("BANDSAUNTER_CIRIUM_APP_ID", "abc") monkeypatch.delenv("BANDSAUNTER_CIRIUM_APP_KEY", raising=False) source = schedules.source_named("cirium") assert source.available() is False assert source.missing() == ["BANDSAUNTER_CIRIUM_APP_KEY"] def test_only_the_ones_with_keys_are_listed(monkeypatch): for kind in schedules.SOURCES: for name in kind.needs: monkeypatch.delenv(name, raising=False) assert schedules.available_sources() == [] monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") assert [s.name for s in schedules.available_sources()] == ["flightaware"] def test_they_are_asked_in_the_order_given(monkeypatch): monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") monkeypatch.setenv("BANDSAUNTER_OAG_KEY", "k") assert [s.name for s in schedules.available_sources(["oag", "flightaware"])] \ == ["oag", "flightaware"] def test_a_name_that_is_not_a_service_is_ignored(): assert schedules.source_named("nonesuch") is None assert schedules.available_sources(["nonesuch"]) == [] # --------------------------------------------------------------------------- # FlightAware AeroAPI # --------------------------------------------------------------------------- AEROAPI = {"flights": [ {"ident": "SWA930", "operator": "SWA", "origin": {"code_icao": "KLAS", "name": "Harry Reid International"}, "destination": {"code_icao": "KMDW", "name": "Chicago Midway"}, "scheduled_off": at(-9), "scheduled_on": at(-6)}, {"ident": "SWA930", "operator": "SWA", "origin": {"code_icao": "KMDW", "name": "Chicago Midway"}, "destination": {"code_icao": "KSAN", "name": "San Diego International"}, "actual_off": at(-1), "estimated_on": at(2)}, {"ident": "SWA930", "operator": "SWA", "origin": {"code_icao": "KSAN", "name": "San Diego International"}, "destination": {"code_icao": "KOAK", "name": "Oakland International"}, "scheduled_off": at(4), "scheduled_on": at(6)}, ]} def _read(source_name, body, when=NOON): """Read one service's answer, without needing a key to do it.""" return schedules.source_named(source_name).read(body, when) def test_flightaware_finds_the_leg_that_is_in_the_air(): """Three legs of one flight number in a day; the one being watched is the one whose window holds the moment.""" got = _read("flightaware", AEROAPI) assert got["origin_code"] == "KMDW" assert got["destination_code"] == "KSAN" assert "Midway" in got["origin"] assert got["stops"] == ("KMDW", "KSAN") def test_flightaware_picks_a_different_leg_at_a_different_hour(): early = _read("flightaware", AEROAPI, when=NOON - 8 * 3600) assert (early["origin_code"], early["destination_code"]) == ("KLAS", "KMDW") late = _read("flightaware", AEROAPI, when=NOON + 5 * 3600) assert (late["origin_code"], late["destination_code"]) == ("KSAN", "KOAK") def test_flightaware_says_nothing_about_a_day_it_has_no_flights_for(): assert _read("flightaware", AEROAPI, when=NOON + 5 * 86_400) is None def test_flightaware_survives_an_answer_it_does_not_recognise(): for body in ({}, {"flights": []}, {"flights": [{}]}, {"flights": [{"origin": None, "destination": None}]}, {"flights": [{"origin": {"code_icao": "KLAS"}}]}): assert _read("flightaware", body) is None # --------------------------------------------------------------------------- # Flightradar24 # --------------------------------------------------------------------------- FR24 = {"data": [ {"fr24_id": "1", "flight": "SWA930", "operating_as": "SWA", "orig_icao": "KMDW", "dest_icao": "KSAN", "datetime_takeoff": at(-1), "datetime_landed": at(2)}, {"fr24_id": "2", "flight": "SWA930", "operating_as": "SWA", "orig_icao": "KSAN", "dest_icao": "KOAK", "datetime_takeoff": at(4), "datetime_landed": at(6)}, ]} def test_flightradar24_finds_the_leg_in_the_air(): got = _read("flightradar24", FR24) assert (got["origin_code"], got["destination_code"]) == ("KMDW", "KSAN") def test_flightradar24_reads_a_bare_list_too(): """Some of its endpoints wrap the rows and some do not.""" got = _read("flightradar24", FR24["data"]) assert got is not None and got["origin_code"] == "KMDW" def test_flightradar24_survives_nonsense(): for body in ({}, {"data": []}, {"data": [{}]}, [], None): assert _read("flightradar24", body) is None # --------------------------------------------------------------------------- # OAG # --------------------------------------------------------------------------- OAG = {"data": [ {"carrier": {"icao": "SWA"}, "departure": {"airport": {"icao": "KMDW", "name": "Chicago Midway"}, "date": {"utc": at(-1)[:10]}, "time": {"utc": at(-1)[11:19]}}, "arrival": {"airport": {"icao": "KSAN", "name": "San Diego"}, "date": {"utc": at(2)[:10]}, "time": {"utc": at(2)[11:19]}}}, ]} def test_oag_finds_the_leg(): got = _read("oag", OAG) assert (got["origin_code"], got["destination_code"]) == ("KMDW", "KSAN") assert "Midway" in got["origin"] def test_oag_survives_nonsense(): for body in ({}, {"data": []}, {"data": [{}]}, {"data": [{"departure": {}, "arrival": {}}]}): assert _read("oag", body) is None def test_oag_needs_an_airline_callsign(monkeypatch): monkeypatch.setenv("BANDSAUNTER_OAG_KEY", "k") source = schedules.source_named("oag") with pytest.raises(schedules.SourceError): source.route("N517HP", NOON) # a registration, not a flight # --------------------------------------------------------------------------- # Cirium # --------------------------------------------------------------------------- CIRIUM = { "scheduledFlights": [ {"carrierFsCode": "WN", "flightNumber": "930", "departureAirportFsCode": "MDW", "arrivalAirportFsCode": "SAN", "departureTime": at(-1), "arrivalTime": at(2)}, ], "appendix": {"airports": [ {"fs": "MDW", "icao": "KMDW", "name": "Chicago Midway", "countryCode": "US", "latitude": 41.786, "longitude": -87.752}, {"fs": "SAN", "icao": "KSAN", "name": "San Diego International", "countryCode": "US", "latitude": 32.733, "longitude": -117.19}, ]}, } def test_cirium_finds_the_leg_and_where_its_airports_are(): got = _read("cirium", CIRIUM) assert (got["origin_code"], got["destination_code"]) == ("KMDW", "KSAN") assert got["origin_country"] == "US" assert got["origin_lat"] == pytest.approx(41.786) assert got["destination_lon"] == pytest.approx(-117.19) def test_cirium_survives_nonsense(): for body in ({}, {"scheduledFlights": []}, {"scheduledFlights": [{}]}, {"scheduledFlights": [{"departureAirportFsCode": "MDW"}], "appendix": {}}): assert _read("cirium", body) is None # --------------------------------------------------------------------------- # Reading the answers # --------------------------------------------------------------------------- def test_a_callsign_splits_into_an_airline_and_a_number(): assert schedules._split_callsign("SWA930") == ("SWA", "930") assert schedules._split_callsign("BAW49") == ("BAW", "49") assert schedules._split_callsign("N517HP")[1] == "" # a registration assert schedules._split_callsign("")[1] == "" @pytest.mark.parametrize("text,ok", [ ("2026-09-04T12:00:00Z", True), ("2026-09-04T12:00:00+00:00", True), ("1788600000", True), ("", False), ("not a time", False), ("2026-13-45", False), ]) def test_the_several_ways_a_time_is_written_are_all_read(text, ok): got = schedules._seconds(text) assert (got > 0) is ok def test_the_leg_whose_window_holds_the_moment_wins(): legs = [(NOON - 7200, NOON - 3600, {"n": "before"}), (NOON - 600, NOON + 600, {"n": "now"}), (NOON + 3600, NOON + 7200, {"n": "after"})] assert schedules._closest(legs, NOON)["n"] == "now" def test_the_nearest_leg_wins_when_none_quite_holds_it(): """A departure runs late and the windows no longer line up; the nearest is a better answer than none, and much better than the first in the list.""" legs = [(NOON - 7 * 3600, NOON - 6 * 3600, {"n": "long before"}), (NOON + 600, NOON + 3600, {"n": "just after"})] assert schedules._closest(legs, NOON)["n"] == "just after" def test_nothing_within_half_a_day_is_not_this_flight(): legs = [(NOON - 40 * 3600, NOON - 39 * 3600, {"n": "yesterday"})] assert schedules._closest(legs, NOON) is None assert schedules._closest([], NOON) is None def test_a_leg_needs_both_ends(): assert schedules._leg("KMDW", "") is None assert schedules._leg("", "KSAN") is None assert schedules._leg("kmdw", "ksan")["origin_code"] == "KMDW" def test_a_leg_comes_back_in_the_shape_the_rest_of_the_program_reads(): from bandsaunter.flights import Flight leg = schedules._leg("EGLL", "KSEA") fields = set(Flight().__dict__) for key in leg: if key in ("stops", "airline"): continue assert key in fields, key assert leg["origin_country"] == "GB" # worked out from the code assert leg["destination_country"] == "US" # --------------------------------------------------------------------------- # And what happens when one falls over # --------------------------------------------------------------------------- def test_a_service_that_fails_does_not_stop_the_next_one(monkeypatch): """A key that has run out of quota should cost that service and not the others.""" monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") monkeypatch.setenv("BANDSAUNTER_FR24_TOKEN", "k") def fetch(self, url, headers=None): if self.name == "flightaware": raise OSError("quota exceeded") return FR24 monkeypatch.setattr(schedules.Schedule, "fetch", fetch) got = schedules.route_for("SWA930", NOON, ["flightaware", "flightradar24"]) assert got is not None assert got["origin_code"] == "KMDW" assert got["source"] == "flightradar24" def test_the_answer_says_which_service_gave_it(monkeypatch): monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") monkeypatch.setattr(schedules.Schedule, "fetch", lambda self, url, headers=None: AEROAPI) got = schedules.route_for("SWA930", NOON, ["flightaware"]) assert got["source"] == "flightaware" def test_the_key_is_sent_the_way_the_service_wants_it(monkeypatch): """Each of them asks for its key somewhere different.""" seen = {} def fetch(self, url, headers=None): seen[self.name] = (url, headers or {}) return {} monkeypatch.setattr(schedules.Schedule, "fetch", fetch) monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "aero-key") monkeypatch.setenv("BANDSAUNTER_FR24_TOKEN", "fr24-token") monkeypatch.setenv("BANDSAUNTER_OAG_KEY", "oag-key") monkeypatch.setenv("BANDSAUNTER_CIRIUM_APP_ID", "cid") monkeypatch.setenv("BANDSAUNTER_CIRIUM_APP_KEY", "ckey") for source in schedules.available_sources(): source.route("SWA930", NOON) assert seen["flightaware"][1]["x-apikey"] == "aero-key" assert "fr24-token" in seen["flightradar24"][1]["Authorization"] assert seen["oag"][1]["Subscription-Key"] == "oag-key" assert "appId=cid" in seen["cirium"][0] assert "appKey=ckey" in seen["cirium"][0] # And no key is ever put in a URL that did not ask for one there. assert "aero-key" not in seen["flightaware"][0] assert "fr24-token" not in seen["flightradar24"][0] assert "oag-key" not in seen["oag"][0] def test_a_key_is_never_written_into_the_settings_file(): """A settings file gets copied between machines and pasted into messages asking for help; an API key does not belong in one.""" from bandsaunter.aircraft import AircraftOptions text = json.dumps(AircraftOptions().to_dict()) for kind in schedules.SOURCES: for name in kind.needs: assert name not in text def test_the_book_asks_the_services_before_the_free_databases(monkeypatch): """They know the leg; the free databases hold one route per number.""" import tempfile from pathlib import Path from bandsaunter.flights import FlightBook monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") monkeypatch.setattr(schedules.Schedule, "fetch", lambda self, url, headers=None: AEROAPI) asked = [] def free(self, url): asked.append(url) raise OSError("should not have been needed") monkeypatch.setattr(FlightBook, "_request", free) book = FlightBook(cache=Path(tempfile.mkdtemp()) / "c.json") entry = book.get("AC0FB6", "SWA930", when=NOON) book.wait(5.0) assert entry.origin_code == "KMDW" assert entry.route_source == "flightaware" assert not any("callsign" in url for url in asked), \ "asked a free database when a schedule service had answered" def test_with_no_keys_the_free_databases_answer_as_before(monkeypatch): import tempfile from pathlib import Path from bandsaunter.flights import FlightBook for kind in schedules.SOURCES: for name in kind.needs: monkeypatch.delenv(name, raising=False) answers = {"adsbdb.com/v0/callsign": {"response": {"flightroute": { "callsign": "SWA930", "origin": {"icao_code": "KHOU", "name": "Hobby", "country_iso_name": "US"}, "destination": {"icao_code": "KSAT", "name": "San Antonio", "country_iso_name": "US"}}}}} def request(self, url): for fragment, body in answers.items(): if fragment in url: return body raise OSError("not found") monkeypatch.setattr(FlightBook, "_request", request) book = FlightBook(cache=Path(tempfile.mkdtemp()) / "c.json") entry = book.get("AC0FB6", "SWA930", when=NOON) book.wait(5.0) assert entry.origin_code == "KHOU" assert entry.route_source == "" # --------------------------------------------------------------------------- # Two things that would go wrong quietly # --------------------------------------------------------------------------- def test_cirium_prefers_the_utc_time_over_the_local_one(): """Its plain departureTime is local and carries no offset, so reading that as UTC puts a leg up to half a day from where it belongs -- which is exactly far enough to pick the wrong leg of the same number.""" body = json.loads(json.dumps(CIRIUM)) flight = body["scheduledFlights"][0] flight["departureTimeUtc"] = at(-1) flight["arrivalTimeUtc"] = at(2) # The local times say the small hours, nine time zones away. flight["departureTime"] = at(-10)[:-1] flight["arrivalTime"] = at(-7)[:-1] got = _read("cirium", body) assert got is not None assert (got["origin_code"], got["destination_code"]) == ("KMDW", "KSAN") def test_a_reader_that_throws_is_a_source_that_does_not_know(monkeypatch): """An answer shaped differently from the documented one is the failure most likely to actually happen. It has to come back as this service not knowing, not as a traceback out of the middle of a scan.""" monkeypatch.setenv("BANDSAUNTER_AEROAPI_KEY", "k") source = schedules.source_named("flightaware") monkeypatch.setattr(type(source), "ask", lambda self, c, w: {"flights": 7}) with pytest.raises(schedules.SourceError): source.route("SWA930", NOON) # and the caller above it turns that into silence, not a crash monkeypatch.setattr(schedules.Schedule, "fetch", lambda self, url, headers=None: {"flights": 7}) assert schedules.route_for("SWA930", NOON) is None