123 lines
6.4 KiB
Python
123 lines
6.4 KiB
Python
"""Compare validated primary-name candidates without inventing report equality."""
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
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 collect_arr_named_day as named, collect_arr_source as source, compare_arr_sources as compare
|
|
from integrations.ohip.profile_reader import ProfileSummaryReader
|
|
from integrations.ohip.rate_info import RateInfoReader
|
|
from tests.test_ohip_named_day import NamedService, DAY, HOTEL, SECRET
|
|
from tests.test_ohip_source_comparison import report
|
|
|
|
|
|
class NamedSourceComparisonTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
self.capture_dir = self.root / "capture"
|
|
self.report_path = self.root / "source.xml"
|
|
self.service = NamedService()
|
|
self.xml = report(count=3)
|
|
for i, row in enumerate(self.xml.findall('.//G_RESERVATION')):
|
|
row.find('FULL_NAME').text = self.service.profiles[f'p{i}']['fullName']
|
|
|
|
def capture(self, transport=None):
|
|
archive = source.Archive(self.capture_dir)
|
|
transport = transport or self.service
|
|
result = named.collect(named.Options(DAY, DAY, DAY, HOTEL, page_size=2, max_profiles=3), archive,
|
|
source.Reader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None),
|
|
RateInfoReader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None),
|
|
ProfileSummaryReader(archive, HOTEL, transport, max_profiles=3, key=SECRET, sleep=lambda _: None))
|
|
self.pin = result['manifest_sha256']
|
|
return result
|
|
|
|
def compare(self, version='v3'):
|
|
if not hasattr(self, 'pin'):
|
|
self.assertTrue(self.capture()['candidate_capture_complete'])
|
|
raw = ET.tostring(self.xml, encoding='utf-8', xml_declaration=True)
|
|
self.report_path.write_bytes(raw)
|
|
self.report_pin = hashlib.sha256(raw).hexdigest()
|
|
with patch.object(source, 'HTTPTransport', side_effect=AssertionError('no network')), \
|
|
patch.object(source, 'load_key', side_effect=AssertionError('no credentials')):
|
|
return compare.compare_files(self.capture_dir, self.pin, self.report_path, self.report_pin,
|
|
capture_version=version)
|
|
|
|
def test_exact_raw_and_trimmed_name_observations_preserve_unaccepted_status(self):
|
|
result, code = self.compare()
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result['version'], 'arr-api-native-comparison/v2')
|
|
for field in ('FULL_NAME_PRIMARY_SUMMARY_EXACT', 'FULL_NAME_PRIMARY_SUMMARY_TRIMMED'):
|
|
self.assertEqual(result['observations'][field]['equal_nonblank'], 3)
|
|
for flag in ('finance_ready', 'report_equivalence_verified', 'source_mapping_verified'):
|
|
self.assertIs(result[flag], False)
|
|
self.assertNotIn('trace_scope_and_order', result['unresolved'])
|
|
|
|
def test_trim_agreement_does_not_hide_exact_whitespace_difference(self):
|
|
field = self.xml.find('.//G_RESERVATION/FULL_NAME')
|
|
field.text = field.text.strip()
|
|
result, code = self.compare()
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result['observations']['FULL_NAME_PRIMARY_SUMMARY_EXACT']['different'], 1)
|
|
self.assertEqual(result['observations']['FULL_NAME_PRIMARY_SUMMARY_TRIMMED']['equal_nonblank'], 3)
|
|
|
|
def test_missing_blank_and_conflicting_summary_names_do_not_fallback(self):
|
|
self.service.profiles['p0'].pop('fullName')
|
|
self.service.profiles['p1']['fullName'] = ' '
|
|
self.service.profiles['p2']['givenName'] = 'MISMATCH'
|
|
result, code = self.compare()
|
|
self.assertEqual(code, 0)
|
|
for field in ('FULL_NAME_PRIMARY_SUMMARY_EXACT', 'FULL_NAME_PRIMARY_SUMMARY_TRIMMED'):
|
|
self.assertEqual(result['observations'][field]['unavailable'], 3)
|
|
self.assertEqual(result['observations'][field]['reasons'], {'api:primary_summary_name_unavailable': 3})
|
|
|
|
def test_swapping_xml_identities_compares_names_by_reservation_not_position(self):
|
|
first, second = self.xml.findall('.//G_RESERVATION/RESV_NAME_ID')[:2]
|
|
first.text, second.text = second.text, first.text
|
|
result, code = self.compare()
|
|
self.assertEqual(code, 0)
|
|
self.assertFalse(result['identity_sequence_equal'])
|
|
self.assertEqual(result['observations']['FULL_NAME_PRIMARY_SUMMARY_EXACT']['different'], 2)
|
|
|
|
def test_cross_hotel_and_duplicate_identity_guards_still_apply(self):
|
|
self.xml = report(count=3, hotel='OTHER')
|
|
result, code = self.compare()
|
|
self.assertEqual((code, result['reason']), (2, 'hotel_or_date_differs'))
|
|
self.xml = report(count=3)
|
|
rows = self.xml.findall('.//G_RESERVATION')
|
|
rows[1].find('RESV_NAME_ID').text = rows[0].find('RESV_NAME_ID').text
|
|
result, code = self.compare()
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result['ambiguous_report_rows'], 2)
|
|
self.assertEqual(result['observations']['FULL_NAME_PRIMARY_SUMMARY_EXACT']['compared_rows'], 1)
|
|
|
|
def test_failed_v3_capture_never_compares_partial_names(self):
|
|
def transport(method, path, body):
|
|
return (503, {}, b'{}') if path.endswith('/profiles/searches') else self.service(method, path, body)
|
|
self.assertFalse(self.capture(transport)['candidate_capture_complete'])
|
|
result, code = self.compare()
|
|
self.assertEqual((code, result['error']), (3, 'incomplete_capture'))
|
|
self.assertNotIn('observations', result)
|
|
|
|
def test_explicit_cli_version_and_output_privacy(self):
|
|
result, code = self.compare(version='v2')
|
|
self.assertEqual((code, result['error']), (3, 'incomplete_capture'))
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
code = compare.main(['--capture-dir', str(self.capture_dir), '--capture-sha256', self.pin,
|
|
'--report', str(self.report_path), '--report-sha256', self.report_pin, '--capture-version', 'v3'])
|
|
self.assertEqual(code, 0)
|
|
for value in ('Surname0', 'Given', 'PRIVATE', SECRET, 'confirmation-id0', '"p0"'):
|
|
self.assertNotIn(value, output.getvalue())
|
|
self.assertEqual(json.loads(output.getvalue())['matched_unique_records'], 3)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|