moneyfyandClaude Opus 5
8edb69fe6a
- simple-icons als kompakter Index im Repository (3.459 Marken, 2 MB gzip),
erzeugt von scripts/vendor_simple_icons.py bzw. `make vendor-icons`
- logo.dev und Brandfetch als optionale Adapter, ohne Schlüssel übersprungen
- Favicon-Fallback und generierter Buchstaben-Avatar als Garantie
- Bei eindeutigem Offline-Treffer unterbleiben Anfragen nach außen komplett
- Cache im Dateisystem nach SHA-256, Auslieferung nur über /api/logos/{id}
mit immutable-Header und ETag
- Markenfarbe aus SVG-Fills bzw. per k-Means (k=4) über 64x64 Pixel, dazu eine
aufgehellte Variante mit mindestens 4,5:1 Kontrast auf dunklem Grund
- Kandidatensuche mit Vorauswahl, Auswahl, Upload und Zurücksetzen
- Bildtyp wird nur noch am Inhalt bestimmt, nicht an der gemeldeten Kopfzeile
- 60 neue Tests, insgesamt 208 grün
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
425 lines
15 KiB
Python
425 lines
15 KiB
Python
"""Tests der Provider-Kette, des Caches und der Logo-Endpunkte."""
|
|
|
|
import io
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from PIL import Image
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.models import LogoAsset, Merchant
|
|
from app.models.enums import LogoSource, LogoStatus
|
|
from app.services.logos import (
|
|
detect_mime,
|
|
extract_domain,
|
|
generate_avatar,
|
|
initials,
|
|
resolve_merchant_logo,
|
|
search_candidates,
|
|
simple_icon_index,
|
|
store_candidate,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def kein_netz(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jeder ausgehende Request lässt den Test scheitern."""
|
|
|
|
class Verboten(httpx.AsyncClient):
|
|
def __init__(self, *args, **kwargs):
|
|
raise AssertionError("Es wurde ein ausgehender Request versucht.")
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", Verboten)
|
|
|
|
|
|
@pytest.fixture
|
|
def netz_ohne_treffer(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
|
"""Ersetzt ausgehende Requests durch 404 und protokolliert die URLs."""
|
|
aufgerufen: list[str] = []
|
|
|
|
class Leer:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def get(self, url, **kwargs):
|
|
aufgerufen.append(str(url))
|
|
return httpx.Response(404, request=httpx.Request("GET", url))
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", Leer)
|
|
return aufgerufen
|
|
|
|
|
|
def png_bytes(farbe: tuple[int, int, int] = (229, 9, 20), size: int = 32) -> bytes:
|
|
image = Image.new("RGBA", (size, size), (*farbe, 255))
|
|
buffer = io.BytesIO()
|
|
image.save(buffer, format="PNG")
|
|
return buffer.getvalue()
|
|
|
|
|
|
async def firma(session: AsyncSession, name: str, domain: str | None = None) -> Merchant:
|
|
from app.services.merchants import normalize_name
|
|
|
|
merchant = Merchant(name=name, normalized_name=normalize_name(name), domain=domain)
|
|
session.add(merchant)
|
|
await session.flush()
|
|
return merchant
|
|
|
|
|
|
# --- simple-icons --------------------------------------------------------------
|
|
|
|
|
|
def test_index_ist_vollstaendig_gevendort() -> None:
|
|
index = simple_icon_index()
|
|
|
|
assert index.version, "Der Index muss eine Version melden."
|
|
assert len(index.icons) > 3000
|
|
assert index.lookup("netflix") is not None
|
|
assert index.lookup("netflix").hex_color.lower() == "e50914"
|
|
|
|
|
|
async def test_netflix_ohne_schluessel_und_ohne_netz(kein_netz: None) -> None:
|
|
"""Akzeptanzkriterium: bekannte Marke ohne API-Keys und ohne Netzzugriff."""
|
|
kandidaten = await search_candidates("Netflix", stop_on_exact=True)
|
|
|
|
bester = kandidaten[0]
|
|
assert bester.source is LogoSource.SIMPLE_ICONS
|
|
assert bester.mime == "image/svg+xml"
|
|
assert bester.brand_color == "#e50914"
|
|
assert bester.score == 1.0
|
|
assert b"<svg" in bester.content
|
|
|
|
|
|
async def test_treffer_ueber_alias_und_domain(kein_netz: None) -> None:
|
|
ueber_domain = await search_candidates("Irgendwas", "spotify.com", stop_on_exact=True)
|
|
assert ueber_domain[0].title == "Spotify"
|
|
|
|
# Rechtsform im Namen darf den Treffer nicht verhindern.
|
|
mit_rechtsform = await search_candidates("Deutsche Telekom AG", stop_on_exact=True)
|
|
assert mit_rechtsform[0].title == "Deutsche Telekom"
|
|
|
|
|
|
async def test_teiltreffer_wird_niedriger_bewertet(netz_ohne_treffer: list[str]) -> None:
|
|
"""'Netflix Standard' ist kein exakter Name, soll aber Netflix finden."""
|
|
kandidaten = await search_candidates("Netflix Standard")
|
|
|
|
treffer = [k for k in kandidaten if k.source is LogoSource.SIMPLE_ICONS]
|
|
assert treffer[0].title == "Netflix"
|
|
assert 0.5 <= treffer[0].score < 0.95
|
|
|
|
|
|
# --- Generierter Avatar --------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("name", "erwartet"),
|
|
[
|
|
("Netflix", "NE"),
|
|
("Deutsche Telekom", "DT"),
|
|
("1&1 Telecom", "1T"),
|
|
("X", "X"),
|
|
("!!!", "?"),
|
|
],
|
|
)
|
|
def test_initialen(name: str, erwartet: str) -> None:
|
|
assert initials(name) == erwartet
|
|
|
|
|
|
def test_avatar_ist_deterministisch() -> None:
|
|
erster = generate_avatar("Knuspertronic GmbH")
|
|
zweiter = generate_avatar("Knuspertronic GmbH")
|
|
|
|
assert erster.sha256 == zweiter.sha256
|
|
assert erster.source is LogoSource.GENERATED
|
|
assert b"KG" in erster.content
|
|
|
|
|
|
def test_avatar_maskiert_sonderzeichen() -> None:
|
|
avatar = generate_avatar("Meier & Söhne")
|
|
|
|
assert b"&" in avatar.content
|
|
assert b"<title>Meier & S\xc3\xb6hne</title>" in avatar.content
|
|
|
|
|
|
async def test_fantasiefirma_endet_beim_avatar(netz_ohne_treffer: list[str]) -> None:
|
|
"""Akzeptanzkriterium: unbekannte Firma endet fehlerfrei beim Avatar."""
|
|
kandidaten = await search_candidates("Knuspertronic GmbH", stop_on_exact=True)
|
|
|
|
assert len(kandidaten) == 1
|
|
assert kandidaten[0].source is LogoSource.GENERATED
|
|
assert kandidaten[0].brand_color is not None
|
|
|
|
|
|
async def test_providerfehler_werden_nicht_durchgereicht(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Ein abstürzender Provider darf die Kette nicht beenden."""
|
|
|
|
class Kaputt:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def get(self, url, **kwargs):
|
|
raise httpx.ConnectError("Netz weg")
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", Kaputt)
|
|
|
|
kandidaten = await search_candidates("Voellig Unbekannt AG")
|
|
|
|
assert kandidaten[-1].source is LogoSource.GENERATED
|
|
|
|
|
|
# --- Hilfsfunktionen -----------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("eingabe", "erwartet"),
|
|
[
|
|
("netflix.com", "netflix.com"),
|
|
("https://www.netflix.com/de/", "netflix.com"),
|
|
("http://sub.example.co.uk/pfad", "sub.example.co.uk"),
|
|
("keine domain", None),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_domain_wird_herausgeloest(eingabe: str | None, erwartet: str | None) -> None:
|
|
assert extract_domain(eingabe) == erwartet
|
|
|
|
|
|
def test_dateityp_wird_an_der_signatur_erkannt() -> None:
|
|
assert detect_mime(png_bytes()) == "image/png"
|
|
assert detect_mime(b'<svg xmlns="http://www.w3.org/2000/svg"></svg>') == "image/svg+xml"
|
|
assert detect_mime(b"\xff\xd8\xff\xe0irgendwas") == "image/jpeg"
|
|
assert detect_mime(b'<?xml version="1.0"?><svg viewBox="0 0 1 1"/>') == "image/svg+xml"
|
|
# Eine behauptete Kopfzeile darf keinen Bildtyp vortäuschen.
|
|
assert detect_mime(b"harmloser text", "image/png") is None
|
|
assert detect_mime(b"%PDF-1.7", "image/svg+xml") is None
|
|
|
|
|
|
# --- Cache ---------------------------------------------------------------------
|
|
|
|
|
|
async def test_gleicher_inhalt_wird_nur_einmal_gespeichert(session: AsyncSession) -> None:
|
|
kandidat = generate_avatar("Knuspertronic")
|
|
|
|
erster = await store_candidate(session, kandidat)
|
|
zweiter = await store_candidate(session, kandidat)
|
|
|
|
assert erster.id == zweiter.id
|
|
anzahl = (await session.execute(select(func.count()).select_from(LogoAsset))).scalar_one()
|
|
assert anzahl == 1
|
|
assert (settings.logo_storage_dir / erster.file_path).exists()
|
|
assert Path(erster.file_path).stem == kandidat.sha256
|
|
|
|
|
|
async def test_zuordnung_setzt_farbe_und_status(session: AsyncSession, kein_netz: None) -> None:
|
|
merchant = await firma(session, "Netflix")
|
|
|
|
asset = await resolve_merchant_logo(session, merchant)
|
|
|
|
assert asset is not None
|
|
assert merchant.logo_asset_id == asset.id
|
|
assert merchant.logo_source is LogoSource.SIMPLE_ICONS
|
|
assert merchant.logo_status is LogoStatus.RESOLVED
|
|
assert merchant.brand_color == "#e50914"
|
|
# Die aufgehellte Variante ist für den dunklen Hintergrund gedacht.
|
|
assert merchant.brand_color_dark != merchant.brand_color
|
|
|
|
|
|
async def test_zweite_zuordnung_erzeugt_kein_neues_asset(
|
|
session: AsyncSession, kein_netz: None
|
|
) -> None:
|
|
"""Akzeptanzkriterium: der zweite Aufruf löst keinen ausgehenden Request aus."""
|
|
merchant = await firma(session, "Netflix")
|
|
|
|
erster = await resolve_merchant_logo(session, merchant)
|
|
zweiter = await resolve_merchant_logo(session, merchant)
|
|
|
|
assert erster.id == zweiter.id
|
|
anzahl = (await session.execute(select(func.count()).select_from(LogoAsset))).scalar_one()
|
|
assert anzahl == 1
|
|
|
|
|
|
async def test_manuelles_logo_wird_nicht_ueberschrieben(
|
|
session: AsyncSession, kein_netz: None
|
|
) -> None:
|
|
merchant = await firma(session, "Netflix")
|
|
merchant.logo_status = LogoStatus.MANUAL
|
|
merchant.logo_asset_id = None
|
|
|
|
ergebnis = await resolve_merchant_logo(session, merchant)
|
|
|
|
assert ergebnis is None
|
|
assert merchant.logo_asset_id is None
|
|
|
|
# Mit force wird sie doch ersetzt.
|
|
erzwungen = await resolve_merchant_logo(session, merchant, force=True)
|
|
assert erzwungen is not None
|
|
|
|
|
|
# --- Endpunkte -----------------------------------------------------------------
|
|
|
|
|
|
async def test_logo_wird_lokal_mit_cache_header_ausgeliefert(
|
|
auth_client: AsyncClient, session: AsyncSession, kein_netz: None
|
|
) -> None:
|
|
merchant = await firma(session, "Netflix")
|
|
asset = await resolve_merchant_logo(session, merchant)
|
|
|
|
antwort = await auth_client.get(f"/api/logos/{asset.id}")
|
|
|
|
assert antwort.status_code == 200
|
|
assert antwort.headers["content-type"].startswith("image/svg+xml")
|
|
assert antwort.headers["cache-control"] == "public, max-age=31536000, immutable"
|
|
assert antwort.headers["etag"] == f'"{asset.sha256}"'
|
|
assert b"<svg" in antwort.content
|
|
|
|
# Mit passendem ETag antwortet der Server ohne Nutzlast.
|
|
erneut = await auth_client.get(
|
|
f"/api/logos/{asset.id}", headers={"If-None-Match": antwort.headers["etag"]}
|
|
)
|
|
assert erneut.status_code == 304
|
|
|
|
|
|
async def test_unbekanntes_logo_liefert_404(auth_client: AsyncClient) -> None:
|
|
antwort = await auth_client.get("/api/logos/999999")
|
|
|
|
assert antwort.status_code == 404
|
|
assert antwort.json()["code"] == "not_found"
|
|
|
|
|
|
async def test_suche_liefert_kandidaten_mit_vorauswahl(
|
|
auth_client: AsyncClient, netz_ohne_treffer: list[str]
|
|
) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json()
|
|
|
|
ergebnis = (await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/search")).json()
|
|
|
|
kandidaten = ergebnis["candidates"]
|
|
assert 1 <= len(kandidaten) <= 5
|
|
assert kandidaten[0]["source"] == "simple-icons"
|
|
assert kandidaten[0]["is_preselected"] is True
|
|
assert sum(1 for k in kandidaten if k["is_preselected"]) == 1
|
|
assert kandidaten[-1]["source"] == "generated"
|
|
|
|
# Jeder Kandidat liegt bereits im Cache und ist sofort abrufbar.
|
|
for kandidat in kandidaten:
|
|
bild = await auth_client.get(f"/api/logos/{kandidat['candidate_id']}")
|
|
assert bild.status_code == 200
|
|
|
|
# Die Auswahl der Firma wurde durch die Suche noch nicht verändert.
|
|
firma_danach = (await auth_client.get(f"/api/merchants/{angelegt['id']}")).json()
|
|
assert firma_danach["logo_asset_id"] is None
|
|
assert firma_danach["logo_status"] == "pending"
|
|
|
|
|
|
async def test_kandidat_auswaehlen_setzt_manuell(
|
|
auth_client: AsyncClient, netz_ohne_treffer: list[str]
|
|
) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json()
|
|
kandidaten = (await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/search")).json()[
|
|
"candidates"
|
|
]
|
|
|
|
gewaehlt = (
|
|
await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/select",
|
|
json={"candidate_id": kandidaten[0]["candidate_id"]},
|
|
)
|
|
).json()
|
|
|
|
assert gewaehlt["logo_asset_id"] == kandidaten[0]["candidate_id"]
|
|
assert gewaehlt["logo_status"] == "manual"
|
|
assert gewaehlt["brand_color"] == "#e50914"
|
|
|
|
|
|
async def test_auswahl_eines_unbekannten_kandidaten(auth_client: AsyncClient) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json()
|
|
|
|
antwort = await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/select", json={"candidate_id": 999999}
|
|
)
|
|
|
|
assert antwort.status_code == 422
|
|
assert antwort.json()["code"] == "candidate_not_found"
|
|
|
|
|
|
async def test_suche_mit_eigener_domain(
|
|
auth_client: AsyncClient, netz_ohne_treffer: list[str]
|
|
) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Streamdienst"})).json()
|
|
|
|
await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/search", params={"domain": "spotify.com"}
|
|
)
|
|
|
|
# Die Domain wird übernommen und führt zum Treffer über den Domainnamen.
|
|
danach = (await auth_client.get(f"/api/merchants/{angelegt['id']}")).json()
|
|
assert danach["domain"] == "spotify.com"
|
|
|
|
|
|
async def test_upload_setzt_manuell_und_farbe(auth_client: AsyncClient) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Hausverwaltung"})).json()
|
|
|
|
antwort = await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/upload",
|
|
files={"file": ("logo.png", png_bytes((30, 215, 96)), "image/png")},
|
|
)
|
|
|
|
assert antwort.status_code == 200
|
|
body = antwort.json()
|
|
assert body["logo_status"] == "manual"
|
|
assert body["logo_source"] == "upload"
|
|
assert body["logo_asset_id"] is not None
|
|
assert body["brand_color"] is not None
|
|
|
|
bild = await auth_client.get(f"/api/logos/{body['logo_asset_id']}")
|
|
assert bild.headers["content-type"].startswith("image/png")
|
|
|
|
|
|
async def test_upload_lehnt_fremde_dateitypen_ab(auth_client: AsyncClient) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Testfirma"})).json()
|
|
|
|
antwort = await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/upload",
|
|
files={"file": ("schaedlich.pdf", b"%PDF-1.7 kein Bild", "application/pdf")},
|
|
)
|
|
|
|
assert antwort.status_code == 422
|
|
assert antwort.json()["code"] == "unsupported_media_type"
|
|
|
|
|
|
async def test_upload_lehnt_zu_grosse_dateien_ab(auth_client: AsyncClient) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Grossfirma"})).json()
|
|
zu_gross = b"<svg>" + b"x" * (settings.logo_max_upload_bytes + 1) + b"</svg>"
|
|
|
|
antwort = await auth_client.post(
|
|
f"/api/merchants/{angelegt['id']}/logo/upload",
|
|
files={"file": ("gross.svg", zu_gross, "image/svg+xml")},
|
|
)
|
|
|
|
assert antwort.status_code == 422
|
|
assert antwort.json()["code"] == "logo_too_large"
|
|
|
|
|
|
async def test_logo_entfernen_setzt_zurueck(
|
|
auth_client: AsyncClient, netz_ohne_treffer: list[str]
|
|
) -> None:
|
|
angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json()
|
|
await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/resolve")
|
|
|
|
entfernt = (await auth_client.delete(f"/api/merchants/{angelegt['id']}/logo")).json()
|
|
|
|
assert entfernt["logo_asset_id"] is None
|
|
assert entfernt["logo_status"] == "pending"
|
|
assert entfernt["brand_color"] is None
|