lixu
766a573989
- adapters/openfoodfacts.py: OFF API 适配器(限速+User-Agent) + JSONL/.gz dump 读取 - etl/transform.py: 纯函数转换(GTIN 校验/净含量解析+归一/营养 per_100g 双能量/过敏原添加剂清洗/关键词分类映射) - etl/load.py: psycopg upsert(product/food_detail/product_image) + product_source 字段级溯源 - jobs/seed_off.py: CLI(--barcodes API / --dump 文件 / --limit) - 测试: 13 个离线 transform 单测(fixture) + 可跳过的 DB 集成测试 - 依赖: 增加 psycopg[binary] - 实跑: 从 OFF API 拉 5 个真实条码入本地 postgres 验证通过 - docs/etl-openfoodfacts.md: 流程/运行/字段映射 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
62 lines
1.8 KiB
Python
62 lines
1.8 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
|
|
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
|