from __future__ import annotations
import json
import sys
from copy import deepcopy
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from app.models import RECIPE_JSON_SCHEMA, RecipeSpec
from app.quality import local_quality_warnings
from app.source_extractor import extract_from_html
from app.tandoor_service import (
_materialize_created_objects,
resolve_object,
search_variants,
)
class FakeClient:
def __init__(self):
self.objects = {
"food": [
{
"id": 10,
"name": "Kartoffel",
"plural_name": "Kartoffeln",
"properties": [{"x": 1}],
},
{"id": 11, "name": "Zwiebeln", "plural_name": "Zwiebeln"},
{"id": 12, "name": "Schinken", "plural_name": "Schinken"},
],
"unit": [
{"id": 20, "name": "g", "plural_name": "g"},
{"id": 21, "name": "TL", "plural_name": "TL"},
],
"keyword": [],
}
self.next_id = 100
self.posts = []
self.deleted = []
def list_objects(self, endpoint, force_refresh=False):
return deepcopy(self.objects.get(endpoint, []))
def get_json(self, path):
parts = path.strip("/").split("/")
endpoint = parts[-2]
object_id = int(parts[-1])
for obj in self.objects.get(endpoint, []):
if obj["id"] == object_id:
return deepcopy(obj)
raise RuntimeError("not found")
def post_json(self, path, payload):
endpoint = path.strip("/").split("/")[-1]
obj = {"id": self.next_id, **payload}
self.next_id += 1
self.objects.setdefault(endpoint, []).append(obj)
self.posts.append((endpoint, deepcopy(payload)))
return deepcopy(obj)
def forget_cache(self, endpoint):
pass
def delete(self, path):
self.deleted.append(path)
def test_json_ld_extraction():
html = """
Test
Testrezept
Mischen.
"""
source = extract_from_html("https://example.com/rezept.html", html)
assert source.title == "Testrezept"
assert source.structured_recipe["@type"] == "Recipe"
assert source.image_url == "https://example.com/image.jpg"
def test_recipe_validation():
payload = json.loads((ROOT / "examples" / "crispy-chickpeas.json").read_text())
recipe = RecipeSpec.model_validate(payload)
assert recipe.name.startswith("Knusprige")
assert local_quality_warnings(recipe) == []
def test_suspicious_unit():
payload = json.loads((ROOT / "examples" / "crispy-chickpeas.json").read_text())
payload["steps"][0]["ingredients"][0]["unit_name"] = "Limette"
recipe = RecipeSpec.model_validate(payload)
warnings = local_quality_warnings(recipe)
assert any(item["code"] == "suspicious_unit" for item in warnings)
def test_schema_has_strict_objects():
assert RECIPE_JSON_SCHEMA["additionalProperties"] is False
ingredient = (
RECIPE_JSON_SCHEMA["properties"]["steps"]["items"]["properties"]
["ingredients"]["items"]
)
assert ingredient["additionalProperties"] is False
required = set(ingredient["required"])
assert {"create_food", "create_unit", "food_plural_name", "unit_plural_name"} <= required
def test_german_word_variants_and_preselection():
assert "kartoffel" in search_variants("Kartoffeln")
assert "zwiebeln" in search_variants("Zwiebel")
client = FakeClient()
potato = resolve_object(
client,
"food",
"Kartoffeln",
None,
object_type="Food",
)
assert potato["blocking"] is False
assert potato["resolved"]["id"] == 10
assert potato["selected_id"] == 10
onion = resolve_object(
client,
"food",
"Zwiebel",
None,
object_type="Food",
)
assert onion["blocking"] is False
assert onion["resolved"]["id"] == 11
def test_missing_can_be_explicitly_created():
client = FakeClient()
missing = resolve_object(
client,
"food",
"Kochschinkenwürfel",
None,
object_type="Food",
)
# Schinken ist als Teilwort-Kandidat vorhanden und wird vorausgewählt.
assert missing["blocking"] is False
assert missing["candidates"]
create = resolve_object(
client,
"food",
"Tempeh-Crunch",
None,
object_type="Food",
create_requested=True,
plural_name="Tempeh-Crunch",
)
assert create["status"] == "create"
assert create["blocking"] is False
def test_create_objects_are_materialized_for_recipe_payload():
client = FakeClient()
payload = {
"steps": [
{
"ingredients": [
{
"food": {
"name": "Tempeh-Crunch",
"plural_name": None,
"create": True,
"endpoint": "food",
},
"unit": {
"name": "Portion",
"plural_name": "Portionen",
"create": True,
"endpoint": "unit",
},
}
]
}
]
}
final, created = _materialize_created_objects(client, payload)
ingredient = final["steps"][0]["ingredients"][0]
assert ingredient["food"]["id"] == 100
assert ingredient["unit"]["id"] == 101
assert len(created) == 2
assert client.posts == [
("food", {"name": "Tempeh-Crunch"}),
("unit", {"name": "Portion", "plural_name": "Portionen"}),
]
if __name__ == "__main__":
test_json_ld_extraction()
test_recipe_validation()
test_suspicious_unit()
test_schema_has_strict_objects()
test_german_word_variants_and_preselection()
test_missing_can_be_explicitly_created()
test_create_objects_are_materialized_for_recipe_payload()
print("Alle Selbsttests erfolgreich.")