novaalphastrikeomegaz663
c090bcdc9b
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
import httpx
|
|
|
|
from opengoods.adapters.openfoodfacts import OpenFoodFactsAdapter, is_cn_gs1
|
|
|
|
|
|
def _adapter(handler, **kwargs):
|
|
client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
return OpenFoodFactsAdapter(client=client, min_interval=0, backoff_base=0, **kwargs)
|
|
|
|
|
|
def test_is_cn_gs1():
|
|
assert is_cn_gs1("6901234567892") # GS1 China prefix
|
|
assert is_cn_gs1("690")
|
|
assert not is_cn_gs1("3017624010701") # France
|
|
assert not is_cn_gs1("5449000000996") # Belgium
|
|
assert not is_cn_gs1("")
|
|
assert not is_cn_gs1(None)
|
|
assert not is_cn_gs1("69") # too short to carry a prefix digit
|
|
|
|
|
|
def test_fetch_by_country_passes_filter_and_paginates():
|
|
seen_params = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen_params.append(dict(request.url.params))
|
|
page = int(request.url.params.get("page", "1"))
|
|
if page == 1:
|
|
return httpx.Response(
|
|
200,
|
|
json={"products": [{"code": "6901"}, {"code": "6902"}]},
|
|
)
|
|
return httpx.Response(200, json={"products": []})
|
|
|
|
adapter = _adapter(handler)
|
|
out = list(adapter.fetch_by_country("china", page_size=2, max_pages=5))
|
|
assert [p["code"] for p in out] == ["6901", "6902"]
|
|
assert seen_params[0]["countries_tags_en"] == "china"
|
|
assert seen_params[0]["sort_by"] == "unique_scans_n"
|
|
|
|
|
|
def test_fetch_by_country_dedupes_across_pages():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
page = int(request.url.params.get("page", "1"))
|
|
if page == 1:
|
|
return httpx.Response(200, json={"products": [{"code": "6901"}, {"code": "6902"}]})
|
|
if page == 2:
|
|
# OFF popularity ordering is unstable; a repeat appears on page 2
|
|
return httpx.Response(200, json={"products": [{"code": "6902"}, {"code": "6903"}]})
|
|
return httpx.Response(200, json={"products": []})
|
|
|
|
adapter = _adapter(handler)
|
|
out = [p["code"] for p in adapter.fetch_by_country("china", page_size=2, max_pages=5)]
|
|
assert out == ["6901", "6902", "6903"]
|