Files
ARR-2.0-0918/tests/test_ohip_source_fields.py

397 lines
23 KiB
Python

"""Guarded source selectors; no fixture certifies report equivalence."""
import copy
from dataclasses import replace
from datetime import date
from decimal import Decimal, localcontext
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
import xml.etree.ElementTree as ET
from integrations.ohip import audit_arr_day as audit
from integrations.ohip import collect_arr_day as day
from integrations.ohip import collect_arr_source as capture
from integrations.ohip import rate_info
from integrations.ohip import source_fields as fields
from integrations.ohip import arr_xml
from integrations.ohip import validate_arr_xml
from integrations.ohip.arr_xml_contract import SourceDocument
from tests.test_ohip_arr_collection import DAY, HOTEL, SECRET, envelope
from tests.test_ohip_arr_xml import row as synthetic_row
from tests.test_arr_opera_daily_ingest import core
from tests.test_ohip_day_capture import DayService
def detail():
return {'hotelId': HOTEL, 'reservationIdList': [
{'type': 'External', 'id': 'external'}, {'type': 'Confirmation', 'id': ' 000123 '},
{'type': 'Reservation', 'id': 'id0'}], 'roomStay': {
'arrivalDate': DAY, 'departureDate': '2026-09-16',
'roomRates': [
{'start': '2026-09-16', 'end': '2026-09-16', 'ratePlanCode': 'WRONG_DAY',
'rates': {'rate': [{'base': {'currencyCode': 'USD', 'amountBeforeTax': 999}}]}},
{'start': DAY, 'end': DAY, 'ratePlanCode': ' unknown_code ',
'rates': {'rate': [{'base': {'currencyCode': 'USD', 'amountBeforeTax': 888}}]}}]}}
def rate_parts():
return ({'operation_id': 'searchRateInfo', 'method': 'POST',
'path': '/api/v1/reservations/rate-info/searches', 'body': {
'detailDate': DAY, 'summaryInfo': False, 'id': 'id0', 'type': 'Reservation'}},
envelope('searchRateInfo', {'detail': {'totalRateAmount': Decimal('123.4500'),
'revenue': {'currencyCode': 'USD'}}}))
def count_detail():
record = detail()
record['roomStay']['guestCounts'] = {'adults': 2, 'children': 3, 'childAges': [8],
'childBuckets': {'ageBucket1': 99}}
record['roomStay']['roomRates'][1].update(guestCounts={'adults': 2, 'children': 3}, numberOfUnits=2)
record['roomStay']['roomRates'][0].update(guestCounts={'adults': 99, 'children': 99}, numberOfUnits=99)
return record
def note(text, **fields):
return {'comment': {'type': 'GEN', 'notificationLocation': 'RESERVATION',
'text': {'value': text}, **fields}}
class NoteCountSelectorTests(unittest.TestCase):
def test_gen_selection_includes_both_internal_states_and_independent_confidential_flag(self):
record = {'comments': [note('internal', internal=True, confidential=False),
note('external', internal=False, confidential=True), note('flags absent'),
note('other type', type='OTH'), note('other location', notificationLocation='PROFILE'),
note('reversed example', type='RESERVATION', notificationLocation='GEN'),
note('case changed', type='gen'), note('padded', type=' GEN')]}
original = copy.deepcopy(record)
self.assertEqual(fields.reservation_gen_notes(record), ('internal', 'external', 'flags absent'))
self.assertEqual(record, original)
def test_note_blanks_duplicates_unicode_cr_and_array_order_are_preserved(self):
values = ('', ' \t ', ' Second\r\n中ไทย & < > ', 'First', 'First')
record = {'comments': [note(value, internal=True) for value in values]}
for index, item in enumerate(record['comments']):
item['createDateTime'] = f'2026-09-15T{23-index:02d}:00:00Z'
item['commentId'] = {'id': str(50-index)}
self.assertEqual(fields.reservation_gen_notes(record), values)
record['comments'].reverse()
self.assertEqual(fields.reservation_gen_notes(record), values[::-1])
def test_explicit_empty_or_only_other_categories_differs_from_missing_comments(self):
self.assertEqual(fields.reservation_gen_notes({'comments': []}), ())
self.assertEqual(fields.reservation_gen_notes({'comments': [note(None, type='OTHER')]}), ())
for record in ({}, {'comments': None}, {'comments': {}}, {'comments': ''}, None,
{'comments': [None]}, {'comments': [{}]}, {'comments': [{'comment': []}]}):
with self.subTest(record=record), self.assertRaises(capture.CollectionError):
fields.reservation_gen_notes(record)
def test_unknown_category_shape_cannot_silently_mean_unselected(self):
for key in ('type', 'notificationLocation'):
for value in (None, '', ' ', 0, False, [], {}):
record = {'comments': [note('PRIVATE')]}
record['comments'][0]['comment'][key] = value
with self.subTest(key=key, value=value), self.assertRaises(capture.CollectionError):
fields.reservation_gen_notes(record)
record = {'comments': [note('PRIVATE')]}
del record['comments'][0]['comment'][key]
with self.assertRaises(capture.CollectionError):
fields.reservation_gen_notes(record)
def test_selected_text_and_present_flags_are_strict_without_private_error_text(self):
malformed = [note(None), note(123), note(False), note([]), note({})]
for value in (None, '', [], {'value': None}, {}):
item = note('PRIVATE')
item['comment']['text'] = value
malformed.append(item)
item = note('PRIVATE')
del item['comment']['text']
malformed.append(item)
for key in ('internal', 'confidential'):
for value in (None, 0, 1, 'true', 'false', []):
malformed.append(note('PRIVATE', **{key: value}))
for item in malformed:
with self.subTest(item=item), self.assertRaises(capture.CollectionError) as error:
fields.reservation_gen_notes({'comments': [item]})
self.assertNotIn('PRIVATE', str(error.exception))
def test_agreed_counts_keep_day_only_integers_and_do_not_derive_child_total(self):
record = count_detail()
before = copy.deepcopy(record)
self.assertEqual(fields.agreed_guest_counts(record, DAY), fields.GuestCounts(2, 3))
self.assertEqual(fields.arrival_number_of_units(record, DAY), 2)
self.assertEqual(record, before)
for counts in (record['roomStay']['guestCounts'], record['roomStay']['roomRates'][1]['guestCounts']):
counts.update(adults=0, children=0)
record['roomStay']['roomRates'][1]['numberOfUnits'] = 0
self.assertEqual(fields.agreed_guest_counts(record, DAY), fields.GuestCounts(0, 0))
self.assertEqual(fields.arrival_number_of_units(record, DAY), 0)
def test_missing_or_different_counts_do_not_fallback_to_the_other_level(self):
for level in ('stay', 'day'):
for change in ('absent', 'null', 'bad_container', 'missing_adults', 'missing_children', 'different'):
record = count_detail()
target = record['roomStay'] if level == 'stay' else record['roomStay']['roomRates'][1]
if change == 'absent':
target.pop('guestCounts')
elif change == 'null':
target['guestCounts'] = None
elif change == 'bad_container':
target['guestCounts'] = []
elif change.startswith('missing_'):
target['guestCounts'].pop(change.removeprefix('missing_'))
else:
target['guestCounts']['adults'] = 1
with self.subTest(level=level, change=change), self.assertRaises(capture.CollectionError):
fields.agreed_guest_counts(record, DAY)
def test_counts_reject_boolean_fraction_string_negative_and_missing_units(self):
for value in (False, True, 1.0, Decimal('1'), '1', -1, None, [], {}):
for key in ('adults', 'children'):
record = count_detail()
record['roomStay']['guestCounts'][key] = value
record['roomStay']['roomRates'][1]['guestCounts'][key] = value
with self.subTest(key=key, value=value), self.assertRaises(capture.CollectionError):
fields.agreed_guest_counts(record, DAY)
record['roomStay']['roomRates'][1]['numberOfUnits'] = value
with self.subTest(units=value), self.assertRaises(capture.CollectionError):
fields.arrival_number_of_units(record, DAY)
record = count_detail()
record['roomStay']['roomRates'][1].pop('numberOfUnits')
with self.assertRaises(capture.CollectionError):
fields.arrival_number_of_units(record, DAY)
def test_no_shared_room_business_decision_or_duplicate_segment_guess(self):
for shared in ('Primary', 'NonPrimary', None):
record = count_detail()
record['roomStay']['primaryShareType'] = shared
record['sharedGuests'] = [{'reservationId': {'id': 'OTHER', 'type': 'Reservation'}}]
self.assertEqual(fields.arrival_number_of_units(record, DAY), 2)
self.assertEqual(fields.agreed_guest_counts(record, DAY), fields.GuestCounts(2, 3))
record['roomStay']['roomRates'].append(copy.deepcopy(record['roomStay']['roomRates'][1]))
for selector in (fields.arrival_number_of_units, fields.agreed_guest_counts):
with self.subTest(shared=shared), self.assertRaises(capture.CollectionError):
selector(record, DAY)
def test_synthetic_notes_survive_xml_and_retain_processor_first_nonempty_behavior(self):
# This seam test deliberately demonstrates why API order is not report acceptance.
for values, first in [(('', ' \t ', ' Group-B\r\nLine ', 'Group-A'), 'Group-B\r\nLine'),
(('Group-A', ' Group-B\r\nLine '), 'Group-A'), (('', ' '), '')]:
chosen = fields.reservation_gen_notes({'comments': [note(value) for value in values]})
synthetic = synthetic_row(notes=chosen, traces=fields.traces(), arrival=DAY, departure=DAY)
document = SourceDocument(HOTEL, date.fromisoformat(DAY), (synthetic,))
payload = arr_xml.serialize(document)
verification = validate_arr_xml.verify(document, payload)
self.assertFalse(verification['report_equivalence_verified'])
root = ET.fromstring(payload)
node = root.find('./LIST_G_GROUP_BY1/G_GROUP_BY1/LIST_G_RESERVATION/G_RESERVATION')
self.assertEqual([item.text or '' for item in node.findall(
'./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID/RES_COMMENT')], list(values))
parsed = core.source_record(node, 1)
self.assertEqual(parsed['RES_COMMENT'], first)
self.assertEqual(parsed['_GROUP_CODE_KEY'], first.upper() or None)
self.assertEqual(parsed['TRACE_TEXT'], '')
class SourceFieldTests(unittest.TestCase):
def test_handwritten_selection_preserves_code_text_and_identity(self):
record = detail()
self.assertEqual(fields.confirmation_no(record), ' 000123 ')
self.assertEqual(fields.stay_dates(record, DAY), fields.StayDates(date(2026, 9, 15), date(2026, 9, 16)))
self.assertEqual(fields.single_day_rate_index(record, DAY), 1)
self.assertEqual(fields.arrival_rate_code(record, DAY), ' unknown_code ')
self.assertEqual(fields.traces(), ())
def test_confirmation_never_uses_internal_external_or_duplicate_id(self):
for ids in [[], [{'type': 'Reservation', 'id': 'id0'}],
[{'type': 'Reservation', 'id': 'id0'}, {'type': 'External', 'id': 'display'}],
[{'type': 'Reservation', 'id': 'id0'}, {'type': 'Confirmation', 'id': 'same'},
{'type': 'Confirmation', 'id': 'same'}],
[{'type': 'Reservation', 'id': 'id0'}, None],
[{'type': 'Reservation', 'id': 'id0'}, {'id': 'untyped'}]]:
with self.subTest(ids=ids), self.assertRaises(capture.CollectionError):
fields.confirmation_no({**detail(), 'reservationIdList': ids})
def test_explicit_blank_text_remains_distinct_from_missing_and_bad_type(self):
record = detail()
for text in ('', ' ', '\t'):
record['reservationIdList'][1]['id'] = text
record['roomStay']['roomRates'][1]['ratePlanCode'] = text
self.assertEqual(fields.confirmation_no(record), text)
self.assertEqual(fields.arrival_rate_code(record, DAY), text)
for value in (None, 123, False, [], {}):
record['reservationIdList'][1]['id'] = value
record['roomStay']['roomRates'][1]['ratePlanCode'] = value
with self.subTest(value=value):
with self.assertRaises(capture.CollectionError):
fields.confirmation_no(record)
with self.assertRaises(capture.CollectionError):
fields.arrival_rate_code(record, DAY)
del record['reservationIdList'][1]['id']
del record['roomStay']['roomRates'][1]['ratePlanCode']
with self.assertRaises(capture.CollectionError):
fields.confirmation_no(record)
with self.assertRaises(capture.CollectionError):
fields.arrival_rate_code(record, DAY)
def test_dates_support_zero_night_full_year_and_do_not_use_clock_or_times(self):
for value in ('0001-01-01', '1965-01-01', '2065-01-01', '9999-12-31'):
record = detail()
record['roomStay'].update(arrivalDate=value, departureDate=value,
expectedTimes={'reservationExpectedArrivalTime': 'invalid'},
originalTimeSpan={'start': '2000-01-01'}, actualDepartureDate='2000-01-01')
actual = fields.stay_dates(record, value)
self.assertEqual(actual.arrival, actual.departure)
self.assertEqual(actual.arrival.isoformat(), value)
def test_bad_dates_mismatches_and_negative_stay_are_explicit_errors(self):
for key in ('arrivalDate', 'departureDate'):
for value in (None, '', '20260915', '26-09-15', '2026-02-30', DAY+'T00:00:00Z', False):
record = detail()
record['roomStay'][key] = value
with self.subTest(key=key, value=value), self.assertRaises(capture.CollectionError):
fields.stay_dates(record, DAY)
with self.assertRaisesRegex(capture.CollectionError, 'arrival_date_mismatch'):
fields.stay_dates(detail(), '2026-09-14')
record = detail()
record['roomStay']['departureDate'] = '2026-09-14'
with self.assertRaisesRegex(capture.CollectionError, 'negative_stay'):
fields.stay_dates(record, DAY)
def test_every_segment_checked_no_range_guess_or_first_match_fallback(self):
for mode in ('range', 'other_bad_date', 'missing', 'duplicate', 'invalid_item', 'empty', 'no_day'):
record = detail()
rates = record['roomStay']['roomRates']
if mode == 'range':
rates[0]['start'] = '2026-09-14'
elif mode == 'other_bad_date':
rates[0]['end'] = 'nonsense'
elif mode == 'missing':
del rates[0]['start']
elif mode == 'duplicate':
rates.append(copy.deepcopy(rates[1]))
elif mode == 'invalid_item':
rates.append(None)
elif mode == 'empty':
rates.clear()
else:
rates.pop()
with self.subTest(mode=mode), self.assertRaises(capture.CollectionError):
fields.arrival_rate_code(record, DAY)
def test_exact_raw_day_rate_zero_and_large_decimal_preserved(self):
request, response = rate_parts()
options = day.Options(DAY, DAY, DAY, HOTEL)
for amount in (0, Decimal('123.4500'), Decimal('900719925474099312345678901234567890.0123456789')):
response['data']['detail']['totalRateAmount'] = amount
with localcontext() as ctx:
ctx.prec = 6
actual = fields.effective_rate(detail(), request, response, options)
self.assertEqual(str(actual.amount), str(amount))
self.assertEqual(actual.currency, 'USD')
def test_rate_request_cannot_rebind_identity_day_operation_or_quote(self):
options = day.Options(DAY, DAY, DAY, HOTEL)
mutations = [lambda q: q.update(operation_id='getRateInfo'),
lambda q: q.update(method='GET'), lambda q: q.update(path='/other'),
lambda q: q['body'].update(id='id1'),
lambda q: q['body'].update(type='Confirmation'),
lambda q: q['body'].update(detailDate='2026-09-14'),
lambda q: q['body'].update(summaryInfo=0),
lambda q: q['body'].update(ratePlanCode='requote')]
for change in mutations:
request, response = rate_parts()
change(request)
with self.subTest(change=change), self.assertRaisesRegex(capture.CollectionError, 'rate_request_mismatch'):
fields.effective_rate(detail(), request, response, options)
request, response = rate_parts()
with self.assertRaises(capture.CollectionError):
fields.effective_rate(detail(), request, response, replace(options, rate_date='2026-09-14'))
def test_rate_response_envelope_warnings_and_currency_checked_again(self):
changes = [lambda r: r.update(hotel_id='OTHER'), lambda r: r.update(operation_id='getRateInfo'),
lambda r: r.pop('oracle_request_id'), lambda r: r.update(warnings=['warning']),
lambda r: r['data']['detail']['revenue'].update(currencyCode='EUR'),
lambda r: r.update(data={}), lambda r: r.update(data={'summary': {}})]
for change in changes:
request, response = rate_parts()
change(response)
with self.subTest(change=change), self.assertRaises(capture.CollectionError):
fields.effective_rate(detail(), request, response, day.Options(DAY, DAY, DAY, HOTEL))
request, response = rate_parts()
record = detail()
record['roomStay']['roomRates'][0]['rates']['rate'][0]['base'].pop('currencyCode')
with self.assertRaisesRegex(capture.CollectionError, 'missing_detail_rate_currency'):
fields.effective_rate(record, request, response, day.Options(DAY, DAY, DAY, HOTEL))
def test_hidden_and_missing_rate_never_falls_back_to_base_or_zero(self):
for amount in (None, '0', False, -1, 1.2, Decimal('NaN'), Decimal('Infinity')):
request, response = rate_parts()
response['data']['detail']['totalRateAmount'] = amount
with self.subTest(amount=amount), self.assertRaises(capture.CollectionError):
fields.effective_rate(detail(), request, response, day.Options(DAY, DAY, DAY, HOTEL))
request, response = rate_parts()
response['data']['detail']['rateSuppressed'] = True
with self.assertRaisesRegex(capture.CollectionError, 'rate_suppressed'):
fields.effective_rate(detail(), request, response, day.Options(DAY, DAY, DAY, HOTEL))
def test_inputs_unchanged_no_whitelist_status_or_business_field_gate(self):
for status in ('Cancelled', 'NoShow', 'Reserved', 'InHouse', 'CheckedOut'):
record = detail()
record.update(reservationStatus=status, traces=[{'traceText': 'PRIVATE'}])
record['roomStay']['roomRates'][1].update(numberOfUnits=0, roomId=None, stayProfiles=None)
request, response = rate_parts()
before = copy.deepcopy((record, request, response))
fields.confirmation_no(record)
fields.stay_dates(record, DAY)
fields.arrival_rate_code(record, DAY)
fields.effective_rate(record, request, response, day.Options(DAY, DAY, DAY, HOTEL))
self.assertEqual((record, request, response), before)
self.assertEqual(fields.traces(), ())
def test_pinned_capture_replay_to_raw_field_selection_offline(self):
with tempfile.TemporaryDirectory() as temp:
directory = Path(temp) / 'capture'
service = DayService()
for row in service.base.rows:
row['comments'] = [note(''), note(' \r\nSYNTHETIC NOTE ', internal=True)]
row['roomStay'].update(departureDate=DAY, guestCounts={'adults': 2, 'children': 0})
row['roomStay']['roomRates'][0].update(start=DAY, end=DAY, ratePlanCode='UNKNOWN',
guestCounts={'adults': 2, 'children': 0}, numberOfUnits=1)
options = day.Options(DAY, DAY, DAY, HOTEL, page_size=2)
sink = capture.Archive(directory)
amount = b'900719925474099312345678901234567890.0123456789'
def transport(method, path, body):
status, headers, raw = service(method, path, body)
if path.endswith('/rate-info/searches'):
raw = raw.replace(b'"totalRateAmount": 0', b'"totalRateAmount": '+amount)
return status, headers, raw
result = day.collect(options, sink,
capture.Reader(sink, HOTEL, transport, key=SECRET, sleep=lambda _: None),
rate_info.RateInfoReader(sink, HOTEL, transport, key=SECRET, sleep=lambda _: None))
archive = audit.VerifiedArchive(directory, result['manifest_sha256'])
with patch.object(capture, 'HTTPTransport', side_effect=AssertionError('no network')), \
patch.object(capture, 'load_key', side_effect=AssertionError('no credentials')):
_, records, assessments = audit.replay(archive)
for record, assessment in zip(records, assessments['records']):
request_file = assessment['rate_request']
request = archive.document(request_file)
raw = archive.read(request_file[:-5] + '.response.bin')
capture.strict_json(raw)
response = json.loads(raw, parse_float=Decimal)
chosen = fields.effective_rate(record, request, response, options)
self.assertEqual(str(chosen.amount), assessment['effective_rate'])
self.assertEqual(fields.arrival_rate_code(record, DAY), 'UNKNOWN')
self.assertEqual(fields.stay_dates(record, DAY).arrival, date(2026, 9, 15))
self.assertEqual(fields.confirmation_no(record), 'confirmation-'+assessment['reservation_id'])
self.assertEqual(fields.agreed_guest_counts(record, DAY), fields.GuestCounts(2, 0))
self.assertEqual(fields.arrival_number_of_units(record, DAY), 1)
self.assertEqual(fields.reservation_gen_notes(record), ('', ' \r\nSYNTHETIC NOTE '))
self.assertEqual(fields.traces(), ())
self.assertEqual(assessments['records'][0]['effective_rate'], amount.decode())
if __name__ == '__main__':
unittest.main()