232 lines
12 KiB
Python
232 lines
12 KiB
Python
"""Room/profile source agreement is not native ARR display acceptance."""
|
|
import copy
|
|
import unittest
|
|
|
|
from integrations.ohip import source_fields as fields
|
|
from integrations.ohip.collect_arr_source import CollectionError
|
|
from tests.test_ohip_source_fields import detail
|
|
from tests.test_ohip_arr_collection import DAY
|
|
|
|
|
|
def rooms(number=' 0007A '):
|
|
record = detail()
|
|
record['roomStay'].update(currentRoomInfo={'roomId': number, 'suggestedRoomNumbers': ['OTHER']})
|
|
record['roomStay']['roomRates'][0]['roomId'] = 'OTHER_DAY'
|
|
record['roomStay']['roomRates'][1]['roomId'] = number
|
|
search = {'hotelId': record['hotelId'], 'reservationIdList': copy.deepcopy(record['reservationIdList']),
|
|
'roomStay': {'arrivalDate': DAY, 'roomId': number}}
|
|
return search, record
|
|
|
|
|
|
def profile(role='Company', identity='profile-001', name=' C- RAW NAME\r\nSECOND LINE '):
|
|
return {'reservationProfileType': role, 'profileIdList': [{'type': 'Profile', 'id': identity}],
|
|
'profile': {'company': {'companyName': name}}}
|
|
|
|
|
|
def profiles(*items):
|
|
record = detail()
|
|
record['reservationProfiles'] = {'reservationProfile': copy.deepcopy(list(items))}
|
|
record['roomStay']['roomRates'][1]['stayProfiles'] = copy.deepcopy(list(items))
|
|
record['roomStay']['roomRates'][0]['stayProfiles'] = [profile(name='OTHER DAY')]
|
|
return record
|
|
|
|
|
|
def room_types(value=' 001A '):
|
|
search, record = rooms()
|
|
search['roomStay'].update(roomType=value, roomTypeCharged='CHARGED')
|
|
record['roomStay']['currentRoomInfo'].update(roomType=value, description='DESCRIPTION')
|
|
record['roomStay']['roomRates'][0].update(roomType='OTHER_DAY')
|
|
record['roomStay']['roomRates'][1].update(roomType=value, roomTypeCharged='CHARGED')
|
|
return search, record
|
|
|
|
|
|
class RoomProfileFieldTests(unittest.TestCase):
|
|
def test_room_type_exact_three_layer_agreement_preserves_code(self):
|
|
for value in (' 001A ', '001', 'ไทย', '', ' '):
|
|
search, record = room_types(value)
|
|
before = copy.deepcopy((search, record))
|
|
self.assertEqual(fields.agreed_room_type(search, record, DAY), value)
|
|
self.assertEqual((search, record), before)
|
|
|
|
def test_room_type_missing_invalid_or_conflicting_layer_has_no_fallback(self):
|
|
for level in ('search', 'current', 'day'):
|
|
for value in (None, 1, False, [], {}, 'OTHER'):
|
|
search, record = room_types()
|
|
target = (search['roomStay'] if level == 'search' else record['roomStay']['currentRoomInfo']
|
|
if level == 'current' else record['roomStay']['roomRates'][1])
|
|
target['roomType'] = value
|
|
with self.subTest(level=level, value=value), self.assertRaises(CollectionError):
|
|
fields.agreed_room_type(search, record, DAY)
|
|
del target['roomType']
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_type(search, record, DAY)
|
|
search, record = room_types()
|
|
record['roomStay'].pop('currentRoomInfo')
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_type(search, record, DAY)
|
|
|
|
def test_equal_room_types_still_require_reservation_hotel_and_date_binding(self):
|
|
changes = [lambda s, d: s['reservationIdList'][-1].update(id='OTHER_RESERVATION'),
|
|
lambda s, d: s.update(hotelId='OTHER_HOTEL'),
|
|
lambda s, d: (s.pop('hotelId'), d.pop('hotelId')),
|
|
lambda s, d: s['roomStay'].update(arrivalDate='2026-09-14'),
|
|
lambda s, d: d['roomStay'].update(arrivalDate='2026-09-14'),
|
|
lambda s, d: d['roomStay']['roomRates'][0].update(start=DAY),
|
|
lambda s, d: d['roomStay']['roomRates'].append(
|
|
copy.deepcopy(d['roomStay']['roomRates'][1]))]
|
|
for change in changes:
|
|
search, record = room_types()
|
|
change(search, record)
|
|
with self.subTest(change=change), self.assertRaises(CollectionError):
|
|
fields.agreed_room_type(search, record, DAY)
|
|
|
|
def test_exact_room_agreement_preserves_text_and_explicit_blank(self):
|
|
for value in (' 0007A ', '0007', '7', '', ' ', 'ห้อง中'):
|
|
search, record = rooms(value)
|
|
before = copy.deepcopy((search, record))
|
|
self.assertEqual(fields.agreed_room_number(search, record, DAY), value)
|
|
self.assertEqual((search, record), before)
|
|
|
|
def test_missing_bad_or_different_room_layer_never_falls_back(self):
|
|
for level in ('search', 'current', 'day'):
|
|
for value in (None, 7, False, [], {}, '0007A'):
|
|
search, record = rooms()
|
|
target = (search['roomStay'] if level == 'search' else record['roomStay']['currentRoomInfo']
|
|
if level == 'current' else record['roomStay']['roomRates'][1])
|
|
target['roomId'] = value
|
|
with self.subTest(level=level, value=value), self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
target.pop('roomId')
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
for value in (None, {}, []):
|
|
search, record = rooms()
|
|
record['roomStay']['currentRoomInfo'] = value
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
|
|
def test_same_room_does_not_bind_wrong_reservation_hotel_or_date(self):
|
|
changes = [lambda r: r['reservationIdList'][-1].update(id='other-reservation'),
|
|
lambda r: r.update(hotelId='OTHER_HOTEL'),
|
|
lambda r: r.pop('hotelId'),
|
|
lambda r: r['roomStay'].update(arrivalDate='2026-09-14'),
|
|
lambda r: r['roomStay'].pop('arrivalDate')]
|
|
for change in changes:
|
|
search, record = rooms()
|
|
change(search)
|
|
with self.subTest(change=change), self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
search, record = rooms()
|
|
record.pop('hotelId')
|
|
search.pop('hotelId')
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
|
|
def test_role_is_explicit_no_priority_group_fallback_prefix_or_name_normalization(self):
|
|
record = profiles(*(profile(role=role, identity=role, name=' '+role+'\r\nRAW ')
|
|
for role in ('Company', 'TravelAgent', 'Source', 'Group')))
|
|
before = copy.deepcopy(record)
|
|
for role in ('Company', 'TravelAgent', 'Source', 'Group'):
|
|
actual = fields.agreed_associated_profile(record, DAY, role)
|
|
self.assertEqual(actual, fields.AssociatedProfile(role, role, ' '+role+'\r\nRAW '))
|
|
self.assertNotIn('RAW', repr(actual))
|
|
self.assertNotIn('profile_id=', repr(actual))
|
|
self.assertEqual(record, before)
|
|
group_only = profiles(profile('Group'))
|
|
self.assertIsNone(fields.agreed_associated_profile(group_only, DAY, 'Company'))
|
|
for role in (None, 'company', 'Company ', '', 'Guest', [], True):
|
|
with self.subTest(role=role), self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(record, DAY, role)
|
|
|
|
def test_extra_external_identifiers_do_not_create_false_profile_conflict(self):
|
|
record = profiles(profile())
|
|
top = record['reservationProfiles']['reservationProfile'][0]
|
|
day = record['roomStay']['roomRates'][1]['stayProfiles'][0]
|
|
top['profileIdList'].insert(0, {'type': 'CorporateId', 'id': 'NOT_INTERNAL'})
|
|
day['profileIdList'].append({'type': 'IATA', 'id': 'OTHER_EXTERNAL'})
|
|
day['profile']['company']['extraMetadata'] = 'difference'
|
|
self.assertNotEqual(top, day)
|
|
self.assertEqual(fields.agreed_associated_profile(record, DAY, 'Company'),
|
|
fields.AssociatedProfile('Company', 'profile-001', ' C- RAW NAME\r\nSECOND LINE '))
|
|
|
|
def test_empty_name_absent_role_and_missing_arrays_are_distinct(self):
|
|
self.assertIsNone(fields.agreed_associated_profile(profiles(), DAY, 'Company'))
|
|
for name in ('', ' '):
|
|
selected = fields.agreed_associated_profile(profiles(profile(name=name)), DAY, 'Company')
|
|
self.assertEqual(selected.name, name)
|
|
self.assertIsNotNone(selected)
|
|
for layer in ('reservation', 'day'):
|
|
for value in (None, {}, '', False):
|
|
record = profiles()
|
|
target = record['reservationProfiles'] if layer == 'reservation' else record['roomStay']['roomRates'][1]
|
|
key = 'reservationProfile' if layer == 'reservation' else 'stayProfiles'
|
|
target[key] = value
|
|
with self.subTest(layer=layer, value=value), self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
target.pop(key)
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
record = profiles()
|
|
record.pop('reservationProfiles')
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
|
|
def test_internal_identity_name_or_role_presence_conflicts_are_explicit(self):
|
|
changes = [lambda x: x['profileIdList'][0].update(id='other-id'),
|
|
lambda x: x['profile']['company'].update(companyName='different'),
|
|
lambda x: x.update(reservationProfileType='Source')]
|
|
for change in changes:
|
|
record = profiles(profile())
|
|
change(record['roomStay']['roomRates'][1]['stayProfiles'][0])
|
|
with self.subTest(change=change), self.assertRaisesRegex(CollectionError, 'profile_disagreement'):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
record = profiles(profile())
|
|
record['roomStay']['roomRates'][1]['stayProfiles'].clear()
|
|
with self.assertRaisesRegex(CollectionError, 'profile_disagreement'):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
|
|
def test_duplicate_roles_or_internal_ids_rejected_even_with_identical_values(self):
|
|
record = profiles(profile(), profile())
|
|
with self.assertRaisesRegex(CollectionError, 'ambiguous_associated_profile'):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
for ids in ([], [{'type': 'CorporateId', 'id': 'profile-001'}],
|
|
[{'type': 'Profile', 'id': 'profile-001'}]*2,
|
|
[{'type': 'Profile', 'id': ' ' }], [{'type': 'Profile', 'id': 1}],
|
|
[None], [{'id': 'profile-001'}]):
|
|
p = profile()
|
|
p['profileIdList'] = ids
|
|
with self.subTest(ids=ids), self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(profiles(p), DAY, 'Company')
|
|
|
|
def test_unknown_role_or_malformed_selected_profile_is_not_an_empty_company(self):
|
|
for role in (None, 'company', '', [], 'NewRole'):
|
|
with self.subTest(role=role), self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(profiles(profile(role)), DAY, 'Company')
|
|
for value in (None, {}, [], {'company': None}, {'company': {}},
|
|
{'company': {'companyName': None}}, {'company': {'companyName': 1}}):
|
|
p = profile()
|
|
p['profile'] = value
|
|
with self.subTest(value=value), self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(profiles(p), DAY, 'Company')
|
|
|
|
def test_both_selectors_keep_strict_day_segment_guard(self):
|
|
for mode in ('duplicate', 'range', 'wrong_arrival'):
|
|
search, record = rooms()
|
|
record['reservationProfiles'] = {'reservationProfile': []}
|
|
record['roomStay']['roomRates'][1]['stayProfiles'] = []
|
|
if mode == 'duplicate':
|
|
record['roomStay']['roomRates'].append(copy.deepcopy(record['roomStay']['roomRates'][1]))
|
|
elif mode == 'range':
|
|
record['roomStay']['roomRates'][0]['start'] = DAY
|
|
else:
|
|
record['roomStay']['arrivalDate'] = '2026-09-14'
|
|
with self.subTest(mode=mode):
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_room_number(search, record, DAY)
|
|
with self.assertRaises(CollectionError):
|
|
fields.agreed_associated_profile(record, DAY, 'Company')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|