mitai-jinkendo/backend/tests/test_bls_parser.py

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_components_official_bezeichnung_and_empty_name():
    data = _xlsx([
        ["Component code", "Bezeichnung", "English", "Einheit"],
        ["PROT625", "", "Protein", "g/100g"],
        ["FAT", "Fett", "Fat", "g/100g"],
    ])
    attrs = parse_components_xlsx(data)
    by_key = {a["attr_key"]: a for a in attrs}
    assert by_key["PROT625"]["name_de"] == "Protein (N×6,25)"
    assert by_key["FAT"]["name_de"] == "Fett"


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