loyaly-catalogue/backend/tests/test_elec_database.py

"""Database tests against the local electronics_catalog_test database:
constraints, append-only history, verification rule, views and the API.
Skipped when the local Postgres container is not running."""
from __future__ import annotations

from decimal import Decimal

import psycopg
import pytest

from app.electronics.collector import Collector, RunOptions
from app.electronics.db import repository as repo
from app.electronics.db.connection import connect
from app.electronics.models import Listing
from app.electronics.normalise.title_parser import parse_title, variant_key


def _listing(site: str, sku: str, title: str, *, price=None, source_type="search_snippet",
             evidence=None) -> Listing:
    p = parse_title(title, "mobiles")
    l = Listing(site_domain=site, source_sku=sku, source_url=f"https://www.{site}/p/{sku}",
                source_type=source_type, brand_slug=p.brand.brand_slug, category="mobiles", title=title,
                evidence_text=evidence or f"{title} ₹{price}", confidence=0.5, parser="test",
                model=p.model, ram_gb=p.ram_gb, storage_gb=p.storage_gb, price=price)
    l.model_norm, l.variant_key = p.model_norm, variant_key(p, "mobiles")
    return l


def test_settings_guard_refuses_remote_database():
    from app.infrastructure.settings import _guard_local_database

    with pytest.raises(RuntimeError, match="not a local host"):
        _guard_local_database("31.97.228.132", "electronics_catalog")
    with pytest.raises(RuntimeError, match="expected 'electronics_catalog'"):
        _guard_local_database("localhost", "pgvector")
    _guard_local_database("localhost", "electronics_catalog")


def test_settings_guard_remote_opt_in_is_exact():
    from app.infrastructure.settings import _guard_local_database

    prod = dict(remote_hosts=frozenset({"31.97.228.132"}), remote_names=frozenset({"loyalycatalogue"}))
    # Listed host + name, with the flag on: allowed.
    _guard_local_database("31.97.228.132", "loyalycatalogue", allow_remote=True, **prod)
    # Same host without the flag: still refused.
    with pytest.raises(RuntimeError, match="not a local host"):
        _guard_local_database("31.97.228.132", "loyalycatalogue", **prod)
    # Flag on but another host or another database: refused.
    with pytest.raises(RuntimeError, match="not a local host"):
        _guard_local_database("10.0.0.9", "loyalycatalogue", allow_remote=True, **prod)
    with pytest.raises(RuntimeError, match="not in ELEC_REMOTE_DB_NAMES"):
        _guard_local_database("31.97.228.132", "pgvector", allow_remote=True, **prod)


def test_constraints_reject_fabricated_rows(db):
    ids = repo.id_maps()
    with connect() as conn:
        base = dict(site=ids["site"]["croma.com"], brand=ids["brand"]["samsung"], cat=ids["category"]["mobiles"])
        bad_rows = [
            ("no URL", "INSERT INTO elec.source_listing (site_id, source_sku, source_url, source_type, brand_id, "
             "category_id, title, evidence_text, confidence, parser) VALUES (%(site)s,'x','not-a-url',"
             "'search_snippet',%(brand)s,%(cat)s,'t','e',0.5,'t')"),
            ("no evidence", "INSERT INTO elec.source_listing (site_id, source_sku, source_url, source_type, brand_id, "
             "category_id, title, evidence_text, confidence, parser) VALUES (%(site)s,'x','https://a.in/x',"
             "'search_snippet',%(brand)s,%(cat)s,'t','',0.5,'t')"),
            ("absurd price", "INSERT INTO elec.source_listing (site_id, source_sku, source_url, source_type, brand_id, "
             "category_id, title, evidence_text, confidence, parser, price) VALUES (%(site)s,'x','https://a.in/x',"
             "'search_snippet',%(brand)s,%(cat)s,'t','e',0.5,'t', 5)"),
            ("non-INR", "INSERT INTO elec.source_listing (site_id, source_sku, source_url, source_type, brand_id, "
             "category_id, title, evidence_text, confidence, parser, currency) VALUES (%(site)s,'x','https://a.in/x',"
             "'search_snippet',%(brand)s,%(cat)s,'t','e',0.5,'t','USD')"),
            ("pincode claim", "INSERT INTO elec.source_listing (site_id, source_sku, source_url, source_type, brand_id, "
             "category_id, title, evidence_text, confidence, parser, pincode_applied) VALUES (%(site)s,'x',"
             "'https://a.in/x','search_snippet',%(brand)s,%(cat)s,'t','e',0.5,'t',TRUE)"),
            ("bad grade", "UPDATE elec.site SET probe_outcome = 'D' WHERE id = %(site)s"),
        ]
        for label, sql in bad_rows:
            with pytest.raises(psycopg.errors.CheckViolation):
                with conn.transaction():
                    conn.execute(sql, base)
                pytest.fail(label)


