335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""Offline browser and parser coverage, with no requests to Google."""
|
||
|
||
import base64
|
||
import json
|
||
import typing
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from urllib.parse import parse_qs, urlsplit
|
||
|
||
import pytest
|
||
from playwright.sync_api import Page, Response
|
||
|
||
from agenda import (
|
||
airport_lookup,
|
||
conference_flights,
|
||
flight_search_cache,
|
||
google_flights,
|
||
)
|
||
|
||
|
||
def itinerary(airline: str = "BA", price: int | None = 424) -> list[typing.Any]:
|
||
"""A Google itinerary with omitted zero minutes and overnight arrival."""
|
||
leg: list[typing.Any] = [None] * 23
|
||
leg[3], leg[6] = "LHR", "LAX"
|
||
leg[8], leg[10] = [23], [None, 30]
|
||
leg[20], leg[21] = [2027, 3, 30], [2027, 3, 31]
|
||
leg[22] = [airline, "135", "BA"]
|
||
details: list[typing.Any] = [None] * 10
|
||
details[2], details[9] = [leg], 670
|
||
return [details, [[None, price]] if price is not None else None]
|
||
|
||
|
||
def result_script() -> str:
|
||
"""Represent best and other groups, airline metadata, and an unpriced fare."""
|
||
payload: list[typing.Any] = [None] * 8
|
||
payload[2] = [[itinerary()]]
|
||
payload[3] = [[itinerary("AA", None)]]
|
||
payload[7] = [None, [[], [["BA", "British Airways"], ["AA", "American Airlines"]]]]
|
||
return (
|
||
"AF_initDataCallback({key: 'ds:1', data:"
|
||
+ json.dumps(payload)
|
||
+ ", sideChannel: {}});"
|
||
)
|
||
|
||
|
||
def test_results_and_sparse_times() -> None:
|
||
"""Best flights retain order; codeshares, missing fares and midnight parse."""
|
||
rows = google_flights.parse_results(result_script(), "https://example.com")
|
||
assert [row["price"] for row in rows] == [424, None]
|
||
assert rows[0]["departure"] == "2027-03-30T23:00:00"
|
||
assert rows[0]["arrival"] == "2027-03-31T00:30:00"
|
||
assert rows[0]["currency"] == "GBP"
|
||
assert rows[0]["duration"] == 670
|
||
assert rows[1]["legs"][0]["airline"] == "American Airlines"
|
||
assert rows[1]["legs"][0]["operating_airline_code"] == "BA"
|
||
assert conference_flights.flight_rank(rows[1]) == (0, 0)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"script", ["no data", "data:{}", "data:[null,null,[[[[]]]],null]"]
|
||
)
|
||
def test_unrecognized_data_is_error(script: str) -> None:
|
||
"""Broken parsing must not silently cache a successful empty search."""
|
||
with pytest.raises(ValueError):
|
||
google_flights.parse_results(script, "https://example.com")
|
||
|
||
|
||
def test_search_url() -> None:
|
||
"""London includes all six airports; locale, currency and date are explicit."""
|
||
url = google_flights.search_url("LON", "LAX", date(2027, 3, 30), 1)
|
||
query = parse_qs(urlsplit(url).query)
|
||
assert query["hl"] == ["en-GB"]
|
||
assert query["gl"] == ["GB"]
|
||
assert query["curr"] == ["GBP"]
|
||
encoded = query["tfs"][0]
|
||
data = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
|
||
assert all(
|
||
code.encode() in data for code in (*google_flights.LONDON_AIRPORTS, "LAX")
|
||
)
|
||
assert b"2027-03-30" in data
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"query,code",
|
||
[
|
||
("SFO", "SFO"),
|
||
("sfo", "SFO"),
|
||
("Copenhagen, Denmark", "CPH"),
|
||
("Sacramento, CA", "SMF"),
|
||
("London", "LON"),
|
||
],
|
||
)
|
||
def test_local_airports(query: str, code: str) -> None:
|
||
"""Exact codes avoid fuzzy matching; useful scheduled airports come first."""
|
||
assert airport_lookup.matches(query)[0]["code"] == code
|
||
assert conference_flights.resolve_airport(query)[0] == code
|
||
|
||
|
||
def test_unknown_iata_is_not_fuzzy() -> None:
|
||
"""Three letters must not accidentally resolve to another airport."""
|
||
assert airport_lookup.matches("ZZZ") == []
|
||
with pytest.raises(ValueError, match="Unknown IATA"):
|
||
conference_flights.resolve_airport("ZZZ")
|
||
|
||
|
||
def test_browser_429_stops_without_retry(
|
||
tmp_path: Path, monkeypatch: typing.Any
|
||
) -> None:
|
||
"""A page navigation's 429 blocks other searches without a second navigation."""
|
||
calls: list[str] = []
|
||
|
||
def goto(url: str, **kwargs: typing.Any) -> typing.Any:
|
||
calls.append(url)
|
||
return SimpleNamespace(status=429)
|
||
|
||
browser = google_flights.BrowserSearch(tmp_path)
|
||
monkeypatch.setattr(browser, "start", lambda: SimpleNamespace(goto=goto))
|
||
with pytest.raises(flight_search_cache.RateLimitCooldownError):
|
||
browser.search("LON", "LAX", date(2027, 3, 30), 1)
|
||
with pytest.raises(flight_search_cache.RateLimitCooldownError):
|
||
browser.search("LON", "LAX", date(2027, 3, 31), 1)
|
||
assert len(calls) == 1
|
||
assert flight_search_cache.cooldown_message(tmp_path)
|
||
|
||
|
||
def test_stops_fallback_and_ba_ranking(tmp_path: Path, monkeypatch: typing.Any) -> None:
|
||
"""Only fall back to two stops for London; BA preference keeps other airlines."""
|
||
calls: list[int] = []
|
||
|
||
def search(
|
||
origin: str, destination: str, day: date, max_stops: int
|
||
) -> list[dict[str, typing.Any]]:
|
||
calls.append(max_stops)
|
||
return [
|
||
{"stops": 2, "legs": [{"airline_code": "BA"}]},
|
||
{"stops": 1, "legs": [{"airline_code": "AA"}]},
|
||
{"stops": 1, "legs": [{"airline_code": "BA"}]},
|
||
{"stops": 0, "legs": [{"airline_code": "AA"}]},
|
||
]
|
||
|
||
monkeypatch.setattr(google_flights, "search", search)
|
||
rows = conference_flights.fetch_day(
|
||
"LON", "LAX", date(2027, 3, 30), False, tmp_path
|
||
)
|
||
assert calls == [1]
|
||
assert [(r["stops"], r["legs"][0]["airline_code"]) for r in rows] == [
|
||
(0, "AA"),
|
||
(1, "BA"),
|
||
(1, "AA"),
|
||
]
|
||
calls.clear()
|
||
|
||
def no_results(*args: typing.Any) -> list[dict[str, typing.Any]]:
|
||
calls.append(args[-1])
|
||
return []
|
||
|
||
monkeypatch.setattr(google_flights, "search", no_results)
|
||
assert (
|
||
conference_flights.fetch_day("LON", "LAX", date(2027, 3, 30), False, tmp_path)
|
||
== []
|
||
)
|
||
assert calls == [1, 2]
|
||
calls.clear()
|
||
assert (
|
||
conference_flights.fetch_day("BRS", "CPH", date(2027, 3, 30), True, tmp_path)
|
||
== []
|
||
)
|
||
assert calls == [0]
|
||
|
||
|
||
def test_data_request_429_uses_shared_cooldown(tmp_path: Path) -> None:
|
||
"""A 429 in the loaded page's Google data request pauses all searches too."""
|
||
browser = google_flights.BrowserSearch(tmp_path)
|
||
response = SimpleNamespace(
|
||
status=429, url="https://www.google.com/_/FlightsUi/data/batchexecute"
|
||
)
|
||
browser.check_response(typing.cast(Response, response))
|
||
with pytest.raises(flight_search_cache.RateLimitCooldownError):
|
||
browser.check_cooldown()
|
||
assert (
|
||
flight_search_cache.read_state(tmp_path / "rate-limit.json")["transport"]
|
||
== flight_search_cache.TRANSPORT
|
||
)
|
||
|
||
|
||
def test_browser_errors_show_diagnostic_stderr() -> None:
|
||
"""Long Playwright launch commands must not hide Chromium's actual failure."""
|
||
exc = RuntimeError(
|
||
"BrowserType.launch_persistent_context: Target closed\nBrowser logs:\n"
|
||
"<launching> /usr/bin/chromium "
|
||
+ "--long-option " * 200
|
||
+ "\n[pid=123][err] grep: /proc/cpuinfo: No such file or directory\n"
|
||
"[pid=123][err] The hardware lacks SSE3 support.\n"
|
||
)
|
||
details = conference_flights.error_detail(exc)
|
||
assert "Target closed" in details
|
||
assert "/proc/cpuinfo: No such file or directory" in details
|
||
assert "SSE3" in details
|
||
assert "--long-option" not in details
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"query", ["Malmö", "Malmo", "Malmö, Sweden", "MMX,CPH", "mmx, cph"]
|
||
)
|
||
def test_malmo_airport_group(query: str) -> None:
|
||
"""Malmö includes nearby Copenhagen across the border; explicit codes still work."""
|
||
assert conference_flights.resolve_airport(query)[0] == "MMX,CPH"
|
||
assert airport_lookup.matches(query)[0]["code"] == "MMX,CPH"
|
||
assert conference_flights.resolve_airport("MMX")[0] == "MMX"
|
||
assert conference_flights.resolve_airport("CPH")[0] == "CPH"
|
||
|
||
|
||
@pytest.mark.parametrize("origin,destination", [("BRS", "MMX,CPH"), ("MMX,CPH", "BRS")])
|
||
def test_combined_airport_search_url(origin: str, destination: str) -> None:
|
||
"""Outbound and return searches include each airport as its own protobuf entry."""
|
||
query = parse_qs(
|
||
urlsplit(
|
||
google_flights.search_url(origin, destination, date(2026, 10, 19), 0)
|
||
).query
|
||
)
|
||
encoded = query["tfs"][0]
|
||
data = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
|
||
assert b"MMX,CPH" not in data
|
||
assert all(code.encode() in data for code in ("MMX", "CPH", "BRS"))
|
||
assert data.count(b"MMX") == data.count(b"CPH") == 1
|
||
assert b"\x28\x00" in data # Explicit non-stop filter.
|
||
|
||
|
||
def test_bristol_copenhagen_serves_malmo(
|
||
tmp_path: Path, monkeypatch: typing.Any
|
||
) -> None:
|
||
"""CPH flights satisfy the Bristol preference without searching London or MMX separately."""
|
||
calls: list[tuple[str, str, bool]] = []
|
||
|
||
def search(
|
||
origin: str, destination: str, day: date, direct: bool
|
||
) -> list[dict[str, typing.Any]]:
|
||
calls.append((origin, destination, direct))
|
||
return [
|
||
{
|
||
"arrival": day.isoformat() + "T18:00:00",
|
||
"legs": [
|
||
{
|
||
"origin": "BRS" if origin == "BRS" else "CPH",
|
||
"destination": "CPH" if origin == "BRS" else "BRS",
|
||
}
|
||
],
|
||
}
|
||
]
|
||
|
||
monkeypatch.setattr(conference_flights, "search_day", search)
|
||
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
||
path = conference_flights.cache_path(str(tmp_path), start, end, "MMX,CPH", True)
|
||
result = conference_flights.lookup(path, start, end, "MMX,CPH", True)
|
||
assert calls == [("BRS", "MMX,CPH", True), ("MMX,CPH", "BRS", True)]
|
||
assert [row["origin"] for row in result["searches"]] == ["BRS"]
|
||
|
||
|
||
@pytest.mark.parametrize("query", ["Bern", "Bern, Switzerland", "BRN,BSL"])
|
||
def test_bern_includes_basel(query: str) -> None:
|
||
"""Bern's destination group includes Basel, while BSL alone stays explicit."""
|
||
code, name = conference_flights.resolve_airport(query)
|
||
assert code == "BRN,BSL"
|
||
assert "Basel" in name
|
||
assert conference_flights.resolve_airport("BSL")[0] == "BSL"
|
||
|
||
|
||
@pytest.mark.parametrize("query", ["Bonn", "Bonn, Germany", "CGN,DUS"])
|
||
def test_bonn_includes_dusseldorf(query: str) -> None:
|
||
"""Bonn uses Cologne/Bonn and Düsseldorf; DUS remains individually selectable."""
|
||
code, name = conference_flights.resolve_airport(query)
|
||
assert code == "CGN,DUS"
|
||
assert "Düsseldorf" in name
|
||
assert conference_flights.resolve_airport("DUS")[0] == "DUS"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"script",
|
||
[
|
||
'data:["Requested flight date is too far in the future."]',
|
||
'data:{"error":"Dates are too far in the future"}',
|
||
],
|
||
)
|
||
def test_provider_future_date_error(script: str) -> None:
|
||
"""Explicit provider date rejections aren't mistaken for unknown result formats."""
|
||
with pytest.raises(
|
||
google_flights.FlightDateUnavailableError, match="too far in the future"
|
||
):
|
||
google_flights.parse_results(script, "https://example.com")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"text,expected",
|
||
[
|
||
("Requested flight date is too far in the future.", "future"),
|
||
("No flights found", "empty"),
|
||
(
|
||
"No non-stop flights found There might not be daily non-stop flights to Bologna (BLQ). Try changing your dates, or search flights with more stops.",
|
||
"empty",
|
||
),
|
||
("No nonstop flights found", "empty"),
|
||
("No non‑stop flights found", "empty"),
|
||
("No direct flights found", "empty"),
|
||
("Something went wrong", "error"),
|
||
],
|
||
)
|
||
def test_unusable_page_data(tmp_path: Path, text: str, expected: str) -> None:
|
||
"""Only confirmed empty pages become empty results; unrelated failures remain errors."""
|
||
page = SimpleNamespace(
|
||
locator=lambda selector: SimpleNamespace(inner_text=lambda: text)
|
||
)
|
||
browser = google_flights.BrowserSearch(tmp_path)
|
||
args = (
|
||
typing.cast(Page, page),
|
||
"data:[]",
|
||
"https://example.com",
|
||
date(2027, 9, 12),
|
||
)
|
||
if expected == "future":
|
||
with pytest.raises(google_flights.FlightDateUnavailableError):
|
||
browser.read_results(*args)
|
||
elif expected == "empty":
|
||
assert browser.read_results(*args) == []
|
||
else:
|
||
with pytest.raises(ValueError, match="12 Sep 2027"):
|
||
browser.read_results(*args)
|
||
|
||
|
||
@pytest.mark.parametrize("query", ["Funen", "Funen, Denmark", "BLL,CPH"])
|
||
def test_funen_includes_copenhagen(query: str) -> None:
|
||
"""Funen considers both Billund and Copenhagen in a single search."""
|
||
code, name = conference_flights.resolve_airport(query)
|
||
assert code == "BLL,CPH"
|
||
assert "Billund" in name and "Copenhagen" in name
|