import copy from decimal import Decimal import hashlib 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 source from integrations.ohip import compare_arr_sources as compare from integrations.ohip.rate_info import RateInfoReader from tests.test_ohip_day_capture import DayService, DAY, HOTEL def service(count=1): result = DayService(count) for i, row in enumerate(result.base.rows): stay = row["roomStay"] stay.update(departureDate="2026-09-17", guestCounts={"adults": 2, "children": 0}, currentRoomInfo={"roomId": "0101"}) stay["roomRates"][0].update(start=DAY, end=DAY, roomId="0101", ratePlanCode="SYNTHETIC", numberOfUnits=1, roomType="TEST", stayProfiles=[]) row["reservationGuest"] = {"fullName": "PRIVATE_SURNAME, PRIVATE_GIVEN"} row["reservationGuests"] = [{"primary": True, "profileInfo": {"profile": {"customer": {"personName": [ {"nameType": "Primary", "surname": "PRIVATE_SURNAME", "givenName": "PRIVATE_GIVEN"}]}}}}] row["reservationProfiles"] = {"reservationProfile": [{"reservationProfileType": "Company", "profile": {"company": {"companyName": "PRIVATE_COMPANY"}}}]} row["comments"] = [{"comment": {"type": "GEN", "notificationLocation": "RESERVATION", "internal": True, "text": {"value": "PRIVATE_FIRST_NOTE"}}}] row["traces"] = [{"traceText": "PRIVATE_TRACE"}] return result def report(count=1, hotel=HOTEL, day=DAY): root = ET.Element("RES_DETAIL") group = ET.SubElement(ET.SubElement(root, "LIST_G_GROUP_BY1"), "G_GROUP_BY1") ET.SubElement(group, "GROUPBY1_SORT_COL").text = day.replace("-", "") yyyy, mm, dd = day.split("-") ET.SubElement(group, "GROUPBY1_COL").text = f"{dd}-{mm}-{yyyy[2:]}" records = ET.SubElement(group, "LIST_G_RESERVATION") for i in range(count): row = ET.SubElement(records, "G_RESERVATION") values = {"RESORT": hotel, "RESV_NAME_ID": f"id{i}", "ROWNUM": str(99-i), "CONFIRMATION_NO": f"confirmation-id{i}", "TRUNC_BEGIN": "15-SEP-26", "TRUNC_END": "17-SEP-26", "ADULTS": "2", "CHILDREN": "0", "CF_CHILDREN": "0", "NO_OF_ROOMS": "1", "DISP_ROOM_NO": "0101", "RATE_CODE": "SYNTHETIC", "ROOM_CATEGORY_LABEL": "TEST", "EFFECTIVE_RATE_AMOUNT": str(i), "CURRENCY_CODE": "USD", "SHORT_RESV_STATUS": "CKIN", "FULL_NAME": "PRIVATE_SURNAME, PRIVATE_GIVEN", "COMPANY_NAME": "PRIVATE_COMPANY"} for field, text in values.items(): ET.SubElement(row, field).text = text notes = ET.SubElement(row, "LIST_G_COMMENT_RESV_NAME_ID") note = ET.SubElement(notes, "G_COMMENT_RESV_NAME_ID") ET.SubElement(note, "RES_COMMENT").text = "PRIVATE_FIRST_NOTE" traces = ET.SubElement(row, "LIST_G_DEPT_ID") ET.SubElement(ET.SubElement(traces, "G_DEPT_ID"), "TRACE_TEXT").text = "PRIVATE_TRACE" return root class SourceComparisonTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.capture_dir = self.root / "capture" self.report_path = self.root / "source.xml" self.service = service() self.xml = report() def capture(self, transport=None): archive = source.Archive(self.capture_dir) transport = transport or self.service result = day.collect(day.Options(DAY, DAY, DAY, HOTEL, page_size=2), archive, source.Reader(archive, HOTEL, transport, sleep=lambda _: None), RateInfoReader(archive, HOTEL, transport, sleep=lambda _: None)) self.assertTrue(result["candidate_capture_complete"]) return result["manifest_sha256"] def run_compare(self, capture_pin=None, report_pin=None): raw = self.xml if isinstance(self.xml, bytes) else ET.tostring(self.xml, encoding="utf-8", xml_declaration=True) self.report_path.write_bytes(raw) pin = capture_pin or self.capture() with patch.object(source, "HTTPTransport", side_effect=AssertionError("offline only")), \ patch.object(source, "load_key", side_effect=AssertionError("offline only")): return compare.compare_files(self.capture_dir, pin, self.report_path, report_pin or hashlib.sha256(raw).hexdigest()) def row(self): return self.xml.find(".//G_RESERVATION") def assert_observation(self, output, name, category, count=1): self.assertEqual(output["status"], "compared_observations") self.assertEqual(output["observations"][name][category], count) def test_join_compare_exact_fields_zero_and_privacy_without_acceptance_claim(self): result, code = self.run_compare() self.assertEqual(code, 0) self.assertEqual(result["matched_unique_records"], 1) self.assertTrue(result["identity_sequence_equal"]) for field in ("CONFIRMATION_NO", "ARRIVAL", "DEPARTURE", "ADULTS_STAY", "CHILDREN_STAY", "EFFECTIVE_RATE_AMOUNT", "CURRENCY_CODE", "ROOM_CURRENT", "ROOM_SINGLE_DAY_SEGMENT", "RATE_CODE_SINGLE_DAY_SEGMENT", "ROOM_COUNT_SINGLE_DAY_SEGMENT", "FULL_NAME_SEARCH", "COMPANY_RESERVATION_COMPANY", "GEN_RESERVATION_ORDERED_TEXTS", "GEN_RESERVATION_FIRST_NONEMPTY", "ALL_TRACES_ORDERED_TEXTS"): self.assert_observation(result, field, "equal_nonblank") for flag in ("finance_ready", "report_equivalence_verified", "source_mapping_verified"): self.assertFalse(result[flag]) self.assertEqual(result["network_calls"], 0) text = json.dumps(result) for private in ("PRIVATE_SURNAME", "PRIVATE_GIVEN", "PRIVATE_COMPANY", "PRIVATE_FIRST_NOTE", "PRIVATE_TRACE", "confirmation-id0", '"id0"', '"0101"'): self.assertNotIn(private, text) def test_hotel_or_group_date_mismatch_never_compares_rows(self): pin = self.capture() for kwargs in ({"hotel": "OTHER_HOTEL"}, {"day": "2026-09-14"}): self.xml = report(**kwargs) result, code = self.run_compare(capture_pin=pin) self.assertEqual((code, result["status"], result["reason"]), (2, "not_comparable", "hotel_or_date_differs")) self.assertNotIn("observations", result) def test_report_row_arrival_cannot_hide_behind_group_date(self): self.row().find("TRUNC_BEGIN").text = "14-SEP-26" result, code = self.run_compare() self.assertEqual(code, 2) self.assertEqual(result["reason"], "report_row_date_missing_or_different") def test_api_only_and_report_only_rows_not_paired_by_confirmation_room_or_name(self): self.service = service(2) self.xml = report(2) self.xml.findall('.//G_RESERVATION')[1].find('RESV_NAME_ID').text = 'foreign-id' result, code = self.run_compare() self.assertEqual((result["matched_unique_records"], result["api_only_records"], result["report_only_rows"]), (1, 1, 1)) self.assertEqual(result["api_only_statuses"], {"InHouse": 1}) self.assertFalse(result["identity_sequence_equal"]) def test_duplicate_native_ids_are_ambiguous_not_reused_for_multiple_rows(self): records = self.xml.find('.//LIST_G_RESERVATION') records.append(copy.deepcopy(self.row())) result, code = self.run_compare() self.assertEqual(code, 0) self.assertEqual(result["ambiguous_report_ids"], 1) self.assertEqual(result["ambiguous_report_rows"], 2) self.assertEqual(result["matched_unique_records"], 0) self.assertEqual(result["observations"], {}) self.assertIsNone(result["matched_identity_order_equal"]) def test_physical_order_checked_independently_of_rownum(self): self.service = service(2) self.xml = report(2) records = self.xml.find('.//LIST_G_RESERVATION') records[:] = list(reversed(records)) result, code = self.run_compare() self.assertEqual(result["matched_unique_records"], 2) self.assertFalse(result["identity_sequence_equal"]) self.assertFalse(result["matched_identity_order_equal"]) self.assert_observation(result, "CONFIRMATION_NO", "equal_nonblank", 2) def test_children_uses_cf_children_not_similarly_named_raw_field(self): self.row().find('CHILDREN').text = '99' result, _ = self.run_compare() self.assert_observation(result, "CHILDREN_STAY", "equal_nonblank") def test_room_leading_zero_is_significant(self): self.row().find('DISP_ROOM_NO').text = '101' result, _ = self.run_compare() self.assert_observation(result, "ROOM_CURRENT", "different") self.assert_observation(result, "ROOM_SINGLE_DAY_SEGMENT", "different") def test_duplicate_confirmation_is_unavailable_even_if_first_matches(self): self.service.base.rows[0]['reservationIdList'].append({'type': 'Confirmation', 'id': 'other'}) result, _ = self.run_compare() self.assert_observation(result, "CONFIRMATION_NO", "unavailable") def test_hidden_or_empty_price_never_compared_as_zero(self): for index, data in enumerate(({}, {"detail": {"rateSuppressed": True}})): self.capture_dir = self.root / str(index) self.service.rate_edit = lambda _, body, data=data: data result, _ = self.run_compare() self.assert_observation(result, "EFFECTIVE_RATE_AMOUNT", "unavailable") self.assert_observation(result, "EFFECTIVE_RATE_AMOUNT", "equal_nonblank", 0) def test_decimal_money_is_compared_without_binary_float_rounding(self): self.row().find('EFFECTIVE_RATE_AMOUNT').text = '9007199254740993.01' def transport(method, path, body): status, headers, raw = self.service(method, path, body) if path.endswith('rate-info/searches'): raw = raw.replace(b'"totalRateAmount": 0', b'"totalRateAmount": 9007199254740993.01') return status, headers, raw result, _ = self.run_compare(capture_pin=self.capture(transport)) self.assert_observation(result, "EFFECTIVE_RATE_AMOUNT", "equal_nonblank") def test_company_roles_are_separate_and_no_fallback_priority_is_adopted(self): profiles = self.service.base.rows[0]['reservationProfiles']['reservationProfile'] profiles.insert(0, {'reservationProfileType': 'Group', 'profile': {'company': {'companyName': 'PRIVATE_GROUP'}}}) result, _ = self.run_compare() self.assert_observation(result, "COMPANY_RESERVATION_COMPANY", "equal_nonblank") self.assert_observation(result, "COMPANY_RESERVATION_GROUP", "different") self.assertFalse(result['source_mapping_verified']) def test_duplicate_company_role_does_not_choose_first(self): profiles = self.service.base.rows[0]['reservationProfiles']['reservationProfile'] profiles.append(copy.deepcopy(profiles[0])) result, _ = self.run_compare() self.assert_observation(result, "COMPANY_RESERVATION_COMPANY", "unavailable") def test_display_name_missing_is_not_replaced_by_name_format_hypothesis(self): self.service.base.rows[0].pop('reservationGuest') result, _ = self.run_compare() self.assert_observation(result, "FULL_NAME_SEARCH", "unavailable") self.assert_observation(result, "FULL_NAME_SURNAME_COMMA_GIVEN", "equal_nonblank") self.assert_observation(result, "FULL_NAME_GIVEN_SURNAME", "different") self.assertFalse(result['source_mapping_verified']) def test_multiple_primary_guests_are_not_arbitrarily_chosen(self): guests = self.service.base.rows[0]['reservationGuests'] guests.append(copy.deepcopy(guests[0])) result, _ = self.run_compare() self.assert_observation(result, "FULL_NAME_SURNAME_COMMA_GIVEN", "unavailable") def test_note_scope_internal_and_first_nonempty_and_order_checked_separately(self): row = self.service.base.rows[0] row['comments'].insert(0, {'comment': {'type':'OTHER','notificationLocation':'RESERVATION', 'text':{'value':'PRIVATE_OTHER'}}}) row['comments'].insert(1, {'comment': {'type':'GEN','notificationLocation':'RESERVATION', 'text':{'value':' '}}}) row['comments'].append({'comment': {'type':'GEN','notificationLocation':'RESERVATION', 'text':{'value':'PRIVATE_SECOND'}}}) result, _ = self.run_compare() self.assert_observation(result, "GEN_RESERVATION_FIRST_NONEMPTY", "equal_nonblank") self.assert_observation(result, "GEN_RESERVATION_ORDERED_TEXTS", "different") self.assert_observation(result, "ALL_RESERVATION_NOTES_FIRST_NONEMPTY", "different") def test_reversed_trace_order_is_not_hidden_by_set_comparison(self): self.service.base.rows[0]['traces'].append({'traceText':'PRIVATE_SECOND_TRACE'}) traces = self.row().find('LIST_G_DEPT_ID') first = ET.Element('G_DEPT_ID') ET.SubElement(first,'TRACE_TEXT').text = 'PRIVATE_SECOND_TRACE' traces.insert(0, first) result, _ = self.run_compare() self.assert_observation(result, "ALL_TRACES_ORDERED_TEXTS", "different") self.assert_observation(result, "ALL_TRACES_FIRST_NONEMPTY", "different") def test_general_intervals_and_ambiguous_same_day_segments_remain_unavailable(self): for index, mutate in enumerate((lambda segments: segments[0].update(end='2026-09-17'), lambda segments: segments.append(copy.deepcopy(segments[0])))): self.capture_dir = self.root / str(index) self.service = service() mutate(self.service.base.rows[0]['roomStay']['roomRates']) result, _ = self.run_compare() self.assert_observation(result, "RATE_CODE_SINGLE_DAY_SEGMENT", "unavailable") def test_missing_optional_containers_are_unknown_not_proven_empty(self): row = self.service.base.rows[0] row.pop('comments') row.pop('traces') row.pop('reservationProfiles') row['roomStay']['guestCounts'] = None row['reservationGuests'][0]['profileInfo'] = None result, code = self.run_compare() self.assertEqual(code, 0) for name in ('GEN_RESERVATION_FIRST_NONEMPTY','ALL_TRACES_FIRST_NONEMPTY','COMPANY_RESERVATION_COMPANY', 'ADULTS_STAY','FULL_NAME_SURNAME_GIVEN'): self.assert_observation(result,name,'unavailable') def test_missing_xml_list_is_empty_but_duplicate_scalar_is_ambiguous(self): self.row().remove(self.row().find('LIST_G_DEPT_ID')) ET.SubElement(self.row(), 'ADULTS').text='2' result, _ = self.run_compare() self.assert_observation(result,'ALL_TRACES_FIRST_NONEMPTY','different') self.assert_observation(result,'ADULTS_STAY','unavailable') def test_missing_xml_optional_lists_equal_explicit_empty_api_arrays(self): for name in ('LIST_G_DEPT_ID', 'LIST_G_COMMENT_RESV_NAME_ID'): self.row().remove(self.row().find(name)) self.service.base.rows[0].update(traces=[], comments=[]) result, _ = self.run_compare() for name in ('ALL_TRACES', 'GEN_RESERVATION', 'TRACES_WITHOUT_RESOLUTION_MARKERS'): self.assert_observation(result, name + '_FIRST_NONEMPTY', 'both_blank') self.assert_observation(result, name + '_ORDERED_TEXTS', 'both_blank') def test_duplicate_xml_optional_list_is_ambiguous(self): ET.SubElement(self.row(), 'LIST_G_DEPT_ID') result, _ = self.run_compare() self.assert_observation(result, 'ALL_TRACES_FIRST_NONEMPTY', 'unavailable') def test_xml_processor_numeric_forms_do_not_relax_api_number_types(self): for text, value in (('1,000', '1000'), ('.5', '0.5'), ('1.', '1'), ('1e3', '1000')): with self.subTest(text=text): self.assertEqual(compare.xml_number(text).value, Decimal(value)) self.assertIsNotNone(compare.number(text, api_integer=True).issue) for text in ('1,000', '.5', '1.'): self.assertIsNotNone(compare.number(text).issue) for text in ('NaN', 'Infinity', '', '-1', 'two'): self.assertIsNotNone(compare.xml_number(text).issue) self.assertIsNotNone(compare.xml_number('.5', integer=True).issue) def test_xml_processor_date_forms_do_not_relax_api_iso_dates(self): pin = self.capture() for text in ('15-SEP-26', '15-09-26', '20260915', '2026-09-15', '09/15/2026'): with self.subTest(text=text): self.row().find('TRUNC_BEGIN').text = text result, code = self.run_compare(capture_pin=pin) self.assertEqual(code, 0) self.assert_observation(result, 'ARRIVAL', 'equal_nonblank') if text != DAY: self.assertIsNotNone(compare.api_date(text).issue) self.assertIsNotNone(compare.report_date('31-02-26').issue) def test_zero_rooms_equality_never_implies_processor_validity(self): self.row().find('NO_OF_ROOMS').text = '0' self.service.base.rows[0]['roomStay']['roomRates'][0]['numberOfUnits'] = 0 result, _ = self.run_compare() self.assert_observation(result, 'ROOM_COUNT_SINGLE_DAY_SEGMENT', 'equal_nonblank') self.assertFalse(result['finance_ready']) self.assertFalse(result['source_mapping_verified']) def test_prefixed_company_hypotheses_remain_separate_for_each_role_and_level(self): for index, (role, prefix) in enumerate(compare.ROLE_PREFIX_HYPOTHESES.items()): self.capture_dir = self.root / str(index) self.service = service() profile = self.service.base.rows[0]['reservationProfiles']['reservationProfile'][0] profile['reservationProfileType'] = role self.service.base.rows[0]['roomStay']['roomRates'][0]['stayProfiles'] = [copy.deepcopy(profile)] self.row().find('COMPANY_NAME').text = prefix + 'PRIVATE_COMPANY' result, _ = self.run_compare() for level in ('RESERVATION', 'SINGLE_DAY'): name = 'COMPANY_' + level + '_' + role.upper() self.assert_observation(result, name, 'different') self.assert_observation(result, name + '_PREFIXED', 'equal_nonblank') self.assertFalse(result['source_mapping_verified']) self.assertFalse(result['native_observations']['company_role_priority_verified']) def test_prefix_does_not_turn_absent_role_into_nonempty_company(self): self.row().find('COMPANY_NAME').text = '' self.service.base.rows[0]['reservationProfiles']['reservationProfile'] = [] result, _ = self.run_compare() self.assert_observation(result, 'COMPANY_RESERVATION_COMPANY_PREFIXED', 'both_blank') self.assertEqual(result['native_observations']['company_display_buckets'], {'blank': 1}) def test_ambiguous_company_is_not_repaired_by_display_prefix(self): profiles = self.service.base.rows[0]['reservationProfiles']['reservationProfile'] profiles.append(copy.deepcopy(profiles[0])) self.row().find('COMPANY_NAME').text = 'C- PRIVATE_COMPANY' result, _ = self.run_compare() self.assert_observation(result, 'COMPANY_RESERVATION_COMPANY_PREFIXED', 'unavailable') def test_literal_prefix_in_api_name_and_multiline_display_not_silently_normalized(self): profile = self.service.base.rows[0]['reservationProfiles']['reservationProfile'][0] profile['profile']['company']['companyName'] = 'C- PRIVATE_COMPANY' self.row().find('COMPANY_NAME').text = 'C- PRIVATE_COMPANY' result, _ = self.run_compare() self.assert_observation(result, 'COMPANY_RESERVATION_COMPANY', 'equal_nonblank') self.assert_observation(result, 'COMPANY_RESERVATION_COMPANY_PREFIXED', 'different') candidate = compare.prefixed_company(compare.Candidate('PRIVATE_ONE\nPRIVATE_TWO'), 'Company') self.assertEqual(candidate.issue, 'multiline_company_display_unresolved') def test_daily_and_stay_guest_counts_are_compared_without_fallback(self): segment = self.service.base.rows[0]['roomStay']['roomRates'][0] segment['guestCounts'] = {'adults': 1, 'children': 2} result, _ = self.run_compare() for name in ('ADULTS', 'CHILDREN'): self.assert_observation(result, name + '_STAY', 'equal_nonblank') self.assert_observation(result, name + '_SINGLE_DAY_SEGMENT', 'different') self.assertEqual(len(result['observations']), 42) self.capture_dir = self.root / 'missing-counts' segment.pop('guestCounts') result, _ = self.run_compare() self.assert_observation(result, 'ADULTS_SINGLE_DAY_SEGMENT', 'unavailable') def test_cross_hotel_native_summary_is_not_cross_hotel_field_comparison(self): self.xml = report(hotel='OTHER_HOTEL') self.row().find('COMPANY_NAME').text = 'T- PRIVATE_COMPANY' ET.SubElement(self.row(), 'FULL_NAME_NO_SHR_IND').text = 'PRIVATE_SURNAME, PRIVATE_GIVEN' note = self.row().find('.//G_COMMENT_RESV_NAME_ID') ET.SubElement(note, 'RES_COMMENT_TYPE').text = 'CAS' ET.SubElement(note, 'RES_COMMENT_DESCRIPTION').text = 'GENERAL' result, code = self.run_compare() self.assertEqual(code, 2) self.assertNotIn('observations', result) shape = result['native_observations'] self.assertEqual(shape['scope'], 'native_input_only_not_mapping_acceptance') self.assertEqual(shape['company_display_buckets'], {'T-': 1}) self.assertEqual(shape['note_type_buckets'], {'CAS': 1}) self.assertEqual(shape['note_description_buckets'], {'GENERAL': 1}) self.assertEqual(shape['nonblank_full_name_equals_no_share_name'], 1) self.assertEqual(shape['trace_elements'], 1) self.assertIsNone(shape['lexical_casefold_ascending_hypotheses']['DISP_ROOM_NO']) def test_native_summary_never_echoes_arbitrary_note_types_or_company_prefixes(self): self.row().find('COMPANY_NAME').text = 'PRIVATE_PREFIX: PRIVATE_COMPANY' note = self.row().find('.//G_COMMENT_RESV_NAME_ID') ET.SubElement(note, 'RES_COMMENT_TYPE').text = 'PRIVATE_TYPE' ET.SubElement(note, 'RES_COMMENT_DESCRIPTION').text = 'PRIVATE_DESCRIPTION' result, _ = self.run_compare() shape = result['native_observations'] self.assertEqual(shape['company_display_buckets'], {'other_nonblank': 1}) self.assertEqual(shape['note_type_buckets'], {'OTHER_OR_MISSING': 1}) self.assertEqual(shape['note_description_buckets'], {'OTHER_OR_MISSING': 1}) self.assertNotIn('PRIVATE', json.dumps(result)) def test_observed_room_order_is_separate_from_configured_sort_and_rownum(self): self.service, self.xml = service(3), report(3) for row, room in zip(self.xml.findall('.//G_RESERVATION'), ('001', '002', '010')): row.find('DISP_ROOM_NO').text = room pin = self.capture() result, _ = self.run_compare(capture_pin=pin) shape = result['native_observations'] self.assertTrue(shape['lexical_casefold_ascending_hypotheses']['DISP_ROOM_NO']) self.assertEqual(shape['rownum_matches_physical_position'], 0) self.assertFalse(shape['configured_sort_verified']) self.row().find('DISP_ROOM_NO').text = '' result, _ = self.run_compare(capture_pin=pin) self.assertIsNone(result['native_observations']['lexical_casefold_ascending_hypotheses']['DISP_ROOM_NO']) def test_resolution_markers_are_a_separate_trace_hypothesis(self): traces = self.service.base.rows[0]['traces'] traces[0]['resolveInfo'] = {} traces.insert(0, {'traceText': 'PRIVATE_RESOLVED_TRACE', 'resolveInfo': {'resolvedOn': DAY}}) traces.append({'traceText': 'PRIVATE_RESOLVER_TRACE', 'resolveInfo': {'resolvedBy': 'PRIVATE_USER'}}) result, _ = self.run_compare() self.assert_observation(result, 'ALL_TRACES_FIRST_NONEMPTY', 'different') self.assert_observation(result, 'TRACES_WITHOUT_RESOLUTION_MARKERS_FIRST_NONEMPTY', 'equal_nonblank') self.assert_observation(result, 'TRACES_WITHOUT_RESOLUTION_MARKERS_ORDERED_TEXTS', 'equal_nonblank') self.assertFalse(result['source_mapping_verified']) for private in ('PRIVATE_RESOLVED_TRACE', 'PRIVATE_RESOLVER_TRACE', 'PRIVATE_USER'): self.assertNotIn(private, json.dumps(result)) def test_malformed_trace_resolution_does_not_invent_open_status(self): for index, info in enumerate((None, [], {'resolvedOn': False}, {'resolvedBy': {}})): self.capture_dir = self.root / str(index) self.service.base.rows[0]['traces'][0]['resolveInfo'] = info result, _ = self.run_compare() self.assert_observation(result, 'ALL_TRACES_FIRST_NONEMPTY', 'equal_nonblank') self.assert_observation(result, 'TRACES_WITHOUT_RESOLUTION_MARKERS_FIRST_NONEMPTY', 'unavailable') def test_problem_positions_are_bounded_but_counts_are_complete(self): self.service = service(25) self.xml = report(25) for row in self.xml.findall('.//G_RESERVATION'): row.find('FULL_NAME').text = 'PRIVATE_CHANGED_NAME' result, _ = self.run_compare() observation = result['observations']['FULL_NAME_SEARCH'] self.assertEqual(observation['different'], 25) self.assertEqual(observation['problem_report_positions'], list(range(1, 21))) def test_bad_numeric_text_never_defaults_to_zero(self): self.row().find('CF_CHILDREN').text = '' self.row().find('EFFECTIVE_RATE_AMOUNT').text = 'NaN' result, _ = self.run_compare() self.assert_observation(result,'CHILDREN_STAY','unavailable') self.assert_observation(result,'EFFECTIVE_RATE_AMOUNT','unavailable') def test_pin_and_raw_tampering_refused(self): pin=self.capture() for kind in ('report','capture'): result,code=self.run_compare(capture_pin='0'*64 if kind=='capture' else pin, report_pin='0'*64 if kind=='report' else None) self.assertEqual((code,result['status']),(3,'invalid_input')) target=self.capture_dir/'rate-000001.response.bin' target.write_bytes(target.read_bytes().replace(b'USD',b'THB')) result,code=self.run_compare(capture_pin=pin) self.assertEqual((code,result['error']),(3,'archive_hash_mismatch')) def test_malformed_xml_and_entities_are_refused_without_value_leak(self): pin=self.capture() for raw in (b'PRIVATE_BROKEN', b']>'): self.xml=raw result,code=self.run_compare(capture_pin=pin) self.assertEqual((code,result['status']),(3,'invalid_input')) self.assertNotIn('PRIVATE',json.dumps(result)) def test_differences_return_positions_without_prices_names_or_ids(self): self.row().find('FULL_NAME').text='PRIVATE_CHANGED_NAME' self.row().find('EFFECTIVE_RATE_AMOUNT').text='123456789' result,_=self.run_compare() self.assertEqual(result['observations']['FULL_NAME_SEARCH']['problem_report_positions'],[1]) text=json.dumps(result) self.assertNotIn('PRIVATE_CHANGED_NAME',text) self.assertNotIn('123456789',text) if __name__ == '__main__': unittest.main()