"""Typed BlockCode observations must not be confused with Block IDs/names.""" 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 block(code=' 0007-ไทย ', identity='block-001'): return {'blockIdList': [{'type': 'BlockCode', 'id': code}, {'type': 'Block', 'id': identity}], 'blockName': 'NAME_IS_NOT_CODE'} def records(code=' 0007-ไทย '): record = detail() record['roomStay']['roomRates'][0]['reservationBlock'] = block('OTHER_DAY') record['roomStay']['roomRates'][1]['reservationBlock'] = block(code) search = {'hotelId': record['hotelId'], 'reservationIdList': copy.deepcopy(record['reservationIdList']), 'roomStay': {'arrivalDate': DAY, 'reservationBlock': block(code)}} return search, record def blocks(search, record): return (search['roomStay']['reservationBlock'], record['roomStay']['roomRates'][1]['reservationBlock']) class BlockFieldTests(unittest.TestCase): def test_typed_code_is_not_id_name_first_identifier_or_other_day(self): search, record = records() left, right = blocks(search, record) left['blockIdList'].insert(0, {'type': 'ExternalReference', 'id': 'NOT_CODE'}) right['blockIdList'].reverse() right['blockName'] = 'DIFFERENT_NAME_NOT_USED' before = copy.deepcopy((search, record)) self.assertEqual(fields.agreed_block_code(search, record, DAY), ' 0007-ไทย ') self.assertEqual((search, record), before) def test_both_missing_is_a_gap_and_never_a_blank_report(self): for replacement in (None, {}, [], '', False): search, record = records() search['roomStay']['reservationBlock'] = replacement record['roomStay']['roomRates'][1]['reservationBlock'] = replacement with self.subTest(value=replacement), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) search, record = records() del search['roomStay']['reservationBlock'] del record['roomStay']['roomRates'][1]['reservationBlock'] with self.assertRaisesRegex(CollectionError, 'missing_or_invalid_reservation_block'): fields.agreed_block_code(search, record, DAY) def test_explicit_blank_code_preserved_but_missing_null_and_numeric_rejected(self): for code in ('', ' '): search, record = records(code) self.assertEqual(fields.agreed_block_code(search, record, DAY), code) for code in (None, 7, False, [], {}): search, record = records(code) with self.subTest(code=code), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) search, record = records() for item in blocks(search, record): del item['blockIdList'][0]['id'] with self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) def test_same_code_on_different_blocks_is_rejected(self): for index, value in ((0, 'OTHER_CODE'), (1, 'OTHER_BLOCK_ID')): search, record = records() blocks(search, record)[1]['blockIdList'][index]['id'] = value with self.subTest(index=index), self.assertRaisesRegex(CollectionError, 'disagreement'): fields.agreed_block_code(search, record, DAY) def test_duplicate_or_missing_typed_ids_cannot_choose_first_or_fall_back(self): for kind in ('Block', 'BlockCode'): for mode in ('duplicate', 'missing', 'untyped', 'wrong_case'): search, record = records() for b in blocks(search, record): item = next(x for x in b['blockIdList'] if x['type'] == kind) if mode == 'duplicate': b['blockIdList'].append(copy.deepcopy(item)) elif mode == 'missing': b['blockIdList'].remove(item) elif mode == 'untyped': del item['type'] else: item['type'] = kind.lower() with self.subTest(kind=kind, mode=mode), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) def test_exact_hotel_and_optional_id_context_binding(self): search, record = records() left, right = blocks(search, record) left['hotelId'] = record['hotelId'] right['blockIdList'][0]['idContext'] = 'OPERA' self.assertEqual(fields.agreed_block_code(search, record, DAY), ' 0007-ไทย ') for layer in (0, 1): for value in ('OTHER', None, '', False): search, record = records() blocks(search, record)[layer]['hotelId'] = value with self.subTest(layer=layer, hotel=value), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) for kind_index in (0, 1): search, record = records() blocks(search, record)[layer]['blockIdList'][kind_index]['idContext'] = 'EXTERNAL' with self.subTest(layer=layer, kind_index=kind_index), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) def test_reservation_hotel_day_and_segment_binding_remain_required(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 = records() change(search, record) with self.subTest(change=change), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) def test_unusable_block_identity_or_id_collection_is_rejected(self): for identity in (None, '', ' ', 1, False, [], {}): search, record = records() for b in blocks(search, record): b['blockIdList'][1]['id'] = identity with self.subTest(identity=identity), self.assertRaises(CollectionError): fields.agreed_block_code(search, record, DAY) for value in (None, {}, [], [None], [{'type': False, 'id': 'PRIVATE'}]): search, record = records() blocks(search, record)[0]['blockIdList'] = value with self.subTest(value=value), self.assertRaises(CollectionError) as caught: fields.agreed_block_code(search, record, DAY) self.assertNotIn('PRIVATE', str(caught.exception)) if __name__ == '__main__': unittest.main()