485 lines
27 KiB
Python
485 lines
27 KiB
Python
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'<RES_DETAIL>PRIVATE_BROKEN', b'<!DOCTYPE RES_DETAIL [<!ENTITY x "PRIVATE_ENTITY">]><RES_DETAIL/>'):
|
|
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()
|