diff --git a/README.md b/README.md index 157f582..3b55039 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,10 @@ flight gflight JFK LHR --dep 2026-08-15 --return 2026-08-22 # IATA autocomplete flight airport LON + +# overlay PointsPath award prices on cash search results (requires PointsPath +# subscription + one-time `flight auth pp login`) +flight fare JFK LHR --dep 2026-08-15 --pp ``` Every result-printing command supports: @@ -55,6 +59,57 @@ Every result-printing command supports: - **Time-of-day filters** (`--depart-times`, `--return-times`): `morning,evening` etc. - **Calendar-mode duration ranges** (`-d 5-7`): one search returns prices for 5-, 6-, and 7-night trips at every starting day. +## PointsPath integration + +`flight fare ... --pp` overlays award prices from [PointsPath](https://pointspath.com) onto the cash itineraries Matrix returns. For each cash flight, you see the airline-native miles cost, taxes, the banks whose points transfer to that program, and cents-per-mile valuation. Round-trips render one table per leg. + +```sh +# one-way with award overlay +flight fare JFK LHR --dep 2026-08-15 --pp + +# limit the cabin set (default: Economy + Business) +flight fare JFK LHR --dep 2026-08-15 --pp --pp-cabin Economy + +# limit the airline set (default: discovered from your account's enabled list) +flight fare JFK LHR --dep 2026-08-15 --pp --pp-airlines United,Delta,American + +# award-only listing (still runs Matrix to show the cash table; pass --pp-only +# to skip the cash render) +flight fare JFK LHR --dep 2026-08-15 --pp --pp-only +``` + +### Setup + +PointsPath requires a paid subscription (free tier is the browser extension only). Today the only login mode is to import tokens captured from a logged-in Chrome session: + +```sh +# 1. Capture your Supabase tokens from the browser. The extension is on +# pointspath.com under cookies named `sb-hxjqzkcirzhjvtubefie-auth-token.0/.1`; +# reassemble and save as JSON like: +# {"access_token": "...", "refresh_token": "...", "user": {"email": "..."}} +# 2. Import into the local token store: +flight auth pp login --tokens-file ~/Downloads/pp_tokens.json +flight auth pp whoami # confirm +``` + +A browser-based `flight auth pp login` is planned but not yet shipped; once you have tokens cached, refresh is automatic for the lifetime of your refresh token. + +### How airline selection works + +On each `--pp` invocation (cached for 24h / 7d respectively): + +1. `GET /api/pricing-info` — universe of supported airlines + their transfer-partner banks +2. `GET /api/extension-config` — your account's enabled feature flags +3. The airlines fanned out are: pricing-info entries minus those with `enable=0` in the feature flags. Always-on airlines (American, Delta, United, JetBlue, Alaska) have no toggle and are always included. + +Pass `--pp-airlines United,Delta,...` to skip discovery and call only the named set. + +### What it doesn't do + +- Browser-based login (use `--tokens-file` for now) +- `calendar --pp` (lowest-fare-calendar overlay) — fan-out is N days × M airlines; deserves its own design +- Match against airlines we don't yet support (the few in pricing-info but not enabled for your tier are silently skipped) + ## Architecture The codebase is a small pydantic discriminated union with match-based @@ -72,9 +127,16 @@ src/flight_cli/ cli.py typer commands models.py response models _http.py httpx + curl_cffi + aiolimiter + stamina + pp/ PointsPath integration (--pp on `fare` + `auth pp` subapp) + auth.py Supabase JWT store + refresh + client.py airline-search / pricing-info / extension-config (cached) + match.py cash↔award join by (flight#, date) + cli.py auth subapp + augmenter for `fare` + models.py PointsPath response shapes tests/ fixtures/ captured SPA wire bodies (golden files) test_wire_round_trip.py + pp/ PointsPath model + match + helper unit tests ``` Run tests with `pytest tests/`. diff --git a/pyproject.toml b/pyproject.toml index a494926..d10a4d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,11 +90,12 @@ ignore = [ "N", # test function names may mirror wire-field names (camelCase) "PLC0415", # internal lazy imports are fine for accessing private helpers ] -# DIVERGE: wire.py mirrors Matrix's JSON shape directly with camelCase attrs -# (instead of snake_case + Field(alias=...) on every field). N815 doesn't -# apply to models.py either — its alias= strings aren't attr names — or to -# domain.py, which is snake_case throughout. +# DIVERGE: wire.py and pp/models.py mirror their respective upstream JSON +# shapes directly with camelCase attrs (instead of snake_case + Field(alias=...) +# on every field). Profile-B-edge: reverse-engineered, undocumented upstreams +# (Matrix, PointsPath) — the attr names ARE the contract here. "src/flight_cli/wire.py" = ["N815"] +"src/flight_cli/pp/models.py" = ["N815"] [tool.ruff.format] docstring-code-format = true diff --git a/src/flight_cli/cli.py b/src/flight_cli/cli.py index 4672e7c..d94ec7a 100644 --- a/src/flight_cli/cli.py +++ b/src/flight_cli/cli.py @@ -37,6 +37,7 @@ ) from .links import google_flights_url, matrix_deep_link from .log import configure as configure_logging +from .pp.cli import LegQuery, auth_app, run_pp_for_search if TYPE_CHECKING: from .models import CalendarResult, Location, SearchResult, Slice @@ -44,6 +45,7 @@ # Tuple-length sentinels for `--slice` parser (`ORIGIN-DEST:DATE[:r=...:e=...]`). _SLICE_MIN_PARTS = 2 _SLICE_MAX_PARTS = 3 +_ROUND_TRIP_LEGS = 2 # 2 legs = round-trip; 1 = one-way; >2 = multi-city # Matrix returns prices as 'USD877.00' (ISO-4217 prefix + decimal). We split # the prefix off for rendering so tables can show the currency once in the @@ -67,6 +69,7 @@ def _amount(s: str | None) -> str: app = typer.Typer( add_completion=False, rich_markup_mode="rich", help="CLI for ITA Matrix's Alkali backend." ) +app.add_typer(auth_app, name="auth") console = Console() err = Console(stderr=True) @@ -403,6 +406,30 @@ def fare( matrix_url: bool = typer.Option(True, "--matrix-url/--no-matrix-url"), google_url: bool = typer.Option(True, "--google-url/--no-google-url"), no_cache: bool = typer.Option(False, "--no-cache"), + pp: bool = typer.Option( + False, + "--pp", + help="Augment results with PointsPath award prices (requires `auth pp login`).", + ), + pp_only: bool = typer.Option( + False, + "--pp-only", + help="Show only PointsPath award availability; skip Matrix table render.", + ), + pp_airlines: str | None = typer.Option( + None, + "--pp-airlines", + help=( + "CSV of PointsPath airline names (e.g. United,Delta). " + "Default: discovered from your account's enabled airline set " + "via /api/extension-config + /api/pricing-info." + ), + ), + pp_cabin: str | None = typer.Option( + None, + "--pp-cabin", + help="CSV of cabins to query (Economy,Business,First). Default: Economy,Business.", + ), ) -> None: """Specific-date search. One leg = one-way, two = round-trip, N = multi-city.""" if slice_specs: @@ -451,10 +478,49 @@ def fare( search = SpecificDateSearch(legs=legs, options=opts) # SpecificDateSearch → SearchResult by client._parse_response dispatch. res = cast("SearchResult", _run(search, rps, impersonate, no_cache)) - if json_out: + if json_out and not pp: sys.stdout.write(json.dumps(res.raw, indent=2)) return - _render_search(res) + if not pp_only: + _render_search(res) + if pp: + # Build one PP query per Matrix leg. PP's airline-search is per-direction, + # so a round-trip → 2 queries, multi-city → N queries. Each LegQuery + # carries its slice_index so the matcher knows which Itinerary slice to + # join against. + pp_legs: list[LegQuery] = [] + for i, leg in enumerate(legs): + if not leg.date or not leg.origins or not leg.destinations: + continue # shouldn't happen for SpecificDateSearch; defensive + n = len(legs) + label_kind = ( + "outbound" + if i == 0 and n > 1 + else "return" + if i == 1 and n == _ROUND_TRIP_LEGS + else f"leg {i + 1}" + if n > _ROUND_TRIP_LEGS + else "one-way" + ) + iso = leg.date.isoformat() + pp_legs.append( + LegQuery( + origin=leg.origins[0], + destination=leg.destinations[0], + date=iso, + slice_index=i, + label=f"{label_kind} {leg.origins[0]}→{leg.destinations[0]} {iso}", + ) + ) + run_pp_for_search( + res, + legs=pp_legs, + num_passengers=adults + children + seniors + youth, + airlines=pp_airlines, + cabins=pp_cabin, + pp_only=pp_only, + json_out=json_out, + ) _emit_urls(search, matrix_url=matrix_url, google_url=google_url) diff --git a/src/flight_cli/pp/__init__.py b/src/flight_cli/pp/__init__.py new file mode 100644 index 0000000..01238b9 --- /dev/null +++ b/src/flight_cli/pp/__init__.py @@ -0,0 +1,6 @@ +"""PointsPath integration: award prices joined to ITA Matrix cash itineraries. + +Architecture mirrors the rest of flight_cli — Pydantic models in models.py, +async HTTP in client.py, CLI surface in cli.py. Auth + token refresh is +isolated in auth.py so the rest of the package stays unaware of Supabase. +""" diff --git a/src/flight_cli/pp/auth.py b/src/flight_cli/pp/auth.py new file mode 100644 index 0000000..1f86328 --- /dev/null +++ b/src/flight_cli/pp/auth.py @@ -0,0 +1,207 @@ +"""Token store + Supabase JWT refresh + login-flow plumbing for PointsPath. + +Storage layout + ~/.config/flight-cli/pp.json → {access_token, refresh_token, expires_at, user_email} + PP_ACCESS_TOKEN env var → takes precedence (CI / scripting / one-shot) + PP_SUPABASE_ANON_KEY env var → override the bundled anon key (default works) + +The Supabase anon key below is the public key shipped in PointsPath's browser +extension and Next.js bundle — anyone who installs the extension sees it. It is +not a secret and rotates on the order of years (the JWT exp is in 2034). +""" + +from __future__ import annotations + +import base64 +import json +import os +import time +from dataclasses import dataclass +from http import HTTPStatus +from pathlib import Path +from typing import Any, cast + +import httpx + +_JWT_PARTS = 3 # header.payload.signature +_JsonDict = dict[str, Any] + +SUPABASE_URL = "https://hxjqzkcirzhjvtubefie.supabase.co" + +# Public anon JWT extracted from extension chunk (role=anon, exp=2028+). +# Override with PP_SUPABASE_ANON_KEY if PointsPath rotates it. +DEFAULT_SUPABASE_ANON_KEY = ( + "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9." + "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Imh4anF6a2NpcnpoanZ0dWJlZmllIiwicm9sZSI6" + "ImFub24iLCJpYXQiOjE3MTI5MDM2NTEsImV4cCI6MjAyODQ3OTY1MX0." + "eTj23z4l_XxbVLhdaeJzXJzTvR06j_CdGsl9atohvj0" +) + +CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))) / "flight-cli" +TOKENS_PATH = CONFIG_DIR / "pp.json" + +# Refresh when token has < 60s left, to avoid mid-request expiry. +REFRESH_LEEWAY_SECS = 60 + + +class PPAuthError(Exception): + """Auth failure (no tokens, refresh failed, expired without refresh token).""" + + +@dataclass +class Tokens: + access_token: str + refresh_token: str + expires_at: int # Unix seconds + user_email: str | None = None + + def to_json(self) -> _JsonDict: + return { + "access_token": self.access_token, + "refresh_token": self.refresh_token, + "expires_at": self.expires_at, + "user_email": self.user_email, + } + + @classmethod + def from_json(cls, d: _JsonDict) -> Tokens: + return cls( + access_token=d["access_token"], + refresh_token=d.get("refresh_token", ""), + expires_at=int(d.get("expires_at") or 0), + user_email=d.get("user_email"), + ) + + def needs_refresh(self) -> bool: + return self.expires_at - time.time() < REFRESH_LEEWAY_SECS + + def jwt_claims(self) -> _JsonDict: + """Decode the access_token JWT payload. No signature verification — + we only use this to display issuer/email/expiry, never for trust.""" + parts = self.access_token.split(".") + if len(parts) != _JWT_PARTS: + return {} + payload = parts[1] + payload += "=" * (-len(payload) % 4) + return cast("_JsonDict", json.loads(base64.urlsafe_b64decode(payload))) + + +def _anon_key() -> str: + return os.environ.get("PP_SUPABASE_ANON_KEY") or DEFAULT_SUPABASE_ANON_KEY + + +# ─────────────────────────────── store ───────────────────────────────────── + + +def load_tokens() -> Tokens | None: + """Load tokens from env override, then disk. Returns None if neither present.""" + env_access = os.environ.get("PP_ACCESS_TOKEN") + if env_access: + # Env-override mode: no refresh, no expiry tracking. Caller must + # provide a fresh token. We surface this as a Tokens with expires_at=0 + # so needs_refresh() returns True; refresh() will then try to use + # PP_REFRESH_TOKEN if present. + return Tokens( + access_token=env_access, + refresh_token=os.environ.get("PP_REFRESH_TOKEN", ""), + expires_at=0, + user_email=None, + ) + if not TOKENS_PATH.exists(): + return None + try: + return Tokens.from_json(cast("_JsonDict", json.loads(TOKENS_PATH.read_text()))) + except (OSError, json.JSONDecodeError, KeyError): + return None + + +def save_tokens(t: Tokens) -> None: + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + TOKENS_PATH.write_text(json.dumps(t.to_json(), indent=2)) + # 0600 — contains a bearer token to a paid user account. + TOKENS_PATH.chmod(0o600) + + +def clear_tokens() -> bool: + """Remove the on-disk store. Returns True if a file was removed.""" + if TOKENS_PATH.exists(): + TOKENS_PATH.unlink() + return True + return False + + +# ─────────────────────────────── refresh ─────────────────────────────────── + + +def refresh(tokens: Tokens) -> Tokens: + """Exchange refresh_token for a new access_token via Supabase auth. + + Mutates the on-disk store as a side effect when tokens load from disk. + Raises PPAuthError if refresh_token is missing or refresh fails. + """ + if not tokens.refresh_token: + raise PPAuthError("No refresh_token available — re-run `flight-cli auth pp login`.") + r = httpx.post( + f"{SUPABASE_URL}/auth/v1/token", + params={"grant_type": "refresh_token"}, + headers={ + "apikey": _anon_key(), + "content-type": "application/json", + }, + json={"refresh_token": tokens.refresh_token}, + timeout=20, + ) + if r.status_code != HTTPStatus.OK: + raise PPAuthError(f"Supabase refresh failed: HTTP {r.status_code} {r.text[:200]}") + data: _JsonDict = r.json() + new = Tokens( + access_token=data["access_token"], + refresh_token=data.get("refresh_token") or tokens.refresh_token, + expires_at=int(time.time()) + int(data.get("expires_in", 3600)), + user_email=(cast("_JsonDict", data.get("user") or {})).get("email") or tokens.user_email, + ) + # Only persist if the on-disk store is the source of truth (not env override). + if not os.environ.get("PP_ACCESS_TOKEN"): + save_tokens(new) + return new + + +def get_valid_tokens() -> Tokens: + """Return tokens that are good for at least the next ~minute. Refreshes + on the fly if stale. Raises PPAuthError if no tokens are available.""" + t = load_tokens() + if t is None: + raise PPAuthError( + "No PointsPath tokens. Run `flight-cli auth pp login --tokens-file ...` " + "or set PP_ACCESS_TOKEN." + ) + if t.needs_refresh(): + t = refresh(t) + return t + + +# ──────────────────────────── login helpers ──────────────────────────────── + + +def import_from_tokens_file(path: Path) -> Tokens: + """Import tokens from a JSON file (e.g., one captured via CDP cookie sniffing). + + Accepts the shape we already produce in /tmp/pp_tokens.json: + {access_token, refresh_token, supabase_url?, user?} + """ + raw: _JsonDict = json.loads(Path(path).read_text()) + access: str = raw["access_token"] + parts = access.split(".") + if len(parts) != _JWT_PARTS: + raise PPAuthError("access_token is not a JWT (expected 3 dot-separated parts)") + payload = parts[1] + "=" * (-len(parts[1]) % 4) + claims: _JsonDict = json.loads(base64.urlsafe_b64decode(payload)) + user: _JsonDict = raw.get("user") or {} + t = Tokens( + access_token=access, + refresh_token=raw.get("refresh_token") or "", + expires_at=int(claims.get("exp") or 0), + user_email=user.get("email") or claims.get("email"), + ) + save_tokens(t) + return t diff --git a/src/flight_cli/pp/cli.py b/src/flight_cli/pp/cli.py new file mode 100644 index 0000000..81beca5 --- /dev/null +++ b/src/flight_cli/pp/cli.py @@ -0,0 +1,538 @@ +"""CLI surface for PointsPath: `auth pp ...` subcommands + `--pp` augmentation. + +The augmentation entry point (`run_pp_for_search`) is what the existing `fare` +command calls when `--pp` is on. It runs PP airline-search in parallel with the +finished Matrix result, joins via match.py, and renders. +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path # noqa: TC003 - typer evaluates annotations at runtime +from typing import TYPE_CHECKING, Annotated, Any + +import anyio +import typer +from rich.console import Console +from rich.table import Table + +from .auth import ( + TOKENS_PATH, + PPAuthError, + clear_tokens, + get_valid_tokens, + import_from_tokens_file, + load_tokens, +) +from .client import ( + DEFAULT_AIRLINES, + DEFAULT_CABINS, + PPClient, + SearchSpec, + enabled_airlines, +) +from .match import MatchedFare, join + +if TYPE_CHECKING: + from ..models import SearchResult + from .match import AwardOption + from .models import AirlineSearchResponse, PricingInfoResponse + + +@dataclass +class LegQuery: + """One leg's PP query. slice_index points at the corresponding Slice in + each Itinerary returned by Matrix; label is the user-facing leg name.""" + + origin: str + destination: str + date: str # YYYY-MM-DD + slice_index: int + label: str # e.g. "outbound JFK→LHR" or "return LHR→JFK" + + +console = Console() +err = Console(stderr=True) + + +# ─────────────────────────── auth pp subcommands ──────────────────────────── + +auth_app = typer.Typer( + add_completion=False, + no_args_is_help=True, + help="Auth helpers per provider. Today: PointsPath only.", +) +pp_auth_app = typer.Typer( + add_completion=False, + no_args_is_help=True, + help="PointsPath token management.", +) +auth_app.add_typer(pp_auth_app, name="pp") + + +@pp_auth_app.command("login") +def pp_login( + tokens_file: Annotated[ + Path | None, + typer.Option( + "--tokens-file", + "-f", + help="Import tokens from a JSON file (e.g. one captured via CDP cookie sniff).", + ), + ] = None, +) -> None: + """Save tokens to ~/.config/flight-cli/pp.json. + + Today only `--tokens-file` is wired. Browser-based login is planned but + not yet implemented; for now, capture tokens from your authenticated Chrome + session and pass the JSON file in. + """ + if tokens_file is None: + err.print("[yellow]Browser-based login isn't implemented yet.[/]") + err.print( + "Workaround: capture your Supabase tokens from your authenticated " + "Chrome session and pass them in:\n" + " flight-cli auth pp login --tokens-file /path/to/pp_tokens.json\n" + "Expected file shape: " + '{"access_token": "...", "refresh_token": "...", "user": {"email": "..."}}' + ) + raise typer.Exit(2) + try: + t = import_from_tokens_file(tokens_file) + except (PPAuthError, OSError, json.JSONDecodeError, KeyError) as e: + err.print(f"[red]Login failed: {e}[/]") + raise typer.Exit(1) from e + when = datetime.fromtimestamp(t.expires_at).isoformat() if t.expires_at else "?" + console.print( + f"[green]Saved[/] tokens for [bold]{t.user_email or '?'}[/] " + f"to {TOKENS_PATH}\n access_token expires: {when}" + ) + + +@pp_auth_app.command("whoami") +def pp_whoami() -> None: + """Print the authenticated user, expiry, and token store path.""" + t = load_tokens() + if t is None: + err.print("[yellow]Not logged in.[/] Run `flight-cli auth pp login --tokens-file ...`.") + raise typer.Exit(1) + claims = t.jwt_claims() + when = datetime.fromtimestamp(t.expires_at).isoformat() if t.expires_at else "?" + console.print(f"email: [bold]{t.user_email or claims.get('email') or '?'}[/]") + console.print(f"sub: {claims.get('sub', '?')}") + console.print(f"role: {claims.get('role', '?')}") + console.print(f"expires: {when}") + console.print(f"store: {TOKENS_PATH}") + + +@pp_auth_app.command("logout") +def pp_logout() -> None: + """Delete the on-disk PointsPath token store.""" + if clear_tokens(): + console.print(f"[green]Deleted[/] {TOKENS_PATH}") + else: + console.print(f"Nothing to delete ({TOKENS_PATH} doesn't exist).") + + +# ───────────────────── augmentation entry point for `fare` ────────────────── + + +def _parse_csv(s: str | None, default: tuple[str, ...]) -> tuple[str, ...]: + if not s: + return default + return tuple(part.strip() for part in s.split(",") if part.strip()) + + +_CABIN_ALIASES = { + "y": "Economy", + "economy": "Economy", + "coach": "Economy", + "main": "Economy", + "w": "Premium economy", + "premium": "Premium economy", + "premiumeconomy": "Premium economy", + "j": "Business", + "business": "Business", + "f": "First", + "first": "First", +} + + +def _normalize_cabin(c: str) -> str: + k = c.strip().lower().replace(" ", "").replace("-", "") + return _CABIN_ALIASES.get(k, c) + + +def run_pp_for_search( + res: SearchResult, + *, + legs: list[LegQuery], + num_passengers: int = 1, + airlines: str | None = None, + cabins: str | None = None, + pp_only: bool = False, + json_out: bool = False, +) -> None: + """Called by the `fare` command when --pp is on. For each leg, runs PP + queries (one per airline x cabin), joins against the corresponding Slice + in each Itinerary, and renders one table per leg. + + Errors are non-fatal — print and continue so the user still sees their + Matrix results. + """ + try: + get_valid_tokens() # validate + refresh up-front, surface a clear error + except PPAuthError as e: + err.print(f"[red]--pp: {e}[/]") + return + + cabin_list = tuple(_normalize_cabin(c) for c in _parse_csv(cabins, DEFAULT_CABINS)) + # Resolve airline list: explicit --pp-airlines wins; otherwise fetch the + # tier-enabled set from extension-config + pricing-info (cached). Static + # DEFAULT_AIRLINES is the last-resort fallback if discovery fails. + explicit_airlines = _parse_csv(airlines, ()) if airlines else None + + async def _go() -> tuple[list[dict[str, AirlineSearchResponse]], PricingInfoResponse]: + return await _gather_pp( + legs=legs, + num_passengers=num_passengers, + explicit_airlines=explicit_airlines, + cabins=cabin_list, + ) + + try: + per_leg, pricing = anyio.run(_go) + except Exception as e: # noqa: BLE001 — surface anything to user, don't crash CLI + err.print(f"[red]--pp: PointsPath query failed: {e}[/]") + return + + if pp_only: + # JSON: dump per-leg award listings; pretty: one PP-only table per leg. + if json_out: + sys.stdout.write(_serialize_pp_only_per_leg(per_leg, legs)) + return + for leg, merged in zip(legs, per_leg, strict=True): + console.print(f"\n[bold]Leg: {leg.label}[/]") + _render_pp_only(merged, pricing) + return + + # Cash + award per leg. + matches_per_leg: list[list[MatchedFare]] = [ + join(res, merged, pricing, slice_index=leg.slice_index) + for leg, merged in zip(legs, per_leg, strict=True) + ] + if json_out: + sys.stdout.write(_serialize_matches_per_leg(matches_per_leg, legs)) + return + for leg, matches in zip(legs, matches_per_leg, strict=True): + console.print(f"\n[bold]Leg: {leg.label}[/]") + _render_matches(matches, cabin_list, slice_index=leg.slice_index) + + +async def _gather_pp( + *, + legs: list[LegQuery], + num_passengers: int, + explicit_airlines: tuple[str, ...] | None, + cabins: tuple[str, ...], +) -> tuple[list[dict[str, AirlineSearchResponse]], PricingInfoResponse]: + """Per leg, fan out PP airline-search across (airlines x cabins). Returns + one merged airline→response map per leg, plus the shared pricing-info.""" + async with await PPClient.create() as c: + pricing = await c.pricing_info() + if explicit_airlines: + airlines = explicit_airlines + else: + try: + ext_cfg = await c.extension_config() + airlines = enabled_airlines(pricing, ext_cfg) + if not airlines: + airlines = DEFAULT_AIRLINES + except Exception: # noqa: BLE001 — fall back to static default + airlines = DEFAULT_AIRLINES + per_leg: list[dict[str, AirlineSearchResponse]] = [] + for leg in legs: + merged: dict[str, AirlineSearchResponse] = {} + for cabin in cabins: + spec = SearchSpec( + origin=leg.origin, + destination=leg.destination, + date=leg.date, + return_date="", + is_round_trip_return=False, + num_passengers=num_passengers, + cabin_class=cabin, + enable_matching=False, + ) + per_airline = await c.airline_search_many(spec, airlines) + for airline, resp in per_airline.items(): + if airline not in merged: + merged[airline] = resp + continue + # Append flights from later cabin queries; the matcher + # tolerates same-key duplicates without double-counting. + merged[airline].outboundFlights.extend(resp.outboundFlights) + per_leg.append(merged) + return per_leg, pricing + + +# ──────────────────────────── render: matched ────────────────────────────── + +_CASH_NUM_RE = __import__("re").compile(r"[\d,]*\d+(?:\.\d+)?") + + +def _parse_cash(s: str | None) -> float | None: + """Pull the first numeric value out of strings like 'USD530.00', '$1,078', + '1,078 USD'. Returns None if nothing parseable found.""" + 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 + + +def _cents_per_mile(cash_usd: float, miles: int, tax_usd: float) -> float | None: + if miles <= 0: + return None + net = max(cash_usd - tax_usd, 0.0) + return (net / miles) * 100 + + +_MILES_K_THRESHOLD = 1000 # render as "30.0k" once we cross 1000 miles + + +def _fmt_miles(n: int) -> str: + return f"{n / 1000:.1f}k" if n >= _MILES_K_THRESHOLD else str(n) + + +def _fmt_award_cell(award_options: list[AwardOption], want_cabin: str) -> str: + """Render the best (lowest miles) offer across airlines for one cabin.""" + best: tuple[int, float, str, list[str]] | None = None + for ao in award_options: + for ca in ao.cabins: + if ca.cabin != want_cabin: + continue + key = (ca.miles, ca.tax_usd, ao.airline, ao.funding_banks) + if best is None or key[0] < best[0]: + best = key + if best is None: + return "—" + miles, tax, airline, _banks = best + return f"{_fmt_miles(miles)} {airline} + ${tax:.0f}" + + +def _fmt_funding(award_options: list[AwardOption]) -> str: + banks: list[str] = [] + seen: set[str] = set() + for ao in award_options: + for b in ao.funding_banks: + if b not in seen: + seen.add(b) + banks.append(b) + return ", ".join(banks) if banks else "" + + +def _dedupe_per_leg(matches: list[MatchedFare], slice_index: int = 0) -> list[MatchedFare]: + """Matrix returns the cross-product of outbound x return itineraries, so + the same leg-flight surfaces in many rows. Collapse to one row per + (flight_number, departure_date), keeping the row with cheapest cash. + """ + best: dict[tuple[str, str], MatchedFare] = {} + for m in matches: + itn = m.itinerary.itinerary + if not itn or len(itn.slices) <= slice_index: + continue + s = itn.slices[slice_index] + if not s.flights or not s.departure: + continue + key = (s.flights[0].upper().replace(" ", ""), (s.departure or "")[:10]) + cash = _parse_cash(m.itinerary.price) or float("inf") + existing = best.get(key) + if existing is None or (_parse_cash(existing.itinerary.price) or float("inf")) > cash: + best[key] = m + # Preserve original order (cheapest cash first, since `solutions` is sorted). + seen: set[tuple[str, str]] = set() + out: list[MatchedFare] = [] + for m in matches: + itn = m.itinerary.itinerary + if not itn or len(itn.slices) <= slice_index: + continue + s = itn.slices[slice_index] + if not s.flights or not s.departure: + continue + key = (s.flights[0].upper().replace(" ", ""), (s.departure or "")[:10]) + if key in seen: + continue + if best.get(key) is m: + seen.add(key) + out.append(m) + return out + + +def _render_matches( + matches: list[MatchedFare], cabin_list: tuple[str, ...], *, slice_index: int = 0 +) -> None: + matches = _dedupe_per_leg(matches, slice_index=slice_index) + if not matches: + console.print("[yellow]No matched fares.[/]") + return + t = Table( + title="Cash + award (matched on flight # x date)", + show_header=True, + header_style="bold cyan", + ) + t.add_column("flight") + t.add_column("price", justify="right") + for cab in cabin_list: + t.add_column(cab, justify="right") + t.add_column("¢/mi (Y)", justify="right") + t.add_column("funded by") + + for m in matches: + itn = m.itinerary.itinerary + if not itn or not itn.slices or len(itn.slices) <= slice_index: + continue + s = itn.slices[slice_index] + flight = (s.flights or ["?"])[0] + cash_str = m.itinerary.price or "—" + # Parse cash price for cpm calc. Matrix returns "USD530.00", "$1,078", + # "1,078 USD" etc. — strip currency tokens and grab the first number. + cash_val: float | None = _parse_cash(cash_str) + + cells = [flight, cash_str] + cpm_str = "—" + for cab in cabin_list: + cells.append(_fmt_award_cell(m.awards, cab)) + if cab == "Economy" and cash_val is not None: + # CPM uses cheapest economy across airlines. + best_y: tuple[int, float] | None = None + for ao in m.awards: + for ca in ao.cabins: + if ca.cabin == "Economy" and (best_y is None or ca.miles < best_y[0]): + best_y = (ca.miles, ca.tax_usd) + if best_y is not None: + cpm = _cents_per_mile(cash_val, best_y[0], best_y[1]) + if cpm is not None: + cpm_str = f"{cpm:.1f}¢" + cells.append(cpm_str) + cells.append(_fmt_funding(m.awards)) + t.add_row(*cells) + console.print(t) + + +# ──────────────────────────── render: pp-only ────────────────────────────── + + +def _render_pp_only( + merged: dict[str, AirlineSearchResponse], + pricing: PricingInfoResponse, +) -> None: + rows: list[tuple[str, str, str, str, str, int, float, str]] = [] + pricing_idx = {p.airline: p for p in pricing.pricingInfos} + for airline, resp in merged.items(): + pi = pricing_idx.get(airline) + banks = ", ".join(b.bank for b in (pi.bankPointsInfos if pi else [])) + for of in resp.outboundFlights: + for c in of.perCabinMilesPricing: + pp = c.perPassengerPricing + if not pp or pp.perPassengerMilesAmount <= 0: + continue + rows.append( + ( + airline, + of.firstFlightNumber, + f"{of.origin}→{of.destination}", + of.localDepartureDateTime[:16], + c.cabinClass, + pp.perPassengerMilesAmount, + pp.perPassengerTaxAmountUsd, + banks, + ) + ) + rows.sort(key=lambda r: (r[3], r[5])) # by departure, then miles + t = Table(title="PointsPath award availability", show_header=True, header_style="bold cyan") + for col in ("airline", "flight", "route", "departs", "cabin", "miles", "tax", "funded by"): + t.add_column(col) + for r in rows: + t.add_row(r[0], r[1], r[2], r[3], r[4], _fmt_miles(r[5]), f"${r[6]:.0f}", r[7]) + console.print(t) + + +# ─────────────────────────────── json shapes ─────────────────────────────── + + +def _serialize_matches(matches: list[MatchedFare]) -> str: + out: list[dict[str, Any]] = [] + for m in matches: + itn = m.itinerary.itinerary + s = itn.slices[0] if itn and itn.slices else None + out.append( + { + "flight": (s.flights[0] if s and s.flights else None), + "departure": (s.departure if s else None), + "origin": (s.origin.code if s and s.origin else None), + "destination": (s.destination.code if s and s.destination else None), + "cash_price": m.itinerary.price, + "awards": [ + { + "airline": ao.airline, + "miles_to_cash_ratio": ao.miles_to_cash_ratio, + "funding_banks": ao.funding_banks, + "matched_origin": ao.flight.origin, + "matched_destination": ao.flight.destination, + "matched_departure": ao.flight.localDepartureDateTime, + "cabins": [ + { + "cabin": c.cabin, + "miles": c.miles, + "tax_usd": c.tax_usd, + "tax_currency": c.tax_currency, + } + for c in ao.cabins + ], + } + for ao in m.awards + ], + } + ) + return json.dumps(out, indent=2) + + +def _serialize_matches_per_leg( + matches_per_leg: list[list[MatchedFare]], legs: list[LegQuery] +) -> str: + return json.dumps( + [ + { + "leg": leg.label, + "slice_index": leg.slice_index, + "matches": json.loads(_serialize_matches(matches)), + } + for leg, matches in zip(legs, matches_per_leg, strict=True) + ], + indent=2, + ) + + +def _serialize_pp_only_per_leg( + per_leg: list[dict[str, AirlineSearchResponse]], + legs: list[LegQuery], +) -> str: + return json.dumps( + [ + { + "leg": leg.label, + "slice_index": leg.slice_index, + "by_airline": {a: r.model_dump() for a, r in merged.items()}, + } + for leg, merged in zip(legs, per_leg, strict=True) + ], + indent=2, + ) diff --git a/src/flight_cli/pp/client.py b/src/flight_cli/pp/client.py new file mode 100644 index 0000000..6133337 --- /dev/null +++ b/src/flight_cli/pp/client.py @@ -0,0 +1,327 @@ +"""Async PointsPath HTTP client. + +Three endpoints matter: + POST /api/airline-search one POST per airline (request body holds route + cabin) + GET /api/pricing-info airline catalog + transfer-partner mapping (cached 24h) + GET /api/extension-config per-tier feature flags (cached 7d) — drives airline filter + +401 → refresh tokens once → retry. Anything else propagates. +""" + +from __future__ import annotations + +import json +import re +import time +from dataclasses import dataclass +from http import HTTPStatus +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import anyio +import httpx +import structlog + +from .auth import Tokens, get_valid_tokens +from .auth import refresh as refresh_tokens +from .models import AirlineSearchResponse, PricingInfoResponse + +if TYPE_CHECKING: + from types import TracebackType + + from structlog.stdlib import BoundLogger + +log: BoundLogger = structlog.get_logger(__name__) # pyright: ignore[reportAny] + +_JsonDict = dict[str, Any] + +API_BASE = "https://api.pointspath.com" + +PRICING_CACHE = Path.home() / ".cache" / "flight-cli" / "pp_pricing.json" +PRICING_TTL_SECS = 24 * 3600 + +EXT_CONFIG_CACHE = Path.home() / ".cache" / "flight-cli" / "pp_extension_config.json" +EXT_CONFIG_TTL_SECS = 7 * 24 * 3600 +EXT_CONFIG_VERSION = "1.10.4" + +# Airlines observed firing in a single GFlights international search. Used as +# the default fan-out set when --pp-airlines isn't provided. Real catalog +# probably differs by Pro tier — the extension's /api/extension-config feature +# flags drive this — but starting with what we observed avoids guessing. +DEFAULT_AIRLINES: tuple[str, ...] = ( + "AirCanada", + "AirFrance", + "Alaska", + "American", + "Avianca", + "Delta", + "Etihad", + "JetBlue", + "Qantas", + "Qatar", + "United", + "VirginAtlantic", + "VirginAustralia", +) + +DEFAULT_CABINS: tuple[str, ...] = ("Economy", "Business") + +# PointsPath fans out concurrent queries; mirror that but bound it. The server +# already 500s under load (we saw 2/40 in our session); 5 is a safe default. +DEFAULT_CONCURRENCY = 5 + + +@dataclass +class CashFlightHint: + """Subset of GFlights itinerary data the airline-search endpoint expects + when enableGoogleFlightMatching=True. Unused when matching is False.""" + + origin: str + dest: str + start_dt: str # "YYYY-MM-DD HH:MM" + end_dt: str + flight_id: str + airline: str + google_airlines: list[str] + num_connections: int + first_flight_number: str + cash_price_usd: int + raw_cash_price: str + + def to_payload(self) -> _JsonDict: + return { + "origin": self.origin, + "dest": self.dest, + "startDateTime": self.start_dt, + "endDateTime": self.end_dt, + "flightId": self.flight_id, + "airline": self.airline, + "googleAirlines": self.google_airlines, + "numConnections": self.num_connections, + "hasCarryOnBaggage": False, + "firstFlightNumber": self.first_flight_number, + "cashPrice": self.cash_price_usd, + "rawCashPriceString": self.raw_cash_price, + } + + +@dataclass +class SearchSpec: + origin: str + destination: str + date: str # YYYY-MM-DD + return_date: str = "" + is_round_trip_return: bool = False + num_passengers: int = 1 + cabin_class: str = "Economy" + currency: str = "USD" + enable_matching: bool = False + cash_hints: tuple[CashFlightHint, ...] = () + disable_cache: bool = False + + +def _payload(spec: SearchSpec, airline: str) -> _JsonDict: + return { + "selectedFlightNumber": "", + "selectedFlightPricing": [], + "originAirport": spec.origin, + "destinationAirport": spec.destination, + "date": spec.date, + "returnDate": spec.return_date, + "numPassengers": spec.num_passengers, + "googleFlightCurrencyCode": spec.currency, + "cabinClass": spec.cabin_class, + "isRoundTripReturn": spec.is_round_trip_return, + "airline": airline, + "enableGoogleFlightMatching": spec.enable_matching, + "googleFlightDetails": [h.to_payload() for h in spec.cash_hints], + "disableCache": spec.disable_cache, + } + + +class PPClient: + """Thin async wrapper. Construct with `await PPClient.create()` so token + refresh runs once up front.""" + + def __init__( + self, + tokens: Tokens, + *, + timeout: float = 25.0, + concurrency: int = DEFAULT_CONCURRENCY, + ) -> None: + self._tokens = tokens + self._sem = anyio.Semaphore(concurrency) + self._client = httpx.AsyncClient( + base_url=API_BASE, + timeout=timeout, + headers={"user-agent": _UA}, + ) + + @classmethod + async def create(cls, **kw: Any) -> PPClient: + # get_valid_tokens is sync (httpx.post for refresh) — fine; called once. + return cls(get_valid_tokens(), **kw) + + async def aclose(self) -> None: + await self._client.aclose() + + # context-manager sugar + async def __aenter__(self) -> PPClient: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + await self.aclose() + + def _auth_headers(self) -> dict[str, str]: + return { + "authorization": f"Bearer {self._tokens.access_token}", + "content-type": "application/json", + } + + async def _request( + self, + method: str, + path: str, + *, + json_body: _JsonDict | None = None, + params: _JsonDict | None = None, + ) -> httpx.Response: + async with self._sem: + r = await self._client.request( + method, + path, + json=json_body, + params=params, + headers=self._auth_headers(), + ) + if r.status_code == HTTPStatus.UNAUTHORIZED: + log.info("pp_token_refresh", reason="401_retry_once") + self._tokens = refresh_tokens(self._tokens) + async with self._sem: + r = await self._client.request( + method, + path, + json=json_body, + params=params, + headers=self._auth_headers(), + ) + return r + + async def airline_search(self, spec: SearchSpec, airline: str) -> AirlineSearchResponse: + r = await self._request("POST", "/api/airline-search", json_body=_payload(spec, airline)) + # 204 = airline has nothing for this route+date; return empty model. + if r.status_code == HTTPStatus.NO_CONTENT or not r.content: + return AirlineSearchResponse() + if r.status_code >= HTTPStatus.BAD_REQUEST: + log.warning( + "pp_airline_search_failed", + airline=airline, + status=r.status_code, + body=r.text[:200], + ) + return AirlineSearchResponse() + return AirlineSearchResponse.model_validate(r.json()) + + async def airline_search_many( + self, + spec: SearchSpec, + airlines: tuple[str, ...], + ) -> dict[str, AirlineSearchResponse]: + """Fan out one request per airline; concurrency-bounded by the semaphore.""" + out: dict[str, AirlineSearchResponse] = {} + + async def runner(airline: str) -> None: + try: + out[airline] = await self.airline_search(spec, airline) + except Exception as e: # noqa: BLE001 - per-airline failures are non-fatal + log.warning("pp_airline_search_exception", airline=airline, error=str(e)) + + async with anyio.create_task_group() as tg: + for a in airlines: + tg.start_soon(runner, a) + return out + + async def pricing_info(self, *, force_refresh: bool = False) -> PricingInfoResponse: + if not force_refresh and PRICING_CACHE.exists(): + age = time.time() - PRICING_CACHE.stat().st_mtime + if age < PRICING_TTL_SECS: + return PricingInfoResponse.model_validate(json.loads(PRICING_CACHE.read_text())) + r = await self._request("GET", "/api/pricing-info") + r.raise_for_status() + PRICING_CACHE.parent.mkdir(parents=True, exist_ok=True) + PRICING_CACHE.write_text(r.text) + return PricingInfoResponse.model_validate(r.json()) + + async def extension_config(self, *, force_refresh: bool = False) -> dict[str, Any]: + """Fetch /api/extension-config?v=. Cached 7d on disk. + + The shape is large and version-tagged; we only consume `featureFlags` + currently, but the full body is cached so future code paths can read + other parts (e.g. valuation overrides) without re-fetching. + """ + if not force_refresh and EXT_CONFIG_CACHE.exists(): + age = time.time() - EXT_CONFIG_CACHE.stat().st_mtime + if age < EXT_CONFIG_TTL_SECS: + return json.loads(EXT_CONFIG_CACHE.read_text()) + r = await self._request( + "GET", + "/api/extension-config", + params={"v": EXT_CONFIG_VERSION}, + ) + r.raise_for_status() + EXT_CONFIG_CACHE.parent.mkdir(parents=True, exist_ok=True) + EXT_CONFIG_CACHE.write_text(r.text) + return r.json() + + +# Match `enable` exactly, or `enableV` (the +# Vn pattern is how PointsPath versions individual airline integrations — e.g. +# `enableAirFranceV2`). Sub-feature flags like `enableDeltaTakeOff15` or +# `enableSpiritSaversClub` carry trailing words and so don't match. +_AIRLINE_FLAG_RE = re.compile(r"^enable(?P[A-Z][A-Za-z]+?)(?:V\d+)?$") + + +def enabled_airlines( + pricing: PricingInfoResponse, + ext_config: dict[str, Any], +) -> tuple[str, ...]: + """Pick the airline-search call set: pricing-info airlines (universe) + intersected with extension-config feature flags. + + Rule: an airline is enabled if either (a) no `enable` flag exists + (always-on, e.g. American, Delta, United, JetBlue, Alaska), or (b) the + flag exists with value 1. Falsy flags (e.g. `enableSingapore=0`) suppress + the airline. + """ + flags: dict[str, int] = ext_config.get("featureFlags", {}) or {} + + # Build a name → truthy-or-not lookup, ignoring sub-feature flags. + flag_for_airline: dict[str, bool] = {} + for flag, val in flags.items(): + m = _AIRLINE_FLAG_RE.match(flag) + if not m: + continue + # Last writer wins on collisions (e.g. enableAirFrance + enableAirFranceV2). + # In practice PP uses one form per airline at a time. + flag_for_airline[m.group("name").lower()] = bool(val) + + enabled: list[str] = [] + for p in pricing.pricingInfos: + v = flag_for_airline.get(p.airline.lower()) + if v is False: + continue # explicitly disabled + enabled.append(p.airline) + return tuple(enabled) + + +_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/148.0.0.0 Safari/537.36" +) diff --git a/src/flight_cli/pp/match.py b/src/flight_cli/pp/match.py new file mode 100644 index 0000000..ecbcf22 --- /dev/null +++ b/src/flight_cli/pp/match.py @@ -0,0 +1,168 @@ +"""Join Matrix cash itineraries to PointsPath award flights. + +Match key: (normalized first-segment flight number, ISO departure date). +Same key can appear at most once per side per day, so a dict-lookup is enough. + +Outputs MatchedFare records, one per cash itinerary, with optional award +data attached. Caller renders. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from ..models import Itinerary, SearchResult + from .models import ( + AirlineSearchResponse, + OutboundFlight, + PerCabinMilesPricing, + PricingInfo, + PricingInfoResponse, + ) + +MatchKey = tuple[str, str] # (FLIGHT_NUMBER_UPPER_NOSPACE, "YYYY-MM-DD") + + +def _norm_fn(fn: str | None) -> str: + return (fn or "").upper().replace(" ", "") + + +def _iso_date(s: str | None) -> str: + """Best-effort isolate the YYYY-MM-DD prefix from various formats.""" + if not s: + return "" + # PointsPath: "2026-06-09T22:00:00" + # Matrix: "2026-06-09T22:00" / "2026-06-09 22:00" + s = s.replace(" ", "T") + return s[:10] + + +def cash_match_key(it: Itinerary, slice_index: int = 0) -> MatchKey | None: + """Build the match key from a Matrix itinerary's slice's first flight. + + Default slice_index=0 = outbound leg. For round-trips pass 1 to match the + return leg; for multi-city pass 2, 3, etc. + """ + itn = it.itinerary + if not itn or not itn.slices or slice_index >= len(itn.slices): + return None + s = itn.slices[slice_index] + flights = s.flights or [] + if not flights: + return None + fn = _norm_fn(flights[0]) + dep = _iso_date(s.departure) + if not fn or not dep: + return None + return (fn, dep) + + +def award_match_key(of: OutboundFlight) -> MatchKey: + return (_norm_fn(of.firstFlightNumber), _iso_date(of.localDepartureDateTime)) + + +@dataclass +class CabinAward: + """One cabin's award price for a single flight.""" + + cabin: str # "Economy" / "Business" / etc. + miles: int + tax_usd: float + tax_currency: str + is_basic_economy: bool | None = None + + +@dataclass +class AwardOption: + """All cabin offerings for a single matched flight, plus transfer info.""" + + airline: str # PointsPath canonical name (e.g. "United") + miles_to_cash_ratio: float # PointsPath valuation (¢/mi) + flight: OutboundFlight + cabins: list[CabinAward] = field(default_factory=list) + funding_banks: list[str] = field(default_factory=list) + + +@dataclass +class MatchedFare: + """One cash itinerary with zero-or-more award options attached.""" + + itinerary: Itinerary + awards: list[AwardOption] = field(default_factory=list) + + +def _cabin_awards(pricing: list[PerCabinMilesPricing]) -> list[CabinAward]: + out: list[CabinAward] = [] + for p in pricing: + pp = p.perPassengerPricing + if not pp or pp.perPassengerMilesAmount <= 0: + continue + out.append( + CabinAward( + cabin=p.cabinClass, + miles=pp.perPassengerMilesAmount, + tax_usd=pp.perPassengerTaxAmountUsd, + tax_currency=pp.taxCurrencyCode or "USD", + is_basic_economy=pp.isBasicEconomyFare, + ) + ) + return out + + +def _index_pricing(pi: PricingInfoResponse) -> dict[str, PricingInfo]: + return {p.airline: p for p in pi.pricingInfos} + + +def join( + search: SearchResult, + award_by_airline: dict[str, AirlineSearchResponse], + pricing: PricingInfoResponse, + *, + slice_index: int = 0, + use_inbound: bool = False, +) -> list[MatchedFare]: + """Outer-join cash itineraries onto award flights by (flight#, date). + + Cash itineraries with no award match keep an empty `awards` list — caller + decides whether to render them or filter to inner-join. + + `slice_index` selects which leg of each Itinerary to match against (0 for + outbound, 1 for return on a round-trip, etc). + + `use_inbound` reads from `inboundFlights` instead of `outboundFlights` — + set when joining the return leg of a round-trip query whose response is + a single bidirectional record. + """ + pricing_idx = _index_pricing(pricing) + + # Build award index. One key may surface from multiple airlines (codeshares), + # so we keep a list per key. + award_idx: dict[MatchKey, list[tuple[str, OutboundFlight]]] = {} + for airline, resp in award_by_airline.items(): + flights = resp.inboundFlights if use_inbound else resp.outboundFlights + for of in flights: + k = award_match_key(of) + if not k[0]: + continue + award_idx.setdefault(k, []).append((airline, of)) + + out: list[MatchedFare] = [] + for it in search.solutions: + k = cash_match_key(it, slice_index=slice_index) + awards: list[AwardOption] = [] + if k and k in award_idx: + for airline, of in award_idx[k]: + pi = pricing_idx.get(airline) + awards.append( + AwardOption( + airline=airline, + miles_to_cash_ratio=pi.milesToCashRatio if pi else 0.0, + flight=of, + cabins=_cabin_awards(of.perCabinMilesPricing), + funding_banks=[b.bank for b in (pi.bankPointsInfos if pi else [])], + ) + ) + out.append(MatchedFare(itinerary=it, awards=awards)) + return out diff --git a/src/flight_cli/pp/models.py b/src/flight_cli/pp/models.py new file mode 100644 index 0000000..e038892 --- /dev/null +++ b/src/flight_cli/pp/models.py @@ -0,0 +1,104 @@ +"""Pydantic v2 shapes for PointsPath API responses. + +Mirrors the `_Loose`-extra style from src/flight_cli/models.py — PointsPath +adds and removes fields without notice; we capture the parts we use and +let the rest pass through. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, ConfigDict, field_validator + + +class _Loose(BaseModel): + # DIVERGE Profile-B edge: PointsPath is reverse-engineered, adds fields + # without notice. Same justification as flight_cli.models._Loose for + # Matrix responses. + model_config = ConfigDict(extra="ignore", populate_by_name=True) + + +def _none_to_empty_list(v: Any) -> Any: + """PointsPath returns null for some list fields (`bankPointsInfos`, + `outboundFlights` for shoulder dates, etc). Coerce to [] so consumers + don't crash.""" + return [] if v is None else v + + +# ─────────────────────────── /api/airline-search ─────────────────────────── + + +class PerPassengerPricing(_Loose): + perPassengerMilesAmount: int = 0 + perPassengerTaxAmountUsd: float = 0.0 + taxCurrencyCode: str = "USD" + isBasicEconomyFare: bool | None = None + + +class OneWayPricing(_Loose): + perPassengerMilesAmount: int = 0 + perPassengerTaxAmountUsd: float = 0.0 + taxCurrencyCode: str = "" + isBasicEconomyFare: bool | None = None + + +class SelectedFlightState(_Loose): + oneWayPricing: OneWayPricing | None = None + airlineState: dict[str, Any] | None = None + + +class PerCabinMilesPricing(_Loose): + cabinClass: str + perPassengerPricing: PerPassengerPricing | None = None + selectedFlightState: SelectedFlightState | None = None + + +class OutboundFlight(_Loose): + origin: str + destination: str + localDepartureDateTime: str # ISO local without TZ, e.g. "2026-06-09T22:00:00" + localArrivalDateTime: str + firstFlightNumber: str + googleAirlineName: str | None = None + numConnections: int = 0 + externalId: str | None = None + matchedGoogleFlightId: str | None = None + matchedGoogleFlightCashPriceUsd: float | None = None + perCabinMilesPricing: list[PerCabinMilesPricing] = [] + + _none_pricing = field_validator("perCabinMilesPricing", mode="before")(_none_to_empty_list) + + +class AirlineSearchResponse(_Loose): + outboundFlights: list[OutboundFlight] = [] + inboundFlights: list[OutboundFlight] = [] + roundTripReturnState: dict[str, Any] | None = None + + _none_out = field_validator("outboundFlights", mode="before")(_none_to_empty_list) + _none_in = field_validator("inboundFlights", mode="before")(_none_to_empty_list) + + +# ─────────────────────────── /api/pricing-info ───────────────────────────── + + +class BankPointsInfo(_Loose): + bank: str + conversionValue: float = 1.0 + defaultConversionValue: float = 1.0 + isBonusActive: bool = False + conversionExpiryDate: str | None = None + + +class PricingInfo(_Loose): + airline: str + milesToCashRatio: float = 0.0 # PointsPath's valuation, e.g. 0.0125 = 1.25¢/mi + bankPointsInfos: list[BankPointsInfo] = [] + + _none_banks = field_validator("bankPointsInfos", mode="before")(_none_to_empty_list) + + +class PricingInfoResponse(_Loose): + pricingInfos: list[PricingInfo] = [] + + _none_pi = field_validator("pricingInfos", mode="before")(_none_to_empty_list) diff --git a/tests/pp/__init__.py b/tests/pp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/pp/fixtures/airline_search_united.json b/tests/pp/fixtures/airline_search_united.json new file mode 100644 index 0000000..f178992 --- /dev/null +++ b/tests/pp/fixtures/airline_search_united.json @@ -0,0 +1,51 @@ +{ + "outboundFlights": [ + { + "origin": "EWR", + "destination": "LHR", + "localDepartureDateTime": "2026-06-09T22:00:00", + "localArrivalDateTime": "2026-06-10T10:25:00", + "firstFlightNumber": "UA146", + "googleAirlineName": "United", + "numConnections": 0, + "externalId": "115b7fa5-c02b-4fe5-b54a-eb83c070c849", + "matchedGoogleFlightId": null, + "matchedGoogleFlightCashPriceUsd": null, + "perCabinMilesPricing": [ + { + "cabinClass": "Economy", + "perPassengerPricing": { + "perPassengerMilesAmount": 47000, + "perPassengerTaxAmountUsd": 282.06, + "taxCurrencyCode": "USD", + "isBasicEconomyFare": false + }, + "selectedFlightState": { + "oneWayPricing": { + "perPassengerMilesAmount": 23500, + "perPassengerTaxAmountUsd": 33.57, + "taxCurrencyCode": "USD", + "isBasicEconomyFare": false + }, + "airlineState": null + } + }, + { + "cabinClass": "Business", + "perPassengerPricing": null, + "selectedFlightState": { + "oneWayPricing": { + "perPassengerMilesAmount": 0, + "perPassengerTaxAmountUsd": 0, + "taxCurrencyCode": "", + "isBasicEconomyFare": null + }, + "airlineState": null + } + } + ] + } + ], + "inboundFlights": null, + "roundTripReturnState": null +} diff --git a/tests/pp/fixtures/extension_config.json b/tests/pp/fixtures/extension_config.json new file mode 100644 index 0000000..30f36d5 --- /dev/null +++ b/tests/pp/fixtures/extension_config.json @@ -0,0 +1,30 @@ +{ + "status": true, + "latestSafariExtensionVersion": "1.10.3", + "featureFlags": { + "enableAerLingus": 1, + "enableAirCanada": 1, + "enableAirCanadaHiddenMatching": 0, + "enableAirFranceV2": 1, + "enableAvianca": 1, + "enableDeltaTakeOff15": 1, + "enableDeltaOneWayBMPApi": 0, + "enableEmirates": 1, + "enableEtihad": 1, + "enableIberia": 0, + "enableQantas": 1, + "enableQatar": 1, + "enableSingapore": 0, + "enableSpirit": 0, + "enableSpiritSaversClub": 1, + "enableTapAirPortugal": 1, + "enableTurkish": 1, + "enableUnitedAwardToolApi": 0, + "enableUnitedWebApi": 0, + "enableVirginAtlantic": 1, + "enableVirginAustralia": 1, + "enableBasicEconomyIndicator": 1, + "enableProgramValuations": 1, + "enableManualPointsWallet": 1 + } +} diff --git a/tests/pp/fixtures/pricing_info.json b/tests/pp/fixtures/pricing_info.json new file mode 100644 index 0000000..f541897 --- /dev/null +++ b/tests/pp/fixtures/pricing_info.json @@ -0,0 +1,45 @@ +{ + "pricingInfos": [ + { + "airline": "United", + "milesToCashRatio": 0.0125, + "bankPointsInfos": [ + {"bank": "Chase", "conversionValue": 1, "defaultConversionValue": 1, "isBonusActive": false, "conversionExpiryDate": null}, + {"bank": "Bilt", "conversionValue": 1, "defaultConversionValue": 1, "isBonusActive": false, "conversionExpiryDate": null} + ] + }, + { + "airline": "Delta", + "milesToCashRatio": 0.011, + "bankPointsInfos": [ + {"bank": "Amex", "conversionValue": 1, "defaultConversionValue": 1, "isBonusActive": false, "conversionExpiryDate": null} + ] + }, + { + "airline": "American", + "milesToCashRatio": 0.013, + "bankPointsInfos": null + }, + { + "airline": "AirFrance", + "milesToCashRatio": 0.0125, + "bankPointsInfos": [ + {"bank": "Chase", "conversionValue": 0.8333, "defaultConversionValue": 1, "isBonusActive": true, "conversionExpiryDate": "2026-05-28T00:00:00Z"} + ] + }, + { + "airline": "Iberia", + "milesToCashRatio": 0.014, + "bankPointsInfos": [ + {"bank": "Citi", "conversionValue": 1, "defaultConversionValue": 1, "isBonusActive": false, "conversionExpiryDate": null} + ] + }, + { + "airline": "Singapore", + "milesToCashRatio": 0.013, + "bankPointsInfos": [ + {"bank": "Amex", "conversionValue": 1, "defaultConversionValue": 1, "isBonusActive": false, "conversionExpiryDate": null} + ] + } + ] +} diff --git a/tests/pp/test_cli.py b/tests/pp/test_cli.py new file mode 100644 index 0000000..e1a898b --- /dev/null +++ b/tests/pp/test_cli.py @@ -0,0 +1,84 @@ +# pyright: reportPrivateUsage=false +"""Tests for the small parsing/normalization helpers in pp.cli.""" + +from __future__ import annotations + +import pytest + +from flight_cli.pp.cli import _normalize_cabin, _parse_cash, _parse_csv + +# ───────────────────────────── _parse_cash ───────────────────────────────── + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("USD530.00", 530.0), # Matrix's normal format + ("$1,078", 1078.0), # Dollar prefix, comma thousand-sep + ("1,078 USD", 1078.0), # Currency suffix + ("USD 530.00", 530.0), # With space + ("USD12345.67", 12345.67), # Long number + ("$0.50", 0.5), # Sub-dollar + ("EUR250", 250.0), # Foreign currency + ("530", 530.0), # Bare number + ], +) +def test_parse_cash_recognises_common_formats(raw: str, expected: float) -> None: + assert _parse_cash(raw) == expected + + +@pytest.mark.parametrize("raw", [None, "", "—", "-", "no price"]) +def test_parse_cash_returns_none_for_empty_or_unparseable(raw: str | None) -> None: + assert _parse_cash(raw) is None + + +# ──────────────────────────── _normalize_cabin ───────────────────────────── + + +@pytest.mark.parametrize( + "alias,expected", + [ + ("y", "Economy"), + ("Y", "Economy"), + ("economy", "Economy"), + ("Economy", "Economy"), + ("coach", "Economy"), + ("Main", "Economy"), + ("j", "Business"), + ("Business", "Business"), + ("business", "Business"), + ("BUSINESS", "Business"), + ("first", "First"), + ("F", "First"), + ("premium", "Premium economy"), + ("premium-economy", "Premium economy"), + ("Premium Economy", "Premium economy"), + ], +) +def test_normalize_cabin_canonicalizes_aliases(alias: str, expected: str) -> None: + assert _normalize_cabin(alias) == expected + + +def test_normalize_cabin_passes_through_unknown() -> None: + """Unknown values pass through verbatim — the API call will silently + return 0 results rather than crashing on a typo.""" + assert _normalize_cabin("Cargo") == "Cargo" + + +# ──────────────────────────────── _parse_csv ─────────────────────────────── + + +def test_parse_csv_strips_whitespace() -> None: + assert _parse_csv(" a , b , c ", ()) == ("a", "b", "c") + + +def test_parse_csv_drops_empty_entries() -> None: + assert _parse_csv("a,,b,", ()) == ("a", "b") + + +def test_parse_csv_returns_default_when_none() -> None: + assert _parse_csv(None, ("default",)) == ("default",) + + +def test_parse_csv_returns_default_when_empty_string(): + assert _parse_csv("", ("default",)) == ("default",) diff --git a/tests/pp/test_client.py b/tests/pp/test_client.py new file mode 100644 index 0000000..f42b301 --- /dev/null +++ b/tests/pp/test_client.py @@ -0,0 +1,71 @@ +"""Tests for the PointsPath HTTP client's pure helpers (no live API).""" + +from __future__ import annotations + +import json +import pathlib +from typing import Any + +from flight_cli.pp.client import enabled_airlines +from flight_cli.pp.models import PricingInfoResponse + +FIX = pathlib.Path(__file__).parent / "fixtures" + + +def _pricing() -> PricingInfoResponse: + return PricingInfoResponse.model_validate(json.loads((FIX / "pricing_info.json").read_text())) + + +def _ext_config() -> dict[str, Any]: + return json.loads((FIX / "extension_config.json").read_text()) + + +# ────────────────────────── enabled_airlines() ────────────────────────────── + + +def test_enabled_excludes_explicitly_disabled(): + """`enableSingapore=0` and `enableIberia=0` should suppress those + airlines even though they appear in pricing-info.""" + enabled = enabled_airlines(_pricing(), _ext_config()) + assert "Iberia" not in enabled + assert "Singapore" not in enabled + + +def test_enabled_includes_always_on_airlines(): + """American has no `enable` flag at all — should still be included + (always-on tier).""" + enabled = enabled_airlines(_pricing(), _ext_config()) + assert "American" in enabled + + +def test_enabled_includes_versioned_flag_match(): + """AirFrance is gated by `enableAirFranceV2=1` — the V suffix should + match without needing a separate alias map.""" + enabled = enabled_airlines(_pricing(), _ext_config()) + assert "AirFrance" in enabled + + +def test_enabled_ignores_subfeature_flags(): + """`enableDeltaTakeOff15` is a sub-feature flag, not an airline toggle. + United has `enableUnitedAwardToolApi=0` — that's a sub-feature too, not + a disable for the airline itself. Both airlines should remain enabled.""" + enabled = enabled_airlines(_pricing(), _ext_config()) + assert "Delta" in enabled + assert "United" in enabled + + +def test_enabled_handles_missing_feature_flags_section(): + """Some response variants may omit featureFlags; should fall back to + universe (assume everything is always-on).""" + enabled = enabled_airlines(_pricing(), {"status": True}) + universe = {p.airline for p in _pricing().pricingInfos} + assert set(enabled) == universe + + +def test_enabled_returns_pricing_order(): + """Order matters for deterministic CLI output and for tests; the helper + walks pricingInfos in order, so the result preserves it.""" + enabled = enabled_airlines(_pricing(), _ext_config()) + pricing_order = [p.airline for p in _pricing().pricingInfos] + expected = [a for a in pricing_order if a in set(enabled)] + assert list(enabled) == expected diff --git a/tests/pp/test_match.py b/tests/pp/test_match.py new file mode 100644 index 0000000..f71a124 --- /dev/null +++ b/tests/pp/test_match.py @@ -0,0 +1,214 @@ +# pyright: reportCallIssue=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. +"""Tests for the cash-itinerary ↔ award-flight matcher.""" + +from __future__ import annotations + +import json +import pathlib + +from flight_cli.models import ( + Itinerary, + ItineraryDetails, + SearchResult, + Slice, + SliceEndpoint, +) +from flight_cli.pp.match import ( + award_match_key, + cash_match_key, + join, +) +from flight_cli.pp.models import ( + AirlineSearchResponse, + OutboundFlight, + PricingInfoResponse, +) + +FIX = pathlib.Path(__file__).parent / "fixtures" + + +def _itin(*slices_data: tuple[str, str, str, str]) -> Itinerary: + """Build an Itinerary with the given slices. Each tuple is + (flight_number, departure_iso, origin_iata, destination_iata).""" + 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="USD500.00", + itinerary=ItineraryDetails(slices=slcs, carriers=[]), + ) + + +def _award( + fn: str, + dep: str, + *, + miles: int = 47000, + tax: float = 250.0, + cabin: str = "Economy", + origin: str = "EWR", + dest: str = "LHR", +) -> OutboundFlight: + return OutboundFlight.model_validate( + { + "origin": origin, + "destination": dest, + "localDepartureDateTime": dep, + "localArrivalDateTime": dep, + "firstFlightNumber": fn, + "perCabinMilesPricing": [ + { + "cabinClass": cabin, + "perPassengerPricing": { + "perPassengerMilesAmount": miles, + "perPassengerTaxAmountUsd": tax, + "taxCurrencyCode": "USD", + }, + } + ], + } + ) + + +# ─────────────────────────── key normalization ───────────────────────────── + + +def test_cash_match_key_uppercases_and_strips_whitespace(): + it = _itin(("ua 146", "2026-06-09T22:00:00", "JFK", "LHR")) + assert cash_match_key(it) == ("UA146", "2026-06-09") + + +def test_cash_match_key_empty_when_no_flights(): + it = Itinerary(itinerary=ItineraryDetails(slices=[Slice(flights=[])])) + assert cash_match_key(it) is None + + +def test_cash_match_key_handles_space_separated_iso(): + """Some Matrix payloads return 'YYYY-MM-DD HH:MM' instead of ISO 'T'.""" + it = _itin(("UA146", "2026-06-09 22:00", "JFK", "LHR")) + assert cash_match_key(it) == ("UA146", "2026-06-09") + + +def test_cash_match_key_uses_slice_index_for_return_leg(): + it = _itin( + ("UA146", "2026-06-09T22:00:00", "JFK", "LHR"), # outbound + ("UA147", "2026-06-12T10:00:00", "LHR", "JFK"), # return + ) + assert cash_match_key(it, slice_index=0) == ("UA146", "2026-06-09") + assert cash_match_key(it, slice_index=1) == ("UA147", "2026-06-12") + + +def test_cash_match_key_out_of_range_slice_returns_none(): + it = _itin(("UA146", "2026-06-09T22:00:00", "JFK", "LHR")) + assert cash_match_key(it, slice_index=5) is None + + +def test_award_match_key_normalizes_consistently(): + of = _award("ua 146", "2026-06-09T22:00:00") + assert award_match_key(of) == ("UA146", "2026-06-09") + + +# ────────────────────────────── join semantics ───────────────────────────── + + +def _pricing() -> PricingInfoResponse: + return PricingInfoResponse.model_validate(json.loads((FIX / "pricing_info.json").read_text())) + + +def test_join_outer_keeps_unmatched_cash(): + """A cash itinerary whose flight # has no PP award match should still + surface in the output — with empty awards.""" + res = SearchResult( + solutions=[ + _itin(("UA146", "2026-06-09T22:00:00", "JFK", "LHR")), # will match + _itin(("BA178", "2026-06-09T07:50:00", "JFK", "LHR")), # no award + ] + ) + award_resp = AirlineSearchResponse( + outboundFlights=[ + _award("UA146", "2026-06-09T22:00:00"), + ] + ) + matches = join(res, {"United": award_resp}, _pricing()) + assert len(matches) == 2 + assert matches[0].awards and matches[0].awards[0].airline == "United" + assert matches[1].awards == [] # BA178 unmatched but preserved + + +def test_join_attaches_funding_banks_from_pricing(): + res = SearchResult( + solutions=[ + _itin(("UA146", "2026-06-09T22:00:00", "JFK", "LHR")), + ] + ) + award_resp = AirlineSearchResponse( + outboundFlights=[ + _award("UA146", "2026-06-09T22:00:00"), + ] + ) + matches = join(res, {"United": award_resp}, _pricing()) + [option] = matches[0].awards + assert sorted(option.funding_banks) == ["Bilt", "Chase"] + assert option.miles_to_cash_ratio == 0.0125 + + +def test_join_codeshare_surfaces_multiple_airlines(): + """When two airlines (PP-side) report the same flight#xdate as their own + award, both options should attach to the cash itinerary.""" + res = SearchResult( + solutions=[ + _itin(("BA178", "2026-06-09T07:50:00", "JFK", "LHR")), + ] + ) + award_by_airline = { + "American": AirlineSearchResponse( + outboundFlights=[ + _award("BA178", "2026-06-09T07:50:00", miles=30000, tax=308.0), + ] + ), + "British Airways": AirlineSearchResponse( + outboundFlights=[ + _award("BA178", "2026-06-09T07:50:00", miles=50000, tax=750.0), + ] + ), + } + matches = join(res, award_by_airline, _pricing()) + airlines = {ao.airline for ao in matches[0].awards} + assert airlines == {"American", "British Airways"} + + +def test_join_use_inbound_reads_inboundFlights(): + res = SearchResult( + solutions=[ + _itin( + ("UA146", "2026-06-09T22:00:00", "JFK", "LHR"), # outbound + ("UA147", "2026-06-12T10:00:00", "LHR", "JFK"), # return + ), + ] + ) + award_resp = AirlineSearchResponse( + outboundFlights=[], # nothing on outbound + inboundFlights=[_award("UA147", "2026-06-12T10:00:00", origin="LHR", dest="JFK")], + ) + # use_inbound=True flips the index source + matches = join(res, {"United": award_resp}, _pricing(), slice_index=1, use_inbound=True) + assert len(matches[0].awards) == 1 + + +def test_join_empty_award_response(): + res = SearchResult( + solutions=[ + _itin(("UA146", "2026-06-09T22:00:00", "JFK", "LHR")), + ] + ) + matches = join(res, {"United": AirlineSearchResponse()}, _pricing()) + assert matches[0].awards == [] diff --git a/tests/pp/test_models.py b/tests/pp/test_models.py new file mode 100644 index 0000000..c33be05 --- /dev/null +++ b/tests/pp/test_models.py @@ -0,0 +1,95 @@ +"""Pydantic shape tests. Real captured-JSON snippets in fixtures/.""" + +from __future__ import annotations + +import json +import pathlib +from typing import Any + +from flight_cli.pp.models import ( + AirlineSearchResponse, + OutboundFlight, + PricingInfoResponse, +) + +FIX = pathlib.Path(__file__).parent / "fixtures" + + +def _load(name: str) -> dict[str, Any]: + return json.loads((FIX / name).read_text()) + + +def test_airline_search_parses_real_capture(): + r = AirlineSearchResponse.model_validate(_load("airline_search_united.json")) + assert len(r.outboundFlights) == 1 + f = r.outboundFlights[0] + assert f.firstFlightNumber == "UA146" + assert f.origin == "EWR" + assert f.destination == "LHR" + assert f.localDepartureDateTime.startswith("2026-06-09T22:00") + # First cabin has full pricing + economy = f.perCabinMilesPricing[0] + assert economy.cabinClass == "Economy" + assert economy.perPassengerPricing is not None + assert economy.perPassengerPricing.perPassengerMilesAmount == 47000 + # Second cabin has perPassengerPricing=null (no availability) + business = f.perCabinMilesPricing[1] + assert business.cabinClass == "Business" + assert business.perPassengerPricing is None + + +def test_airline_search_inbound_null_coerced(): + """PointsPath returns inboundFlights=null on one-way responses; we coerce + to [] so callers can iterate without None-checks.""" + r = AirlineSearchResponse.model_validate(_load("airline_search_united.json")) + assert r.inboundFlights == [] + + +def test_outbound_flight_pricing_null_coerced(): + """`perCabinMilesPricing` arrives as null for some routes; coerce to [].""" + f = OutboundFlight.model_validate( + { + "origin": "JFK", + "destination": "LHR", + "localDepartureDateTime": "2026-06-09T22:00:00", + "localArrivalDateTime": "2026-06-10T10:25:00", + "firstFlightNumber": "UA146", + "perCabinMilesPricing": None, + } + ) + assert f.perCabinMilesPricing == [] + + +def test_pricing_info_parses_real_capture(): + p = PricingInfoResponse.model_validate(_load("pricing_info.json")) + by_airline = {pi.airline: pi for pi in p.pricingInfos} + assert "United" in by_airline + united = by_airline["United"] + assert united.milesToCashRatio == 0.0125 + banks = {b.bank for b in united.bankPointsInfos} + assert banks == {"Chase", "Bilt"} + + +def test_pricing_info_null_bankPointsInfos_coerced(): + """American's `bankPointsInfos` is null in our fixture (we observed this + in real responses). Should coerce to [] so consumers don't crash.""" + p = PricingInfoResponse.model_validate(_load("pricing_info.json")) + american = next(pi for pi in p.pricingInfos if pi.airline == "American") + assert american.bankPointsInfos == [] + + +def test_pricing_info_active_bonus_preserved(): + """Active transfer bonuses round-trip through parsing — caller relies on + `isBonusActive` and `conversionExpiryDate` to surface deal urgency.""" + p = PricingInfoResponse.model_validate(_load("pricing_info.json")) + af = next(pi for pi in p.pricingInfos if pi.airline == "AirFrance") + chase = next(b for b in af.bankPointsInfos if b.bank == "Chase") + assert chase.isBonusActive is True + assert chase.conversionExpiryDate == "2026-05-28T00:00:00Z" + assert chase.conversionValue == 0.8333 + + +def test_pricing_info_response_handles_null_top_level_list(): + """Defensive: even pricingInfos itself can come back null on errors.""" + p = PricingInfoResponse.model_validate({"pricingInfos": None}) + assert p.pricingInfos == []