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>
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
import httpx
|
|
|
|
from opengoods.adapters.openfoodfacts import OpenFoodFactsAdapter
|
|
|
|
|
|
def _product(code, lm):
|
|
return {"code": code, "product_name": f"P{code}", "last_modified_t": lm}
|
|
|
|
|
|
def _adapter(pages):
|
|
"""Build an adapter whose search endpoint serves the given pages."""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
page = int(request.url.params.get("page", "1"))
|
|
products = pages.get(page, [])
|
|
return httpx.Response(200, json={"products": products, "page": page})
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
return OpenFoodFactsAdapter(client=client, min_interval=0)
|
|
|
|
|
|
def test_incremental_yields_only_newer_and_stops_at_watermark():
|
|
pages = {
|
|
1: [_product("1", 300), _product("2", 250), _product("3", 100)],
|
|
}
|
|
adapter = _adapter(pages)
|
|
got = list(adapter.fetch_modified_since(200, page_size=3, max_pages=5))
|
|
codes = [p["code"] for p in got]
|
|
assert codes == ["1", "2"] # 100 <= 200 stops iteration
|
|
|
|
|
|
def test_incremental_paginates_until_short_page():
|
|
pages = {
|
|
1: [_product("1", 900), _product("2", 800)],
|
|
2: [_product("3", 700)], # short page -> stop after
|
|
}
|
|
adapter = _adapter(pages)
|
|
got = list(adapter.fetch_modified_since(0, page_size=2, max_pages=5))
|
|
assert [p["code"] for p in got] == ["1", "2", "3"]
|
|
|
|
|
|
def test_incremental_empty_first_page():
|
|
adapter = _adapter({1: []})
|
|
assert list(adapter.fetch_modified_since(0, page_size=10, max_pages=3)) == []
|