Vorrania/backend/tests/test_product_images.py

import json

import pytest

from app.models import BaseUnit, Product, ProductImage, Role, User
from app.routers import transfer
from app.services import images

EIN_PIXEL = b"\x89PNG\r\n\x1a\n-nicht-echt-aber-egal"


def _artikel(db, url=None):
    p = Product(name="Pesto", base_unit=BaseUnit.gram, package_size=195, image_url=url)
    db.add(p)
    db.commit()
    db.refresh(p)
    return p


def test_image_version_spiegelt_bildzeitstempel(db):
    from app.crud import product_to_out

    p = _artikel(db)
    assert product_to_out(db, p).image_version is None   # kein Bild
    img = ProductImage(product_id=p.id, content_type="image/png", data=EIN_PIXEL)
    db.add(img)
    db.commit()
    db.refresh(img)
    assert product_to_out(db, p).image_version == int(img.updated_at.timestamp())


def test_ohne_bildadresse_gibt_es_nichts_zu_holen(db, monkeypatch):
    monkeypatch.setattr(images, "fetch", lambda url: (_ for _ in ()).throw(AssertionError))
    assert images.ensure(db, _artikel(db)) is None


def test_bild_wird_einmal_geholt_und_danach_lokal_geliefert(db, monkeypatch):
    aufrufe = []

    def gefaelscht(url):
        aufrufe.append(url)
        return EIN_PIXEL, "image/png"

    monkeypatch.setattr(images, "fetch", gefaelscht)
    artikel = _artikel(db, "https://example.org/pesto.png")

    erstes = images.ensure(db, artikel)
    assert erstes is not None
    assert erstes.data == EIN_PIXEL
    assert erstes.content_type == "image/png"

    # Zweiter Aufruf darf nicht erneut ins Netz gehen - genau das ist der Zweck.
    images.ensure(db, artikel)
    assert aufrufe == ["https://example.org/pesto.png"]


def test_ein_nicht_erreichbares_bild_bleibt_folgenlos(db, monkeypatch):
    monkeypatch.setattr(images, "fetch", lambda url: None)
    artikel = _artikel(db, "https://example.org/weg.png")
    assert images.ensure(db, artikel) is None
    assert db.get(ProductImage, artikel.id) is None


@pytest.fixture()
def user(db):
    person = User(username="tester", password_hash="x", role=Role.admin)
    db.add(person)
    db.commit()
    db.refresh(person)
    return person


def test_backup_import_uebernimmt_die_bildadresse(db, user):
    """Sie stand im Backup, ging beim Einlesen aber verloren.

    Der Test geht bewusst durch _import_json und nicht nur durch
    _get_or_create_product: Genau zwischen beiden fiel die Adresse heraus.
    """
    backup = {
        "products": [
            {
                "name": "Pesto",
                "brand": "Barilla",
                "image_url": "https://example.org/pesto.jpg",
                "unit": "Gramm",
                "package_size": 195,
                "lots": [],
            }
        ]
    }
    transfer._import_json(db, json.dumps(backup).encode("utf-8"), user, "add")
    db.commit()

    artikel = db.query(Product).filter(Product.name == "Pesto").one()
    assert artikel.image_url == "https://example.org/pesto.jpg"


def test_import_ueberschreibt_kein_vorhandenes_bild(db, user):
    vorhanden = Product(
        name="Pesto", base_unit=BaseUnit.gram, image_url="https://eigenes.example/bild.jpg"
    )
    db.add(vorhanden)
    db.commit()

    backup = {"products": [{"name": "Pesto", "image_url": "https://off.example/anderes.jpg",
                            "unit": "Gramm", "lots": []}]}
    transfer._import_json(db, json.dumps(backup).encode("utf-8"), user, "add")
    db.commit()

    db.refresh(vorhanden)
    assert vorhanden.image_url == "https://eigenes.example/bild.jpg"


def test_speichern_holt_nach_wenn_die_kopie_fehlt(db, user, monkeypatch):
    """Der haeufigste Fall: Adresse stimmt laengst, der Abruf schlug nur einmal fehl.

    Wuerde nur auf eine geaenderte Adresse geachtet, bliebe so ein Artikel
    dauerhaft ohne Bild - "Uebernehmen" waere dann folgenlos.
    """
    from app.routers.products import update_product
    from app.schemas import ProductUpdate

    monkeypatch.setattr(images, "fetch", lambda url: (EIN_PIXEL, "image/png"))
    artikel = _artikel(db, "https://example.org/pesto.png")
    assert db.get(ProductImage, artikel.id) is None

    update_product(
        artikel.id,
        ProductUpdate(image_url="https://example.org/pesto.png"),
        db=db,
        _=user,
    )
    assert db.get(ProductImage, artikel.id) is not None


def test_fetch_geht_nur_an_http_adressen(monkeypatch):
    """Ohne diese Schranke waere eine 'file://'-Adresse ein Weg ins Dateisystem."""
    monkeypatch.setattr(
        images.httpx, "get", lambda *a, **k: (_ for _ in ()).throw(AssertionError)
    )
    assert images.fetch("file:///etc/passwd") is None
    assert images.fetch("") is None