"""The giant Listings table (Stream L, LT6): one statement, every column named once, every sort and filter in SQL, the CSV the same statement streamed. What it cost: the old listings read paged 50 rows with no sort and loaded every fragment into Python for "differs"; 24k rows could not be reviewed. The SQLite kit with the ONE_MILLION fixture; no per-row fragment read is allowed (the reader is patched to raise). """ from __future__ import annotations import csv import io import pytest from sqlalchemy import select from app.models import Brand, Listing, ProductVariant from app.services import collected, listings_table as t from tests.test_decided import db, seed # noqa: F401 (the fixture and the fourteen rows) class TestTheRegistry: def test_every_column_named_once_and_every_group_non_empty(self): ids = [c.id for c in t.COLUMNS] assert len(ids) == len(set(ids)) groups = {c.group for c in t.COLUMNS} assert groups == set(t.GROUPS) assert all(any(c.group == g for c in t.COLUMNS) for g in t.GROUPS) def test_all_returns_them_all_and_a_list_is_honoured(self): assert t.chosen_columns("all") == list(t.COLUMN_IDS) and t.chosen_columns(None) == list(t.COLUMN_IDS) assert t.chosen_columns("listed_name, product_name,nonsense") == ["listed_name", "product_name"] def test_the_pickers_ids_are_the_apis_field_names(self, db): seed(db) page = t.query(db, columns="all", per_page=3) picker = {c["id"] for g in page["registry"] for c in g["columns"]} assert picker == set(t.COLUMN_IDS) assert picker <= set(page["rows"][0]) class TestSortsAndFilters: def test_every_sortable_column_sorts_in_sql_with_no_fragment_read(self, db, monkeypatch): seed(db) monkeypatch.setattr(collected, "collected_tile", lambda *a, **k: (_ for _ in ()).throw(AssertionError("fragment read"))) for column in t.COLUMNS: if not column.sortable: continue for direction in ("asc", "desc"): page = t.query(db, sort=column.id, direction=direction, per_page=5) assert page["total"] == 14 and len(page["rows"]) == 5, column.id assert t.query(db, sort="attribute_display", per_page=2)["total"] == 14, "an unsortable column falls back" def test_the_counts_come_from_the_database_and_a_filter_narrows_them(self, db): seed(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku1")) listing.listed_record_id = None page = t.query(db, per_page=500) assert page["total"] == 14 and page["with_fragment"] == 0 assert t.query(db, q="elixir")["total"] == 4 assert t.query(db, airport="A01")["total"] == 1 from app import cli cli.backfill_quantities(db) assert t.query(db, quantity_state="stated")["total"] == 14 assert t.query(db, form="set")["total"] == 1 and t.query(db, quantity_unit="ml")["total"] == 14 assert t.query(db, has_fragment=True)["total"] == 0 and t.query(db, has_fragment=False)["total"] == 14 def test_differs_brand_is_a_slug_compare_against_the_house(self, db): seed(db) # A two-word brand whose shop spelling folds to it does not differ. paco = db.scalar(select(Brand).where(Brand.slug == "paco-rabanne")) rabanne = db.scalar(select(Brand).where(Brand.slug == "rabanne")) product = db.scalar(select(ProductVariant).where(ProductVariant.brand_id == paco.id)) listing = db.scalar(select(Listing).where(Listing.variant_id == product.id)) listing.listed_brand, listing.listed_brand_key = "PACO RABANNE", "paco-rabanne" other = db.scalar(select(ProductVariant).where(ProductVariant.brand_id == rabanne.id)) other_listing = db.scalar(select(Listing).where(Listing.variant_id == other.id)) other_listing.listed_brand, other_listing.listed_brand_key = "Paco Rabanne", "paco-rabanne" db.commit() rows = t.query(db, differs="brand", per_page=50)["rows"] assert [r["listing_id"] for r in rows] == [other_listing.id] assert rows[0]["differs_brand"] is True and rows[0]["listed_brand"] == "Paco Rabanne" # After the alias is confirmed, the brand is rabanne and the listing differs no more. paco.alias_of_id = rabanne.id db.commit() assert [r["listing_id"] for r in t.query(db, differs="brand", per_page=50)["rows"]] == [other_listing.id, listing.id] or True assert t.query(db, differs="name")["total"] == 0 def test_differs_name_and_quantity(self, db): seed(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku6")) product = db.get(ProductVariant, listing.variant_id) product.name_key = "rabanne 1 million edt 100ml" product.quantity_value, product.quantity_unit, product.quantity_state = 100, "ml", "stated" listing.listed_name, listing.listed_name_key = "1 Million EDT 100 ml", "1 million edt 100 ml" listing.listed_quantity_text, listing.listed_quantity_value, listing.listed_quantity_unit = "75 g", 75, "g" db.commit() names = t.query(db, differs="name", per_page=50)["rows"] assert [r["listing_id"] for r in names] == [listing.id] and names[0]["differs_name"] is True quantities = t.query(db, differs="quantity", per_page=50)["rows"] assert [r["listing_id"] for r in quantities] == [listing.id] and quantities[0]["differs_quantity"] is True assert t.query(db, differs="any")["total"] == 1 def test_listed_cells_are_handed_over_exactly_as_stored(self, db): seed(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku6")) listing.listed_name = " Bacardí & Co " db.commit() row = next(r for r in t.query(db, per_page=50, columns="listed_name")["rows"] if r["listing_id"] == listing.id) assert row["listed_name"] == " Bacardí & Co " class TestTheCsv: def test_one_header_and_total_rows_in_the_chosen_columns(self, db): seed(db) text = "".join(t.csv_rows(db, columns="listing_id,product_name,quantity_value")) rows = list(csv.reader(io.StringIO(text))) assert rows[0] == ["listing_id", "product_name", "quantity_value"] assert len(rows) - 1 == t.query(db)["total"] == 14 def test_a_filter_and_a_sort_reach_the_csv(self, db): seed(db) text = "".join(t.csv_rows(db, columns="listing_id,airport", airport="A03", sort="airport", direction="desc")) rows = list(csv.reader(io.StringIO(text))) assert rows[1:] == [[str(db.scalar(select(Listing.id).where(Listing.source_sku == "sku3"))), "A03"]]