84 lines
4 KiB
Python
84 lines
4 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from click.testing import CliRunner
|
|
import pytest
|
|
|
|
from ocado_grocy.grocy import imported_products
|
|
from ocado_grocy.history import main
|
|
from ocado_grocy import postprocess
|
|
|
|
|
|
def test_scope_uses_completed_lines_for_selected_order_and_server(tmp_path):
|
|
state = tmp_path / 'journal.sqlite3'
|
|
with sqlite3.connect(state) as db:
|
|
db.execute('CREATE TABLE imports(server,order_id,status,product_id)')
|
|
db.executemany('INSERT INTO imports VALUES (?,?,?,?)', [
|
|
('server','latest','done',1), ('server','older','done',2),
|
|
('server','latest','pending',3), ('other','latest','done',4),
|
|
('server','latest','done',5)])
|
|
class Client:
|
|
url = 'server'
|
|
def get(self, path):
|
|
assert path == '/objects/products'
|
|
return [{'id':i} for i in range(1,5)] # 5 was deleted
|
|
assert imported_products(Client(), state, ['latest']) == [{'id':1}]
|
|
|
|
|
|
@pytest.mark.parametrize('dry_run', [False, True])
|
|
def test_main_runs_enrichment_even_when_no_stock_added_but_never_in_dry_run(tmp_path, monkeypatch, dry_run):
|
|
from ocado_grocy import history
|
|
from test_grocy import receipt
|
|
monkeypatch.setattr(history, 'cached_orders', lambda *a: ([receipt()], []))
|
|
monkeypatch.setattr(history, 'run_import', lambda *a, **kw: {'items':[], 'imported_lines':0})
|
|
calls = []
|
|
monkeypatch.setattr(postprocess, 'enrich_orders', lambda *a, **kw: calls.append((a,kw)) or {'errors':[]})
|
|
args = ['--cached','--archive',str(tmp_path),'--config',str(tmp_path/'config.toml')]
|
|
if dry_run:
|
|
args.append('--dry-run')
|
|
result = CliRunner().invoke(main, args)
|
|
assert result.exit_code == 0, result.output
|
|
assert len(calls) == (0 if dry_run else 1)
|
|
if calls:
|
|
assert calls[0][0][3] == ['123']
|
|
assert calls[0][1] == {'images':True,'openfoodfacts':True}
|
|
|
|
|
|
def test_picture_failure_does_not_prevent_off_and_errors_are_recorded(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
client = SimpleNamespace(url='server')
|
|
products = [{'id':42}]
|
|
monkeypatch.setattr(postprocess, 'Grocy', lambda *a: client)
|
|
monkeypatch.setattr(postprocess, 'imported_products', lambda *a: products)
|
|
def fail_images(*args, **kwargs):
|
|
raise ValueError('image unavailable')
|
|
monkeypatch.setattr(postprocess, 'run_images', fail_images)
|
|
calls = []
|
|
def off(api, selected, cache, contact):
|
|
assert api is client and selected == products
|
|
calls.append(True)
|
|
return {'counts':{'unchanged':1}, 'products':[]}
|
|
monkeypatch.setattr(postprocess, 'run_openfoodfacts', off)
|
|
result = postprocess.enrich_orders({'openfoodfacts':{'contact':'test@example.com'}}, tmp_path/'config.toml',tmp_path,['123'])
|
|
assert calls and result['openfoodfacts'] == {'unchanged':1}
|
|
assert result['errors'] == ['Images: image unavailable']
|
|
assert (tmp_path/'enrichment/123/summary.json').exists()
|
|
|
|
|
|
def test_disabled_enrichment_does_not_access_grocy(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(postprocess, 'Grocy', lambda *a: (_ for _ in ()).throw(AssertionError('Unexpected API access')))
|
|
assert postprocess.enrich_orders({}, tmp_path/'config.toml',tmp_path,['123'],images=False,openfoodfacts=False) == {'errors':[]}
|
|
|
|
|
|
def test_enrichment_failure_reports_stock_completion_and_nonzero_exit(tmp_path, monkeypatch):
|
|
from ocado_grocy import history
|
|
from test_grocy import receipt
|
|
monkeypatch.setattr(history, 'cached_orders', lambda *a: ([receipt()], []))
|
|
monkeypatch.setattr(history, 'run_import', lambda *a, **kw: {'items':[], 'imported_lines':1})
|
|
monkeypatch.setattr(postprocess, 'enrich_orders', lambda *a, **kw: {'errors':['OFF unavailable']})
|
|
result = CliRunner().invoke(main, ['--cached','--archive',str(tmp_path),'--config',str(tmp_path/'config.toml')])
|
|
assert result.exit_code == 1
|
|
assert 'Stock import completed; enrichment needs a retry' in result.output
|
|
import json
|
|
saved = json.loads((tmp_path/'report.json').read_text())
|
|
assert saved['imported_lines'] == 1 and saved['enrichment']['errors'] == ['OFF unavailable']
|