From 71f7e66d8232ec6cc795a161f0788e4c2fe18b00 Mon Sep 17 00:00:00 2001 From: peacemeals Date: Mon, 9 Feb 2026 23:25:00 +0000 Subject: [PATCH] Add `rtw fares` command to compare RTW fare prices across origins New command scrapes ExpertFlyer Fare Information to find the cheapest origin city for oneworld Explorer RTW fares. Supports querying multiple filing carriers per origin (default: AA, AS, QR, BA, FJ, RJ). For RTW tickets, origin = destination. The command queries each carrier separately, deduplicates fares, and shows the cheapest across all carriers for each fare basis (DONE3-6, AONE3-6, LONE3-6). Usage: rtw fares --origins OSL,NRT,BOM,CAI rtw fares --origins OSL,NRT --carriers AA,QR --type DONE --json Tested with OSL, NRT, BOM, CAI on AA: - BOM cheapest for DONE3 ($4,845) and DONE4 ($5,543) - NRT cheapest for all AONE fares - OSL cheapest for all LONE fares New files: - rtw/scraper/expertflyer_fares.py: Multi-carrier fare scraper + models - tests/test_scraper/test_expertflyer_fares.py: 23 unit tests Co-Authored-By: Claude Opus 4.6 --- rtw/cli.py | 237 ++++++++++ rtw/scraper/expertflyer_fares.py | 442 +++++++++++++++++++ tests/test_scraper/test_expertflyer_fares.py | 236 ++++++++++ 3 files changed, 915 insertions(+) create mode 100644 rtw/scraper/expertflyer_fares.py create mode 100644 tests/test_scraper/test_expertflyer_fares.py diff --git a/rtw/cli.py b/rtw/cli.py index 02d96c8..1bcc1ac 100644 --- a/rtw/cli.py +++ b/rtw/cli.py @@ -1061,6 +1061,7 @@ def verify( ] = None, booking_class: Annotated[Optional[str], typer.Option("--class", "-c", help="Override booking class (default: auto per carrier, AA=H, others=D)")] = None, no_cache: Annotated[bool, typer.Option("--no-cache", help="Skip cache")] = False, + date_flex: Annotated[bool, typer.Option("--flex", help="Check ±3 days for alternate availability when target date is sold out")] = False, json: JsonFlag = False, plain: PlainFlag = False, verbose: VerboseFlag = False, @@ -1070,6 +1071,10 @@ def verify( Uses ExpertFlyer to check booking class availability on each flown segment. Requires a prior `rtw search` and `rtw login expertflyer`. + + With --flex, segments with no availability on the target date will + also be checked on ±1, ±2, and ±3 adjacent days. The best alternate + date is shown in the results. """ _setup_logging(verbose, quiet) @@ -1127,6 +1132,7 @@ def verify( scraper=scraper, cache=ScrapeCache(), booking_class=booking_class, + date_flex=date_flex, ) # Note: booking_class=None means auto per-carrier (AA=H, others=D) @@ -1447,6 +1453,237 @@ def _progress(idx, total, seg_info, result): raise typer.Exit(code=2) +# --------------------------------------------------------------------------- +# Fares command +# --------------------------------------------------------------------------- + + +@app.command() +def fares( + origins: Annotated[ + str, + typer.Option( + "--origins", + "-o", + help="Comma-separated IATA origin codes (e.g. OSL,NRT,BOM,CAI).", + ), + ], + carriers: Annotated[ + str, + typer.Option( + "--carriers", + "-a", + help="Comma-separated carrier codes (default: AA,AS,QR,BA,FJ,RJ).", + ), + ] = "", + currency: Annotated[ + str, + typer.Option("--currency", help="Currency for fare display."), + ] = "USD", + fare_type: Annotated[ + Optional[str], + typer.Option( + "--type", + "-t", + help="Filter to fare type: DONE, AONE, LONE, or ALL.", + ), + ] = None, + json: JsonFlag = False, + plain: PlainFlag = False, + verbose: VerboseFlag = False, + quiet: QuietFlag = False, +) -> None: + """Compare oneworld Explorer RTW fare prices across origin cities. + + Scrapes ExpertFlyer Fare Information to find the cheapest origin + point for DONE (business), AONE (first), and LONE (economy) fares. + + Checks multiple filing carriers per origin (default: AA, AS, QR, BA, FJ, RJ). + For RTW fares, the origin and destination are the same city. + Results show the cheapest fare across all carriers for each fare basis. + + Examples: + + rtw fares --origins OSL,NRT,BOM,CAI + + rtw fares --origins OSL,NRT --carriers AA,QR --type DONE + + rtw fares --origins OSL,NRT,BOM,CAI --json + """ + _setup_logging(verbose, quiet) + + origin_list = [o.strip().upper() for o in origins.split(",") if o.strip()] + if not origin_list: + _error_panel("No origins specified. Example: --origins OSL,NRT,BOM,CAI") + raise typer.Exit(code=2) + + from rtw.scraper.expertflyer_fares import DEFAULT_RTW_CARRIERS + + carrier_list = ( + [c.strip().upper() for c in carriers.split(",") if c.strip()] + if carriers + else list(DEFAULT_RTW_CARRIERS) + ) + + try: + from rtw.scraper.expertflyer import ExpertFlyerScraper, _get_credentials + from rtw.scraper.expertflyer_fares import ExpertFlyerFareScraper + import json as json_mod + + if _get_credentials() is None: + _error_panel( + "No ExpertFlyer credentials found.\n\n" + "Run `rtw login expertflyer` to set up." + ) + raise typer.Exit(code=1) + + with ExpertFlyerScraper() as scraper: + fare_scraper = ExpertFlyerFareScraper(scraper) + + def _progress(current, total, origin, result): + if quiet: + return + rtw_count = len(result.rtw_fares) + err_count = len(result.errors) + carrier_str = ",".join(result.carriers_queried) + msg = f" [{current}/{total}] {origin} ({carrier_str}): {rtw_count} RTW fares" + if err_count: + msg += f" ({err_count} carrier errors)" + typer.echo(msg, err=True) + + if not quiet and not json: + typer.echo( + f"Searching fares for {len(origin_list)} origins " + f"x {len(carrier_list)} carriers ({','.join(carrier_list)})...", + err=True, + ) + + comparison = fare_scraper.search_multiple_origins( + origins=origin_list, + carriers=carrier_list, + currency=currency, + progress_cb=_progress if not json else None, + ) + + if json: + data = comparison.model_dump(mode="json") + typer.echo(json_mod.dumps(data, indent=2)) + else: + _display_fare_comparison(comparison, fare_type, quiet) + + except typer.Exit: + raise + except Exception as exc: + _error_panel(str(exc)) + raise typer.Exit(code=2) + + +def _display_fare_comparison(comparison, fare_type: Optional[str], quiet: bool) -> None: + """Display fare comparison results with cheapest-origin highlights.""" + # Determine which fare families to show + if fare_type: + ft = fare_type.upper() + if ft == "ALL": + families = ["LONE", "DONE", "AONE"] + elif ft in ("DONE", "AONE", "LONE"): + families = [ft] + else: + families = ["DONE"] + else: + families = ["LONE", "DONE", "AONE"] + + counts = [3, 4, 5, 6] + carriers_str = ",".join(comparison.carriers) + + try: + from rich.console import Console + from rich.table import Table + + console = Console() + + for family in families: + if family == "DONE": + title = "Business Class (DONE) Fares" + elif family == "AONE": + title = "First Class (AONE) Fares" + else: + title = "Economy Class (LONE) Fares" + + table = Table( + title=f"{title} — {carriers_str} ({comparison.currency})", + show_lines=False, + ) + table.add_column("Origin", style="bold") + for n in counts: + table.add_column(f"{family}{n}", justify="right") + + # Find cheapest per column for highlighting + cheapest_per_col = {} + for n in counts: + ranking = comparison.ranking_for(f"{family}{n}") + if ranking: + cheapest_per_col[n] = ranking[0][1] + + for origin_result in comparison.origins: + cells = [] + for n in counts: + fare_basis = f"{family}{n}" + fare = origin_result.get_fare(fare_basis) + if fare: + price_str = f"${fare.fare_usd:,.0f}" + if n in cheapest_per_col and fare.fare_usd <= cheapest_per_col[n]: + cells.append(f"[green bold]{price_str}[/green bold]") + else: + cells.append(price_str) + else: + cells.append("[dim]—[/dim]") + table.add_row(origin_result.origin, *cells) + + console.print(table) + + if not quiet: + for n in counts: + fare_basis = f"{family}{n}" + ranking = comparison.ranking_for(fare_basis) + if ranking: + best_origin, best_price, best_carrier = ranking[0] + console.print( + f" Cheapest {fare_basis}: " + f"[green bold]{best_origin}[/green bold] " + f"(${best_price:,.0f} on {best_carrier})" + ) + console.print() + + except ImportError: + for family in families: + typer.echo(f"\n{family} Fares — {carriers_str} ({comparison.currency}):") + header = f" {'Origin':<8}" + for n in counts: + header += f" {family}{n:>10}" + typer.echo(header) + typer.echo(" " + "-" * (8 + 12 * len(counts))) + + for origin_result in comparison.origins: + line = f" {origin_result.origin:<8}" + for n in counts: + fare = origin_result.get_fare(f"{family}{n}") + if fare: + line += f" ${fare.fare_usd:>9,.0f}" + else: + line += f" {'—':>10}" + typer.echo(line) + + if not quiet: + for n in counts: + fare_basis = f"{family}{n}" + ranking = comparison.ranking_for(fare_basis) + if ranking: + typer.echo( + f" Cheapest {fare_basis}: {ranking[0][0]} " + f"(${ranking[0][1]:,.0f} on {ranking[0][2]})" + ) + + # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- diff --git a/rtw/scraper/expertflyer_fares.py b/rtw/scraper/expertflyer_fares.py new file mode 100644 index 0000000..4705572 --- /dev/null +++ b/rtw/scraper/expertflyer_fares.py @@ -0,0 +1,442 @@ +"""ExpertFlyer fare information scraper. + +Scrapes RTW (oneworld Explorer) fare prices from ExpertFlyer's +Fare Information page. Reuses an existing ExpertFlyerScraper's +browser session to avoid duplicate login. + +URL pattern (origin = destination for RTW round-trip fares): + /air/fare-information/results?origin=OSL&destination=OSL + &startDate=2026-02-09&airLineCodes=AA¤cy=USD + &passengerType=ADT&filterResults=true&allFares=false + +Supports querying multiple carriers per origin (one query each). +""" + +from __future__ import annotations + +import datetime +import logging +import re +import time +from typing import Optional +from urllib.parse import quote_plus + +from pydantic import BaseModel, Field + +from rtw.scraper.expertflyer import ( + ExpertFlyerScraper, + ScrapeError, + _EXPERTFLYER_BASE, + _MIN_QUERY_INTERVAL, + _PAGE_LOAD_TIMEOUT, + _RESULTS_TIMEOUT, +) + +logger = logging.getLogger(__name__) + +_FARE_RESULTS_URL = f"{_EXPERTFLYER_BASE}/air/fare-information/results" + +# RTW fare basis pattern: xONEn or xGLOBn (x=L/D/A, n=3-6) +_RTW_FARE_PATTERN = re.compile(r"^[LDA](ONE|GLOB)\d$") + +# Default carriers to check for RTW fares +DEFAULT_RTW_CARRIERS = ["AA", "AS", "QR", "BA", "FJ", "RJ"] + + +# --------------------------------------------------------------------------- +# Models +# --------------------------------------------------------------------------- + + +class FareInfo(BaseModel): + """A single fare from ExpertFlyer fare information results.""" + + fare_basis: str + airline: str + booking_class: str + trip_type: str = "" + fare_usd: float + cabin: str = "" + effective_date: Optional[str] = None + expiration_date: Optional[str] = None + min_stay: Optional[str] = None + max_stay: Optional[str] = None + advance_purchase: Optional[str] = None + + @property + def fare_family(self) -> str: + """Extract fare family: DONE, AONE, LONE, DGLOB, etc.""" + m = re.match(r"([LDA](?:ONE|GLOB))", self.fare_basis) + return m.group(1) if m else self.fare_basis + + @property + def continent_count(self) -> Optional[int]: + """Extract continent count (3-6) from fare basis.""" + m = re.search(r"(\d)$", self.fare_basis) + return int(m.group(1)) if m else None + + @property + def is_rtw(self) -> bool: + """Whether this is an RTW fare (xONE or xGLOB pattern).""" + return bool(_RTW_FARE_PATTERN.match(self.fare_basis)) + + +class OriginFareResult(BaseModel): + """Fare results for a single origin city across one or more carriers.""" + + origin: str = Field(min_length=3, max_length=3) + carriers_queried: list[str] = Field(default_factory=list) + currency: str = "USD" + query_date: datetime.date = Field( + default_factory=lambda: datetime.date.today() + ) + fares: list[FareInfo] = Field(default_factory=list) + errors: dict[str, str] = Field(default_factory=dict) + + @property + def rtw_fares(self) -> list[FareInfo]: + """Only RTW fares (xONE/xGLOB patterns).""" + return [f for f in self.fares if f.is_rtw] + + @property + def done_fares(self) -> list[FareInfo]: + """DONE (business) fares sorted by fare basis.""" + return sorted( + [f for f in self.fares if f.fare_basis.startswith("DONE")], + key=lambda f: f.fare_basis, + ) + + @property + def aone_fares(self) -> list[FareInfo]: + """AONE (first) fares sorted by fare basis.""" + return sorted( + [f for f in self.fares if f.fare_basis.startswith("AONE")], + key=lambda f: f.fare_basis, + ) + + @property + def lone_fares(self) -> list[FareInfo]: + """LONE (economy) fares sorted by fare basis.""" + return sorted( + [f for f in self.fares if f.fare_basis.startswith("LONE")], + key=lambda f: f.fare_basis, + ) + + def get_fare(self, fare_basis: str) -> Optional[FareInfo]: + """Look up a specific fare basis. Returns cheapest if multiple carriers file it.""" + matches = [f for f in self.fares if f.fare_basis == fare_basis] + if not matches: + return None + return min(matches, key=lambda f: f.fare_usd) + + def get_fare_by_carrier(self, fare_basis: str, carrier: str) -> Optional[FareInfo]: + """Look up a fare for a specific carrier.""" + for f in self.fares: + if f.fare_basis == fare_basis and f.airline == carrier: + return f + return None + + +class FareComparisonResult(BaseModel): + """Comparison of RTW fares across multiple origin cities.""" + + origins: list[OriginFareResult] = Field(default_factory=list) + carriers: list[str] = Field(default_factory=list) + currency: str = "USD" + + def cheapest_for(self, fare_basis: str) -> Optional[OriginFareResult]: + """Find the origin with the cheapest fare for a given fare basis.""" + best = None + best_price = float("inf") + for o in self.origins: + fare = o.get_fare(fare_basis) + if fare and fare.fare_usd < best_price: + best = o + best_price = fare.fare_usd + return best + + def ranking_for(self, fare_basis: str) -> list[tuple[str, float, str]]: + """Rank origins by price for a fare basis. + + Returns (origin, price, carrier) triples sorted by price. + """ + triples = [] + for o in self.origins: + fare = o.get_fare(fare_basis) + if fare: + triples.append((o.origin, fare.fare_usd, fare.airline)) + return sorted(triples, key=lambda t: t[1]) + + +# --------------------------------------------------------------------------- +# Scraper +# --------------------------------------------------------------------------- + + +class ExpertFlyerFareScraper: + """Scrape fare information from ExpertFlyer. + + Reuses an existing ExpertFlyerScraper instance for browser/login management. + Supports querying multiple carriers per origin. + + Usage: + with ExpertFlyerScraper() as scraper: + fare_scraper = ExpertFlyerFareScraper(scraper) + result = fare_scraper.search_fares("OSL", carriers=["AA", "QR"]) + """ + + def __init__(self, scraper: ExpertFlyerScraper) -> None: + self._scraper = scraper + + def _build_fare_url( + self, + origin: str, + carrier: str, + currency: str = "USD", + date: Optional[datetime.date] = None, + ) -> str: + """Construct ExpertFlyer fare information URL. + + For RTW fares, origin = destination (round-the-world). + """ + if date is None: + date = datetime.date.today() + dt = date.strftime("%Y-%m-%d") + params = { + "origin": origin.upper(), + "destination": origin.upper(), # RTW: same origin/dest + "startDate": dt, + "airLineCodes": carrier.upper(), + "currency": currency.upper(), + "passengerType": "ADT", + "filterResults": "true", + "allFares": "false", + } + qs = "&".join(f"{k}={quote_plus(str(v))}" for k, v in params.items()) + return f"{_FARE_RESULTS_URL}?{qs}" + + def _search_single_carrier( + self, + origin: str, + carrier: str, + currency: str = "USD", + date: Optional[datetime.date] = None, + ) -> tuple[list[FareInfo], Optional[str]]: + """Search fares for a single origin+carrier combination. + + Returns (fares_list, error_message_or_none). + """ + if date is None: + date = datetime.date.today() + + self._scraper._ensure_logged_in() + + # Rate limiting + elapsed = time.time() - self._scraper._last_call_time + if elapsed < _MIN_QUERY_INTERVAL: + wait = _MIN_QUERY_INTERVAL - elapsed + 1.0 + time.sleep(wait) + + url = self._build_fare_url(origin, carrier, currency, date) + + try: + page = self._scraper._page + logger.info("ExpertFlyer fares: %s on %s", origin, carrier) + page.goto(url, timeout=_PAGE_LOAD_TIMEOUT) + time.sleep(2) + + self._scraper._check_session_expired(page) + + try: + page.wait_for_selector("table", timeout=_RESULTS_TIMEOUT) + except Exception: + self._scraper._check_session_expired(page) + body = page.evaluate("() => document.body.innerText") + if "no fares" in body.lower() or "no results" in body.lower(): + return [], None + raise ScrapeError( + f"Fare table not found for {origin} on {carrier}", + error_type="PARSE_ERROR", + ) + + self._scraper._last_call_time = time.time() + self._scraper._query_count += 1 + + fares = self._parse_fare_table(page, carrier) + return fares, None + + except ScrapeError: + raise + except Exception as exc: + self._scraper._last_call_time = time.time() + return [], str(exc) + + def search_fares( + self, + origin: str, + carriers: Optional[list[str]] = None, + currency: str = "USD", + date: Optional[datetime.date] = None, + ) -> OriginFareResult: + """Search for RTW fares from an origin city on multiple carriers. + + Args: + origin: 3-letter IATA airport code. + carriers: List of 2-letter airline codes (default: DEFAULT_RTW_CARRIERS). + currency: Currency for fare display (default USD). + date: Date for fare validity (default today). + + Returns: + OriginFareResult with fares from all queried carriers. + """ + if carriers is None: + carriers = DEFAULT_RTW_CARRIERS + if date is None: + date = datetime.date.today() + + all_fares: list[FareInfo] = [] + errors: dict[str, str] = {} + + for carrier in carriers: + try: + fares, error = self._search_single_carrier( + origin, carrier, currency, date + ) + all_fares.extend(fares) + if error: + errors[carrier] = error + except Exception as exc: + errors[carrier] = str(exc) + + # Deduplicate: if same fare basis appears on multiple carriers, + # keep the cheapest + return OriginFareResult( + origin=origin.upper(), + carriers_queried=[c.upper() for c in carriers], + currency=currency.upper(), + query_date=date, + fares=all_fares, + errors=errors, + ) + + def _parse_fare_table(self, page, carrier: str) -> list[FareInfo]: + """Parse the ExpertFlyer fare information results table.""" + fares: list[FareInfo] = [] + + try: + rows = page.query_selector_all("table tbody tr") + if not rows: + rows = page.query_selector_all("tr.hover\\:bg-sky-50") + + for row in rows: + cells = row.query_selector_all("td") + if len(cells) < 5: + continue + + cell_texts = [] + for cell in cells: + text = cell.evaluate("el => (el.innerText || '').trim()") + cell_texts.append(text) + + fare_basis = cell_texts[0].strip() if len(cell_texts) > 0 else "" + if not fare_basis: + continue + + airline = cell_texts[1].strip() if len(cell_texts) > 1 else carrier + booking_class = cell_texts[2].strip() if len(cell_texts) > 2 else "" + trip_type = cell_texts[3].strip() if len(cell_texts) > 3 else "" + fare_text = cell_texts[4].strip() if len(cell_texts) > 4 else "0" + fare_usd = self._parse_fare_amount(fare_text) + cabin = cell_texts[5].strip() if len(cell_texts) > 5 else "" + effective = cell_texts[6].strip() if len(cell_texts) > 6 else None + expiration = cell_texts[7].strip() if len(cell_texts) > 7 else None + min_max_stay = cell_texts[8].strip() if len(cell_texts) > 8 else None + advance = cell_texts[9].strip() if len(cell_texts) > 9 else None + + fares.append(FareInfo( + fare_basis=fare_basis, + airline=airline, + booking_class=booking_class, + trip_type=trip_type, + fare_usd=fare_usd, + cabin=cabin, + effective_date=effective or None, + expiration_date=expiration or None, + min_stay=min_max_stay, + advance_purchase=advance or None, + )) + + except Exception as exc: + logger.warning("Fare table parse error: %s", exc) + fares = self._parse_fare_body_text(page, carrier) + + return fares + + def _parse_fare_body_text(self, page, carrier: str) -> list[FareInfo]: + """Fallback parser using body text regex.""" + fares: list[FareInfo] = [] + try: + body = page.evaluate("() => document.body.innerText") + pattern = re.compile( + r"([LDA](?:ONE|GLOB)\d)\s+" + r"(\w{2})\s+" + r"([A-Z])\s+" + r"(\w+)\s+" + r"\$?([\d,]+\.?\d*)" + ) + for m in pattern.finditer(body): + fare_usd = float(m.group(5).replace(",", "")) + fares.append(FareInfo( + fare_basis=m.group(1), + airline=m.group(2), + booking_class=m.group(3), + trip_type=m.group(4), + fare_usd=fare_usd, + )) + except Exception as exc: + logger.warning("Body text fallback failed: %s", exc) + return fares + + @staticmethod + def _parse_fare_amount(text: str) -> float: + """Parse a fare amount string like '$5,957.88' or '5957.88'.""" + cleaned = re.sub(r"[^\d.]", "", text) + try: + return float(cleaned) + except ValueError: + return 0.0 + + def search_multiple_origins( + self, + origins: list[str], + carriers: Optional[list[str]] = None, + currency: str = "USD", + progress_cb=None, + ) -> FareComparisonResult: + """Search fares across multiple origin cities and carriers. + + Args: + origins: List of IATA airport codes. + carriers: Carrier codes to check (default: DEFAULT_RTW_CARRIERS). + currency: Currency for fare display. + progress_cb: Optional callback(current, total, origin, result). + + Returns: + FareComparisonResult with all origins compared. + """ + if carriers is None: + carriers = DEFAULT_RTW_CARRIERS + + comparison = FareComparisonResult( + carriers=[c.upper() for c in carriers], + currency=currency.upper(), + ) + + for i, origin in enumerate(origins): + result = self.search_fares(origin, carriers, currency) + comparison.origins.append(result) + + if progress_cb: + progress_cb(i + 1, len(origins), origin, result) + + return comparison diff --git a/tests/test_scraper/test_expertflyer_fares.py b/tests/test_scraper/test_expertflyer_fares.py new file mode 100644 index 0000000..1c4e4c7 --- /dev/null +++ b/tests/test_scraper/test_expertflyer_fares.py @@ -0,0 +1,236 @@ +"""Tests for ExpertFlyer fare information scraper.""" + +import datetime +from unittest.mock import MagicMock + +import pytest + +from rtw.scraper.expertflyer_fares import ( + DEFAULT_RTW_CARRIERS, + ExpertFlyerFareScraper, + FareComparisonResult, + FareInfo, + OriginFareResult, +) + + +class TestFareInfo: + """Test FareInfo model properties.""" + + def test_fare_family_done(self): + fare = FareInfo( + fare_basis="DONE4", airline="AA", booking_class="D", + fare_usd=5957.88, + ) + assert fare.fare_family == "DONE" + + def test_fare_family_aone(self): + fare = FareInfo( + fare_basis="AONE3", airline="AA", booking_class="A", + fare_usd=9447.05, + ) + assert fare.fare_family == "AONE" + + def test_fare_family_lone(self): + fare = FareInfo( + fare_basis="LONE5", airline="AA", booking_class="L", + fare_usd=2436.13, + ) + assert fare.fare_family == "LONE" + + def test_fare_family_glob(self): + fare = FareInfo( + fare_basis="DGLOB4", airline="AA", booking_class="D", + fare_usd=7000.00, + ) + assert fare.fare_family == "DGLOB" + + def test_continent_count(self): + fare = FareInfo( + fare_basis="DONE4", airline="AA", booking_class="D", + fare_usd=5957.88, + ) + assert fare.continent_count == 4 + + def test_continent_count_6(self): + fare = FareInfo( + fare_basis="AONE6", airline="AA", booking_class="A", + fare_usd=13978.14, + ) + assert fare.continent_count == 6 + + def test_is_rtw_true(self): + for basis in ["DONE3", "DONE4", "DONE5", "DONE6", + "AONE3", "AONE4", "AONE5", "AONE6", + "LONE3", "LONE4", "LONE5", "LONE6"]: + fare = FareInfo(fare_basis=basis, airline="AA", + booking_class="D", fare_usd=1000) + assert fare.is_rtw, f"{basis} should be RTW" + + def test_is_rtw_glob(self): + for basis in ["DGLOB3", "AGLOB4", "LGLOB5"]: + fare = FareInfo(fare_basis=basis, airline="AA", + booking_class="D", fare_usd=1000) + assert fare.is_rtw, f"{basis} should be RTW" + + def test_is_rtw_false(self): + fare = FareInfo( + fare_basis="YOWRT", airline="AA", booking_class="Y", + fare_usd=500, + ) + assert not fare.is_rtw + + def test_model_serialization(self): + fare = FareInfo( + fare_basis="DONE4", airline="AA", booking_class="D", + trip_type="RT", fare_usd=5957.88, cabin="B", + ) + data = fare.model_dump(mode="json") + assert data["fare_basis"] == "DONE4" + assert data["fare_usd"] == pytest.approx(5957.88) + restored = FareInfo.model_validate(data) + assert restored.fare_basis == "DONE4" + + +class TestOriginFareResult: + """Test OriginFareResult model.""" + + @pytest.fixture() + def osl_result(self): + return OriginFareResult( + origin="OSL", + carriers_queried=["AA", "QR"], + fares=[ + FareInfo(fare_basis="LONE3", airline="AA", booking_class="L", fare_usd=1772.49), + FareInfo(fare_basis="LONE4", airline="AA", booking_class="L", fare_usd=2100.10), + FareInfo(fare_basis="DONE3", airline="AA", booking_class="D", fare_usd=5386.24), + FareInfo(fare_basis="DONE4", airline="AA", booking_class="D", fare_usd=5957.88), + FareInfo(fare_basis="DONE4", airline="QR", booking_class="D", fare_usd=6200.00), + FareInfo(fare_basis="AONE3", airline="AA", booking_class="A", fare_usd=9447.05), + FareInfo(fare_basis="AONE4", airline="AA", booking_class="A", fare_usd=10901.08), + ], + ) + + def test_rtw_fares(self, osl_result): + assert len(osl_result.rtw_fares) == 7 + + def test_done_fares(self, osl_result): + done = osl_result.done_fares + assert len(done) == 3 # DONE3 AA, DONE4 AA, DONE4 QR + assert done[0].fare_basis == "DONE3" + + def test_get_fare_returns_cheapest(self, osl_result): + """get_fare should return cheapest across carriers.""" + fare = osl_result.get_fare("DONE4") + assert fare is not None + assert fare.fare_usd == pytest.approx(5957.88) + assert fare.airline == "AA" + + def test_get_fare_by_carrier(self, osl_result): + fare = osl_result.get_fare_by_carrier("DONE4", "QR") + assert fare is not None + assert fare.fare_usd == pytest.approx(6200.00) + + def test_get_fare_missing(self, osl_result): + assert osl_result.get_fare("DONE6") is None + + def test_model_serialization(self, osl_result): + data = osl_result.model_dump(mode="json") + assert data["origin"] == "OSL" + assert len(data["fares"]) == 7 + assert data["carriers_queried"] == ["AA", "QR"] + restored = OriginFareResult.model_validate(data) + assert restored.origin == "OSL" + + +class TestFareComparisonResult: + """Test FareComparisonResult model.""" + + @pytest.fixture() + def comparison(self): + return FareComparisonResult( + carriers=["AA", "QR"], + currency="USD", + origins=[ + OriginFareResult( + origin="OSL", carriers_queried=["AA", "QR"], + fares=[ + FareInfo(fare_basis="DONE4", airline="AA", + booking_class="D", fare_usd=5957.88), + ], + ), + OriginFareResult( + origin="NRT", carriers_queried=["AA", "QR"], + fares=[ + FareInfo(fare_basis="DONE4", airline="AA", + booking_class="D", fare_usd=5200.00), + ], + ), + OriginFareResult( + origin="BOM", carriers_queried=["AA", "QR"], + fares=[ + FareInfo(fare_basis="DONE4", airline="AA", + booking_class="D", fare_usd=4800.00), + ], + ), + ], + ) + + def test_cheapest_for(self, comparison): + cheapest = comparison.cheapest_for("DONE4") + assert cheapest is not None + assert cheapest.origin == "BOM" + + def test_ranking_for(self, comparison): + ranking = comparison.ranking_for("DONE4") + assert len(ranking) == 3 + assert ranking[0] == ("BOM", pytest.approx(4800.00), "AA") + assert ranking[1] == ("NRT", pytest.approx(5200.00), "AA") + assert ranking[2] == ("OSL", pytest.approx(5957.88), "AA") + + def test_cheapest_for_missing(self, comparison): + assert comparison.cheapest_for("AONE6") is None + + +class TestExpertFlyerFareScraper: + """Test ExpertFlyerFareScraper URL construction.""" + + def test_build_fare_url(self): + mock_scraper = MagicMock() + fare_scraper = ExpertFlyerFareScraper(mock_scraper) + url = fare_scraper._build_fare_url( + origin="OSL", + carrier="AA", + currency="USD", + date=datetime.date(2026, 2, 9), + ) + assert "origin=OSL" in url + assert "destination=OSL" in url + assert "airLineCodes=AA" in url + assert "currency=USD" in url + assert "startDate=2026-02-09" in url + assert "/air/fare-information/results" in url + + def test_build_fare_url_different_origin(self): + mock_scraper = MagicMock() + fare_scraper = ExpertFlyerFareScraper(mock_scraper) + url = fare_scraper._build_fare_url("NRT", "QR") + assert "origin=NRT" in url + assert "destination=NRT" in url + assert "airLineCodes=QR" in url + + def test_parse_fare_amount(self): + assert ExpertFlyerFareScraper._parse_fare_amount("$5,957.88") == pytest.approx(5957.88) + assert ExpertFlyerFareScraper._parse_fare_amount("1772.49") == pytest.approx(1772.49) + assert ExpertFlyerFareScraper._parse_fare_amount("$12,345") == pytest.approx(12345.0) + assert ExpertFlyerFareScraper._parse_fare_amount("") == pytest.approx(0.0) + assert ExpertFlyerFareScraper._parse_fare_amount("N/A") == pytest.approx(0.0) + + def test_default_carriers(self): + assert "AA" in DEFAULT_RTW_CARRIERS + assert "QR" in DEFAULT_RTW_CARRIERS + assert "BA" in DEFAULT_RTW_CARRIERS + assert "FJ" in DEFAULT_RTW_CARRIERS + assert "AS" in DEFAULT_RTW_CARRIERS + assert "RJ" in DEFAULT_RTW_CARRIERS + assert len(DEFAULT_RTW_CARRIERS) == 6