def test_listing_validation_rejects_missing_evidence():
    l = _listing("croma.com", "1", "Samsung Galaxy S24 5G (8GB RAM, 256GB)", price=Decimal(74999))
    l.evidence_text = " "
    with pytest.raises(ValueError):
        l.validate()


def test_price_history_is_append_only(db):
    ids = repo.id_maps()
    lid = repo.upsert_listing(_listing("croma.com", "1", "Samsung Galaxy S24 5G (8GB RAM, 256GB)",
                                       price=Decimal(74999)), ids, None)
    with connect() as conn:
        with pytest.raises(psycopg.errors.RaiseException):
            conn.execute("UPDATE elec.price_history SET price = 1000 WHERE listing_id = %s", (lid,))


def test_verification_needs_two_sites_including_a_retailer(db):
    c = Collector.__new__(Collector)  # use store() without network setup
    c.opt = RunOptions(category="mobiles", brands=["samsung"])
    c.ids = repo.id_maps()
    c.run_id = None
    c._touched_products = {}
    from app.electronics.collector import RunStats
    c.stats = RunStats()

    title = "Samsung Galaxy S24 5G (Onyx Black, 8GB RAM, 256GB Storage)"
    c.store(_listing("amazon.in", "B0CS5XW6TN", title, price=Decimal(74999)))
    assert repo.refresh_verification() == {"unverified": 1}
    with connect() as conn:
        assert conn.execute("SELECT count(*) n FROM elec.v_brand_catalog").fetchone()["n"] == 0

    # A second, different platform listing the same variant (written its own way).
    c.store(_listing("flipkart.com", "itm1", "SAMSUNG Galaxy S24 5G (Onyx Black, 256 GB) (8 GB RAM)",
                     price=Decimal(72999)))
    assert repo.refresh_verification() == {"verified": 1}
    with connect() as conn:
        row = conn.execute("SELECT * FROM elec.v_brand_catalog").fetchone()
        offers = conn.execute("SELECT site FROM elec.v_product_availability ORDER BY price").fetchall()
    assert row["platform_count"] == 2 and row["best_price"] == Decimal("72999.00")
    assert [o["site"] for o in offers] == ["Flipkart", "Amazon.in"]


def test_scraped_listing_is_not_downgraded_by_a_snippet(db):
    ids = repo.id_maps()
    title = "Samsung Galaxy S24 5G (8GB RAM, 256GB)"
    scraped = _listing("croma.com", "303838", title, price=Decimal(74999), source_type="scraped_page",
                       evidence='{"price": "74999"}')
    lid = repo.upsert_listing(scraped, ids, None)
    snippet = _listing("croma.com", "303838", title, price=Decimal(69999))
    assert repo.upsert_listing(snippet, ids, None) == lid
    with connect() as conn:
        row = conn.execute("SELECT price, source_type FROM elec.source_listing WHERE id = %s", (lid,)).fetchone()
    assert (row["price"], row["source_type"]) == (Decimal("74999.00"), "scraped_page")


def test_catalogue_api_serves_verified_products(db, client):
    c = Collector.__new__(Collector)
    c.opt = RunOptions(category="mobiles", brands=["samsung"])
    c.ids, c.run_id, c._touched_products = repo.id_maps(), None, {}
    from app.electronics.collector import RunStats
    c.stats = RunStats()
    c.store(_listing("amazon.in", "B0CS5XW6TN", "Samsung Galaxy S24 5G (8GB RAM, 256GB)", price=Decimal(74999)))
    c.store(_listing("poorvika.com", "samsung-galaxy-s24", "Samsung Galaxy S24 5G (8GB RAM, 256GB)",
                     price=Decimal(73999)))
    repo.refresh_verification()

    brands = client.get("/api/elec/brands", params={"category": "mobiles"}).json()
    assert brands[0]["brand_slug"] == "samsung" and brands[0]["product_count"] == 1
    listing = client.get("/api/elec/products", params={"category": "mobiles", "tn_only": True}).json()
    assert listing["total"] == 1
    product = listing["products"][0]
    assert product["best_price"] == "73999.00" and product["sold_by_tn_retailer"] is True
    detail = client.get(f"/api/elec/products/{product['product_id']}").json()
    assert {o["site"] for o in detail["offers"]} == {"Amazon.in", "Poorvika"}
    assert all(o["source_url"].startswith("https://") for o in detail["offers"])
    assert client.get("/api/elec/products/999999").status_code == 404


