"""An executable Python model, NOT ASP.NET Core, HTTP networking or real auth. All identities and data come from synthetic in-process test inputs. No port opens. The metadata selects GET/POST or a PATCH candidate; all are Python simulations. """ import copy,json,re,threading from concurrent.futures import ThreadPoolExecutor def canonical(value):return json.dumps(value,ensure_ascii=False,sort_keys=True,separators=(',',':')) def problem(status,kind,request_id): titles={400:'Invalid request',401:'Authentication required',403:'Access denied',404:'Order not found',409:'Idempotency key conflict',415:'Unsupported media type',422:'Order validation failed',500:'Unexpected failure',503:'Availability unavailable'} return {'status':status,'content_type':'application/problem+json','headers':{'WWW-Authenticate':'Bearer'} if status==401 else {},'body':{'type':'urn:maknow:problem:'+kind,'title':titles[status],'status':status,'instance':'urn:maknow:request:'+request_id,'traceId':'trace-'+request_id}} def duplicate_members(pairs): result={} for k,v in pairs: if k in result:raise ValueError('Duplicate JSON property') result[k]=v return result class ApiModel: def __init__(self,implementation): self.implementation=copy.deepcopy(implementation) self.orders={1:{'id':1,'customerReference':'seed-01','quantity':2}} self.replays={};self.lock=threading.Lock();self.writes=0;self.dependency_calls=0 def send(self,request): # Trace describes this Python model only. trace=['EXCEPTION_HANDLER','ROUTING'];rid=request['request_id'] def reply(value):value=copy.deepcopy(value);value['trace']=list(trace);return value try: method=request['method'];path=request['path'] endpoint_id='GetOrder' if method=='GET' and re.fullmatch(r'/orders/[0-9]+',path) else 'PatchOrder' if method=='PATCH' and re.fullmatch(r'/orders/[0-9]+',path) else 'CreateOrder' if method=='POST' and path=='/orders' else None if endpoint_id is None:return reply(problem(404,'not-found',rid)) ep=next((e for e in self.implementation['endpoints'] if e['endpoint_id']==endpoint_id),None) if ep is None:return reply(problem(404,'not-found',rid)) trace.extend(['AUTHENTICATION','AUTHORIZATION']);principal=request['principal'] if ep['authorization_policy'] is not None: if not principal['authenticated']:return reply(problem(401,'unauthenticated',rid)) if ep['authorization_policy'] not in principal['permissions']:return reply(problem(403,'forbidden',rid)) trace.append('BINDING') if method=='GET': trace.append('HANDLER') if request['dependency']=='FAIL':raise RuntimeError('SYNTHETIC_PRIVATE_STACK') with self.lock:order=copy.deepcopy(self.orders.get(int(path.rsplit('/',1)[1]))) return reply({'status':200,'content_type':'application/json','headers':{},'body':order}) if order else reply(problem(404,'not-found',rid)) if request['content_type']!=ep['request_content_type']:return reply(problem(415,'unsupported-media',rid)) try: data=json.loads(request['raw_body'],object_pairs_hook=duplicate_members,parse_constant=lambda _:(_ for _ in ()).throw(ValueError())) except (ValueError,TypeError):return reply(problem(400,'invalid-request',rid)) if method=='PATCH' and ep['request_schema_ref']=='PatchOrder': if not isinstance(data,dict) or set(data)-{'quantity'} or ('quantity' in data and data['quantity'] is not None and type(data['quantity']) is not int):return reply(problem(400,'invalid-request',rid)) trace.append('VALIDATION') if 'quantity' in data and (data['quantity'] is None or not 1<=data['quantity']<=20):return reply(problem(422,'validation',rid)) trace.append('HANDLER');oid=int(path.rsplit('/',1)[1]) with self.lock: if oid not in self.orders:return reply(problem(404,'not-found',rid)) current=self.orders[oid] if 'quantity' not in data or data['quantity']==current['quantity']: return reply({'status':200,'content_type':'application/json','headers':{},'body':current}) self.dependency_calls+=1 if request['dependency']=='FAIL':raise RuntimeError('SYNTHETIC_PRIVATE_STACK') if request['dependency']=='TIMEOUT':return reply(problem(503,'availability-unavailable',rid)) current['quantity']=data['quantity'];self.writes+=1 return reply({'status':200,'content_type':'application/json','headers':{},'body':current}) if not isinstance(data,dict) or set(data)!= {'customerReference','quantity'} or not isinstance(data.get('customerReference'),str) or type(data.get('quantity')) is not int:return reply(problem(400,'invalid-request',rid)) key=request['idempotency_key'] if not isinstance(key,str) or not key.strip():return reply(problem(400,'invalid-request',rid)) trace.append('VALIDATION') if not data['customerReference'].strip() or len(data['customerReference'])>40 or not 1<=data['quantity']<=20:return reply(problem(422,'validation',rid)) trace.append('HANDLER') fingerprint=canonical(data) scope=(principal['subject'],method,path,key) # One atomic local critical section; no multi-process or durable claim. with self.lock: if self.implementation['idempotency_mechanism']=='PROCESS_LOCKED_MAP' and scope in self.replays: previous_fingerprint,response=self.replays[scope] if previous_fingerprint!=fingerprint:return reply(problem(409,'idempotency-conflict',rid)) return reply(response) self.dependency_calls+=1 if request['dependency']=='FAIL':raise RuntimeError('SYNTHETIC_PRIVATE_STACK') oid=max(self.orders)+1;order={'id':oid,**data};self.orders[oid]=order;self.writes+=1 response={'status':201,'content_type':'application/json','headers':{'Location':f'/orders/{oid}'},'body':copy.deepcopy(order)} if self.implementation['idempotency_mechanism']=='PROCESS_LOCKED_MAP':self.replays[scope]=(fingerprint,copy.deepcopy(response)) return reply(response) except RuntimeError: return reply(problem(500,'unexpected',rid)) def run(self,case): requests=case['requests'] if case.get('concurrent'): with ThreadPoolExecutor(max_workers=len(requests)) as pool:responses=list(pool.map(self.send,requests)) else:responses=[self.send(r) for r in requests] return {'responses':responses,'writes':self.writes,'dependency_calls':self.dependency_calls,'final_orders':[self.orders[k] for k in sorted(self.orders)]}