86 lines
5.0 KiB
Python
86 lines
5.0 KiB
Python
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()
|