import copy import json from pathlib import Path import tempfile import unittest from urllib.parse import parse_qs, urlsplit from integrations.ohip import collect_arr_source as source from integrations.ohip import profile_summary as probe HOTEL='TEST_HOTEL' def detail(): return {'reservationIdList':[{'type':'Reservation','id':'r1'}], 'reservationGuests':[ {'primary':False,'profileInfo':{}}, {'primary':True,'profileInfo':{'profileIdList':[{'type':'Profile','id':'p001'}], 'profile':{'customer':{'personName':[{'nameType':'Alternate','surname':'WRONG'}, {'nameType':'Primary','surname':'测试','givenName':'Name','middleName':''}]}}}}]} def response(): return {'operation_id':'searchProfiles','hotel_id':HOTEL,'oracle_request_id':'synthetic', 'data':{'profileSummaries':{'hasMore':False,'profileInfo':[ {'profileIdList':[{'type':'Profile','id':'p001'}], 'profile':{'formerName':{'name':'测试','givenName':'Name','middleName':'','fullName':'测试, Name', 'nameType':'Primary'},'altName':{'fullName':'UNRELATED'}}}]}}} class ProfileSummaryTests(unittest.TestCase): def test_get_summary_uses_exact_id_and_validates_its_own_operation(self): method, path, raw = probe.get_request('p001') self.assertEqual((method, urlsplit(path).path, raw), ('GET', '/api/v1/profiles', None)) self.assertEqual(parse_qs(urlsplit(path).query), {'profileIds': ['p001'], 'summaryInfo': ['true'], 'limit': ['1'], 'offset': ['0']}) doc = response() doc['operation_id'] = probe.GET_OPERATION guest = probe.primary_guest(detail()) result = probe.inspect_summary(source.json_bytes(doc), guest, HOTEL, operation=probe.GET_OPERATION) self.assertTrue(result['identity_matched']) with self.assertRaisesRegex(source.CollectionError, 'profile_response_context_mismatch'): probe.inspect_summary(source.json_bytes(doc), guest, HOTEL) doc['operation_id'] = probe.OPERATION with self.assertRaisesRegex(source.CollectionError, 'profile_response_context_mismatch'): probe.inspect_summary(source.json_bytes(doc), guest, HOTEL, operation=probe.GET_OPERATION) for identity in ('', '../x', False, 123, None): with self.subTest(identity=identity), self.assertRaises(source.CollectionError): probe.get_request(identity) def test_handwritten_single_id_query_and_exact_component_observations(self): guest=probe.primary_guest(detail()) method,path,raw=probe.request(guest.profile_id) self.assertEqual((method,path),('POST','/api/v1/profiles/searches')) self.assertEqual(json.loads(raw),{'profileIds':['p001'],'summaryInfo':True,'limit':1,'offset':0}) result=probe.inspect_summary(source.json_bytes(response()),guest,HOTEL) self.assertEqual(result['full_name'],'测试, Name') self.assertEqual(result['component_comparisons'],{'surname':'equal','givenName':'equal','middleName':'equal','nameTitle':'both_absent'}) self.assertFalse(result['report_equivalence_verified']) self.assertFalse(result['finance_ready']) self.assertNotIn('p001',repr(guest)) self.assertNotIn('测试',repr(guest)) def test_no_guest_or_primary_name_identity_ambiguity_fallback(self): changes=[lambda d:d['reservationGuests'].append(copy.deepcopy(d['reservationGuests'][1])), lambda d:d['reservationGuests'][1].update(primary=1), lambda d:d['reservationGuests'][1].update(primary=False), lambda d:d['reservationGuests'][1]['profileInfo']['profileIdList'].append({'type':'Profile','id':'p001'}), lambda d:d['reservationGuests'][1]['profileInfo']['profile']['customer']['personName'].append({'nameType':'Primary','surname':'SAME'}), lambda d:d['reservationGuests'][1]['profileInfo']['profile']['customer'].update(personName=[])] for change in changes: row=detail();change(row) with self.subTest(change=change),self.assertRaises(source.CollectionError): probe.primary_guest(row) for value in ('','../x',False,123,None): with self.subTest(value=value),self.assertRaises(source.CollectionError): probe.request(value) def test_exact_returned_profile_and_single_result_required(self): for mode in ('wrong_id','duplicate_id','missing_id','no_result','two_results','has_more','numeric_false'): doc=response();summaries=doc['data']['profileSummaries'];rows=summaries['profileInfo'] if mode=='wrong_id': rows[0]['profileIdList'][0]['id']='p002' elif mode=='duplicate_id': rows[0]['profileIdList'].append({'type':'Profile','id':'p001'}) elif mode=='missing_id': rows[0]['profileIdList'][0]['type']='External' elif mode=='no_result': rows.clear() elif mode=='two_results': rows.append(copy.deepcopy(rows[0])) elif mode=='has_more': summaries['hasMore']=True else: summaries['hasMore']=0 with self.subTest(mode=mode),self.assertRaises(source.CollectionError): probe.inspect_summary(source.json_bytes(doc),probe.primary_guest(detail()),HOTEL) def test_missing_blank_different_and_alternate_names_never_become_success(self): for mode in ('missing','blank','different','alternate'): doc=response();name=doc['data']['profileSummaries']['profileInfo'][0]['profile']['formerName'] if mode=='missing': del name['fullName'] elif mode=='blank': name['fullName']=' ' elif mode=='different': name['givenName']='DIFFERENT' else: name['nameType']='Alternate' result=probe.inspect_summary(source.json_bytes(doc),probe.primary_guest(detail()),HOTEL) self.assertFalse(result['report_equivalence_verified']) if mode=='missing': self.assertEqual(result['full_name_state'],'missing');self.assertIsNone(result['full_name']) if mode=='blank': self.assertEqual(result['full_name_state'],'explicit_blank');self.assertEqual(result['full_name'],' ') if mode=='different': self.assertEqual(result['component_comparisons']['givenName'],'different') if mode=='alternate': self.assertEqual(result['name_type'],'Alternate') def test_component_missing_and_explicit_blank_not_equated(self): doc=response();name=doc['data']['profileSummaries']['profileInfo'][0]['profile']['formerName'] name.pop('middleName');name['nameTitle']='' result=probe.inspect_summary(source.json_bytes(doc),probe.primary_guest(detail()),HOTEL) self.assertEqual(result['component_comparisons']['middleName'],'summary_missing') self.assertEqual(result['component_comparisons']['nameTitle'],'summary_only') def test_response_context_warnings_bad_json_and_invalid_names_refused(self): changes=[lambda d:d.update(hotel_id='OTHER'),lambda d:d.update(operation_id='getProfiles'), lambda d:d.pop('oracle_request_id'),lambda d:d.update(warnings=[{'text':'PRIVATE'}]), lambda d:d['data']['profileSummaries']['profileInfo'][0]['profile']['formerName'].update(fullName=None)] for change in changes: doc=response();change(doc) with self.subTest(change=change),self.assertRaises(source.CollectionError): probe.inspect_summary(source.json_bytes(doc),probe.primary_guest(detail()),HOTEL) for raw in (b'{}',b'{"a":1,"a":2}',b'{"n":NaN}'): with self.assertRaises(source.CollectionError): probe.inspect_summary(raw,probe.primary_guest(detail()),HOTEL) def test_single_attempt_archived_permissions_and_input_unchanged(self): with tempfile.TemporaryDirectory() as tmp: archive=source.Archive(Path(tmp)/'probe');row=detail();original=copy.deepcopy(row);calls=[] raw=source.json_bytes(response()) def transport(method,path,body): calls.append((method,path,body));return 200,{'X-Request-ID':'synthetic-edge'},raw result=probe.read_once(probe.primary_guest(row),HOTEL,transport,archive,1) self.assertTrue(result['identity_matched']);self.assertEqual(len(calls),1);self.assertEqual(row,original) self.assertEqual((archive.path/'profile-01.response.bin').read_bytes(),raw) self.assertEqual(archive.path.stat().st_mode&0o777,0o700) self.assertTrue(all(p.stat().st_mode&0o777==0o600 for p in archive.path.iterdir())) def test_http_denial_and_transient_failure_not_retried_or_fallback(self): for status in (403,429,500): with tempfile.TemporaryDirectory() as tmp: archive=source.Archive(Path(tmp)/'probe');calls=[] def transport(*args):calls.append(args);return status,{},b'{}' with self.assertRaisesRegex(source.CollectionError,'profile_http_failure'): probe.read_once(probe.primary_guest(detail()),HOTEL,transport,archive,1) self.assertEqual(len(calls),1) self.assertEqual(json.loads((archive.path/'profile-01.meta.json').read_bytes())['http_status'],status) def test_invalid_index_prevents_request_and_transport_error_is_safe(self): with tempfile.TemporaryDirectory() as tmp: archive=source.Archive(Path(tmp)/'probe');calls=[] def transport(*args):calls.append(args);raise TimeoutError('PRIVATE') for index in (0,4,True): with self.assertRaises(source.CollectionError): probe.read_once(probe.primary_guest(detail()),HOTEL,transport,archive,index) self.assertEqual(calls,[]) with self.assertRaisesRegex(source.CollectionError,'^profile_transport_failure$'): probe.read_once(probe.primary_guest(detail()),HOTEL,transport,archive,1) self.assertNotIn('PRIVATE',(archive.path/'profile-01.meta.json').read_text()) if __name__=='__main__':unittest.main()