John Doe
a35bcd6647
- OFF incremental fetch via search API + persistent watermark (ingest_state, migration 0004) - GS1 barcode supplement adapter (offline mapping + GS1-style API) filling only gaps with field-level provenance - Non-GTIN dedup with canonical selection + merge_log; field-level conflict resolution (source trust > recency) - Quality scoring (0.4 completeness + 0.3 source trust + 0.2 multi-source + 0.1 freshness) wired into load/merge - Jobs: update_off, dedup, schedule; docs/ingestion-management.md - 19 new tests (pure + DB-integration), ruff clean Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
22 lines
730 B
Python
22 lines
730 B
Python
import json
|
|
from pathlib import Path
|
|
|
|
from opengoods.etl.load import ensure_source, load_record
|
|
from opengoods.etl.quality import compute_quality
|
|
from opengoods.etl.transform import transform
|
|
|
|
FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "off_product.json").read_text())
|
|
|
|
|
|
def test_quality_score_set_on_load(db_conn):
|
|
source_id = ensure_source(db_conn)
|
|
rec = transform(FIXTURE)
|
|
pid = load_record(db_conn, rec, source_id, FIXTURE)
|
|
|
|
stored = float(
|
|
db_conn.execute("SELECT quality_score FROM product WHERE id = %s", (pid,)).fetchone()[0]
|
|
)
|
|
assert 0.0 < stored <= 1.0
|
|
# The persisted value matches a fresh recomputation.
|
|
assert abs(stored - compute_quality(db_conn, pid)) < 1e-9
|