"""The product line page's read (Stream K5; plan W1, W20). One line, "1 Million", with three product variants: the EDT and the Elixir at 100 ml, each at Heathrow and Paris (two comparison units, a tie), and the Parfum at 50 ml at Singapore only, also listed by an online catalogue. Each W20 case is one test. What it cost before: a variant page printed "cheapest at" whichever shop it found, the shopper's own airports or not, a catalogue shop could price the page, and a card chose its bottle by whatever list order a query returned. SQLite kit, no network.""" from __future__ import annotations from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.models import (Award, Base, Brand, Listing, PriceObservation, ProductLine, ProductVariant, Redirect, Retailer, Shop, Source) from app.models.places import Place, ShopPlace from app.services import catalog_queries as cq NOW = datetime(2026, 9, 17, 8, 0, tzinfo=UTC) TABLES = [Brand.__table__, ProductLine.__table__, ProductVariant.__table__, Retailer.__table__, Shop.__table__, Listing.__table__, PriceObservation.__table__, Award.__table__, Source.__table__, Place.__table__, ShopPlace.__table__, Redirect.__table__] EDT, ELIXIR, PARFUM = 101, 102, 103 @pytest.fixture def db(monkeypatch): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) # Verification lives in JSONB tables the kit cannot build; nothing is blocked here. monkeypatch.setattr(cq, "blocked_shop_ids", lambda db: []) with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s: s.add_all([Brand(id=1, slug="rabanne", name="Rabanne"), Retailer(id=1, slug="wdf", name="World Duty Free"), Retailer(id=2, slug="catalogue", name="The Catalogue")]) s.flush() s.add(ProductLine(id=1, brand_id=1, key="1 million", slug="rabanne-1-million", name="1 Million")) s.add_all([Shop(id=1, retailer_id=1, code="LHR1", iata="LHR", name="Heathrow T5", city="London", currency="GBP", visible=True), Shop(id=2, retailer_id=1, code="CDG1", iata="CDG", name="Paris CDG", city="Paris", currency="EUR", visible=True), Shop(id=3, retailer_id=1, code="SIN1", iata="SIN", name="Changi T3", city="Singapore", currency="SGD", visible=True), Shop(id=4, retailer_id=2, code="ONLINE", iata=None, name="The Catalogue", currency="EUR", is_catalogue_only=True, visible=True)]) s.flush() for vid, name, ml, concentration in ((EDT, "1 Million Eau de Toilette 100 ml", 100, "eau de toilette"), (ELIXIR, "1 Million Elixir 100 ml", 100, "elixir"), (PARFUM, "1 Million Parfum 50 ml", 50, "parfum")): s.add(ProductVariant(id=vid, name=name, brand="Rabanne", brand_id=1, vertical="beauty", category="Perfume", match_key=f"k{vid}", product_line_id=1, quantity_value=ml, quantity_unit="ml", quantity_state="stated", quantity_ml=ml, form="single", attributes={"attribute": concentration, "attribute_kind": "concentration"})) s.flush() listing_id = 0 for vid, shop_id, price, usd in ((EDT, 1, 80, 101.0), (EDT, 2, 95, 104.0), (ELIXIR, 1, 110, 139.0), (ELIXIR, 2, 120, 131.0), (PARFUM, 3, 150, 112.0), (PARFUM, 4, 70, 76.0)): listing_id += 1 s.add(Listing(id=listing_id, variant_id=vid, shop_id=shop_id, source_sku=f"sku{listing_id}")) s.flush() s.add(PriceObservation(listing_id=listing_id, price=price, currency="GBP", price_usd=usd, price_type="duty_free", in_stock=True, observed_at=NOW - timedelta(days=1))) s.commit() yield s def line(db, variant=None, airports=None): return cq.get_product_line(db, "rabanne-1-million", variant, airports) def test_the_line_holds_every_variant_with_its_attributes_and_units(db): d = line(db) assert [v.id for v in d.variants] == [EDT, ELIXIR, PARFUM] assert {v.id: v.units for v in d.variants} == {EDT: 2, ELIXIR: 2, PARFUM: 1}, "a catalogue is never a comparison unit" assert {v.id: v.airports for v in d.variants} == {EDT: ["CDG", "LHR"], ELIXIR: ["CDG", "LHR"], PARFUM: ["SIN"]} kinds = {s.kind: [o.text for o in s.options] for s in d.selectors} assert kinds == {"quantity": ["100 ml", "50 ml"], "concentration": ["eau de toilette", "elixir", "parfum"]} assert d.path == "/products/rabanne-1-million" and d.brand == "Rabanne" and d.indexed is False def test_case_2_several_variants_and_none_chosen_asks_for_a_variant(db): block = line(db).price_block assert block.case == "ask_variant" and block.prices == [] def test_case_3_no_airports_and_a_variant_at_several_places_prints_every_shop_s_price_here(db): """Rian, 17 Sep: *"What I'm looking for is a price at that airport displayed on that page."* The case named the airports and linked away to them; it prints them now, with their dates.""" block = line(db, ELIXIR).price_block assert block.case == "every_place" assert [p.shop_iata for p in block.prices] == ["CDG", "LHR"] assert all(p.observed_at is not None for p in block.prices), "a price always carries its date" assert [a.iata for a in block.elsewhere] == ["CDG", "LHR"] def test_case_1_no_airports_and_a_variant_at_one_place_prints_that_price_and_shop_never_the_catalogues(db): block = line(db, PARFUM).price_block assert block.case == "single_place" assert [(p.shop_iata, p.price) for p in block.prices] == [("SIN", 150.0)], "the cheaper online catalogue never prices a page" assert block.prices[0].observed_at is not None def test_case_4_chosen_airports_that_stock_it_print_those_prices_and_name_the_ones_that_do_not(db): block = line(db, EDT, ["LHR", "SIN"]).price_block assert block.case == "chosen_airports" assert [p.shop_iata for p in block.prices] == ["LHR"] and block.missing_airports == ["SIN"] def test_case_5_chosen_airports_that_stock_none_say_so_and_offer_where_it_is_even_at_one_place(db): block = line(db, PARFUM, ["LHR", "CDG"]).price_block assert block.case == "none_at_chosen" and block.prices == [] assert block.missing_airports == ["LHR", "CDG"] and [a.iata for a in block.elsewhere] == ["SIN"] assert block.elsewhere[0].path == "/airports/changi-sin-singapore" def test_airports_with_no_variant_chosen_name_the_variants_stocked_there(db): block = line(db, None, ["SIN"]).price_block assert block.case == "ask_variant" and block.variants_at_airports == [PARFUM] def test_the_representative_is_the_most_units_then_the_lowest_id_on_a_tie(db): """The Rabanne tie: the EDT and the Elixir at 100 ml each at two places; the card stands for the lower id, whichever one a collection touched last.""" assert line(db).representative_variant_id == EDT db.get(PriceObservation, 3).observed_at = NOW db.commit() assert line(db).representative_variant_id == EDT def test_a_merged_away_variant_follows_to_its_survivor_and_a_foreign_one_is_not_chosen(db): db.add(ProductVariant(id=999, name="1 Million EDT 100ml", brand="Rabanne", brand_id=1, match_key="old", merged_into_id=EDT, product_line_id=1)) db.commit() assert line(db, 999).chosen_variant_id == EDT assert line(db, 424242).chosen_variant_id is None def test_an_alias_line_or_a_retired_slug_forwards_and_a_hidden_line_answers_nothing(db): db.add(ProductLine(id=2, brand_id=1, key="one million", slug="rabanne-one-million", name="One Million", alias_of_id=1)) db.add(Redirect(from_slug="paco-rabanne-1-million", kind="product_line", to_slug="rabanne-1-million")) db.commit() assert cq.line_by_slug(db, "rabanne-one-million") == (None, "rabanne-1-million") assert cq.line_by_slug(db, "paco-rabanne-1-million") == (None, "rabanne-1-million") assert cq.get_product_line(db, "rabanne-one-million") is None db.get(ProductLine, 1).hidden = True db.commit() assert cq.line_by_slug(db, "rabanne-1-million") == (None, None) and line(db) is None def test_a_card_deep_links_to_its_line_variant_and_the_airports_its_comparison_shows(db): """Plan W1's card contract: the card opens exactly the comparison it promised. Its priced rows are `(label, iata, usd, in_stock)`, cheapest first; a catalogue row has no code; four are shown.""" priced = [("LHR", "LHR", 101.0, True), ("Online", None, 76.0, True), ("CDG", "CDG", 104.0, True), ("SIN", "SIN", 112.0, True), ("DXB", "DXB", 120.0, True)] assert cq.card_path("rabanne-1-million", EDT, priced) == "/products/rabanne-1-million?variant=101&airports=LHR,CDG,SIN" assert cq.card_path(None, EDT, priced) is None assert cq.line_slugs_of(db, [EDT, PARFUM, 555]) == {EDT: "rabanne-1-million", PARFUM: "rabanne-1-million"} block = line(db, EDT, ["LHR", "CDG", "SIN"]).price_block assert block.case == "chosen_airports" and [p.shop_iata for p in block.prices] == ["LHR", "CDG"] and block.missing_airports == ["SIN"] def test_the_address_helpers_read_leniently_and_the_spa_mirrors_them(): """urls.py line_path and read_airports are mirrored by lib/urls.ts linePath and readAirports: a server link and a client link to one view must be the same string, or a card's comparison would open at a different address than the server rendered.""" import pathlib from app.services import urls assert urls.read_airports("lhr, CDG,xx,LHR,sin2,,JFK") == ["LHR", "CDG", "JFK"] assert urls.read_airports(None) == [] and len(urls.read_airports(",".join(["AAA", "BBB", "CCC"] * 5))) == 3 assert urls.line_path("rabanne-1-million") == "/products/rabanne-1-million" assert urls.line_path("rabanne-1-million", 101, ["lhr", "CDG"]) == "/products/rabanne-1-million?variant=101&airports=LHR,CDG" ts = (pathlib.Path(__file__).resolve().parents[1] / "web" / "src" / "lib" / "urls.ts").read_text() assert "export function linePath(" in ts and "export function readAirports(" in ts assert "`variant=${variantId}`" in ts and "`airports=${codes.join(\",\")}`" in ts and "/^[A-Z]{3}$/" in ts class TestThePriceBlockRianAskedFor: """The four states rian walked through on staging, 17 Sep, with a line page open. *"it says 'choose your quantity to see its price' but there is no ability to choose it right there in that section... if you choose the quantity, now it shows you a list of airports you can select. But if you click it, that takes you to the airport page. What I'm looking for is a price at that airport displayed on that page. So if the user has no airports selected, we assume they are not looking for comparisons but rather just looking for pricing for the different variants at different airports. And even if they do have airports selected, we should have some sort of option to browse the prices at different airports (not up front, but still accessible without deselecting their airports)."* What it cost: the block asked for a choice it gave no way to make, because it offered only the variants stocked at the shopper's airports and a shopper with no airports has none of those. """ def test_nothing_chosen_offers_every_variant_in_the_block_itself(self, db): block = line(db).price_block assert block.case == "ask_variant" and block.prices == [] assert block.variants_offered == [EDT, ELIXIR, PARFUM], "the choice is offered here, not only above" assert block.variants_at_airports == [], "and none is marked, because no airport was chosen" def test_nothing_chosen_with_airports_marks_which_are_stocked_there_and_still_offers_the_rest(self, db): block = line(db, airports=["SIN"]).price_block assert block.variants_at_airports == [PARFUM] assert block.variants_offered == [EDT, ELIXIR, PARFUM], "the other sizes are still choosable" def test_a_bottle_chosen_with_no_airports_prints_every_shop_s_price_with_its_date(self, db): block = line(db, ELIXIR).price_block assert block.case == "every_place" assert [(p.shop_iata, p.price) for p in block.prices] == [("CDG", 120.0), ("LHR", 110.0)] assert all(p.observed_at is not None for p in block.prices) assert [a.iata for a in block.elsewhere] == ["CDG", "LHR"], "the airports are still named, and now priced" def test_airports_chosen_keeps_the_comparison_and_carries_the_other_shops_beside_it(self, db): block = line(db, ELIXIR, ["LHR"]).price_block assert block.case == "chosen_airports" assert [p.shop_iata for p in block.prices] == ["LHR"], "the comparison is unchanged" assert [p.shop_iata for p in block.elsewhere_prices] == ["CDG"], "and CDG is one disclosure away" assert all(p.observed_at is not None for p in block.elsewhere_prices) def test_a_shop_the_shopper_did_not_choose_is_never_mixed_into_the_comparison(self, db): block = line(db, ELIXIR, ["LHR"]).price_block assert {p.shop_iata for p in block.prices} & {p.shop_iata for p in block.elsewhere_prices} == set() assert len(block.prices) + len(block.elsewhere_prices) == 2 def test_an_airport_that_stocks_none_of_it_offers_the_prices_rather_than_a_link_away(self, db): block = line(db, PARFUM, ["LHR"]).price_block assert block.case == "none_at_chosen" and block.missing_airports == ["LHR"] assert [p.shop_iata for p in block.elsewhere_prices] == ["SIN"], "what it does cost, here on the page" def test_a_catalogue_shop_never_prices_the_page_in_any_state(self, db): for block in (line(db, PARFUM).price_block, line(db, PARFUM, ["LHR"]).price_block): everything = [*block.prices, *block.elsewhere_prices] assert everything and not any(p.is_catalogue_only for p in everything)