"""These are the shapes a parser misread leaves behind; the audit lists them for the human review; each case here is a real row measured on the 2026-09-05 dump. """ from collections import Counter from datetime import UTC, datetime from app.services.audit import ( THRESHOLDS, LatestRow, _live_products_stmt, brand_folds, capacity_disagreement, cross_shop_ratios, duplicate_kind, duplicate_rows, is_multipack, same_day_flips, stock_smell, ) from app.services.merges import Group def row(**overrides) -> LatestRow: """A LatestRow with every field defaulted, for tests that only care about a few.""" base = dict( listing_id=1, variant_id=1, shop_id=1, shop_code="EZE", shop_currency="USD", visible=True, source_slug="src", price=10.0, currency="USD", price_usd=10.0, was_price=None, in_stock=True, observed_at=datetime(2026, 9, 5, tzinfo=UTC), fx_source=None, url="https://example.com/p", source_sku="sku-1", name="Product", brand="Brand", gtin="0000000000000", quantity_ml=750, vertical="liquor", category="whisky", ) base.update(overrides) return LatestRow(**base) class TestIsMultipack: def test_bottles_x_count_is_a_pack(self): """"Budweiser 24 Bottles x 330ml" is a case, not a single 24 ml bottle.""" assert is_multipack("Budweiser 24 Bottles x 330ml") is True def test_upper_x_with_a_comma_decimal_is_a_pack(self): assert is_multipack("Somersby Pear 4,5% 18X0.33l") is True def test_lower_x_with_a_dot_decimal_is_a_pack(self): assert is_multipack("Tuborg Grøn 4,6% 24x0.33L DS") is True def test_twin_pack_is_a_pack(self): assert is_multipack("Hendrick's Gin Twin Pack") is True def test_a_bare_large_format_is_not_explained_by_its_name(self): """A 6-litre Moët is listed for the review; the name doesn't say pack.""" assert is_multipack("Moët & Chandon Impérial Brut 6L") is False def test_the_7000ml_glenfiddich_name_says_nothing_about_a_pack(self): """The record behind the 7,000 ml Glenfiddich: "0.70cl" read as 7 litres, and two 7-litre product variants were published with prices.""" assert is_multipack("15yo Sherry Cask") is False def test_none_and_empty_are_not_multipacks(self): assert is_multipack(None) is False assert is_multipack("") is False class TestDuplicateKind: def test_two_different_barcodes_cannot_be_told_apart_by_the_key(self): assert duplicate_kind(["5010327000206", "5010327000213"]) == "gtin_differs" def test_one_barcode_present_means_the_rest_should_fold_into_it(self): assert duplicate_kind(["5010327000206", None]) == "gtin_one" def test_no_barcode_leaves_the_key_alone_to_decide(self): assert duplicate_kind([None, None]) == "gtin_none" def test_the_same_barcode_twice_is_one_barcode(self): assert duplicate_kind(["5010327000206", "5010327000206", None]) == "gtin_one" class TestSameDayFlips: def test_two_prices_one_day_no_was_price_either_side_is_a_flip(self): observations = [ (1913, 199.99, None, datetime(2026, 8, 21, 10, 0, tzinfo=UTC)), (1913, 209.99, None, datetime(2026, 8, 21, 15, 0, tzinfo=UTC)), ] flips = same_day_flips(observations) assert flips == [{"listing_id": 1913, "day": "2026-08-21", "prices": [199.99, 209.99]}] def test_a_was_price_on_either_row_explains_it_as_a_promotion(self): observations = [ (1913, 199.99, None, datetime(2026, 8, 21, 10, 0, tzinfo=UTC)), (1913, 209.99, 219.99, datetime(2026, 8, 21, 15, 0, tzinfo=UTC)), ] assert same_day_flips(observations) == [] def test_a_one_cent_flip_is_a_rounding_drift_and_still_flagged(self): observations = [ (8641, 56.52, None, datetime(2026, 8, 23, 9, 0, tzinfo=UTC)), (8641, 56.53, None, datetime(2026, 8, 23, 20, 0, tzinfo=UTC)), ] flips = same_day_flips(observations) assert len(flips) == 1 assert flips[0]["listing_id"] == 8641 assert flips[0]["day"] == "2026-08-23" assert flips[0]["prices"] == [56.52, 56.53] def test_two_prices_on_two_different_days_is_not_a_same_day_flip(self): observations = [ (42, 10.0, None, datetime(2026, 8, 21, 10, 0, tzinfo=UTC)), (42, 20.0, None, datetime(2026, 8, 22, 10, 0, tzinfo=UTC)), ] assert same_day_flips(observations) == [] def test_the_same_price_twice_in_one_day_is_not_a_flip(self): observations = [ (42, 10.0, None, datetime(2026, 8, 21, 10, 0, tzinfo=UTC)), (42, 10.0, None, datetime(2026, 8, 21, 18, 0, tzinfo=UTC)), ] assert same_day_flips(observations) == [] class TestCrossShopRatios: def test_a_four_x_spread_is_listed_with_cheapest_and_dearest(self): rows = [ row(variant_id=101, shop_id=1, shop_code="EZE", price_usd=11.0, name="Campari Bitter 1L", brand="Campari"), row(variant_id=101, shop_id=2, shop_code="JFK", price_usd=44.0, name="Campari Bitter 1L", brand="Campari"), ] out = cross_shop_ratios(rows) assert len(out) == 1 assert out[0]["ratio"] == 4.0 assert out[0]["cheapest"]["shop"] == "EZE" assert out[0]["dearest"]["shop"] == "JFK" assert out[0]["shops"] == 2 def test_a_non_round_ratio_is_rounded_to_two_places(self): rows = [ row(variant_id=102, shop_id=1, shop_code="ZRH", price_usd=20.57, name="Gratien & Meyer Cuvee Flamme Brut 75cl"), row(variant_id=102, shop_id=2, shop_code="JFK", price_usd=77.0, name="Gratien & Meyer Cuvee Flamme Brut 75cl"), ] out = cross_shop_ratios(rows) assert len(out) == 1 assert out[0]["ratio"] == round(77.0 / 20.57, 2) def test_a_spread_under_the_limit_is_not_listed(self): rows = [ row(variant_id=103, shop_id=1, price_usd=20.0), row(variant_id=103, shop_id=2, price_usd=30.0), ] assert cross_shop_ratios(rows) == [] def test_a_product_at_one_shop_only_is_not_listed(self): rows = [row(variant_id=104, shop_id=1, price_usd=50.0)] assert cross_shop_ratios(rows) == [] def test_a_shops_own_in_stock_row_wins_over_its_cheaper_out_of_stock_row(self): rows = [ row(variant_id=105, shop_id=1, shop_code="AAA", price_usd=10.0, in_stock=False), row(variant_id=105, shop_id=1, shop_code="AAA", price_usd=12.0, in_stock=True), row(variant_id=105, shop_id=2, shop_code="BBB", price_usd=40.0, in_stock=True), ] out = cross_shop_ratios(rows) assert len(out) == 1 assert out[0]["cheapest"]["shop"] == "AAA" assert out[0]["cheapest"]["price_usd"] == 12.0 assert out[0]["ratio"] == round(40.0 / 12.0, 2) def test_a_row_with_no_usd_price_is_ignored(self): rows = [ row(variant_id=106, shop_id=1, shop_code="A", price_usd=None), row(variant_id=106, shop_id=1, shop_code="A", price_usd=15.0), row(variant_id=106, shop_id=2, shop_code="B", price_usd=60.0), ] out = cross_shop_ratios(rows) assert len(out) == 1 assert out[0]["ratio"] == 4.0 def test_results_are_sorted_by_ratio_descending(self): rows = [ row(variant_id=101, shop_id=1, shop_code="EZE", price_usd=11.0), row(variant_id=101, shop_id=2, shop_code="JFK", price_usd=44.0), row(variant_id=102, shop_id=1, shop_code="ZRH", price_usd=20.57), row(variant_id=102, shop_id=2, shop_code="JFK", price_usd=77.0), ] out = cross_shop_ratios(rows) assert [r["variant_id"] for r in out] == [101, 102] assert out[0]["ratio"] > out[1]["ratio"] class TestStockSmell: def test_every_row_true_for_two_weeks_is_dubais_record(self): """Dubai's API was read as in stock for every row for two weeks.""" counts = {"dubai-duty-free": Counter(true=1489)} assert stock_smell(counts) == ["dubai-duty-free"] def test_a_source_with_some_false_rows_is_reporting_stock(self): counts = {"extime-paris": Counter(true=6984, false=48)} assert stock_smell(counts) == [] def test_all_unknown_is_honest_not_smelly(self): counts = {"avolta-ath": Counter(unknown=170)} assert stock_smell(counts) == [] def test_a_source_under_the_minimum_sample_is_not_flagged(self): counts = {"tiny": Counter(true=5)} assert stock_smell(counts) == [] def test_output_is_sorted(self): counts = {"zeta": Counter(true=30), "alpha": Counter(true=25)} assert stock_smell(counts) == ["alpha", "zeta"] class TestCapacityDisagreement: def test_a_declared_capacity_the_per_litre_price_confirms_is_not_a_disagreement(self): attribute = { "capacity": "100", "capacity_unit": "cl", "price_per_quantity": {"price": 50, "unit": "l", "quantity": 1}, } assert capacity_disagreement(attribute, 50.0) is None def test_a_declared_70cl_against_an_implied_litre_is_a_disagreement(self): attribute = { "capacity": "70", "capacity_unit": "cl", "price_per_quantity": {"price": 50, "unit": "l", "quantity": 1}, } assert capacity_disagreement(attribute, 50.0) == {"declared_ml": 700, "implied_ml": 1000} def test_no_second_reading_means_no_disagreement_can_be_found(self): attribute = {"capacity": "70", "capacity_unit": "cl"} assert capacity_disagreement(attribute, 50.0) is None def test_700_vs_750_is_within_tolerance(self): attribute = { "capacity": "75", "capacity_unit": "cl", "price_per_quantity": {"price": 50, "unit": "l", "quantity": 1}, } assert capacity_disagreement(attribute, 35.0) is None class TestBrandFolds: def test_three_spellings_of_hendricks_and_two_of_ciroc_fold_separately(self): product_variants = [ ("Hendrick's", 1), ("Hendricks", 2), ("HENDRICK’S", 3), ("Hendrick's", 4), ("Cîroc", 5), ("CÎROC™", 6), ("Glen Moray", 7), ("Glenmorangie", 8), ] out = brand_folds(product_variants) assert len(out) == 2 hendricks, ciroc = out assert len(hendricks["spellings"]) == 3 assert hendricks["product_variants"] == 4 assert len(ciroc["spellings"]) == 2 assert not any({"Glen Moray", "Glenmorangie"} <= set(f["spellings"]) for f in out) def test_a_trailing_category_word_is_no_longer_a_fold(self): """K11.3: "Tanqueray" and "Tanqueray Gin" are two brand rows until a person joins them, so the audit sees no fold to report -- the join is a proposal, not a key.""" assert brand_folds([("Tanqueray", 5), ("Tanqueray Gin", 6)]) == [] def test_no_brand_and_empty_brand_fold_nothing(self): assert brand_folds([(None, 1), ("", 2)]) == [] class TestBrandFoldsAfterTheBrandsTable: """After migration #3 the fold is a fact (`product_variants.brand_id`), so the list is what the fold MISSED, not what it would do; on the first copy after `backfill brands` the old reading still showed all 122 folds as pending.""" def test_a_key_whose_products_share_one_brand_row_is_folded_and_drops_off(self): product_variants = [("Hendrick's", 1), ("Hendricks", 2), ("HENDRICK'S", 3)] assert brand_folds(product_variants, {1: 40, 2: 40, 3: 40}) == [] def test_a_product_with_no_brand_row_keeps_its_key_on_the_list(self): product_variants = [("Hendrick's", 1), ("Hendricks", 2), ("HENDRICK'S", 3)] (fold,) = brand_folds(product_variants, {1: 40, 2: 40, 3: None}) assert fold["key"] == "hendricks" and fold["unfolded"] is True and fold["brand_ids"] == [40] def test_two_brand_rows_under_one_key_is_a_fold_disagreement(self): (fold,) = brand_folds([("Cîroc", 5), ("CÎROC™", 6)], {5: 7, 6: 8}) assert fold["brand_ids"] == [7, 8] and fold["unfolded"] is False def test_without_brand_ids_the_pre_migration_reading_is_unchanged(self): out = brand_folds([("Hendrick's", 1), ("Hendricks", 2)]) assert len(out) == 1 and out[0]["brand_ids"] is None def test_a_single_spelling_is_never_a_fold_whatever_its_rows_say(self): assert brand_folds([("Bacardi", 1), ("Bacardi", 2)], {1: 3, 2: None}) == [] def _product(pid, name, gtin=None, attributes=None): from types import SimpleNamespace return SimpleNamespace(id=pid, name=name, gtin=gtin, attributes=attributes or {}) class TestDuplicateRowsAreTheMergeRules: """The metric counts what `backfill merges` would fold and nothing else. "Boss Bottled 10cl" against "Bottled Gift Set" (one key, one barcode) was listed as a merge to make while the merge rules had queued it for a human.""" def test_a_clean_pair_is_mergeable_and_a_set_against_its_bottle_is_not(self): clean = Group("hendricks-gin|1000", [_product(1, "Hendrick's Gin 1L", "5010327703053"), _product(2, "Hendricks Gin 1L")]) queued = Group("hugo-boss-bottled|100", [_product(7600, "Boss Bottled 10cl"), _product(12124, "Bottled Gift Set", "3614229824742")], conflicts=["set_vs_single"]) rows = duplicate_rows([clean, queued], {1: "https://h/1"}) mergeable = [r for r in rows if not r["conflicts"]] assert [r["match_key"] for r in mergeable] == ["hendricks-gin|1000"] assert rows[0]["kind"] == "gtin_one" and rows[0]["product_variants"][0]["url"] == "https://h/1" assert rows[1]["kind"] == "gtin_one" and rows[1]["conflicts"] == ["set_vs_single"] def test_two_barcodes_read_as_the_rules_conflict_not_as_a_kind_alone(self): group = Group("k|750", [_product(1, "A", "1"), _product(2, "A", "2")], conflicts=["gtin_differs"]) (row,) = duplicate_rows([group], {}) assert row["kind"] == "gtin_differs" and row["conflicts"] == ["gtin_differs"] class TestTombstonesCountNowhere: def test_the_products_the_audit_reads_are_the_live_ones(self): from sqlalchemy.dialects import postgresql sql = str(_live_products_stmt().compile(dialect=postgresql.dialect())) assert "product_variants.merged_into_id IS NULL" in sql assert "product_variants.brand_id" in sql def test_the_thresholds_are_the_post_backfill_numbers(self): """Measured on a copy of the 5 Sep dump after migration #3's backfills: the merge rules leave nothing to fold, two large formats remain unexplained, and every brand spelling has its row.""" assert THRESHOLDS["duplicate_groups"] == 0 assert THRESHOLDS["oversize_singles"] == 2 assert THRESHOLDS["brand_folds"] == 0