def test_implausible_snippet_prices_are_flagged(db):
    c = Collector.__new__(Collector)
    c.opt = RunOptions(category="mobiles", brands=["samsung"])
    c.ids, c.run_id, c._touched_products = repo.id_maps(), None, {}
    from app.electronics.collector import RunStats
    c.stats = RunStats()
    title = "Samsung Galaxy S24 5G (8GB RAM, 256GB)"
    c.store(_listing("poorvika.com", "s24", title, price=Decimal(74999), source_type="scraped_page",
                     evidence='{"price": "74999"}'))
    c.store(_listing("flipkart.com", "itm1", title, price=Decimal(129999)))   # 73% above the page price
    c.store(_listing("amazon.in", "B0X", title, price=Decimal(72999)))        # plausible
    repo.refresh_verification()
    with connect() as conn:
        flagged = {r["site"]: r["price_outlier"] for r in conn.execute(
            "SELECT site, price_outlier FROM elec.v_product_availability")}
        best = conn.execute("SELECT price FROM elec.v_best_price").fetchone()["price"]
    assert flagged == {"Poorvika": False, "Flipkart": True, "Amazon.in": False}
    assert best == Decimal("74999.00")  # page price preferred; the outlier never wins


def test_google_price_lookup_only_trusts_the_same_page(db, monkeypatch):
    from app.electronics import price_lookup
    from app.electronics.search.providers import SearchHit

    c = Collector.__new__(Collector)
    c.opt = RunOptions(category="mobiles", brands=["samsung"])
    c.ids, c.run_id, c._touched_products = repo.id_maps(), None, {}
    from app.electronics.collector import RunStats
    c.stats = RunStats()
    title = "Samsung Galaxy S24 5G (8GB RAM, 256GB)"
    c.store(_listing("amazon.in", "B0CS5XW6TN", title))                 # no price yet
    c.store(_listing("poorvika.com", "s24", title, price=Decimal(74999), source_type="scraped_page",
                     evidence='{"price": "74999"}'))
    offer = {"price": "72999", "currency": "INR", "availability": "InStock", "raw": {"price": "72999"}}
    hits = [
        # Another product page on the same site, with a price: must be ignored.
        SearchHit("https://www.amazon.in/other/dp/B0OTHER123", "Samsung Galaxy S24 Ultra", "", "google", 0,
                  offer={**offer, "price": "129999"}),
        SearchHit("https://www.amazon.in/Samsung-Galaxy/dp/B0CS5XW6TN/ref=x", title, "", "google", 1, offer=offer),
    ]

    class FakeGoogle:
        enabled, error = True, None

    class FakeEngine:
        def __init__(self, budget):
            self.google = FakeGoogle()

        def text(self, query, max_results=10, providers="default"):
            assert providers == "google"
            return hits

    monkeypatch.setattr(price_lookup, "SearchEngine", FakeEngine)
    stats = price_lookup.lookup_prices(limit=5)
    assert stats["priced"] == 1
    with connect() as conn:
        row = conn.execute("SELECT price, parser, evidence_text FROM elec.source_listing WHERE source_sku = 'B0CS5XW6TN'").fetchone()
    assert row["price"] == Decimal("72999.00") and row["parser"].endswith("pagemap") and "72999" in row["evidence_text"]


def test_google_disables_itself_on_a_rejected_key(monkeypatch):
    import app.electronics.search.providers as providers

    calls = []

    class Resp:
        status_code = 403
        text = "forbidden"

        def json(self):
            return {"error": {"message": "This project does not have the access to Custom Search JSON API."}}

    monkeypatch.setattr(providers, "USE_GOOGLE_CSE", True)
    monkeypatch.setattr(providers.requests, "get", lambda *a, **k: calls.append(1) or Resp())
    g = providers.GoogleCseProvider(quota_left=lambda: 100)
    g._pacer.interval = 0
    assert g.text("q") is None and g.text("q2") is None
    assert calls == [1] and not g.enabled and "Custom Search JSON API" in g.error