michael/boehmitools-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: initial import

Browse Source
This commit is contained in:
michael
2026-07-24 21:37:03 +02:00
commit 45bc449ea0
53 changed files with 12329 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/tandoor-ai-import/tests/integration_mock.py
+85
View File
@@ -0,0 +1,85 @@
from __future__ import annotations
import json, os, threading
from copy import deepcopy
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from app.models import RecipeSpec
from app.tandoor_service import resolve_recipe, import_recipe
state = {
'foods': [
{'id': 1, 'name': 'Kartoffel', 'plural_name': 'Kartoffeln', 'properties': [{'x':1}]},
{'id': 2, 'name': 'Schinken', 'plural_name': 'Schinken', 'properties': []},
{'id': 3, 'name': 'Zwiebeln', 'plural_name': 'Zwiebeln', 'properties': []},
],
'units': [{'id': 10, 'name': 'g', 'plural_name': 'g'}],
'keywords': [],
'recipe': None,
'next_food_id': 100,
}
class H(BaseHTTPRequestHandler):
def sendj(self, x, status=200):
b=json.dumps(x).encode(); self.send_response(status); self.send_header('Content-Type','application/json'); self.send_header('Content-Length',str(len(b))); self.end_headers(); self.wfile.write(b)
def do_GET(self):
p=urlparse(self.path).path
if p == '/api/food/': return self.sendj({'results':state['foods'],'next':None})
if p == '/api/unit/': return self.sendj({'results':state['units'],'next':None})
if p == '/api/keyword/': return self.sendj({'results':state['keywords'],'next':None})
if p == '/api/recipe/': return self.sendj({'results':[],'next':None})
if p == '/api/recipe/999/': return self.sendj(state['recipe'])
if p.startswith('/api/food/'):
oid=int(p.rstrip('/').split('/')[-1]); obj=next(x for x in state['foods'] if x['id']==oid); return self.sendj(obj)
if p.startswith('/api/unit/'):
oid=int(p.rstrip('/').split('/')[-1]); obj=next(x for x in state['units'] if x['id']==oid); return self.sendj(obj)
return self.sendj({'detail':'not found'},404)
def do_POST(self):
n=int(self.headers.get('Content-Length','0')); data=json.loads(self.rfile.read(n) or b'{}')
if self.path == '/api/food/':
obj={'id':state['next_food_id'], **data}; state['next_food_id'] += 1; state['foods'].append(obj); return self.sendj(obj,201)
if self.path == '/api/recipe/':
r=deepcopy(data); r['id']=999; iid=2000; sid=1000
for st in r['steps']:
st['id']=sid; sid += 1
for ing in st['ingredients']:
ing['id']=iid; iid += 1
fid=ing['food']['id']; ing['food']=next(x for x in state['foods'] if x['id']==fid)
if ing.get('unit'):
uid=ing['unit']['id']; ing['unit']=next(x for x in state['units'] if x['id']==uid)
state['recipe']=r; return self.sendj(r,201)
return self.sendj({'detail':'not found'},404)
def do_PUT(self): return self.sendj({'image':'ok'})
def do_DELETE(self): self.send_response(204); self.end_headers()
def log_message(self,*args): pass
srv=HTTPServer(('127.0.0.1',0),H); threading.Thread(target=srv.serve_forever,daemon=True).start()
os.environ['TANDOOR_URL']=f'http://127.0.0.1:{srv.server_port}'
os.environ['TANDOOR_TOKEN']='x'
os.environ['TANDOOR_AUTH_SCHEME']='Bearer'
raw={
'schema_version':1,'name':'Bratkartoffeln','description':'','source_url':'https://example.com/r','image_url':None,
'servings':2,'servings_text':'2 Portionen','working_time':10,'waiting_time':30,'keywords':[],'confidence':'high','warnings':[],
'steps':[{'name':'Bratkartoffeln','instruction':'Braten.','time':30,'ingredients':[
{'food_name':'Kartoffeln','preferred_food_id':None,'create_food':False,'food_plural_name':None,'amount':500,'amount_max':None,'unit_name':'g','preferred_unit_id':None,'create_unit':False,'unit_plural_name':None,'note':'','no_amount':False,'original_text':'500 g Kartoffeln'},
{'food_name':'Kochschinken','preferred_food_id':None,'create_food':False,'food_plural_name':None,'amount':100,'amount_max':None,'unit_name':'g','preferred_unit_id':None,'create_unit':False,'unit_plural_name':None,'note':'','no_amount':False,'original_text':'100 g Kochschinken'},
{'food_name':'Zauberkrume','preferred_food_id':None,'create_food':True,'food_plural_name':'Zauberkrumen','amount':20,'amount_max':None,'unit_name':'g','preferred_unit_id':None,'create_unit':False,'unit_plural_name':None,'note':'','no_amount':False,'original_text':'20 g Zauberkrume'},
]}]
}
recipe=RecipeSpec.model_validate(raw)
res=resolve_recipe(recipe)
assert not res['blocking'], res
m=res['mappings']
assert m[0]['food']['resolved']['name']=='Kartoffel', m[0]
assert m[1]['food']['resolved']['name']=='Schinken', m[1]
assert m[2]['food']['status']=='create', m[2]
out=import_recipe(recipe,res,import_image=False,force_duplicate=False)
assert out['status']=='imported', out
assert any(x['name']=='Zauberkrume' for x in out['created_objects']), out
assert state['recipe']['steps'][0]['ingredients'][2]['food']['name']=='Zauberkrume'
print('Integrationstest erfolgreich:', out['recipe_url'])
srv.shutdown()
plugins/tandoor-ai-import/tests/selftest.py
+215
View File
@@ -0,0 +1,215 @@
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 = """
<html><head>
<title>Test</title>
<script type="application/ld+json">
{
"@context": "https://schema.org",
"@type": "Recipe",
"name": "Testrezept",
"image": "https://example.com/image.jpg",
"recipeIngredient": ["1 TL Salz"],
"recipeInstructions": ["Mischen."]
}
</script>
</head><body><main><h1>Testrezept</h1><p>Mischen.</p></main></body></html>
"""
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.")