Lars
a07a668de2
Prüfung und Apply laufen serverseitig, die UI pollt nur noch den Status. Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
import time
|
|
from io import BytesIO
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from bls.import_service import should_persist_value
|
|
from bls.jobs import create_and_check, get_job
|
|
from bls.parser import parse_components_xlsx, parse_foods_xlsx
|
|
|
|
|
|
def _xlsx(rows):
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
for row in rows:
|
|
ws.append(row)
|
|
buf = BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_parse_components_dynamic():
|
|
data = _xlsx([
|
|
["Code", "Name_DE", "Name_EN", "Einheit"],
|
|
["ENERCC", "Energie", "Energy", "kcal/100g"],
|
|
["NA", "Natrium", "Sodium", "mg/100g"],
|
|
])
|
|
attrs = parse_components_xlsx(data)
|
|
keys = {a["attr_key"] for a in attrs}
|
|
assert "ENERCC" in keys
|
|
assert "NA" in keys
|
|
|
|
|
|
def test_parse_foods_keeps_bls_code():
|
|
data = _xlsx([
|
|
["BLS Code", "Name", "Food name", "ENERCC Energie [kcal/100g]", "ENERCC Herkunft", "ENERCC Referenz"],
|
|
["C131000", "Hafer roh", "Oats raw", 350, "Analyse", "MRI"],
|
|
])
|
|
parsed = parse_foods_xlsx(data)
|
|
assert parsed["foods"][0]["bls_code"] == "C131000"
|
|
assert parsed["foods"][0]["name_de"] == "Hafer roh"
|
|
vals = {v["attr_key"]: v["value_num"] for v in parsed["foods"][0]["values"]}
|
|
assert vals.get("ENERCC") == 350
|
|
|
|
|
|
def test_persist_only_numeric_or_trace():
|
|
assert should_persist_value({"value_num": 1.2, "is_trace": False})
|
|
assert should_persist_value({"value_num": None, "is_trace": True})
|
|
assert not should_persist_value({"value_num": None, "is_trace": False})
|
|
|
|
|
|
def test_import_job_check_does_not_need_http():
|
|
data = _xlsx([
|
|
["BLS Code", "Name", "Food name", "ENERCC Energie [kcal/100g]", "ENERCC Herkunft", "ENERCC Referenz"],
|
|
["C131000", "Hafer roh", "Oats raw", 350, "Analyse", "MRI"],
|
|
])
|
|
job_id = create_and_check("foods", data)
|
|
job = None
|
|
for _ in range(80):
|
|
job = get_job(job_id)
|
|
if job and job["status"] in ("checked", "error"):
|
|
break
|
|
time.sleep(0.05)
|
|
assert job is not None
|
|
assert job["status"] == "checked", job.get("error")
|
|
assert job["check"]["foods"] == 1
|