"""Independent DTO/type oracle; no business formatter imported, no network or DB.""" import copy import json import re import unittest from pathlib import Path BASE_FIELDS = 'id supplement_no meal_order_id meal_order_no user_id department_uuid department_name restaurant_id equipment_code consume_time meal_date meal_times order_price reason marketing_strategy_id pay_price discount_amount management_fee cash subsidy gift create_by create_time update_time'.split() MONEY_FIELDS = 'order_price pay_price discount_amount management_fee cash subsidy gift'.split() def expected_row(source, detail=False): physical = source['supplement'][0] row = {key: physical[key] for key in BASE_FIELDS} # Expected contract from actual MySQL declared DEC10,2 and fixture, never wire-coercion. for key in MONEY_FIELDS: assert float(physical[key]) == 0.0 row[key] = '0.00' for key in ['consume_time', 'create_time', 'update_time']: value = physical[key] assert re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.000000)?', value) row[key] = value[:19] for key in ['id', 'meal_order_id', 'user_id', 'restaurant_id', 'meal_times', 'marketing_strategy_id']: assert type(physical[key]) is int order = source['order'][0] assert order['refund_status'] == 0 for key in ['refund_amount', 'cash_refund_amount', 'subsidy_refund_amount', 'gift_refund_amount']: assert float(order[key]) == 0 row[key] = '0.00' row.update(status=20, user_name=source['user'][0]['name'], restaurant_name=source['restaurant'][0]['name'], equipment_name=None, pay_amount='0.00', gift_net_amount='0.00', status_text='补单成功', payment_composition_text='零金额', meal_times_text='午餐') assert source['equipment'] == [] if detail: row['marketing_strategy_name'] = '未匹配消费策略' return row def diff(expected, actual, path='$'): if type(expected) is not type(actual): return [f'{path}: type {type(actual).__name__} != {type(expected).__name__}'] if isinstance(expected, dict): if set(expected) != set(actual): return [f'{path}: keys {sorted(set(actual)-set(expected))} extra / {sorted(set(expected)-set(actual))} missing'] return sum((diff(v, actual[k], path+'.'+k) for k, v in expected.items()), []) if isinstance(expected, list): if len(expected) != len(actual): return [f'{path}: length'] return sum((diff(v, actual[i], path+f'[{i}]') for i, v in enumerate(expected)), []) return [] if expected == actual else [f'{path}: value {actual!r} != {expected!r}'] def expected_response(label, source): if label == 'detail-missing': return {'code':'2', 'message':'补单记录不存在或无权查看', 'data':[]} if label == 'detail-positive': data = expected_row(source, True) else: total = 0 if label == 'list-status40' else 1 # TP5 Paginator clamps current_page to last_page, independently of SQL page2 offset. data = {'total':total,'per_page':10,'current_page':1,'last_page':1 if total else 0, 'data':[expected_row(source)] if label == 'list-positive' else []} return {'code':0,'message':'查询成功','data':data} def check_wire(record): if not record['source_unchanged'] or record['source_before'] != record['source_after']: return ['source drift'] if record['kind'] == 'unreachable': if record['native']: return ['alias native SQL not zero'] group=record['path'].split('/')[2] module=('billingrecordp' if group.startswith('billing') else 'cashrefundp' if group.startswith('cashRefund') else 'supplementmealorderp') return (['HTTP status'] if record.get('status')!=404 else [])+diff({'code':'2','message':'module not exists:'+module,'data':[]},json.loads(record['body'])) actual = json.loads(record['body']) expected = expected_response(record['label'], record['source_before']) if 'trace_id' in actual: if not isinstance(actual['trace_id'], str) or not re.fullmatch('[a-f0-9]{13}', actual['trace_id']): return ['invalid trace_id'] expected['trace_id'] = actual['trace_id'] return (['HTTP status'] if record.get('status')!=200 else [])+diff(expected, actual) class OracleTest(unittest.TestCase): def setUp(self): fixture=json.loads((Path(__file__).parent/'fixture-plan.json').read_text())['lanes']['B'] fixture['id']=101 self.source={'supplement':[fixture], 'order':[{'refund_status':0, **dict.fromkeys(['refund_amount','cash_refund_amount','subsidy_refund_amount','gift_refund_amount'],'0.00')}], 'user':[{'name':'CDX-user'}], 'restaurant':[{'name':'CDX-restaurant'}], 'equipment':[]} def test_all_valid_and_mutated_full_contract(self): count=0 for label in ['list-positive','list-page2','list-status40','detail-positive','detail-missing']: body=expected_response(label,self.source) record={'label':label,'kind':'business','status':200,'body':json.dumps(body,ensure_ascii=False),'source_before':self.source,'source_after':self.source,'source_unchanged':True,'native':[]} self.assertEqual(check_wire(record),[]) for key in body: modified=copy.deepcopy(body); del modified[key] self.assertTrue(check_wire({**record,'body':json.dumps(modified)}));count+=1 if label in ['list-positive','detail-positive']: row=body['data']['data'][0] if label=='list-positive' else body['data'] for key,value in row.items(): modified=copy.deepcopy(body); target=modified['data']['data'][0] if label=='list-positive' else modified['data'] target[key]=str(value) if not isinstance(value,str) else value+'!' self.assertTrue(check_wire({**record,'body':json.dumps(modified)}),key);count+=1 for bad in ['bad',123,'a'*12,'g'*13]: self.assertTrue(check_wire({**record,'body':json.dumps({**body,'trace_id':bad})}));count+=1 self.assertTrue(check_wire({**record,'source_unchanged':False}));count+=1 self.assertTrue(check_wire({**record,'status':500})) self.assertEqual(count,115) def test_aliases_are_exact_non_business_failures(self): for group in ['billingRecordLower','billingrecordLowercase','cashRefundLower','supplementMealOrderLower']: module='billingrecordp' if group.startswith('billing') else 'cashrefundp' if group.startswith('cashRefund') else 'supplementmealorderp' body={'code':'2','message':'module not exists:'+module,'data':[]} r={'label':'alias','path':'/p/'+group+'/detail','kind':'unreachable','status':404,'body':json.dumps(body),'native':[],'source_unchanged':True,'source_before':self.source,'source_after':self.source} self.assertEqual(check_wire(r),[]) for bad in [{'code':0,'message':'success','data':[]},{**body,'code':2},{**body,'message':'您尚未登录'},{**body,'data':{}},{**body,'extra':0}]:self.assertTrue(check_wire({**r,'body':json.dumps(bad)})) self.assertTrue(check_wire({**r,'native':[{'layer':'gate denied'}]})) self.assertTrue(check_wire({**r,'status':200})) if __name__ == '__main__': unittest.main()