"""Reading prices off the Avolta storefront platform.
Three shipped incidents live here: a promoted product recorded at its
pre-promotion price, a whole store recorded in the wrong currency, and a
multi-size product recorded at its smallest variant's price.
"""
from pathlib import Path
import pytest
from app.services.collectors.avolta import (
_CONFIGURABLE_SUFFIX,
STORES,
AvoltaCollector,
declared_currency,
parse_product_variants,
size_label,
variant_prices,
)
FIXTURES = Path(__file__).parent / "fixtures"
def tile(*, sku="5000267023625", name="Some Whisky 1L", price_block="", url="https://x/en/1/p"):
return f''''''
class TestPriceExtraction:
def test_prefers_the_selling_price_over_the_crossed_out_one(self):
"""This platform renders oldPrice BEFORE finalPrice, so taking the first
price on the tile published a 20%-off Laphroaig at its full price."""
html = tile(price_block='''
£47.79
£38.23''')
item = parse_product_variants(html)[0]
assert item["price"] == 38.23 # rounded to the displayed pence
assert item["was_price"] == 47.79
def test_a_plain_tile_still_parses(self):
html = tile(price_block='£299')
item = parse_product_variants(html)[0]
assert item["price"] == 299.0
assert item["was_price"] is None
def test_a_was_price_below_the_selling_price_is_dropped(self):
"""Some shops render a 'was' equal to or under the current price;
showing that as a saving would be wrong."""
html = tile(price_block='''
£10
£12''')
assert parse_product_variants(html)[0]["was_price"] is None
def test_a_tile_without_a_price_is_skipped(self):
assert parse_product_variants(tile(price_block="")) == []
class TestCurrency:
def test_the_page_declaration_beats_the_symbol(self):
"""One store displays 'US$308.00' beside 'MX$5,390.00'. Reading the '$'
and applying the store's default recorded the dollar figure as pesos,
publishing that whole shop at a seventeenth of its real prices."""
assert declared_currency('') == "USD"
def test_reads_the_json_form_too(self):
assert declared_currency('{"priceCurrency":"CHF"}') == "CHF"
def test_absent_declaration_returns_none(self):
"""None means 'fall back to the symbol map', not 'no currency'."""
assert declared_currency("no currency here
") is None
class TestConfigurableProducts:
"""A multi-size tile advertises its CHEAPEST variant. Heathrow lists Johnnie
Walker Blue Label 1L at the 20cl price; the per-size truth is only on the
product page."""
@pytest.fixture
def product_page(self):
return (FIXTURES / "avolta_configurable_product.html").read_text()
def test_reads_the_price_of_each_size(self, product_page):
assert variant_prices(product_page) == {1000: (199.0, 200.69), 200: (54.59, None)}
def test_the_litre_is_not_the_advertised_tile_price(self, product_page):
"""The regression in one line: the tile said 54.59, the litre is 199."""
assert variant_prices(product_page)[1000][0] != 54.59
def test_unparseable_markup_yields_nothing(self):
"""Empty means 'could not resolve', and the collector then keeps the tile
price rather than dropping the listing."""
assert variant_prices("no config here") == {}
def test_parent_skus_are_recognised(self):
assert "2020675100063-P".endswith(_CONFIGURABLE_SUFFIX)
assert not "2070364000006".endswith(_CONFIGURABLE_SUFFIX)
class TestVariantEmission:
"""A multi-size tile is a family of product variants; each priced size becomes its
own listing. Which size a tile's own price means differs per store, so the
tile price is never published as-is for configurables."""
def test_size_labels_read_like_bottles(self):
assert size_label(1500) == "1.5L"
assert size_label(750) == "75cl"
assert size_label(200) == "20cl"
assert size_label(50) == "5cl"
assert size_label(1000) == "1L"
def test_variant_listings_carry_no_barcode(self):
from app.services.collectors.avolta import STORES, AvoltaCollector
collector = AvoltaCollector(STORES[0])
item = {"sku": "5490771-P::1500", "name": "Armand de Brignac Gold 1.5L",
"brand": "Armand de Brignac", "url": "https://x/en/1/armand-15l"}
listing = collector._listing(item, "CHF", "https://x/en/1/champagne", 598.4, None)
assert listing.gtin is None
assert listing.source_sku == "5490771-P::1500"
assert listing.quantity_ml == 1500 # from the name suffix
def test_a_parent_sku_with_real_ean_digits_still_resolves_for_simple_tiles(self):
from app.services.collectors.avolta import STORES, AvoltaCollector
collector = AvoltaCollector(STORES[0])
item = {"sku": "5000267023625", "name": "JW Black 1L", "brand": None,
"url": "https://x/en/1/jw-black-1l"}
listing = collector._listing(item, "GBP", "https://x/en/1/whisky", 30.0, None)
assert listing.gtin == "5000267023625"
class TestLasVegasStore:
"""Las Vegas (LAS), `lasvegas.shopdutyfree.com`, configured from the scout window
`.logs/runs/window-2026-09-19c-scout.md` (19 Sep 2026): the robots.txt as read that day
and the champagne grid trimmed to its fifteen tiles. Every number here is what the scout
counted on the real page, so a configuration that drifts from its host fails in seconds
here rather than three hours into a run at Crawl-delay 60."""
STORE = next(s for s in STORES if s.code == "LAS")
ROBOTS = (FIXTURES / "scout" / "lasvegas-shopdutyfree-com_robots.txt").read_text()
GRID = (FIXTURES / "scout" / "lasvegas-shopdutyfree-com_listing.html").read_text()
FIRST_CATEGORY = "https://lasvegas.shopdutyfree.com/en/6/wines-spirits/champagne-sparkling-wine"
def test_registered_on_the_avolta_platform(self):
from app.services.collector_view import platform_of
from app.services.collectors.registry import COLLECTORS
assert "avolta-las" in COLLECTORS
assert platform_of("avolta-las") == "Avolta (Magento)"
def test_the_shop_carries_its_airport_country_and_currency(self):
shop = AvoltaCollector(self.STORE).shops()[0]
assert (shop.code, shop.iata, shop.country, shop.currency) == ("LAS", "LAS", "United States", "USD")
assert shop.city == "Las Vegas"
def test_robots_permits_the_home_page_and_pages_beyond_the_first(self):
"""The collector derives `page_one_only` from `?p=2` on the first category; this host
forbids no query string, so its drinks categories page. Pinned to the saved file so a
later Disallow shows up as a changed fixture, not a silent one-page walk."""
from app.services.collectors.robots import parse
robots = parse(self.ROBOTS, host="https://lasvegas.shopdutyfree.com")
assert robots.allows("/en/")
assert robots.allows(f"{self.FIRST_CATEGORY}?p=2")
assert not robots.allows("/en/6/catalogsearch/result/")
assert robots.crawl_delay == 60.0 and robots.delay_for(1.0) == 60.0
def test_the_grid_parses_to_the_scouts_fifteen_tiles(self):
items = parse_product_variants(self.GRID)
assert len(items) == 15
assert sum(item["sku"].endswith(_CONFIGURABLE_SUFFIX) for item in items) == 4
by_sku = {item["sku"]: item for item in items}
# A plain tile: one typed finalPrice, no crossed-out figure.
assert by_sku["2022699130002"]["price"] == 220.0 and by_sku["2022699130002"]["was_price"] is None
assert by_sku["2022699130002"]["brand"] == "Dom Pérignon"
assert by_sku["2022699130002"]["url"] == "https://lasvegas.shopdutyfree.com/en/6/dom-perignon-champagne-75cl"
def test_a_tile_showing_a_member_price_records_it_as_the_typed_final_price(self):
"""One tile in fifteen labels its oldPrice "Regular price" (59.00) and its finalPrice
"Club Avolta price" (47.20), both shown. The parser's rule is the typed finalPrice, so
the loyalty figure becomes the price and the regular one the was-price. Whether that
is the price the shopper pays is on the running list for Stream A; this pins what the
unchanged parser does with the real tile, so a rule change is visible here."""
by_sku = {item["sku"]: item for item in parse_product_variants(self.GRID)}
assert by_sku["88110551056"]["price"] == 47.2
assert by_sku["88110551056"]["was_price"] == 59.0
def test_the_page_declares_the_configured_currency(self):
assert declared_currency(self.GRID) == "USD" == self.STORE.currency
def test_the_barcode_share_is_the_scouts(self):
"""Five of fifteen tile SKUs are barcodes by the scout's own measure (`gtin_from_sku`
on the tile SKU, parents included); two of the five are `-P` parents whose per-size
rows carry no barcode (`_listing`), so the rows a run mints carry three."""
from app.services.normalize import gtin_from_sku
items = parse_product_variants(self.GRID)
with_barcode = [item["sku"] for item in items if gtin_from_sku(item["sku"])]
assert len(with_barcode) == 5 and round(len(with_barcode) / len(items), 2) == 0.33
assert sorted(with_barcode) == [
"3049610004500-P", "3049614152337", "3185370023082", "3185370146842", "3282946017152-P",
]
def test_a_tile_naming_no_size_takes_it_from_its_address(self):
"""Veuve Clicquot's tile says "Yellow Label"; the address ends "-15l", which is 1.5
litres. The multi-size tile is a family, so the row is its per-size shape."""
item = next(i for i in parse_product_variants(self.GRID) if i["sku"] == "3049610004500-P")
assert item["name"] == "Yellow Label"
collector = AvoltaCollector(self.STORE)
row = collector._listing(
{**item, "sku": f"{item['sku']}::1500"}, "USD", self.FIRST_CATEGORY, 67.5, None,
)
assert row.quantity_ml == 1500 and row.gtin is None and row.shop_code == "LAS"
assert row.name == "Veuve Clicquot Yellow Label" and row.currency == "USD"
def test_the_store_asks_for_its_held_listings(self):
assert AvoltaCollector(self.STORE).wants_held_listings is True
def test_the_airport_has_its_declared_address(self):
from app.services import urls
assert urls.AIRPORT_SLUGS["LAS"] == "las-vegas-las"
assert urls.airport_path("LAS", "Las Vegas") == "/airports/las-vegas-las"
assert urls.parse_airport_slug("las-vegas-las") == "LAS"
assert urls.parse_airport_slug("las-las-vegas") == "LAS"
class TestLondonGatwickStore:
"""London Gatwick (LGW), `london-gatwick.worlddutyfree.com`, configured from the scout window
`.logs/runs/window-2026-09-19c-scout.md` (19 Sep 2026): the robots.txt as read that day and
the aperitifs grid trimmed to its nine tiles. Every number here is what the scout counted on
the real page. The robots matter most: `Disallow: /*?` makes this a page-one-only store like
Heathrow, and a configuration that assumed paging would spend thirty seconds a request
walking pages the host forbids."""
STORE = next(s for s in STORES if s.code == "LGW")
ROBOTS = (FIXTURES / "scout" / "london-gatwick-worlddutyfree-com_robots.txt").read_text()
GRID = (FIXTURES / "scout" / "london-gatwick-worlddutyfree-com_listing.html").read_text()
FIRST_CATEGORY = "https://london-gatwick.worlddutyfree.com/en/99/liquor/aperitifs-digestifs"
def test_registered_on_the_avolta_platform(self):
from app.services.collector_view import platform_of
from app.services.collectors.registry import COLLECTORS
assert "avolta-lgw" in COLLECTORS
assert platform_of("avolta-lgw") == "Avolta (Magento)"
def test_the_shop_carries_its_airport_country_and_currency(self):
shop = AvoltaCollector(self.STORE).shops()[0]
assert (shop.code, shop.iata, shop.country, shop.currency) == ("LGW", "LGW", "United Kingdom", "GBP")
assert shop.city == "London"
def test_robots_permits_the_home_page_but_no_query_string(self):
"""`Disallow: /*?` refuses `?p=2`, so the collector derives `page_one_only` and reads
page one of each category, then re-reads the product URLs the database holds
(`_reread_held`). Two category ids are closed outright. Pinned to the saved file so a
change either way shows up as a changed fixture."""
from app.services.collectors.robots import parse
robots = parse(self.ROBOTS, host="https://london-gatwick.worlddutyfree.com")
assert robots.allows("/en/")
assert robots.allows(self.FIRST_CATEGORY)
assert not robots.allows(f"{self.FIRST_CATEGORY}?p=2")
assert not robots.allows("/en/174/liquor") and not robots.allows("/en/175/liquor")
assert robots.crawl_delay == 30.0 and robots.delay_for(1.0) == 30.0
def test_the_grid_parses_to_the_scouts_nine_tiles(self):
items = parse_product_variants(self.GRID)
assert len(items) == 9
assert sum(item["sku"].endswith(_CONFIGURABLE_SUFFIX) for item in items) == 1
by_sku = {item["sku"]: item for item in items}
# A plain tile: one typed finalPrice, no crossed-out figure, a barcode SKU.
assert by_sku["8000040000802"]["price"] == 17.79 and by_sku["8000040000802"]["was_price"] is None
assert by_sku["8000040000802"]["brand"] == "Campari"
assert by_sku["8000040000802"]["url"] == "https://london-gatwick.worlddutyfree.com/en/99/campari-bitter-1l"
# The one promoted tile: the typed finalPrice with the crossed-out oldPrice as was-price.
assert by_sku["4290990"]["price"] == 76.49 and by_sku["4290990"]["was_price"] == 84.99
def test_the_page_declares_the_configured_currency(self):
assert declared_currency(self.GRID) == "GBP" == self.STORE.currency
def test_the_barcode_share_is_the_scouts(self):
"""Two of nine tile SKUs are barcodes by the scout's measure (`gtin_from_sku` on the
tile SKU); the other seven are the platform's own article numbers, which match only
within the operator's other shops."""
from app.services.normalize import gtin_from_sku
items = parse_product_variants(self.GRID)
with_barcode = [item["sku"] for item in items if gtin_from_sku(item["sku"])]
assert len(with_barcode) == 2 and round(len(with_barcode) / len(items), 2) == 0.22
assert sorted(with_barcode) == ["5601194700070", "8000040000802"]
def test_a_tile_naming_no_size_takes_it_from_its_address(self):
"""Aperol's multi-size tile says "Italian Spritz Cocktail"; the address ends "-1l".
The multi-size tile is a family, so the row is its per-size shape and carries no
barcode minted from the parent SKU."""
item = next(i for i in parse_product_variants(self.GRID) if i["sku"] == "6493123-P")
assert item["name"] == "Italian Spritz Cocktail"
collector = AvoltaCollector(self.STORE)
row = collector._listing(
{**item, "sku": f"{item['sku']}::1000"}, "GBP", self.FIRST_CATEGORY, 14.19, None,
)
assert row.quantity_ml == 1000 and row.gtin is None and row.shop_code == "LGW"
assert row.name == "Aperol Italian Spritz Cocktail" and row.currency == "GBP"
def test_the_store_asks_for_its_held_listings(self):
"""At a page-one-only store the held-URL re-read is how the second run grows past page
one, so the flag ingest reads must be on."""
assert AvoltaCollector(self.STORE).wants_held_listings is True
def test_the_airport_has_its_declared_address(self):
from app.services import urls
assert urls.AIRPORT_SLUGS["LGW"] == "gatwick-lgw-london"
assert urls.airport_path("LGW", "London") == "/airports/gatwick-lgw-london"
assert urls.parse_airport_slug("gatwick-lgw-london") == "LGW"
assert urls.parse_airport_slug("lgw-london") == "LGW"
class TestHelsinkiStore:
"""Helsinki (HEL), `helsinki.shopdutyfree.com`, configured from the scout window
`.logs/runs/window-2026-09-19c-scout.md` (19 Sep 2026): the robots.txt as read that day and
the aperitifs grid trimmed to its ten tiles. Every number here is what the scout counted on
the real page. The robots carry no Disallow at all, only Crawl-delay 60, so this is a paged
store and a one-page walk here would be a silent loss of most of its catalogue; and Finland
is a new country, so the flag file and both country lines land with the store, or the airport
card shows no band and nobody notices until Adam does."""
STORE = next(s for s in STORES if s.code == "HEL")
ROBOTS = (FIXTURES / "scout" / "helsinki-shopdutyfree-com_robots.txt").read_text()
GRID = (FIXTURES / "scout" / "helsinki-shopdutyfree-com_listing.html").read_text()
FIRST_CATEGORY = "https://helsinki.shopdutyfree.com/en/44/liquor/aperitifs-digestifs"
def test_registered_on_the_avolta_platform(self):
from app.services.collector_view import platform_of
from app.services.collectors.registry import COLLECTORS
assert "avolta-hel" in COLLECTORS
assert platform_of("avolta-hel") == "Avolta (Magento)"
def test_the_shop_carries_its_airport_country_and_currency(self):
shop = AvoltaCollector(self.STORE).shops()[0]
assert (shop.code, shop.iata, shop.country, shop.currency) == ("HEL", "HEL", "Finland", "EUR")
assert shop.city == "Helsinki"
def test_robots_carries_no_disallow_so_the_categories_page(self):
"""The file is a crawl delay and four sitemaps; nothing is closed, so `?p=2` is allowed
and the collector derives `page_one_only` false. Pinned to the saved file so a later
Disallow shows up as a changed fixture, not a silent one-page walk at sixty seconds a
request."""
from app.services.collectors.robots import parse
assert "Disallow" not in self.ROBOTS
robots = parse(self.ROBOTS, host="https://helsinki.shopdutyfree.com")
assert robots.allows("/en/")
assert robots.allows(self.FIRST_CATEGORY)
assert robots.allows(f"{self.FIRST_CATEGORY}?p=2")
assert robots.crawl_delay == 60.0 and robots.delay_for(1.0) == 60.0
def test_the_grid_parses_to_the_scouts_ten_tiles(self):
items = parse_product_variants(self.GRID)
assert len(items) == 10
assert sum(item["sku"].endswith(_CONFIGURABLE_SUFFIX) for item in items) == 3
by_sku = {item["sku"]: item for item in items}
# A plain tile: one typed finalPrice, no crossed-out figure, the platform's own article number.
assert by_sku["6206133"]["price"] == 15.0 and by_sku["6206133"]["was_price"] is None
assert by_sku["6206133"]["brand"] == "Campari" and by_sku["6206133"]["name"] == "Negroni Cocktail Drink 50cl"
assert by_sku["6206133"]["url"] == "https://helsinki.shopdutyfree.com/en/44/campari-negroni-cocktail-drink-50cl"
# The two promoted tiles (a "20% off" flag): the typed finalPrice with the crossed-out oldPrice as was-price.
assert by_sku["8000040000802"]["price"] == 26.8 and by_sku["8000040000802"]["was_price"] == 33.5
assert by_sku["3163937013707"]["price"] == 22.0 and by_sku["3163937013707"]["was_price"] == 27.5
def test_a_hidden_member_block_with_equal_figures_yields_no_was_price(self):
"""Each of the three multi-size tiles carries a hidden Club Avolta block typing a
"Regular price" oldPrice equal to its finalPrice (25.900000 against 25.9). The parser
reads the typed finalPrice and records no was-price for an equal figure, so no member
price and no phantom promotion reaches ingest. The visible, differing case is the Las
Vegas tile and its item on the running list; this pins that Helsinki has none."""
by_sku = {item["sku"]: item for item in parse_product_variants(self.GRID)}
for sku, price in (("6493123-P", 25.9), ("6234815-P", 16.9), ("11034410048-P", 17.0)):
assert by_sku[sku]["price"] == price and by_sku[sku]["was_price"] is None
def test_the_page_declares_the_configured_currency(self):
assert declared_currency(self.GRID) == "EUR" == self.STORE.currency
def test_the_barcode_share_is_the_scouts(self):
"""Two of ten tile SKUs are barcodes by the scout's measure (`gtin_from_sku` on the
tile SKU); the other eight are the platform's own article numbers, which match only
within the operator's other shops."""
from app.services.normalize import gtin_from_sku
items = parse_product_variants(self.GRID)
with_barcode = [item["sku"] for item in items if gtin_from_sku(item["sku"])]
assert len(with_barcode) == 2 and round(len(with_barcode) / len(items), 2) == 0.2
assert sorted(with_barcode) == ["3163937013707", "8000040000802"]
def test_a_tile_naming_no_size_takes_it_from_its_address(self):
"""Collalbrigo's multi-size tile says "Bellini Cipriani"; the address ends "-75cl".
The multi-size tile is a family, so the row is its per-size shape and carries no
barcode minted from the parent SKU."""
item = next(i for i in parse_product_variants(self.GRID) if i["sku"] == "6234815-P")
assert item["name"] == "Bellini Cipriani"
collector = AvoltaCollector(self.STORE)
row = collector._listing(
{**item, "sku": f"{item['sku']}::750"}, "EUR", self.FIRST_CATEGORY, 16.9, None,
)
assert row.quantity_ml == 750 and row.gtin is None and row.shop_code == "HEL"
assert row.name == "Collalbrigo Bellini Cipriani" and row.currency == "EUR"
def test_the_store_asks_for_its_held_listings(self):
assert AvoltaCollector(self.STORE).wants_held_listings is True
def test_the_airport_has_its_declared_address(self):
from app.services import urls
assert urls.AIRPORT_SLUGS["HEL"] == "helsinki-vantaa-hel"
assert urls.airport_path("HEL", "Helsinki") == "/airports/helsinki-vantaa-hel"
assert urls.parse_airport_slug("helsinki-vantaa-hel") == "HEL"
assert urls.parse_airport_slug("hel-helsinki") == "HEL"
def test_finland_has_its_flag_and_region(self):
"""A new country: the flag file and both country lines land together, or the card
shows no band (`test_flags.py` forbids an orphan file; this forbids an orphan line)."""
web = FIXTURES.parents[1] / "web"
src = (web / "src" / "lib" / "countries.ts").read_text()
assert '"Finland": "fi",' in src and '"Finland": "Europe",' in src
flag = web / "public" / "flags" / "fi.svg"
assert flag.is_file() and flag.stat().st_size < 20_000