From 1dcb7dadb8eff80cc7670d67382a18751051b3f3 Mon Sep 17 00:00:00 2001 From: ak2k <19240940+ak2k@users.noreply.github.com> Date: Sun, 17 May 2026 20:57:25 -0400 Subject: [PATCH] Multi-cabin search: --cabin economy,business (work-7ecx) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends `--cabin` to accept a CSV; fans out N parallel single-cabin queries (one per cabin) on both Matrix and gflight; client-side joins itineraries on per-slice (flight-number, date); renders one row per itinerary with one $ column per cabin. '—' marks cabins where the itinerary fell outside that cabin's top-N. New `--sort ` picks the ordering column (default: first cabin in --cabin). Domain stays single-cabin — multi-cabin lives in the orchestration layer, so wire/links/fli_bridge don't need to change. The single-cabin code path is preserved bit-for-bit; only multi-cabin invocations (`len(cabins) > 1`) take the new orchestrator. PP overlay: when --cabin is multi and the user hasn't explicitly set `--provider-opt pp.cabins=`, the PP cabin set is auto-derived from --cabin. Business in --cabin implies First gets added to the PP query — award seekers treat business/first as a paired premium tier, and First availability is rare enough that surfacing it costs nothing while filling a real research gap. The reverse promotion isn't applied: asking for First is an explicit choice we don't second-guess. Caveats noted in two follow-up bd items: - work-r9ht: internal N-bump to widen cabin overlap in the join. - work-338g: unified cash+PP super-table render (PP stays per-leg today). Tests: 33 new (CSV parsing, key/merge/sort, PP cabin derivation). Full gate green: 311 passing. --- src/flight_cli/_multi_cabin.py | 121 ++++++++++ src/flight_cli/cli.py | 427 ++++++++++++++++++++++++++++++++- tests/test_multi_cabin.py | 289 ++++++++++++++++++++++ 3 files changed, 835 insertions(+), 2 deletions(-) create mode 100644 src/flight_cli/_multi_cabin.py create mode 100644 tests/test_multi_cabin.py diff --git a/src/flight_cli/_multi_cabin.py b/src/flight_cli/_multi_cabin.py new file mode 100644 index 0000000..7b7271c --- /dev/null +++ b/src/flight_cli/_multi_cabin.py @@ -0,0 +1,121 @@ +"""Multi-cabin search orchestration: join N single-cabin SearchResults +into rows with one price-per-cabin. + +The domain `Search` stays single-cabin — multi-cabin is an orchestration +concept. The CLI fires N parallel single-cabin queries (one per requested +cabin), then this module joins them on a stable per-itinerary key and +produces rows the renderer can iterate. + +Join key: per-slice `(normalized first flight number, YYYY-MM-DD)` across +all slices. Same shape as `pp.match.cash_match_key`, generalized to the +whole itinerary so round-trips don't collide on outbound alone. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .domain import Cabin + from .models import Itinerary, SearchResult + + +SliceKey = tuple[str, str] # (FLIGHT_NUMBER_UPPER_NOSPACE, "YYYY-MM-DD") +ItineraryKey = tuple[SliceKey, ...] + +_CASH_NUM_RE = re.compile(r"[\d,]*\d+(?:\.\d+)?") + + +def _norm_fn(fn: str | None) -> str: + return (fn or "").upper().replace(" ", "") + + +def _iso_date(s: str | None) -> str: + if not s: + return "" + return s.replace(" ", "T")[:10] + + +def itinerary_key(itin: Itinerary) -> ItineraryKey | None: + """Per-slice (first-flight#, date) tuple across all slices of an + itinerary. None if any slice is missing both anchors — such itineraries + can't be safely joined and are skipped. + """ + details = itin.itinerary + if not details or not details.slices: + return None + slice_keys: list[SliceKey] = [] + for s in details.slices: + flights = s.flights or [] + fn = _norm_fn(flights[0]) if flights else "" + dep = _iso_date(s.departure) + if not fn or not dep: + return None + slice_keys.append((fn, dep)) + return tuple(slice_keys) + + +def parse_price(s: str | None) -> float | None: + """Pull the first numeric value out of strings like 'USD530.00', + '$1,078', '1,078 USD'. Mirrors `pp.cli._parse_cash` — kept local to + avoid a cross-module dep just for one regex.""" + if not s or s in ("—", "-"): + return None + m = _CASH_NUM_RE.search(s) + if not m: + return None + try: + return float(m.group(0).replace(",", "")) + except ValueError: + return None + + +@dataclass +class MultiCabinRow: + """One itinerary observed across one or more cabin queries. + + `itinerary` is the first occurrence we saw (used for render: slices, + carriers, legroom). `prices` maps each cabin we have a price for to + its raw price string (e.g. 'USD623.00'). Cabins with no price are + absent from the dict — renderer treats absence as '—'. + """ + + itinerary: Itinerary + prices: dict[Cabin, str] = field(default_factory=dict) + + +def merge( + results_by_cabin: dict[Cabin, SearchResult], + *, + sort_by: Cabin, + top_n: int, +) -> list[MultiCabinRow]: + """Join itineraries across cabins. Sorted by `sort_by`'s parsed price + (asc); rows missing the sort cabin's price sink to the bottom. Truncated + to `top_n` rows. + + Itineraries that can't be keyed (missing flight# or departure on any + slice) are skipped. + """ + by_key: dict[ItineraryKey, MultiCabinRow] = {} + for cabin, res in results_by_cabin.items(): + for it in res.solutions: + key = itinerary_key(it) + if key is None: + continue + row = by_key.get(key) + if row is None: + row = MultiCabinRow(itinerary=it) + by_key[key] = row + price = it.price + if price: + row.prices[cabin] = price + + def sort_value(row: MultiCabinRow) -> float: + p = parse_price(row.prices.get(sort_by)) + return p if p is not None else float("inf") + + rows = sorted(by_key.values(), key=sort_value) + return rows[:top_n] diff --git a/src/flight_cli/cli.py b/src/flight_cli/cli.py index 2bb26e1..77fd229 100644 --- a/src/flight_cli/cli.py +++ b/src/flight_cli/cli.py @@ -20,11 +20,14 @@ from typing import TYPE_CHECKING, Annotated, Any, cast import anyio +import anyio.to_thread import typer from rich.console import Console from rich.table import Table from . import _config +from ._multi_cabin import MultiCabinRow +from ._multi_cabin import merge as _merge_cabins from .client import MatrixApiError, MatrixClient from .domain import ( Cabin, @@ -169,6 +172,20 @@ def _resolve_cabin(name: str) -> Cabin: raise typer.Exit(2) +def _resolve_cabin_list(csv: str) -> tuple[Cabin, ...]: + """Parse a comma-separated cabin list. Single-cabin invocations still + take this path — they emit a 1-tuple and the dispatcher routes them + back through the single-cabin code path unchanged.""" + tokens = [t.strip() for t in csv.split(",") if t.strip()] + if not tokens: + err.print("[red]--cabin must name at least one cabin.[/]") + raise typer.Exit(2) + seen: dict[Cabin, None] = {} + for t in tokens: + seen.setdefault(_resolve_cabin(t), None) + return tuple(seen) + + def _build_options( *, cabin: str, @@ -811,6 +828,338 @@ def _run_gflight_path( ) +# ─────────────────────────── multi-cabin orchestration ───────────────────── + + +def _derive_pp_cabins(cash_cabins: tuple[Cabin, ...]) -> tuple[str, ...]: + """Map cash cabin list → PP cabin list for the PP overlay. + + Adds First when Business is requested but First isn't: award seekers + treat business/first as a paired premium tier, and First is rare enough + that surfacing it costs almost nothing while filling a real research + gap. The reverse promotion (First → +Business) isn't applied — asking + for First means the user has already made that call. + """ + out: list[str] = [_CABIN_TO_PP_NAME[c] for c in cash_cabins] + if Cabin.BUSINESS in cash_cabins and Cabin.FIRST not in cash_cabins: + out.append(_CABIN_TO_PP_NAME[Cabin.FIRST]) + return tuple(out) + + +def _pp_cabins_for_multi(sel: ProviderSelection, cabins: tuple[Cabin, ...]) -> str | None: + """PP cabins for a multi-cabin search. User's `--provider-opt pp.cabins=` + wins; otherwise derive from `--cabin` with the business→+first rule.""" + user_set = sel.pp_cabins() + if user_set is not None: + return user_set + return ",".join(_derive_pp_cabins(cabins)) + + +def _run_matrix_multi( + *, + legs: tuple[Leg, ...], + opts: SearchOptions, + cabins: tuple[Cabin, ...], + rps: float, + impersonate: str, + no_cache: bool, +) -> dict[Cabin, SearchResult]: + """Fan out N parallel Matrix queries (one per cabin), one shared client. + + Per-cabin failures are soft: log + omit from the result dict (renderer + shows that column as all '—'). Connection-level / auth errors still + propagate so the user sees real outages. + """ + results: dict[Cabin, SearchResult] = {} + + async def query_cabin(client: MatrixClient, cab: Cabin) -> None: + cabin_opts = opts.model_copy(update={"cabin": cab}) + search = SpecificDateSearch(legs=legs, options=cabin_opts) + try: + res = await client.execute(search, cache=not no_cache) + except MatrixApiError as e: + err.print(f"[yellow]Matrix {cab.value} query failed ({e.kind}): {e.message}[/]") + return + results[cab] = cast("SearchResult", res) + + async def go() -> None: + async with ( + MatrixClient(rps=rps, impersonate=impersonate) as client, + anyio.create_task_group() as tg, + ): + for cab in cabins: + tg.start_soon(query_cabin, client, cab) + + try: + anyio.run(go) + except MatrixApiError as e: + err.print(f"[red]Matrix returned an error ({e.kind}):[/] {e.message}") + if e.request_id: + err.print(f"[dim]request_id: {e.request_id}[/]") + raise typer.Exit(1) from e + return results + + +def _run_gflight_multi( + *, + legs: tuple[Leg, ...], + opts: SearchOptions, + cabins: tuple[Cabin, ...], + top_n: int, +) -> dict[Cabin, list[Any]]: + """Fan out N parallel gflight queries (one per cabin). fli is sync, so + each query runs in a worker thread via `anyio.to_thread.run_sync`.""" + from ._gflight_ids import search_with_ids # noqa: PLC0415 + from .fli_bridge import to_fli_filter # noqa: PLC0415 + + results: dict[Cabin, list[Any]] = {} + + def query_sync(cab: Cabin) -> list[Any]: + cabin_opts = opts.model_copy(update={"cabin": cab}) + search = SpecificDateSearch(legs=legs, options=cabin_opts) + return search_with_ids(to_fli_filter(search), top_n=top_n) or [] + + async def query_cabin(cab: Cabin) -> None: + try: + results[cab] = await anyio.to_thread.run_sync(query_sync, cab) + except Exception as e: # noqa: BLE001 — fli has no documented exception surface + err.print(f"[yellow]Google Flights {cab.value} query failed: {e}[/]") + + async def go() -> None: + async with anyio.create_task_group() as tg: + for cab in cabins: + tg.start_soon(query_cabin, cab) + + anyio.run(go) + return results + + +def _gflight_to_search_result_per_cabin( + results_by_cabin: dict[Cabin, list[Any]], +) -> dict[Cabin, SearchResult]: + """Adapt gflight's duck-typed fli results into the SearchResult shape so + the merge/render path is backend-agnostic. Reuses pp.gflight_adapter.""" + from .pp.gflight_adapter import fli_results_to_search_result # noqa: PLC0415 + + return {cab: fli_results_to_search_result(res) for cab, res in results_by_cabin.items()} + + +def _render_multi_cabin_search( + rows: list[MultiCabinRow], + *, + cabins: tuple[Cabin, ...], + sort_by: Cabin, + title_prefix: str = "Itineraries", +) -> None: + """Render multi-cabin merged rows. One row per itinerary, one $ column + per requested cabin, '—' for missing.""" + if not rows: + console.print("[yellow]No itineraries.[/]") + return + # Use the first present price to surface a currency tag in the title. + ccy = "" + for row in rows: + for p in row.prices.values(): + ccy_candidate, _ = _split_price(p) + if ccy_candidate: + ccy = ccy_candidate + break + if ccy: + break + ccy_tag = f" ({ccy})" if ccy else "" + cabin_labels = "+".join(_CABIN_TO_LETTER[c] for c in cabins) + + t = Table( + title=f"{title_prefix} · {cabin_labels} (sorted by {_CABIN_TO_LETTER[sort_by]}){ccy_tag}", + show_header=True, + header_style="bold green", + ) + t.add_column("#", justify="right") + t.add_column("carriers") + t.add_column("outbound") + t.add_column("return") + for cab in cabins: + t.add_column(f"{_CABIN_TO_LETTER[cab]} $", justify="right") + + for i, row in enumerate(rows, 1): + itn = row.itinerary.itinerary + slcs: list[Slice] = itn.slices if itn else [] + carriers = ",".join((c.code or "?") for c in (itn.carriers if itn else [])) + + def _fmt(s: Slice) -> str: + dep = s.departure or "" + arr = s.arrival or "" + dur_min = s.duration or 0 + dur = f"{dur_min // 60}h{dur_min % 60:02d}m" if dur_min else "" + o = (s.origin.code if s.origin else None) or "?" + d = (s.destination.code if s.destination else None) or "?" + head = f"{o}→{d} {'/'.join(s.flights) or '?'} {dep[:16]}→{arr[:16]} {dur}" + tail = _fmt_legroom_lines(s) + return f"{head}\n{tail}" if tail else head + + out_cell = _fmt(slcs[0]) if slcs else "—" + ret_cell = _fmt(slcs[1]) if len(slcs) > 1 else "—" + price_cells = [_amount(row.prices.get(cab)) for cab in cabins] + t.add_row(str(i), carriers or "?", out_cell, ret_cell, *price_cells) + console.print(t) + + +def _validate_sort_cabin(sort_by: Cabin, cabins: tuple[Cabin, ...]) -> None: + if sort_by not in cabins: + names = ", ".join(c.value for c in cabins) + err.print(f"[red]--sort {sort_by.value!r} must be one of --cabin: {names}[/]") + raise typer.Exit(2) + + +def _run_matrix_path_multi( + *, + legs: tuple[Leg, ...], + opts: SearchOptions, + cabins: tuple[Cabin, ...], + sort_by: Cabin, + top_n: int, + rps: float, + impersonate: str, + no_cache: bool, + json_out: bool, + matrix_url: bool, + google_url: bool, + run_pp: bool, + sel: ProviderSelection, +) -> None: + """Matrix multi-cabin: N parallel cabin queries → client-side join → render.""" + results_by_cabin = _run_matrix_multi( + legs=legs, + opts=opts, + cabins=cabins, + rps=_resolve_rps(rps), + impersonate=_resolve_impersonate(impersonate), + no_cache=_resolve_no_cache(no_cache), + ) + if not results_by_cabin: + err.print("[red]All cabin queries failed.[/]") + raise typer.Exit(1) + + if json_out and not run_pp: + # JSON shape: {cabin: raw} so consumers can re-merge if they want. + sys.stdout.write( + json.dumps({c.value: r.raw for c, r in results_by_cabin.items()}, indent=2) + ) + return + + rows = _merge_cabins(results_by_cabin, sort_by=sort_by, top_n=top_n) + if not sel.awards_only: + _render_multi_cabin_search(rows, cabins=cabins, sort_by=sort_by) + + if run_pp: + # PP runs once against the merged result so award flights match against + # the full itinerary set we just rendered. Pick any one of the cabin + # results to source slices for cash_hints / matched-id lookups — + # itineraries that survived the merge are the union of all. + merged = _merge_results_into_one(results_by_cabin, rows) + p = opts.pax + run_pp_for_search( + merged, + legs=_build_pp_legs(legs), + num_passengers=p.adults + p.children + p.seniors + p.youth, + airlines=sel.pp_airlines(), + cabins=_pp_cabins_for_multi(sel, cabins), + pp_only=sel.awards_only, + json_out=json_out, + provider_filter=sel.provider_filter, + seats_sources=sel.seats_sources(), + ) + + # Deep links are cabin-specific (Matrix's URL encodes one cabin). Emit the + # link for the sort cabin — it's the "primary" surface in the rendered + # table and the one a user is most likely to click through to. + sort_opts = opts.model_copy(update={"cabin": sort_by}) + _emit_urls( + SpecificDateSearch(legs=legs, options=sort_opts), + matrix_url=matrix_url, + google_url=google_url, + ) + + +def _run_gflight_path_multi( + *, + legs: tuple[Leg, ...], + opts: SearchOptions, + cabins: tuple[Cabin, ...], + sort_by: Cabin, + top_n: int, + json_out: bool, + run_pp: bool, + sel: ProviderSelection, +) -> None: + """Google Flights multi-cabin: N parallel cabin queries (threadpool) → join → render.""" + fli_by_cabin = _run_gflight_multi(legs=legs, opts=opts, cabins=cabins, top_n=top_n) + if not fli_by_cabin: + err.print("[red]All Google Flights cabin queries failed.[/]") + raise typer.Exit(1) + + if json_out and not run_pp: + out: dict[str, Any] = {} + for cab, fli_results in fli_by_cabin.items(): + cab_dumped: list[Any] = [] + for r in fli_results: + items: list[Any] = list(r) if isinstance(r, tuple) else [r] # pyright: ignore[reportUnknownArgumentType] + dumped = [ + {**g.flight.model_dump(mode="json"), "flight_id": g.flight_id} for g in items + ] + cab_dumped.append(dumped if isinstance(r, tuple) else dumped[0]) + out[cab.value] = cab_dumped + sys.stdout.write(json.dumps(out, indent=2, default=str)) + return + + results_by_cabin = _gflight_to_search_result_per_cabin(fli_by_cabin) + rows = _merge_cabins(results_by_cabin, sort_by=sort_by, top_n=top_n) + if not sel.awards_only: + _render_multi_cabin_search( + rows, cabins=cabins, sort_by=sort_by, title_prefix="Google Flights" + ) + + if run_pp: + merged = _merge_results_into_one(results_by_cabin, rows) + p = opts.pax + run_pp_for_search( + merged, + legs=_build_pp_legs(legs), + num_passengers=p.adults + p.children + p.seniors + p.youth, + airlines=sel.pp_airlines(), + cabins=_pp_cabins_for_multi(sel, cabins), + pp_only=sel.awards_only, + json_out=json_out, + provider_filter=sel.provider_filter, + seats_sources=sel.seats_sources(), + ) + + +def _merge_results_into_one( + results_by_cabin: dict[Cabin, SearchResult], + rows: list[MultiCabinRow], +) -> SearchResult: + """Build a single SearchResult whose `solutions` are the merged itineraries + in render order. Used as input to `run_pp_for_search` — PP matches by + flight#+date, so any per-cabin price difference doesn't affect the match + (PP attaches awards to flights, not fares).""" + # SearchResult is TYPE_CHECKING-only at module top — need a runtime import. + from .models import SearchResult # noqa: PLC0415 + + # Pick any one of the per-cabin results to seed the carrier_stop_matrix / + # currency_notice — PP only reads `.solutions`, so the rest doesn't matter. + seed = next(iter(results_by_cabin.values())) + return SearchResult( + solutionCount=len(rows), + solutions=[r.itinerary for r in rows], + carrierStopMatrix=seed.carrier_stop_matrix, + currencyNotice=seed.currency_notice, + session=seed.session, + solutionSet=seed.solution_set, + raw=seed.raw, + ) + + def _render_gflight_table(results: list[Any], *, legs: tuple[Leg, ...], top_n: int) -> None: """Render fli results as a rich table. Duck-typed: fli has no type stubs. @@ -865,6 +1214,20 @@ def _render_gflight_table(results: list[Any], *, legs: tuple[Leg, ...], top_n: i # Angled Flat aren't comparable on pitch alone) so those stay as text. _LEGROOM_AS_COLOR = {"BELOW": "red", "ABOVE": "green"} _CABIN_LETTER = {"ECONOMY": "Y", "PREMIUM": "W", "BUSINESS": "J", "FIRST": "F"} +# Domain Cabin enum → human label and PP API cabin string. Used for +# multi-cabin column headers and PP cabin derivation. +_CABIN_TO_LETTER: dict[Cabin, str] = { + Cabin.COACH: "Y", + Cabin.PREMIUM_COACH: "W", + Cabin.BUSINESS: "J", + Cabin.FIRST: "F", +} +_CABIN_TO_PP_NAME: dict[Cabin, str] = { + Cabin.COACH: "Economy", + Cabin.PREMIUM_COACH: "Premium economy", + Cabin.BUSINESS: "Business", + Cabin.FIRST: "First", +} # 📶 for wifi is the only emoji (2-col) — wifi is the highest-value binary signal # and 📶 is universally read at-a-glance where ≋ is not. Power and video keep # 1-col Unicode pairs so the plug-vs-USB and stream-vs-ondemand distinctions @@ -1087,7 +1450,25 @@ def search( rich_help_panel=_GROUP_BACKEND, ), ] = BACKEND_AUTO, - cabin: str = typer.Option("economy", "--cabin", rich_help_panel=_GROUP_ITINERARY), + cabin: str = typer.Option( + "economy", + "--cabin", + help=( + "Cabin, or comma list for multi-cabin compare ('economy,business'). " + "Multi-cabin renders one $ column per cabin; '—' means the itinerary " + "wasn't in that cabin's top-N (cabin unavailable OR priced out). " + "Bump -n for broader overlap across cabins." + ), + rich_help_panel=_GROUP_ITINERARY, + ), + sort_cabin: Annotated[ + str | None, + typer.Option( + "--sort", + help="Cabin to sort multi-cabin results by. Default: first in --cabin.", + rich_help_panel=_GROUP_ITINERARY, + ), + ] = None, adults: int = typer.Option(1, "--adults", rich_help_panel=_GROUP_ITINERARY), children: int = typer.Option(0, "--children", rich_help_panel=_GROUP_ITINERARY), seniors: int = typer.Option(0, "--seniors", rich_help_panel=_GROUP_ITINERARY), @@ -1279,8 +1660,19 @@ def search( err.print("[red]Specify --slice ... or origin destination --dep[/]") raise typer.Exit(2) + cabins_tuple = _resolve_cabin_list(cabin) + # _resolve_cabin_list raises typer.Exit on empty input, so cabins_tuple is + # never empty here. Bind `first_cabin` before any len-narrowing branches so + # basedpyright keeps the `tuple[Cabin, ...]` → Cabin inference. + first_cabin = cabins_tuple[0] + sort_by = _resolve_cabin(sort_cabin) if sort_cabin else first_cabin + if len(cabins_tuple) > 1: + _validate_sort_cabin(sort_by, cabins_tuple) + + # `_build_options` seeds with the first cabin in the list; multi-cabin + # orchestrators clone opts per cabin via `model_copy(update={"cabin": ...})`. opts = _build_options( - cabin=cabin, + cabin=first_cabin.value, adults=adults, children=children, seniors=seniors, @@ -1294,6 +1686,37 @@ def search( ) run_awards = _should_run_awards(sel) + + if len(cabins_tuple) > 1: + if resolved == BACKEND_GFLIGHT: + _run_gflight_path_multi( + legs=legs, + opts=opts, + cabins=cabins_tuple, + sort_by=sort_by, + top_n=page_size, + json_out=json_out, + run_pp=run_awards, + sel=sel, + ) + return + _run_matrix_path_multi( + legs=legs, + opts=opts, + cabins=cabins_tuple, + sort_by=sort_by, + top_n=page_size, + rps=_resolve_rps(rps), + impersonate=_resolve_impersonate(impersonate), + no_cache=_resolve_no_cache(no_cache), + json_out=json_out, + matrix_url=matrix_url, + google_url=google_url, + run_pp=run_awards, + sel=sel, + ) + return + if resolved == BACKEND_GFLIGHT: _run_gflight_path( legs=legs, diff --git a/tests/test_multi_cabin.py b/tests/test_multi_cabin.py new file mode 100644 index 0000000..6a7c995 --- /dev/null +++ b/tests/test_multi_cabin.py @@ -0,0 +1,289 @@ +# pyright: reportCallIssue=false, reportPrivateUsage=false, reportOptionalMemberAccess=false +# DIVERGE: pydantic Field(alias=...) confuses basedpyright into thinking +# alias names are required kwargs. The tests rely on populate_by_name=True +# (set on _Loose); silence the rule rather than reformat every constructor. +# Private-usage suppression: tests intentionally drive `_resolve_cabin_list` +# and `_derive_pp_cabins` (module-private helpers) — they're the units we're +# unit-testing. Optional-member-access: the test fixtures always build +# itineraries with `.itinerary` populated, but pydantic's `Optional` typing +# requires a narrow at every access — noise that drowns out real errors. +"""Tests for multi-cabin merge logic, CLI cabin-list parsing, and PP +cabin auto-derivation.""" + +from __future__ import annotations + +import pytest +import typer + +from flight_cli._multi_cabin import ( + MultiCabinRow, + itinerary_key, + merge, + parse_price, +) +from flight_cli.cli import ( + _derive_pp_cabins, + _resolve_cabin_list, +) +from flight_cli.domain import Cabin +from flight_cli.models import ( + Itinerary, + ItineraryDetails, + ItineraryExt, + SearchResult, + Slice, + SliceEndpoint, +) + +# ─────────────────────────── itinerary builders ──────────────────────────── + + +def _itin( + *slices_data: tuple[str, str, str, str], + price: str = "USD500.00", +) -> Itinerary: + """Build an Itinerary. Each slices_data tuple is + (flight_number, departure_iso, origin, destination).""" + slcs = [ + Slice( + flights=[fn], + departure=dep, + origin=SliceEndpoint(code=o), + destination=SliceEndpoint(code=d), + ) + for fn, dep, o, d in slices_data + ] + return Itinerary( + displayTotal=price, + ext=ItineraryExt(price=price), + itinerary=ItineraryDetails(slices=slcs, carriers=[]), + ) + + +def _result(*itins: Itinerary) -> SearchResult: + return SearchResult(solutionCount=len(itins), solutions=list(itins)) + + +# ─────────────────────────── _resolve_cabin_list ─────────────────────────── + + +def test_resolve_cabin_list_csv_basic(): + assert _resolve_cabin_list("economy,business") == (Cabin.COACH, Cabin.BUSINESS) + + +def test_resolve_cabin_list_short_aliases(): + assert _resolve_cabin_list("y,j,f") == (Cabin.COACH, Cabin.BUSINESS, Cabin.FIRST) + + +def test_resolve_cabin_list_dedup_preserves_order(): + assert _resolve_cabin_list("business,economy,business") == (Cabin.BUSINESS, Cabin.COACH) + + +def test_resolve_cabin_list_strips_whitespace_and_empties(): + assert _resolve_cabin_list(" economy , , business ") == (Cabin.COACH, Cabin.BUSINESS) + + +def test_resolve_cabin_list_single_cabin_returns_singleton(): + assert _resolve_cabin_list("economy") == (Cabin.COACH,) + + +def test_resolve_cabin_list_empty_errors(): + with pytest.raises(typer.Exit): + _resolve_cabin_list(",,") + + +def test_resolve_cabin_list_unknown_token_errors(): + with pytest.raises(typer.Exit): + _resolve_cabin_list("economy,nonsense") + + +# ────────────────────────────── itinerary_key ────────────────────────────── + + +def test_itinerary_key_one_way(): + it = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR")) + assert itinerary_key(it) == (("AA100", "2026-08-15"),) + + +def test_itinerary_key_round_trip_distinct_keys(): + out = _itin( + ("AA100", "2026-08-15T09:00", "JFK", "LHR"), + ("AA200", "2026-08-22T18:00", "LHR", "JFK"), + ) + ret_swapped = _itin( + # Same return-first ordering produces a different tuple — the test + # locks in that slice order matters (outbound + return aren't + # interchangeable; the round trip is the unit). + ("AA200", "2026-08-22T18:00", "LHR", "JFK"), + ("AA100", "2026-08-15T09:00", "JFK", "LHR"), + ) + assert itinerary_key(out) != itinerary_key(ret_swapped) + + +def test_itinerary_key_normalizes_flight_numbers(): + it = _itin((" aa 100 ", "2026-08-15T09:00", "JFK", "LHR")) + assert itinerary_key(it) == (("AA100", "2026-08-15"),) + + +def test_itinerary_key_missing_flights_returns_none(): + it = Itinerary( + itinerary=ItineraryDetails( + slices=[Slice(flights=[], departure="2026-08-15T09:00")], carriers=[] + ), + ) + assert itinerary_key(it) is None + + +def test_itinerary_key_missing_departure_returns_none(): + it = _itin(("AA100", "", "JFK", "LHR")) + assert itinerary_key(it) is None + + +# ────────────────────────────── parse_price ──────────────────────────────── + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("USD530.00", 530.00), + ("$1,078", 1078.0), + ("1,078 USD", 1078.0), + ("EUR99.99", 99.99), + (None, None), + ("", None), + ("—", None), + ("free", None), + ], +) +def test_parse_price(raw: str | None, expected: float | None): + assert parse_price(raw) == expected + + +# ──────────────────────────────── merge ──────────────────────────────────── + + +def test_merge_full_overlap(): + a = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD600.00") + b = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD3000.00") + rows = merge( + {Cabin.COACH: _result(a), Cabin.BUSINESS: _result(b)}, + sort_by=Cabin.COACH, + top_n=10, + ) + assert len(rows) == 1 + assert rows[0].prices == {Cabin.COACH: "USD600.00", Cabin.BUSINESS: "USD3000.00"} + + +def test_merge_partial_overlap_missing_filled_with_absent_keys(): + a = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD600.00") + b = _itin(("BA200", "2026-08-15T11:00", "JFK", "LHR"), price="USD3500.00") + rows = merge( + {Cabin.COACH: _result(a), Cabin.BUSINESS: _result(b)}, + sort_by=Cabin.COACH, + top_n=10, + ) + assert len(rows) == 2 + by_carrier = {row.itinerary.itinerary.slices[0].flights[0]: row for row in rows} + assert by_carrier["AA100"].prices == {Cabin.COACH: "USD600.00"} + assert by_carrier["BA200"].prices == {Cabin.BUSINESS: "USD3500.00"} + + +def test_merge_sort_by_missing_sinks_to_bottom(): + has_econ = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD800.00") + no_econ = _itin(("BA200", "2026-08-15T11:00", "JFK", "LHR"), price="USD9000.00") + cheap_econ = _itin(("DL300", "2026-08-15T10:00", "JFK", "LHR"), price="USD600.00") + # BUSINESS-only result contains the no-econ flight. + rows = merge( + { + Cabin.COACH: _result(has_econ, cheap_econ), + Cabin.BUSINESS: _result(no_econ), + }, + sort_by=Cabin.COACH, + top_n=10, + ) + flight_nums = [r.itinerary.itinerary.slices[0].flights[0] for r in rows] + assert flight_nums == ["DL300", "AA100", "BA200"] + + +def test_merge_top_n_truncates_after_sort(): + a = _itin(("AA1", "2026-08-15T09:00", "JFK", "LHR"), price="USD100.00") + b = _itin(("BB2", "2026-08-15T10:00", "JFK", "LHR"), price="USD200.00") + c = _itin(("CC3", "2026-08-15T11:00", "JFK", "LHR"), price="USD300.00") + rows = merge( + {Cabin.COACH: _result(c, a, b)}, + sort_by=Cabin.COACH, + top_n=2, + ) + assert [r.itinerary.itinerary.slices[0].flights[0] for r in rows] == ["AA1", "BB2"] + + +def test_merge_skips_unkeyable_itineraries(): + keyed = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD600.00") + unkeyed = Itinerary( + ext=ItineraryExt(price="USD400.00"), + itinerary=ItineraryDetails(slices=[Slice(flights=[], departure="")], carriers=[]), + ) + rows = merge( + {Cabin.COACH: _result(keyed, unkeyed)}, + sort_by=Cabin.COACH, + top_n=10, + ) + assert len(rows) == 1 + + +def test_merge_preserves_first_itinerary_for_render(): + """When the same key appears in two cabins, the renderer uses the first + observed itinerary (its slices, carriers, legroom) — locking that in so + consumers don't see surprise changes if dict iteration order differs.""" + coach_it = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD600.00") + biz_it = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR"), price="USD3000.00") + rows = merge( + {Cabin.COACH: _result(coach_it), Cabin.BUSINESS: _result(biz_it)}, + sort_by=Cabin.COACH, + top_n=10, + ) + # Coach was first; the row's `itinerary` reference must be `coach_it`. + assert rows[0].itinerary is coach_it + + +# ────────────────────────── _derive_pp_cabins ────────────────────────────── + + +def test_derive_pp_cabins_business_promotes_first(): + assert _derive_pp_cabins((Cabin.COACH, Cabin.BUSINESS)) == ("Economy", "Business", "First") + + +def test_derive_pp_cabins_business_only(): + assert _derive_pp_cabins((Cabin.BUSINESS,)) == ("Business", "First") + + +def test_derive_pp_cabins_first_present_no_double_add(): + assert _derive_pp_cabins((Cabin.BUSINESS, Cabin.FIRST)) == ("Business", "First") + + +def test_derive_pp_cabins_first_alone_does_not_promote_business(): + # Asymmetric rule: First → no auto-add of Business. Asking for First is + # an explicit choice; we don't second-guess it. + assert _derive_pp_cabins((Cabin.FIRST,)) == ("First",) + + +def test_derive_pp_cabins_economy_only(): + # Single cabin — preserve order, no promotion. + assert _derive_pp_cabins((Cabin.COACH,)) == ("Economy",) + + +def test_derive_pp_cabins_premium_economy_business(): + assert _derive_pp_cabins((Cabin.PREMIUM_COACH, Cabin.BUSINESS)) == ( + "Premium economy", + "Business", + "First", + ) + + +# ────────────────────────── MultiCabinRow construction ───────────────────── + + +def test_multi_cabin_row_default_prices_empty(): + it = _itin(("AA100", "2026-08-15T09:00", "JFK", "LHR")) + row = MultiCabinRow(itinerary=it) + assert row.prices == {}