"""The awards importer: placing entries, collapsing same-year collisions, and the reconcile by natural key. Before 2026-09-05 `--rebuild` deleted every network row and reinserted, so `awards.id` changed on every run and nothing could be pinned; and two entries landing on one product in one competition-year were resolved by file order, which put whichever "Elijah Craig" entry came first on the Small Batch. The reconcile runs on an in-memory SQLite with only the `awards` table (Stream D's precedent in `test_articles.py`); the matching is pure. """ from dataclasses import dataclass import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.models import Award from app.services.awards_import import SOURCE, _Desired, match_winners, reconcile_awards @dataclass(frozen=True) class P: id: int brand: str | None name: str def winner(product, brand, medal, *, year=2024, slug="nyisc", score=None, competition="New York International Spirits Competition"): return { "competition": competition, "competition_slug": slug, "brand": brand, "product": product, "year": year, "medal": medal, "score": score, } class TestSameYearCollisions: SMALL_BATCH = [P(1921, "Elijah Craig", "Elijah Craig SB Bourb 47% 1L*")] def test_entries_with_the_same_name_and_different_medals_are_ambiguous(self): """Four NYISC 2024 entries named only 'Elijah Craig' (different expressions, one name) all land on the Small Batch: none of them provably is it.""" desired, stats = match_winners(self.SMALL_BATCH, [ winner("Elijah Craig", "Elijah Craig", "Double Gold", score=96), winner("Elijah Craig", "Elijah Craig", "Silver", score=93), winner("Elijah Craig", "Elijah Craig", "Bronze", score=91), ]) assert desired == [] assert stats["ambiguous_same_year"] == 1 and stats["matched"] == 3 def test_entries_with_different_names_and_the_same_medal_are_ambiguous(self): """Don Ramon Platinium Plata and Limited Edition Plata, both Silver, on one bottle: one of them is not this bottle.""" product_variants = [P(7241, "Don Ramón", "Don Ramón Platinium Plata 70cl")] desired, stats = match_winners(product_variants, [ winner("Tequila Don Ramón Platinium Plata", "Tequila Don Ramón", "Silver", year=2021), winner("Tequila Don Ramón Plata", "Tequila Don Ramón", "Silver", year=2021), ]) assert desired == [] and stats["ambiguous_same_year"] == 1 def test_a_duplicated_entry_is_one_row(self): """Chateau Minuty listed 'Minuty Prestige Rosé' twice in NYIWC 2024, Gold both.""" product_variants = [P(7046, "Chateau Minuty", "Chateau Minuty Minuty Prestige Rosé 75cl")] desired, stats = match_winners(product_variants, [ winner("Chateau Minuty - Minuty Prestige Rosé", "Chateau Minuty", "Gold", slug="nyiwc", score=94), winner("Chateau Minuty - Minuty Prestige Rosé", "Chateau Minuty", "Gold", slug="nyiwc", score=94), ]) assert len(desired) == 1 and desired[0].medal == "Gold" assert stats["ambiguous_same_year"] == 0 def test_different_years_of_the_same_competition_are_separate_rows(self): product_variants = [P(1921, "Elijah Craig", "Elijah Craig Small Batch Bourbon 1L")] desired, _ = match_winners(product_variants, [ winner("Elijah Craig Small Batch", "Elijah Craig", "Gold", year=2023), winner("Elijah Craig Small Batch", "Elijah Craig", "Silver", year=2025), ]) assert sorted((d.year, d.medal) for d in desired) == [(2023, "Gold"), (2025, "Silver")] def test_an_entry_with_no_brand_matches_nothing(self): desired, stats = match_winners([P(1, "Oxley", "Oxley Gin 1L")], [ winner("Oxley", None, "Gold"), ]) assert desired == [] and stats["matched"] == 0 @pytest.fixture def db(): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Award.__table__.create(engine) factory = sessionmaker(bind=engine, expire_on_commit=False) session: Session = factory() yield session session.close() def desired(variant_id, slug, year, medal, score=None, competition=None): names = {"nyisc": "New York International Spirits Competition", "bisc": "Berlin International Spirits Competition"} return _Desired( variant_id=variant_id, competition=competition or names.get(slug, "Retailer's pick"), competition_slug=slug, year=year, medal=medal, score=score, ) def seed(db, variant_id, slug, year, medal, score=None, source=SOURCE): want = desired(variant_id, slug, year, medal, score) row = Award( variant_id=variant_id, competition=want.competition, competition_slug=slug, year=year, medal=medal, score=score, is_own_competition=source == SOURCE, source=source, ) db.add(row) db.commit() return row def rows(db): return { (a.variant_id, a.competition_slug, a.year, a.medal): (a.id, a.score, a.source) for a in db.scalars(select(Award)) } class TestReconcileByNaturalKey: def test_rebuild_keeps_ids_inserts_new_and_removes_stale(self, db): kept = seed(db, 1156, "nyisc", 2025, "Gold", 95) stale = seed(db, 1156, "bisc", 2021, "Gold", 94) stats = reconcile_awards(db, [ desired(1156, "nyisc", 2025, "Gold", 95), desired(1156, "bisc", 2024, "Gold", 95), ], rebuild=True) after = rows(db) assert after[(1156, "nyisc", 2025, "Gold")][0] == kept.id assert (1156, "bisc", 2024, "Gold") in after assert (1156, "bisc", 2021, "Gold") not in after assert stale.id not in {v[0] for v in after.values()} assert stats == { "created": 1, "updated": 0, "corrected": 0, "removed": 1, "unchanged": 1, "skipped_existing": 0, } def test_without_rebuild_nothing_is_removed_or_changed(self, db): seed(db, 1156, "bisc", 2021, "Gold", 94) stats = reconcile_awards(db, [desired(1156, "nyisc", 2025, "Gold", 95)], rebuild=False) assert set(rows(db)) == {(1156, "bisc", 2021, "Gold"), (1156, "nyisc", 2025, "Gold")} assert stats["removed"] == 0 and stats["created"] == 1 def test_a_medal_correction_lands_on_the_same_row(self, db): """The database allows one row per product, competition and year; a corrected medal changes the natural key but must not insert a second row (a unique violation) or renumber the first.""" row = seed(db, 50, "bisc", 2023, "Silver", 92) stats = reconcile_awards(db, [desired(50, "bisc", 2023, "Gold", 94)], rebuild=True) after = rows(db) assert after == {(50, "bisc", 2023, "Gold"): (row.id, 94, SOURCE)} assert stats["corrected"] == 1 and stats["created"] == 0 def test_a_medal_correction_is_skipped_without_rebuild(self, db): row = seed(db, 50, "bisc", 2023, "Silver", 92) stats = reconcile_awards(db, [desired(50, "bisc", 2023, "Gold", 94)], rebuild=False) assert rows(db) == {(50, "bisc", 2023, "Silver"): (row.id, 92, SOURCE)} assert stats["skipped_existing"] == 1 def test_a_score_change_updates_in_place(self, db): row = seed(db, 50, "bisc", 2023, "Gold", 93) stats = reconcile_awards(db, [desired(50, "bisc", 2023, "Gold", 94)], rebuild=False) assert rows(db) == {(50, "bisc", 2023, "Gold"): (row.id, 94, SOURCE)} assert stats["updated"] == 1 def test_other_sources_rows_are_never_touched(self, db): retailer = seed(db, 50, None, 2024, "Best Buy", None, source="retailer") # ... and a slot another source holds is not taken over. seed(db, 60, "nyisc", 2024, "Gold", 94, source="retailer") stats = reconcile_awards(db, [desired(60, "nyisc", 2024, "Silver", 92)], rebuild=True) after = rows(db) assert after[(50, None, 2024, "Best Buy")] == (retailer.id, None, "retailer") assert (60, "nyisc", 2024, "Gold") in after and (60, "nyisc", 2024, "Silver") not in after assert stats["removed"] == 0 and stats["skipped_existing"] == 1 def test_natural_key_ignores_medal_case(self, db): row = seed(db, 50, "bisc", 2023, "gold", 94) stats = reconcile_awards(db, [desired(50, "bisc", 2023, "Gold", 94)], rebuild=True) assert stats["unchanged"] == 1 and rows(db)[(50, "bisc", 2023, "gold")][0] == row.id