goods/ingestion/tests/test_bypos.py

"""Tests for the bypos-collector adapter/transform and DB loader.

The pure-function tests run anywhere; the DB roundtrip is skipped automatically
when no database is reachable or migrations are not applied.
"""

from __future__ import annotations

from decimal import Decimal

import pytest

from opengoods.adapters.bypos import transform_bypos

psycopg = pytest.importorskip("psycopg")

# A real central-library "hit" row as emitted by the collector.
HIT = {
    "barcode": "6901028941068",
    "name": "泰山合悦",
    "spec": "20支",
    "unit": "盒",
    "area": "广东",
    "manufacturer": "",
    "license": "",
    "in_price": "22.50",
    "sell_price": "28.00",
    "status": "hit",
    "retmsg": "获取商品信息成功",
    "fetched_at": "2026-06-24T02:08:34Z",
    "source": "zc.bypos.net",
}

# A hit with a real mass/volume spec that should normalize to net content.
HIT_VOLUME = {
    "barcode": "6920459905012",
    "name": "康师傅冰红茶490ml",
    "spec": "490毫升",
    "unit": "瓶",
    "area": "浙江杭州",
    "manufacturer": "",
    "license": "",
    "in_price": "2.20",
    "sell_price": "3.00",
    "status": "hit",
    "fetched_at": "2026-06-24T02:08:34Z",
    "source": "zc.bypos.net",
}


def test_transform_skips_non_hit():
    assert transform_bypos({**HIT, "status": "miss", "name": ""}) is None
    assert transform_bypos({**HIT, "status": "invalid"}) is None


def test_transform_skips_missing_name():
    assert transform_bypos({**HIT, "name": "  "}) is None


def test_transform_packaging_spec_kept_raw():
    rec = transform_bypos(HIT)
    assert rec is not None
    assert rec["gtin"] == "6901028941068"
    assert rec["name"] == "泰山合悦"
    # '20支' is a count, not mass/volume -> no net_content, raw spec retained.
    assert rec["net_content_value"] is None
    assert rec["net_content_unit"] is None
    assert rec["attributes"]["spec"] == "20支"
    assert rec["attributes"]["pack_unit"] == "盒"
    assert rec["attributes"]["origin_area"] == "广东"
    assert rec["attributes"]["suggested_in_price"] == 22.5
    assert rec["attributes"]["suggested_retail_price"] == 28.0
    assert rec["msrp"] == Decimal("28.00")
    assert rec["country_of_origin"] == "中国"


def test_transform_volume_spec_normalized():
    rec = transform_bypos(HIT_VOLUME)
    assert rec is not None
    assert rec["net_content_value"] == Decimal("490")
    assert rec["net_content_unit"] == "ml"
    assert rec["net_content_canonical"] == Decimal("490")
    assert rec["attributes"]["origin_area"] == "浙江杭州"


def test_transform_zero_price_dropped():
    rec = transform_bypos({**HIT, "in_price": "0.00", "sell_price": "0.00"})
    assert rec is not None
    assert rec["msrp"] is None
    assert "suggested_in_price" not in rec["attributes"]
    assert "suggested_retail_price" not in rec["attributes"]


def test_transform_invalid_barcode_no_gtin():
    rec = transform_bypos({**HIT, "barcode": "123"})
    assert rec is not None
    assert rec["gtin"] is None
    assert rec["country_of_origin"] is None


# --- DB roundtrip (skipped without a database) ---------------------------------


@pytest.fixture()
def conn():
    from opengoods.etl.load import default_dsn

    try:
        c = psycopg.connect(default_dsn(), connect_timeout=3)
    except psycopg.OperationalError as exc:  # pragma: no cover - env dependent
        pytest.skip(f"no database available: {exc}")
    has_product = c.execute("SELECT to_regclass('public.product') IS NOT NULL").fetchone()[0]
    if not has_product:
        c.close()
        pytest.skip("migrations not applied")
    yield c
    c.rollback()
    c.close()


def test_load_bypos_roundtrip(conn):
    from opengoods.etl.load import ensure_bypos_source, load_bypos_record

    source_id = ensure_bypos_source(conn)
    rec = transform_bypos(HIT_VOLUME)
    product_id = load_bypos_record(conn, rec, source_id, HIT_VOLUME)

    row = conn.execute(
        "SELECT name, gtin, net_content_unit, country_of_origin, attributes ->> 'pack_unit' "
        "FROM product WHERE id = %s",
        (product_id,),
    ).fetchone()
    assert row[0] == "康师傅冰红茶490ml"
    assert row[1] == "6920459905012"
    assert row[2] == "ml"
    assert row[3] == "中国"
    assert row[4] == "瓶"

    msrp = conn.execute(
        "SELECT amount, currency FROM product_msrp WHERE product_id = %s AND source_id = %s",
        (product_id, source_id),
    ).fetchone()
    assert msrp[0] == Decimal("3.00")
    assert msrp[1] == "CNY"

    # Re-import is idempotent: still one MSRP and one provenance row per source.
    load_bypos_record(conn, rec, source_id, HIT_VOLUME)
    counts = conn.execute(
        "SELECT (SELECT count(*) FROM product_msrp WHERE product_id=%s AND source_id=%s), "
        "(SELECT count(*) FROM product_source WHERE product_id=%s AND source_id=%s)",
        (product_id, source_id, product_id, source_id),
    ).fetchone()
    assert counts == (1, 1)

    conn.rollback()