novaalphastrikeomegaz663
044c870df7
- Add retry/backoff (429 + 5xx, Retry-After aware) to the OFF adapter so transient API errors no longer abort a run. - Clamp bounded text fields (serving_size, net_content_unit, country_of_origin) to their column widths in transform; long OFF values previously raised StringDataRightTruncation and rolled back the batch. - Load each record inside a savepoint (load_record_safe) so one malformed source record is skipped instead of aborting the whole import; jobs now report an errored count. - Tests for retry behaviour, serving_size clamping, and per-record isolation. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
"""Integration test for the DB loader.
|
|
|
|
Skipped automatically when no database is reachable (e.g. local runs without
|
|
docker, or CI jobs without a postgres service). Requires migrations applied.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opengoods.etl.load import default_dsn, ensure_source, load_record, load_record_safe
|
|
from opengoods.etl.transform import transform
|
|
|
|
psycopg = pytest.importorskip("psycopg")
|
|
|
|
FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "off_product.json").read_text())
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn():
|
|
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}")
|
|
# ensure schema present
|
|
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_record_roundtrip(conn):
|
|
source_id = ensure_source(conn)
|
|
rec = transform(FIXTURE)
|
|
product_id = load_record(conn, rec, source_id, FIXTURE)
|
|
|
|
row = conn.execute(
|
|
"SELECT name, gtin, net_content_unit FROM product WHERE id = %s", (product_id,)
|
|
).fetchone()
|
|
assert row[0] == "Nutella"
|
|
assert row[1] == "3017624010701"
|
|
assert row[2] == "g"
|
|
|
|
nutri = conn.execute(
|
|
"SELECT nutriments ->> 'energy_kcal' FROM food_detail WHERE product_id = %s",
|
|
(product_id,),
|
|
).fetchone()
|
|
assert nutri[0] == "539.0"
|
|
|
|
prov = conn.execute(
|
|
"SELECT count(*) FROM product_source WHERE product_id = %s", (product_id,)
|
|
).fetchone()
|
|
assert prov[0] >= 1
|
|
|
|
conn.rollback() # keep the test DB clean
|
|
|
|
|
|
def test_load_record_safe_isolates_bad_record(conn):
|
|
source_id = ensure_source(conn)
|
|
|
|
# Unique gtin so the good record is a fresh INSERT, not an upsert/update.
|
|
good = transform(FIXTURE)
|
|
good["gtin"] = "4006381333931"
|
|
assert load_record_safe(conn, good, source_id, FIXTURE) is True
|
|
after_good = conn.execute("SELECT count(*) FROM product").fetchone()[0]
|
|
|
|
# A record whose name violates NOT NULL must not abort the batch.
|
|
bad = dict(good)
|
|
bad["gtin"] = "5000112637922"
|
|
bad["name"] = None
|
|
assert load_record_safe(conn, bad, source_id, {}) is False
|
|
|
|
# The good record survived the bad one's rollback-to-savepoint.
|
|
after_bad = conn.execute("SELECT count(*) FROM product").fetchone()[0]
|
|
assert after_bad == after_good
|
|
|
|
conn.rollback() # keep the test DB clean
|