587 lines
23 KiB
Python
587 lines
23 KiB
Python
"""Conference detail pages, airport resolution and flight caching."""
|
|
|
|
import typing
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
import agenda.conference_detail as detail
|
|
import agenda.conference_flights as flights
|
|
import agenda.conference_list
|
|
import agenda.conference_page as page
|
|
import agenda.flight_search_cache as cache
|
|
import agenda.trip
|
|
import web_view
|
|
from agenda.types import StrDict
|
|
|
|
|
|
def test_airport_suggestions(tmp_path: Path) -> None:
|
|
"""Mappings can be overridden; proximity stays within the conference country."""
|
|
conf: StrDict = {"location": "Brussels", "country": "be"}
|
|
assert detail.suggest_airport(conf, str(tmp_path))[0] == "BRU"
|
|
(tmp_path / "conference_airports.yaml").write_text("be:brussels: CRL\n")
|
|
assert detail.suggest_airport(conf, str(tmp_path))[0] == "CRL"
|
|
(tmp_path / "airports.yaml").write_text(
|
|
yaml.safe_dump(
|
|
{
|
|
"AAA": {
|
|
"city": "Elsewhere",
|
|
"country": "fr",
|
|
"latitude": 50,
|
|
"longitude": 4,
|
|
},
|
|
"BBB": {
|
|
"city": "Another",
|
|
"country": "be",
|
|
"latitude": 50.1,
|
|
"longitude": 4.1,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
assert (
|
|
detail.suggest_airport(
|
|
{"location": "Town", "country": "be", "latitude": 50, "longitude": 4},
|
|
str(tmp_path),
|
|
)[0]
|
|
== "BBB"
|
|
)
|
|
assert detail.suggest_airport({"location": "TBC"}, str(tmp_path))[0] is None
|
|
|
|
|
|
def test_short_haul() -> None:
|
|
"""European geography determines Bristol preference, with a distance limit."""
|
|
assert detail.is_short_haul({"country": "BE"})
|
|
assert not detail.is_short_haul({"country": "us"})
|
|
assert not detail.is_short_haul({"country": "fr", "latitude": -21, "longitude": 55})
|
|
|
|
|
|
def test_page_is_cache_only(tmp_path: Path, monkeypatch: typing.Any) -> None:
|
|
"""GET displays venue/details without initiating a search; POST requires a token."""
|
|
conf: StrDict = {
|
|
"name": "Mapping 2099",
|
|
"location": "Brussels",
|
|
"country": "be",
|
|
"topic": "Maps",
|
|
"start": date(2099, 5, 5),
|
|
"end": date(2099, 5, 6),
|
|
"venue": "Test Venue",
|
|
"address": "123 Test Street",
|
|
"latitude": 50.8,
|
|
"longitude": 4.3,
|
|
"url": "https://example.com/conference",
|
|
"description": "A great conference",
|
|
}
|
|
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump([conf]))
|
|
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
monkeypatch.setitem(web_view.app.config, "DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
monkeypatch.setattr(
|
|
flights, "lookup", lambda *args: pytest.fail("GET made a lookup")
|
|
)
|
|
monkeypatch.setattr(flights, "valid_airport", lambda code: code == "BRU")
|
|
url = f"/conference/2099-05-05/{detail.conference_slug(conf)}"
|
|
with web_view.app.test_client() as client:
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
for value in (
|
|
b"Test Venue",
|
|
b"123 Test Street",
|
|
b"View venue on map",
|
|
b"A great conference",
|
|
b"Look up flights",
|
|
b"Conference website",
|
|
):
|
|
assert value in response.data
|
|
assert client.post(url, data={"airport": "BRU"}).status_code == 400
|
|
assert client.get("/conference/2099-05-05/missing").status_code == 404
|
|
cached: StrDict = {
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"destination": "BRU",
|
|
"searches": [],
|
|
}
|
|
path = flights.cache_path(
|
|
str(tmp_path), conf["start"], conf["end"], "BRU", True
|
|
)
|
|
path.parent.mkdir()
|
|
flights.write_cache(path, cached)
|
|
response = client.get(url)
|
|
assert b"Last updated" in response.data and b"Update flights" in response.data
|
|
with client.session_transaction() as session:
|
|
token = session["conference_flight_token"]
|
|
calls: list[str] = []
|
|
|
|
def lookup(*args: typing.Any) -> StrDict:
|
|
calls.append(args[3])
|
|
return cached
|
|
|
|
monkeypatch.setattr(flights, "lookup", lookup)
|
|
assert (
|
|
client.post(url, data={"airport": "BRU", "csrf_token": token}).status_code
|
|
== 302
|
|
)
|
|
assert calls == ["BRU"]
|
|
monkeypatch.setattr(
|
|
flights,
|
|
"resolve_airport",
|
|
lambda query: ("CPH", "Copenhagen Kastrup Airport"),
|
|
)
|
|
response = client.post(
|
|
url + "?airport=BRU",
|
|
data={"airport": "Copenhagen, Denmark", "csrf_token": token},
|
|
)
|
|
assert response.status_code == 302
|
|
assert response.headers["Location"].endswith("airport=CPH")
|
|
assert calls == ["BRU", "CPH"]
|
|
|
|
|
|
def test_slug_collision(tmp_path: Path) -> None:
|
|
"""Events of one series on one day receive distinct reproducible URLs."""
|
|
conferences = [
|
|
{
|
|
"name": name,
|
|
"series": "maps",
|
|
"location": "Berlin",
|
|
"start": date(2099, 5, 1),
|
|
}
|
|
for name in ("Maps morning", "Maps evening")
|
|
]
|
|
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump(conferences))
|
|
rows = agenda.conference_list.build_conference_list(str(tmp_path), [])
|
|
assert rows[0]["slug"] != rows[1]["slug"]
|
|
assert (
|
|
detail.conference_slug({"name": "FOSDEM", "location": "Brussels"})
|
|
== "fosdem-brussels"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"short_haul,bristol_complete,expected",
|
|
[(True, True, ["BRS"]), (True, False, ["BRS", "LON"]), (False, True, ["LON"])],
|
|
)
|
|
def test_cache_and_fallback(
|
|
tmp_path: Path,
|
|
monkeypatch: typing.Any,
|
|
short_haul: bool,
|
|
bristol_complete: bool,
|
|
expected: list[str],
|
|
) -> None:
|
|
"""Bristol preference, London fallback, and refresh throttling."""
|
|
calls: list[str] = []
|
|
|
|
def search(
|
|
origin: str, destination: str, start: date, end: date, flexible: bool
|
|
) -> StrDict:
|
|
calls.append(origin)
|
|
return {
|
|
"origin": origin,
|
|
"outbound": [{"price": 50}],
|
|
"inbound": [{"price": 60}] if bristol_complete else [],
|
|
}
|
|
|
|
monkeypatch.setattr(flights, "search_origin", search)
|
|
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
|
path = flights.cache_path(str(tmp_path), start, end, "BRU", short_haul)
|
|
result = flights.lookup(path, start, end, "BRU", short_haul)
|
|
assert calls == expected
|
|
assert flights.lookup(path, start, end, "BRU", short_haul) == result
|
|
assert calls == expected
|
|
|
|
|
|
def test_failed_refresh_keeps_results(tmp_path: Path, monkeypatch: typing.Any) -> None:
|
|
"""Failures retain old results and get a short retry cooldown."""
|
|
path = flights.cache_path(
|
|
str(tmp_path), date(2099, 1, 5), date(2099, 1, 6), "BRU", True
|
|
)
|
|
path.parent.mkdir()
|
|
old: StrDict = {
|
|
"updated_at": (datetime.now(timezone.utc) - timedelta(days=1)).isoformat(),
|
|
"destination": "BRU",
|
|
"searches": [{"origin": "BRS", "outbound": [1], "inbound": [2]}],
|
|
}
|
|
flights.write_cache(path, old)
|
|
calls: list[bool] = []
|
|
|
|
def fail(*args: typing.Any, **kwargs: typing.Any) -> StrDict:
|
|
calls.append(True)
|
|
raise RuntimeError("network unavailable")
|
|
|
|
monkeypatch.setattr(flights, "search_origin", fail)
|
|
result = flights.lookup(path, date(2099, 1, 5), date(2099, 1, 6), "BRU", True)
|
|
assert result["updated_at"] == old["updated_at"]
|
|
assert result["searches"] == old["searches"]
|
|
assert result["error"]
|
|
assert "RuntimeError: network unavailable" in result["error_details"]
|
|
flights.lookup(path, date(2099, 1, 5), date(2099, 1, 6), "BRU", True)
|
|
assert len(calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"outbound_offset,inbound_offset,expected_calls",
|
|
[(1, 1, 2), (3, 1, 4), (1, 3, 4), (None, None, 8)],
|
|
)
|
|
def test_flexible_direct_dates(
|
|
monkeypatch: typing.Any,
|
|
outbound_offset: int | None,
|
|
inbound_offset: int | None,
|
|
expected_calls: int,
|
|
) -> None:
|
|
"""Each direction expands only as needed, excluding late outbound arrivals."""
|
|
calls: list[tuple[str, str, date, bool]] = []
|
|
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
|
|
|
def search(origin: str, destination: str, day: date, direct: bool) -> list[StrDict]:
|
|
calls.append((origin, destination, day, direct))
|
|
if origin == "BRS":
|
|
offset = (start - day).days
|
|
if offset == 1 and outbound_offset != 1:
|
|
return [{"arrival": start.isoformat() + "T18:00:00", "price": 30}]
|
|
available = offset == outbound_offset
|
|
else:
|
|
available = (day - end).days == inbound_offset
|
|
return (
|
|
[{"arrival": day.isoformat() + "T18:00:00", "price": 30}]
|
|
if available
|
|
else []
|
|
)
|
|
|
|
monkeypatch.setattr(flights, "search_day", search)
|
|
result = flights.search_origin("BRS", "BRU", start, end, True)
|
|
assert len(calls) == expected_calls
|
|
assert all(call[3] for call in calls)
|
|
assert bool(result["outbound"]) == (outbound_offset is not None)
|
|
assert bool(result["inbound"]) == (inbound_offset is not None)
|
|
assert sum(call[0] == "BRS" for call in calls) == (outbound_offset or 4)
|
|
assert sum(call[0] == "BRU" for call in calls) == (inbound_offset or 4)
|
|
|
|
|
|
def test_bad_cache(tmp_path: Path) -> None:
|
|
"""Invalid JSON and naive timestamps are ignored."""
|
|
path = tmp_path / "cache.json"
|
|
path.write_text('{"updated_at": "2026-01-01T12:00:00", "searches": []}')
|
|
assert flights.read_cache(path) is None
|
|
path.write_text("bad json")
|
|
assert flights.read_cache(path) is None
|
|
|
|
|
|
def test_airport_autocomplete(monkeypatch: typing.Any) -> None:
|
|
"""Autocomplete uses the local index and returns airport names and codes."""
|
|
queries: list[str] = []
|
|
|
|
def matches(query: str) -> list[StrDict]:
|
|
queries.append(query)
|
|
return [{"code": "CPH", "name": "Copenhagen Kastrup Airport"}]
|
|
|
|
monkeypatch.setattr(flights, "airport_matches", matches)
|
|
with web_view.app.test_client() as client:
|
|
assert client.get("/conference/airports?q=C").json == []
|
|
response = client.get("/conference/airports?q=Copenhagen%2C%20Denmark")
|
|
assert response.json == [{"code": "CPH", "name": "Copenhagen Kastrup Airport"}]
|
|
assert queries == ["Copenhagen, Denmark"]
|
|
|
|
|
|
def test_multi_airport_location_mapping(tmp_path: Path) -> None:
|
|
"""Mappings support airport lists, accents, and personal overrides."""
|
|
for location in ("Malmö", "Malmo"):
|
|
assert (
|
|
detail.suggest_airport(
|
|
{"country": "se", "location": location}, str(tmp_path)
|
|
)[0]
|
|
== "MMX,CPH"
|
|
)
|
|
(tmp_path / "conference_airports.yaml").write_text('"se:malmo": [CPH, MMX]\n')
|
|
assert (
|
|
detail.suggest_airport({"country": "se", "location": "Malmö"}, str(tmp_path))[0]
|
|
== "CPH,MMX"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("available_offset", [2, 3, None])
|
|
def test_london_earlier_departure(
|
|
monkeypatch: typing.Any, available_offset: int | None
|
|
) -> None:
|
|
"""London expands the outbound by one day, without repeating the return search."""
|
|
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
|
calls: list[tuple[str, date]] = []
|
|
|
|
def search(origin: str, destination: str, day: date, direct: bool) -> list[StrDict]:
|
|
calls.append((origin, day))
|
|
if origin == "LON" and (start - day).days != available_offset:
|
|
return []
|
|
return [{"arrival": day.isoformat() + "T18:00:00"}]
|
|
|
|
monkeypatch.setattr(flights, "search_day", search)
|
|
result = flights.search_origin("LON", "BKW", start, end, False)
|
|
expected = ["2099-01-03"] if available_offset == 2 else ["2099-01-03", "2099-01-02"]
|
|
assert result["outbound_dates"] == expected
|
|
assert result["inbound_dates"] == ["2099-01-07"]
|
|
assert sum(origin == "BKW" for origin, _ in calls) == 1
|
|
assert bool(result["outbound"]) == (available_offset is not None)
|
|
|
|
|
|
def test_legacy_empty_search_can_expand(
|
|
tmp_path: Path, monkeypatch: typing.Any
|
|
) -> None:
|
|
"""An old fresh empty London cache can immediately try the new earlier date."""
|
|
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
|
path = flights.cache_path(str(tmp_path), start, end, "BKW", False)
|
|
path.parent.mkdir()
|
|
flights.write_cache(
|
|
path,
|
|
{
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"searches": [{"origin": "LON", "outbound": [], "inbound": []}],
|
|
"transport": cache.TRANSPORT,
|
|
},
|
|
)
|
|
calls: list[date] = []
|
|
|
|
def search(origin: str, destination: str, day: date, direct: bool) -> list[StrDict]:
|
|
calls.append(day)
|
|
return []
|
|
|
|
monkeypatch.setattr(flights, "search_day", search)
|
|
result = flights.lookup(path, start, end, "BKW", False)
|
|
assert result["searches"][0]["outbound_dates"] == ["2099-01-03", "2099-01-02"]
|
|
assert len(calls) == 3
|
|
assert flights.lookup(path, start, end, "BKW", False) == result
|
|
assert len(calls) == 3
|
|
|
|
|
|
def test_empty_flights_show_airport_and_dates(
|
|
tmp_path: Path, monkeypatch: typing.Any
|
|
) -> None:
|
|
"""Cached empty results show full airport names and checked departure dates."""
|
|
conf: StrDict = {
|
|
"name": "Mapping 2099",
|
|
"location": "Raleigh",
|
|
"country": "us",
|
|
"start": date(2099, 5, 5),
|
|
"end": date(2099, 5, 6),
|
|
}
|
|
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump([conf]))
|
|
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
monkeypatch.setitem(web_view.app.config, "DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
path = flights.cache_path(str(tmp_path), conf["start"], conf["end"], "BKW", False)
|
|
path.parent.mkdir()
|
|
cached: StrDict = {
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"destination": "BKW",
|
|
"searches": [
|
|
{
|
|
"origin": "LON",
|
|
"outbound": [],
|
|
"inbound": [],
|
|
"outbound_dates": ["2099-05-03", "2099-05-02"],
|
|
"inbound_dates": ["2099-05-07"],
|
|
}
|
|
],
|
|
}
|
|
flights.write_cache(path, cached)
|
|
url = "/conference/2099-05-05/" + detail.conference_slug(conf) + "?airport=BKW"
|
|
with web_view.app.test_client() as client:
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert b"Raleigh County Memorial Airport (BKW)</h3>" in response.data
|
|
for day in ("2099-05-03", "2099-05-02", "2099-05-07"):
|
|
assert page.format_date(day).encode() in response.data
|
|
# Legacy cache data reconstructs the single London departure date.
|
|
del cached["searches"][0]["outbound_dates"]
|
|
del cached["searches"][0]["inbound_dates"]
|
|
flights.write_cache(path, cached)
|
|
response = client.get(url)
|
|
assert page.format_date("2099-05-03").encode() in response.data
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"location,online,eligible",
|
|
[("Belfast", False, True), ("Belfast", True, False), ("London", False, False)],
|
|
)
|
|
def test_belfast_domestic_exception(
|
|
tmp_path: Path, monkeypatch: typing.Any, location: str, online: bool, eligible: bool
|
|
) -> None:
|
|
"""Belfast alone permits UK flight lookups; online events remain excluded."""
|
|
conf: StrDict = {
|
|
"name": "Test conference",
|
|
"location": location,
|
|
"country": "gb",
|
|
"start": date(2099, 5, 5),
|
|
"end": date(2099, 5, 6),
|
|
"online": online,
|
|
}
|
|
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump([conf]))
|
|
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
monkeypatch.setitem(web_view.app.config, "DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
calls: list[tuple[str, bool]] = []
|
|
|
|
def lookup(
|
|
path: Path, start: date, end: date, destination: str, short_haul: bool
|
|
) -> StrDict:
|
|
calls.append((destination, short_haul))
|
|
return {}
|
|
|
|
monkeypatch.setattr(flights, "lookup", lookup)
|
|
url = "/conference/2099-05-05/" + detail.conference_slug(conf)
|
|
with web_view.app.test_client() as client:
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert (b"Look up flights" in response.data) == eligible
|
|
assert calls == []
|
|
if eligible:
|
|
assert b"BFS,BHD" in response.data
|
|
with client.session_transaction() as session:
|
|
token = session["conference_flight_token"]
|
|
response = client.post(url, data={"airport": location, "csrf_token": token})
|
|
assert response.status_code == (302 if eligible else 400)
|
|
assert calls == ([("BFS,BHD", True)] if eligible else [])
|
|
|
|
|
|
def test_personal_airport_names(tmp_path: Path, monkeypatch: typing.Any) -> None:
|
|
"""Personal names apply to codes, groups and autocomplete, with live updates."""
|
|
path = tmp_path / "airports.yaml"
|
|
path.write_text(
|
|
yaml.safe_dump({"BSL": {"name": "My Basel Airport"}, "BRN": {"name": ""}})
|
|
)
|
|
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
with web_view.app.app_context():
|
|
assert flights.resolve_airport("BSL") == ("BSL", "My Basel Airport")
|
|
code, name = flights.resolve_airport("Bern")
|
|
assert code == "BRN,BSL"
|
|
assert name == "Bern Airport / My Basel Airport"
|
|
assert flights.resolve_airport("BRN,BSL")[1] == name
|
|
assert flights.airport_matches("My Basel Airport")[0] == {
|
|
"code": "BSL",
|
|
"name": "My Basel Airport",
|
|
}
|
|
assert any(
|
|
match["code"] == "BSL" for match in flights.airport_matches("EuroAirport")
|
|
)
|
|
with web_view.app.test_client() as client:
|
|
assert client.get("/conference/airports?q=BSL").json == [
|
|
{"code": "BSL", "name": "My Basel Airport"}
|
|
]
|
|
path.write_text(yaml.safe_dump({"BSL": {"name": "Renamed Basel Airport"}}))
|
|
assert client.get("/conference/airports?q=BSL").json == [
|
|
{"code": "BSL", "name": "Renamed Basel Airport"}
|
|
]
|
|
path.unlink()
|
|
fallback = client.get("/conference/airports?q=BSL").json
|
|
assert isinstance(fallback, list)
|
|
assert "EuroAirport" in fallback[0]["name"]
|
|
|
|
|
|
def test_date_limit_banner(tmp_path: Path, monkeypatch: typing.Any) -> None:
|
|
"""A wrapped date rejection asks to retry nearer departure, keeping cached fares."""
|
|
from agenda.google_flights import FlightDateUnavailableError
|
|
|
|
start, end = date(2099, 1, 5), date(2099, 1, 6)
|
|
path = flights.cache_path(str(tmp_path), start, end, "AKJ", False)
|
|
path.parent.mkdir()
|
|
old: StrDict = {
|
|
"updated_at": (datetime.now(timezone.utc) - timedelta(days=1)).isoformat(),
|
|
"destination": "AKJ",
|
|
"searches": [{"origin": "LON", "outbound": [{"price": 100}], "inbound": []}],
|
|
}
|
|
flights.write_cache(path, old)
|
|
|
|
def fail(*args: typing.Any, **kwargs: typing.Any) -> StrDict:
|
|
try:
|
|
raise FlightDateUnavailableError(
|
|
"Requested flight date is too far in the future. Please try again closer to departure."
|
|
)
|
|
except FlightDateUnavailableError as exc:
|
|
raise RuntimeError("AKJ → LON on 2099-01-07") from exc
|
|
|
|
monkeypatch.setattr(flights, "search_origin", fail)
|
|
result = flights.lookup(path, start, end, "AKJ", False)
|
|
assert "closer to departure" in result["error"]
|
|
assert "15 minutes" not in result["error"]
|
|
assert result["searches"] == old["searches"]
|
|
assert result["updated_at"] == old["updated_at"]
|
|
|
|
|
|
def test_previous_parser_error_can_retry_once(
|
|
tmp_path: Path, monkeypatch: typing.Any
|
|
) -> None:
|
|
"""The parser fix unlocks a recent old-format error, but fresh errors still throttle."""
|
|
start, end = date(2099, 6, 2), date(2099, 6, 4)
|
|
path = flights.cache_path(str(tmp_path), start, end, "BLQ", True)
|
|
path.parent.mkdir()
|
|
flights.write_cache(
|
|
path,
|
|
{
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"searches": [],
|
|
"error": "Flight lookup failed",
|
|
"error_details": "ValueError: Unrecognized Google Flights result format",
|
|
"transport": cache.TRANSPORT,
|
|
},
|
|
)
|
|
calls: list[bool] = []
|
|
|
|
def fail(*args: typing.Any, **kwargs: typing.Any) -> StrDict:
|
|
calls.append(True)
|
|
raise ValueError("Unrecognized Google Flights result format")
|
|
|
|
monkeypatch.setattr(flights, "search_origin", fail)
|
|
refreshed = flights.lookup(path, start, end, "BLQ", True)
|
|
assert len(calls) == 1
|
|
assert flights.lookup(path, start, end, "BLQ", True) == refreshed
|
|
assert len(calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status,eligible", [("exact", True), ("tentative", True), ("approximate", False)]
|
|
)
|
|
def test_flight_lookup_with_tentative_dates(
|
|
tmp_path: Path, monkeypatch: typing.Any, status: str, eligible: bool
|
|
) -> None:
|
|
"""Tentative start/end dates support lookup and links; approximate ranges don't."""
|
|
start, end = date(2099, 6, 2), date(2099, 6, 4)
|
|
dates: StrDict = {"status": status}
|
|
dates.update(
|
|
{"earliest": start, "latest": end}
|
|
if status == "approximate"
|
|
else {"start": start, "end": end}
|
|
)
|
|
conf: StrDict = {
|
|
"name": "PyCon test",
|
|
"location": "Bologna",
|
|
"country": "it",
|
|
"dates": dates,
|
|
}
|
|
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump([conf]))
|
|
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
monkeypatch.setitem(web_view.app.config, "DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
calls: list[tuple[date, date, str]] = []
|
|
|
|
def lookup(
|
|
path: Path, departure: date, returning: date, destination: str, short_haul: bool
|
|
) -> StrDict:
|
|
calls.append((departure, returning, destination))
|
|
return {}
|
|
|
|
monkeypatch.setattr(flights, "lookup", lookup)
|
|
url = "/conference/2099-06-02/" + detail.conference_slug(conf)
|
|
with web_view.app.test_client() as client:
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert (b"Look up flights" in response.data) == eligible
|
|
assert (b"Open in Google Flights" in response.data) == eligible
|
|
assert (b"Using tentative conference dates" in response.data) == (
|
|
status == "tentative"
|
|
)
|
|
if status == "tentative":
|
|
assert page.format_date(start).encode() in response.data
|
|
assert page.format_date(end).encode() in response.data
|
|
assert calls == []
|
|
with client.session_transaction() as session:
|
|
token = session["conference_flight_token"]
|
|
response = client.post(url, data={"airport": "BLQ", "csrf_token": token})
|
|
assert response.status_code == (302 if eligible else 400)
|
|
assert calls == ([(start, end, "BLQ")] if eligible else [])
|