174 lines
7.5 KiB
Python
174 lines
7.5 KiB
Python
import json
|
|
import pytest
|
|
from ocado_grocy.openfoodfacts import candidate_info, pack, valid_barcode, apply_match
|
|
from ocado_grocy.product_metadata import read_metadata, replace_metadata
|
|
|
|
|
|
def metadata():
|
|
return {'Ocado product ID':'123', 'Barcode':'', 'ocado':{'brand':'M&S', 'size':{'value':'250g'}}, 'custom':'keep me'}
|
|
|
|
|
|
def product():
|
|
return {'id':42, 'name':'M&S Ginger Snaps', 'qu_id_purchase':1, 'description':'<p>Ocado product ID: 123</p>' + json.dumps(metadata()) + '<p>User comment</p>'}
|
|
|
|
|
|
def candidate():
|
|
return {'code':'00000048', 'product_name':'Ginger Snaps', 'brands':'Marks & Spencer', 'quantity':'250 g', 'ingredients_text':'Ginger & flour', 'nutriments':{'energy-kcal_100g':450}}
|
|
|
|
|
|
def test_matching_rejects_variant_brand_and_pack_changes():
|
|
assert candidate_info(product(), metadata(), candidate())['exact']
|
|
for change in ({'quantity':'300 g'}, {'quantity':'2 x 125g'}, {'brands':'Tesco'}, {'product_name':'Reduced Sugar Ginger Snaps'}, {'code':'00000041'}):
|
|
assert not candidate_info(product(), metadata(), {**candidate(), **change})['exact']
|
|
assert pack('1.25L') == pack('1250ml')
|
|
assert pack('250g') != pack('250ml')
|
|
assert pack('5 per pack') is None
|
|
assert not valid_barcode('123456011')
|
|
|
|
|
|
def test_metadata_survives_sanitizer_and_retains_user_comments():
|
|
p = product()
|
|
m = read_metadata(p['description'])
|
|
m['openfoodfacts'] = {'ingredients':'Ginger & flour < 1%'}
|
|
updated = replace_metadata(p['description'], m)
|
|
assert read_metadata(updated) == m
|
|
assert '<p>User comment</p>' in updated
|
|
assert 'Ginger & flour < 1%' in updated
|
|
|
|
|
|
class API:
|
|
def __init__(self):
|
|
self.p = product()
|
|
self.writes = []
|
|
def get(self, path):
|
|
assert path == '/objects/products/42'
|
|
return self.p.copy()
|
|
def post(self, path, data):
|
|
assert path == '/objects/product_barcodes'
|
|
self.writes.append((path, data))
|
|
def request(self, method, path, data):
|
|
assert method == 'PUT' and path == '/objects/products/42'
|
|
assert set(data) == {'description'}
|
|
self.writes.append((path, data))
|
|
self.p.update(data)
|
|
|
|
|
|
def test_enrichment_is_idempotent_and_never_changes_stock_or_title():
|
|
api = API()
|
|
barcodes = []
|
|
assert apply_match(api, product(), candidate(), barcodes) == 'updated'
|
|
assert apply_match(api, product(), candidate(), barcodes) == 'unchanged'
|
|
assert len(api.writes) == 2
|
|
assert api.p['name'] == product()['name']
|
|
assert read_metadata(api.p['description'])['custom'] == 'keep me'
|
|
assert read_metadata(api.p['description'])['openfoodfacts']['nutriments']['energy-kcal_100g'] == 450
|
|
|
|
|
|
def test_conflicting_barcode_and_deleted_product_do_not_write():
|
|
api = API()
|
|
with pytest.raises(ValueError, match='another Grocy product'):
|
|
apply_match(api, product(), candidate(), [{'barcode':'00000048','product_id':99}])
|
|
api.p = {}
|
|
with pytest.raises(ValueError, match='deleted'):
|
|
apply_match(api, product(), candidate(), [])
|
|
assert not api.writes
|
|
|
|
|
|
def test_two_exact_barcodes_are_ambiguous():
|
|
from ocado_grocy.openfoodfacts import choose_match
|
|
first = candidate()
|
|
second = {**first, 'code':'00000055'}
|
|
ranked, match = choose_match(product(), metadata(), {first['code']:first, second['code']:second})
|
|
assert len(ranked) == 2
|
|
assert match is None
|
|
|
|
|
|
def test_product_search_is_bounded_and_includes_name_and_brand(tmp_path):
|
|
from ocado_grocy.openfoodfacts import OpenFoodFacts
|
|
calls = []
|
|
class OFF(OpenFoodFacts):
|
|
def fetch(self, path, params, **kwargs):
|
|
calls.append(params)
|
|
return {'is_count_exact': True, 'count': 9000,
|
|
'hits': [{**candidate(), 'brands': ['Marks & Spencer']}]}
|
|
candidates, result = OFF(tmp_path, 'test@example.com').search(product(), metadata())
|
|
assert len(calls) == 1
|
|
assert calls[0]['page_size'] == 100 and calls[0]['page'] == 1
|
|
assert 'ginger' in calls[0]['q'] and 'snaps' in calls[0]['q']
|
|
assert 'marks-spencer' in calls[0]['q']
|
|
assert candidates[candidate()['code']]['brands'] == 'Marks & Spencer'
|
|
assert not result['search_complete']
|
|
|
|
|
|
@pytest.mark.parametrize('complete', [True, False])
|
|
def test_matching_searches_only_selected_products_and_requires_complete_results(tmp_path, monkeypatch, complete):
|
|
from ocado_grocy import openfoodfacts as module
|
|
searched = []
|
|
class OFF:
|
|
def __init__(self, *args): pass
|
|
def search(self, p, m, offline):
|
|
searched.append(p['id'])
|
|
return {candidate()['code']: candidate()}, {'search_complete': complete}
|
|
class Client:
|
|
def get(self, path):
|
|
assert path == '/objects/product_barcodes'
|
|
return []
|
|
monkeypatch.setattr(module, 'OpenFoodFacts', OFF)
|
|
nonfood = {**product(), 'id': 43}
|
|
nonfood['description'] = replace_metadata(nonfood['description'],
|
|
{**metadata(), 'ocado': {'brand': 'Miniml'}})
|
|
result = module.run_openfoodfacts(Client(), [nonfood, product(), {**product(), 'id': 44}],
|
|
tmp_path, 'test@example.com', apply=False, limit=1)
|
|
assert searched == [42]
|
|
assert [r['status'] for r in result['products']] == [
|
|
'non_food', 'matched' if complete else 'review', 'not_searched']
|
|
|
|
|
|
@pytest.mark.parametrize('error', [{'timed_out': True}, {'warnings': ['partial']}, {'errors': ['bad query']}])
|
|
def test_failed_search_is_not_reported_as_no_match(tmp_path, error):
|
|
from ocado_grocy.openfoodfacts import OpenFoodFacts
|
|
off = OpenFoodFacts(tmp_path, 'test@example.com')
|
|
off.fetch = lambda *args, **kwargs: error
|
|
with pytest.raises(ValueError, match='Incomplete product search'):
|
|
off.search(product(), metadata())
|
|
|
|
|
|
def test_ms_food_brand_and_missing_barcode_recovery():
|
|
assert candidate_info(product(), metadata(), {**candidate(), 'brands':'M&S Food'})['exact']
|
|
api = API()
|
|
barcodes = []
|
|
apply_match(api, product(), candidate(), barcodes)
|
|
# Simulate an interrupted attempt with metadata present but no barcode row.
|
|
assert apply_match(api, api.p.copy(), candidate(), []) == 'updated'
|
|
assert len([p for p, data in api.writes if p == '/objects/products/42']) == 1
|
|
|
|
|
|
def test_established_barcode_match_survives_a_missing_catalog_entry(tmp_path, monkeypatch):
|
|
from ocado_grocy import openfoodfacts as module
|
|
p = product()
|
|
m = metadata()
|
|
m.update(Barcode=candidate()['code'], openfoodfacts=candidate())
|
|
p['description'] = replace_metadata(p['description'], m)
|
|
class OFF:
|
|
def __init__(self, *a): pass
|
|
def search(self, *a): pytest.fail("Established match must not trigger search")
|
|
def product(self, code, offline):
|
|
assert code == candidate()['code']
|
|
return candidate()
|
|
class Client:
|
|
def get(self, path):
|
|
assert path == '/objects/product_barcodes'
|
|
return []
|
|
monkeypatch.setattr(module, 'OpenFoodFacts', OFF)
|
|
result = module.run_openfoodfacts(Client(), [p], tmp_path, 'test@example.com', apply=False)
|
|
assert result['products'][0]['match_method'] == 'existing_match'
|
|
assert result['products'][0]['status'] == 'matched'
|
|
|
|
|
|
def test_review_ranks_the_right_name_ahead_of_unrelated_matching_pack():
|
|
from ocado_grocy.openfoodfacts import choose_match
|
|
missing_size = {**candidate(), 'quantity':''}
|
|
unrelated = {**candidate(), 'code':'00000055', 'product_name':'Tomato Ketchup'}
|
|
ranked, matched = choose_match(product(), metadata(), {c['code']:c for c in [missing_size, unrelated]})
|
|
assert ranked[0]['code'] == missing_size['code']
|
|
assert matched is None
|