273 lines
13 KiB
Python
273 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from integrations.ohip.compare_report_xml import compare_files
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "integrations/ohip/compare_report_xml.py"
|
|
|
|
|
|
def reservation(identity="PRIVATE_ID_A", room="101", hotel="57106", rownum="7"):
|
|
row = ET.Element("G_RESERVATION")
|
|
for name, value in {
|
|
"RESV_NAME_ID": identity, "RESORT": hotel, "ROOM": room, "ROWNUM": rownum,
|
|
"ARRIVAL": "15-09-26", "RATE_CODE": "WHO1", "EFFECTIVE_RATE_AMOUNT": "100",
|
|
"GUEST_NAME": "PRIVATE_GUEST_NAME",
|
|
}.items():
|
|
ET.SubElement(row, name).text = value
|
|
notes = ET.SubElement(row, "LIST_G_COMMENT_RESV_NAME_ID")
|
|
for value in ("PRIVATE_FIRST_NOTE", "PRIVATE_SECOND_NOTE"):
|
|
note = ET.SubElement(notes, "G_COMMENT_RESV_NAME_ID")
|
|
ET.SubElement(note, "RES_COMMENT").text = value
|
|
ET.SubElement(note, "RES_COMMENT_TYPE").text = "CAS"
|
|
traces = ET.SubElement(row, "LIST_G_DEPT_ID")
|
|
for value in ("PRIVATE_FIRST_TRACE", "PRIVATE_SECOND_TRACE"):
|
|
trace = ET.SubElement(traces, "G_DEPT_ID")
|
|
ET.SubElement(trace, "TRACE_TEXT").text = value
|
|
return row
|
|
|
|
|
|
def report(rows=None, sort_date="20260915", display_date="15-09-26"):
|
|
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 = sort_date
|
|
ET.SubElement(group, "GROUPBY1_COL").text = display_date
|
|
record_list = ET.SubElement(group, "LIST_G_RESERVATION")
|
|
record_list.extend(rows if rows is not None else [reservation()])
|
|
ET.SubElement(root, "RMS_REPORT").text = str(len(record_list))
|
|
return root
|
|
|
|
|
|
def payload(root):
|
|
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
|
|
|
|
|
|
class NativeReportComparisonTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.baseline = Path(self.directory.name) / "baseline.xml"
|
|
self.candidate = Path(self.directory.name) / "candidate.xml"
|
|
|
|
def compare(self, before, after, expected=None):
|
|
self.baseline.write_bytes(before if isinstance(before, bytes) else payload(before))
|
|
self.candidate.write_bytes(after if isinstance(after, bytes) else payload(after))
|
|
pin = expected or hashlib.sha256(self.baseline.read_bytes()).hexdigest()
|
|
return compare_files(self.baseline, self.candidate, pin)
|
|
|
|
def test_identical_and_nonmonotonic_rownum_preserve_source_order(self):
|
|
source = report([reservation(rownum="7"), reservation("PRIVATE_ID_B", rownum="2")])
|
|
result, code = self.compare(source, source)
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(result["byte_identical"])
|
|
self.assertEqual(result["baseline"]["source_records"], 2)
|
|
|
|
def test_indentation_comment_bom_and_attribute_order_are_not_content_changes(self):
|
|
before = report()
|
|
before.set("format", "native")
|
|
before.set("version", "1")
|
|
after = copy.deepcopy(before)
|
|
after.attrib = {"version": "1", "format": "native"}
|
|
ET.indent(after)
|
|
raw = payload(after).replace(b"<RES_DETAIL", b"<!-- file comment --><RES_DETAIL", 1)
|
|
result, code = self.compare(before, b"\xef\xbb\xbf" + raw)
|
|
self.assertEqual(code, 0)
|
|
self.assertFalse(result["byte_identical"])
|
|
self.assertTrue(result["report_metadata_identical"])
|
|
|
|
def test_same_count_reorder_detected_without_rownum_sorting(self):
|
|
a, b = reservation(), reservation("PRIVATE_ID_B", rownum="2")
|
|
result, code = self.compare(report([a, b]), report([copy.deepcopy(b), copy.deepcopy(a)]))
|
|
self.assertEqual(code, 1)
|
|
self.assertFalse(result["identity_sequence_identical"])
|
|
self.assertFalse(result["record_content_sequence_identical"])
|
|
self.assertEqual(result["changed_occurrences"], 0)
|
|
self.assertEqual(result["missing_occurrences"], 0)
|
|
|
|
def test_same_count_replacement_is_missing_and_additional(self):
|
|
result, code = self.compare(report(), report([reservation("PRIVATE_REPLACEMENT")]))
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["missing_occurrences"], 1)
|
|
self.assertEqual(result["additional_occurrences"], 1)
|
|
self.assertEqual(result["missing_baseline_positions"], [1])
|
|
|
|
def test_duplicate_identity_not_collapsed_and_swapped_content_detected(self):
|
|
a, b = reservation(room="101"), reservation(room="102")
|
|
result, code = self.compare(report([a, b]), report([copy.deepcopy(b), copy.deepcopy(a)]))
|
|
self.assertEqual(code, 1)
|
|
self.assertTrue(result["identity_sequence_identical"])
|
|
self.assertFalse(result["record_content_sequence_identical"])
|
|
self.assertEqual(result["changed_occurrences"], 2)
|
|
self.assertEqual(result["baseline"]["repeated_id_occurrences"], 1)
|
|
|
|
def test_missing_duplicate_occurrence_is_reported(self):
|
|
a = reservation()
|
|
result, code = self.compare(report([a, copy.deepcopy(a)]), report())
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["missing_baseline_positions"], [2])
|
|
self.assertEqual(result["changed_occurrences"], 0)
|
|
|
|
def test_note_type_text_and_note_trace_order_are_preserved(self):
|
|
for mode in ("type", "text", "note_order", "trace_order", "note_space"):
|
|
with self.subTest(mode=mode):
|
|
after = report()
|
|
if mode == "type":
|
|
after.find(".//RES_COMMENT_TYPE").text = "GEN"
|
|
elif mode == "text":
|
|
after.find(".//RES_COMMENT").text = "PRIVATE_CHANGED_NOTE"
|
|
elif mode == "note_space":
|
|
after.find(".//RES_COMMENT").text += " "
|
|
else:
|
|
name = "LIST_G_DEPT_ID" if mode == "trace_order" else "LIST_G_COMMENT_RESV_NAME_ID"
|
|
items = after.find(".//" + name)
|
|
items[:] = reversed(list(items))
|
|
result, code = self.compare(report(), after)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["changed_occurrences"], 1)
|
|
expected = "LIST_G_DEPT_ID" if mode == "trace_order" else "LIST_G_COMMENT_RESV_NAME_ID"
|
|
self.assertIn(expected, result["changed_records"][0]["fields"])
|
|
|
|
def test_metadata_only_changes_are_separate(self):
|
|
after = report()
|
|
after.find("RMS_REPORT").text = "2"
|
|
result, code = self.compare(report(), after)
|
|
self.assertEqual(code, 1)
|
|
self.assertFalse(result["report_metadata_identical"])
|
|
self.assertTrue(result["record_content_sequence_identical"])
|
|
|
|
def test_pretty_printed_changed_row_does_not_report_indentation_as_field_change(self):
|
|
after = report()
|
|
after.find(".//EFFECTIVE_RATE_AMOUNT").text = "200"
|
|
ET.indent(after)
|
|
result, code = self.compare(report(), after)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["changed_records"][0]["fields"], ["EFFECTIVE_RATE_AMOUNT"])
|
|
|
|
def test_missing_vs_empty_leaf_and_top_level_field_order(self):
|
|
for mode in ("empty", "order", "amount"):
|
|
with self.subTest(mode=mode):
|
|
after = report()
|
|
row = after.find(".//G_RESERVATION")
|
|
if mode == "empty":
|
|
ET.SubElement(row, "EXTRA_FIELD")
|
|
elif mode == "amount":
|
|
row.find("EFFECTIVE_RATE_AMOUNT").text = "200"
|
|
else:
|
|
row[:] = reversed(list(row))
|
|
result, code = self.compare(report(), after)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["changed_occurrences"], 1)
|
|
|
|
def test_mixed_content_and_processing_instruction_are_not_silently_dropped(self):
|
|
for mode in ("tail", "pi"):
|
|
with self.subTest(mode=mode):
|
|
after = report()
|
|
if mode == "tail":
|
|
after.find(".//RATE_CODE").tail = "PRIVATE_TAIL"
|
|
else:
|
|
after.append(ET.ProcessingInstruction("report", "changed"))
|
|
_, code = self.compare(report(), after)
|
|
self.assertEqual(code, 1)
|
|
|
|
def test_different_hotel_or_group_day_is_not_comparable(self):
|
|
for after in (report([reservation(hotel="OHIPSB02")]), report(sort_date="20260914", display_date="14-09-26")):
|
|
with self.subTest():
|
|
result, code = self.compare(report(), after)
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(result["status"], "not_comparable")
|
|
self.assertNotIn("xml_content_identical", result)
|
|
|
|
def test_invalid_context_is_not_accepted_even_when_both_files_identical(self):
|
|
mixed_date = report()
|
|
mixed_date.find("LIST_G_GROUP_BY1").append(
|
|
report(sort_date="20260914", display_date="14-09-26").find(".//G_GROUP_BY1")
|
|
)
|
|
missing_id = report()
|
|
missing_id.find(".//G_RESERVATION").remove(missing_id.find(".//RESV_NAME_ID"))
|
|
ambiguous_id = report()
|
|
ET.SubElement(ambiguous_id.find(".//G_RESERVATION"), "RESV_NAME_ID").text = "PRIVATE_EXTRA"
|
|
unexpected = report()
|
|
unexpected.append(reservation())
|
|
for invalid in (
|
|
report([]), report([reservation(), reservation(hotel="OTHER")]), mixed_date,
|
|
missing_id, ambiguous_id, unexpected, report(sort_date="20260931"),
|
|
report(display_date="14-09-26"),
|
|
):
|
|
with self.subTest():
|
|
result, code = self.compare(invalid, invalid)
|
|
self.assertEqual(code, 3)
|
|
self.assertEqual(result["status"], "invalid_input")
|
|
|
|
def test_pin_mismatch_checked_before_parsing(self):
|
|
result, code = self.compare(b"PRIVATE_NOT_XML", report(), "0" * 64)
|
|
self.assertEqual(code, 3)
|
|
self.assertEqual(result["error"], "BASELINE_SHA256_MISMATCH")
|
|
|
|
def test_bad_xml_encoding_and_entities_are_safe_errors(self):
|
|
for invalid in (
|
|
b"<html>PRIVATE_LOGIN_PAGE</html>", payload(report())[:-15], b"\xff",
|
|
payload(report()).decode().encode("utf-16"),
|
|
payload(report()).replace(b"utf-8", b"iso-8859-1"),
|
|
b'<!DOCTYPE RES_DETAIL [<!ENTITY e "PRIVATE_ENTITY">]><RES_DETAIL>&e;</RES_DETAIL>',
|
|
):
|
|
with self.subTest():
|
|
result, code = self.compare(report(), invalid)
|
|
self.assertEqual(code, 3)
|
|
self.assertEqual(result["input"], "candidate")
|
|
self.assertNotIn("PRIVATE", json.dumps(result))
|
|
|
|
def test_depth_size_and_unreadable_inputs(self):
|
|
deep = report()
|
|
node = deep
|
|
for _ in range(130):
|
|
node = ET.SubElement(node, "nested")
|
|
self.assertEqual(self.compare(report(), deep)[0]["error"], "XML_TOO_DEEP")
|
|
with patch("integrations.ohip.compare_report_xml.MAX_BYTES", 10):
|
|
self.assertEqual(self.compare(report(), report())[1], 3)
|
|
self.compare(report(), report())
|
|
self.candidate.unlink()
|
|
pin = hashlib.sha256(self.baseline.read_bytes()).hexdigest()
|
|
self.assertEqual(compare_files(self.baseline, self.candidate, pin)[1], 3)
|
|
|
|
def test_details_are_bounded_without_losing_counts(self):
|
|
before = report([reservation(f"PRIVATE_ID_{i}") for i in range(60)])
|
|
after = report([reservation(f"PRIVATE_OTHER_{i}") for i in range(60)])
|
|
result, code = self.compare(before, after)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["missing_occurrences"], 60)
|
|
self.assertEqual(len(result["missing_baseline_positions"]), 50)
|
|
self.assertTrue(result["details_truncated"])
|
|
|
|
def test_cli_emits_no_guest_values_paths_ids_or_arbitrary_field_names(self):
|
|
after = report()
|
|
after.find(".//GUEST_NAME").text = "PRIVATE_CHANGED_GUEST"
|
|
ET.SubElement(after.find(".//G_RESERVATION"), "PRIVATE_FIELD_NAME").text = "PRIVATE_VALUE"
|
|
self.compare(report(), after)
|
|
pin = hashlib.sha256(self.baseline.read_bytes()).hexdigest()
|
|
before_bytes, after_bytes = self.baseline.read_bytes(), self.candidate.read_bytes()
|
|
result = subprocess.run([
|
|
sys.executable, str(SCRIPT), "--baseline", str(self.baseline),
|
|
"--candidate", str(self.candidate), "--baseline-sha256", pin,
|
|
], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertEqual(result.stderr, "")
|
|
self.assertNotIn("PRIVATE", result.stdout)
|
|
self.assertNotIn(self.directory.name, result.stdout)
|
|
self.assertIn("OTHER_FIELD", json.loads(result.stdout)["changed_records"][0]["fields"])
|
|
self.assertEqual(self.baseline.read_bytes(), before_bytes)
|
|
self.assertEqual(self.candidate.read_bytes(), after_bytes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|