141 lines
8.7 KiB
Python
141 lines
8.7 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
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_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()
|