"""Processing XML output tests; no API mapping or production acceptance.""" import argparse from dataclasses import replace from datetime import date, datetime from pathlib import Path import tempfile import unittest from unittest.mock import patch import xml.etree.ElementTree as ET from integrations.ohip import arr_xml as writer from integrations.ohip import arr_xml_contract as contract from integrations.ohip import validate_arr_xml as checker from integrations.ohip.collect_arr_source import CollectionError from integrations.ohip.arr_xml_contract import SourceDocument, SourceRow from tests.test_arr_opera_daily_ingest import core, run_processor, validator as output_validator def row(**changes): fields = dict(reservation_id="SYNTHETIC-ID", block_code="", adults="2", children="0", company_name="T / Q.B.D. TRAVEL GROUP", confirmation_no="CONF-001", room_no="0007", effective_rate="900.00", full_name="SYNTHETIC GUEST", no_of_rooms="1", products="", rate_code="GRPA1", room_category_label="SYNTHETIC", arrival="2026-07-27", departure="2026-07-28", notes=(" ", "FIRST NOTE", "SECOND NOTE"), traces=("", "FIRST TRACE", "SECOND TRACE")) return SourceRow(**(fields | changes)) def document(*rows): return SourceDocument("TEST_HOTEL", date(2026, 7, 27), rows or (row(),)) def tree(payload): return ET.fromstring(payload) def xml_rows(root): return root.find("./LIST_G_GROUP_BY1/G_GROUP_BY1/LIST_G_RESERVATION") class ARRXMLTests(unittest.TestCase): def test_handwritten_field_tags_structure_and_exact_text(self): values = row(block_code="BLOCK", children="3", products="A, B", effective_rate="9007199254740993.0100") expected = document(values) raw = writer.serialize(expected) root = tree(raw) group = root.find("./LIST_G_GROUP_BY1/G_GROUP_BY1") self.assertEqual(group.findtext("GROUPBY1_SORT_COL"), "20260727") self.assertEqual(group.findtext("GROUPBY1_COL"), "2026-07-27") node = xml_rows(root)[0] # Handwritten goldens deliberately do not derive expected tags from SCALARS. gold = {"RESORT": "TEST_HOTEL", "RESV_NAME_ID": "SYNTHETIC-ID", "BLOCK_CODE": "BLOCK", "ADULTS": "2", "CF_CHILDREN": "3", "COMPANY_NAME": "T / Q.B.D. TRAVEL GROUP", "CONFIRMATION_NO": "CONF-001", "DISP_ROOM_NO": "0007", "EFFECTIVE_RATE_AMOUNT": "9007199254740993.0100", "FULL_NAME": "SYNTHETIC GUEST", "NO_OF_ROOMS": "1", "PRODUCTS": "A, B", "RATE_CODE": "GRPA1", "ROOM_CATEGORY_LABEL": "SYNTHETIC", "TRUNC_BEGIN": "2026-07-27", "TRUNC_END": "2026-07-28"} self.assertEqual({child.tag: child.text or "" for child in node if not len(child)}, gold) self.assertEqual([v.text or "" for v in node.findall("./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID/RES_COMMENT")], [" ", "FIRST NOTE", "SECOND NOTE"]) self.assertEqual([v.text or "" for v in node.findall("./LIST_G_DEPT_ID/G_DEPT_ID/TRACE_TEXT")], ["", "FIRST TRACE", "SECOND TRACE"]) report = checker.verify(expected, raw) self.assertTrue(report["serialization_verified"]) for flag in ("source_mapping_verified", "report_equivalence_verified", "finance_ready"): self.assertIs(report[flag], False) def test_deterministic_order_duplicates_unknown_rate_and_invalid_values_preserved(self): expected = document(row(room_no="999"), row(room_no="001", confirmation_no="SECOND"), row(rate_code="NOT-ALLOWED", effective_rate="BAD", company_name="", room_no="", notes=(), traces=())) raw = writer.serialize(expected) self.assertEqual(raw, writer.serialize(expected)) nodes = xml_rows(tree(raw)) self.assertEqual([n.findtext("DISP_ROOM_NO") for n in nodes], ["999", "001", ""]) self.assertEqual([n.findtext("RESV_NAME_ID") for n in nodes], ["SYNTHETIC-ID"] * 3) self.assertEqual(nodes[2].findtext("EFFECTIVE_RATE_AMOUNT"), "BAD") checker.verify(expected, raw) def test_unicode_xml_metacharacters_and_crlf_preserved(self): text = ' 中ไทยé 👑 & "\t\n\r\r\n ]]> ' expected = document(row(full_name=text, notes=(text,), traces=(text,))) raw = writer.serialize(expected) self.assertIn(b" ", raw) self.assertIn(b"<node>&", raw) self.assertEqual(xml_rows(tree(raw))[0].findtext("FULL_NAME"), text) checker.verify(expected, raw) def test_explicit_input_required_without_numeric_or_date_coercion(self): bad_rows = [replace(row(), adults=2), replace(row(), effective_rate=900.0), replace(row(), effective_rate=None), replace(row(), notes=None), replace(row(), traces=[]), replace(row(), notes=(None,)), replace(row(), reservation_id=" "), replace(row(), full_name="bad\x00"), replace(row(), full_name="bad\ud800"), replace(row(), full_name="bad\uffff")] bad_docs = [replace(document(), report_date="2026-07-27"), replace(document(), report_date=datetime(2026, 7, 27)), replace(document(), hotel_id=""), replace(document(), rows=()), replace(document(), rows=[row()]), replace(document(), rows=({},))] for bad in [*(document(r) for r in bad_rows), *bad_docs]: with self.subTest(bad=type(bad)), self.assertRaises(CollectionError): writer.serialize(bad) def test_historical_full_year_no_century_guess(self): for year in (1, 1965, 2065): expected = replace(document(), report_date=date(year, 1, 2)) raw = writer.serialize(expected) self.assertEqual(tree(raw).findtext("./LIST_G_GROUP_BY1/G_GROUP_BY1/GROUPBY1_SORT_COL"), f"{year:04d}0102") checker.verify(expected, raw) def test_independent_validation_does_not_serialize(self): expected = document() raw = writer.serialize(expected) with patch.object(writer, "serialize", side_effect=AssertionError("must not reserialize")): checker.verify(expected, raw) def test_missing_extra_duplicate_and_substituted_fields_rejected(self): expected = document() modifications = [ lambda n: n.remove(n.find("CF_CHILDREN")), lambda n: ET.SubElement(n, "UNKNOWN"), lambda n: ET.SubElement(n, "CF_CHILDREN").__setattr__("text", "0"), lambda n: n.find("CF_CHILDREN").__setattr__("tag", "CHILDREN"), lambda n: n.find("EFFECTIVE_RATE_AMOUNT").__setattr__("text", "900"), lambda n: n.find("FULL_NAME").set("approved", "true"), lambda n: ET.SubElement(n.find("FULL_NAME"), "nested"), lambda n: n.find("RESORT").__setattr__("text", "OTHER"), lambda n: n.find("RESV_NAME_ID").__setattr__("text", "OTHER"), ] for index, modify in enumerate(modifications): root = tree(writer.serialize(expected)) modify(xml_rows(root)[0]) with self.subTest(index=index), self.assertRaises(CollectionError): checker.verify(expected, ET.tostring(root)) def test_omitted_added_reordered_rows_and_notes_rejected(self): expected = document(row(), row(reservation_id="SECOND", full_name="SECOND NAME")) for operation in ("drop", "add", "swap", "drop_note", "swap_notes", "drop_empty_trace", "swap_traces"): root = tree(writer.serialize(expected)) rows = xml_rows(root) if operation == "drop": rows.remove(rows[0]) elif operation == "add": rows.append(rows[0]) elif operation == "swap": rows[:] = list(reversed(rows)) else: notes = rows[0].find("LIST_G_COMMENT_RESV_NAME_ID") if operation == "drop_note": notes.remove(notes[0]) elif operation == "swap_notes": notes[:] = list(reversed(notes)) else: traces = rows[0].find("LIST_G_DEPT_ID") if operation == "swap_traces": traces[1:] = list(reversed(traces[1:])) else: traces.remove(traces[0]) with self.subTest(operation=operation), self.assertRaises(CollectionError): checker.verify(expected, ET.tostring(root)) def test_group_dates_root_mixed_content_and_annotations_rejected(self): expected = document() raw = writer.serialize(expected) bad = [raw.replace(b"20260727", b"20260728"), raw.replace(b"2026-07-27", b"2026-07-28", 1), raw.replace(b"", b"unapproved"), raw.replace(b"", b""), raw.replace(b"", b""), raw + b"", raw + b"", raw.replace(b"utf-8", b"iso-8859-1"), b"\xff", b"", b']>&x;'] for candidate in bad: with self.subTest(size=len(candidate)), self.assertRaises(CollectionError): checker.verify(expected, candidate) def test_input_output_and_parser_limits(self): expected = document() raw = writer.serialize(expected) with patch.object(contract, "MAX_BYTES", 1), self.assertRaises(CollectionError): writer.serialize(expected) with patch.object(writer, "MAX_BYTES", 1), self.assertRaises(CollectionError): writer.serialize(expected) with patch.object(contract, "MAX_ROWS", 1), self.assertRaises(CollectionError): writer.serialize(document(row(), row())) with patch.object(checker, "MAX_ELEMENTS", 1), self.assertRaises(CollectionError): checker.verify(expected, raw) with self.assertRaises(CollectionError): checker.verify(expected, b"" + b"" * 20 + b"" * 20 + b"") def test_actual_handoff_byte_limit_including_escape_expansion(self): from arr_web.contracts import MAX_UPLOAD_BYTES, validate_xml_payload self.assertEqual(contract.MAX_BYTES, MAX_UPLOAD_BYTES) validate_xml_payload(writer.serialize(document())) with self.assertRaisesRegex(CollectionError, "arr_xml_payload_budget"): checker.verify(document(), b" " * (MAX_UPLOAD_BYTES + 1)) # Input fits, but escaped XML exceeds the handoff limit. expected = document(row(full_name="&" * (MAX_UPLOAD_BYTES // 5 + 1))) with self.assertRaisesRegex(CollectionError, "arr_xml_output_budget_exceeded"): writer.serialize(expected) def test_existing_processor_retains_semantic_rules_and_validation(self): expected = document(row(rate_code="NOT-ALLOWED", effective_rate="BAD", room_no=""), row(room_no="0007", notes=(" ", "GROUP-A", "IGNORED-LATER"), departure="2026-07-30"), row(room_no="0007", confirmation_no="DUPLICATE", notes=("OTHER",)), row(room_no="0008", company_name="T- Rainbow Holiday Service", rate_code="LBMS", departure="2026-07-27")) raw = writer.serialize(expected) checker.verify(expected, raw) with tempfile.TemporaryDirectory() as root: code, xml_path, output, result, structured = run_processor(raw.decode(), Path(root)) self.assertEqual(code, 0) self.assertEqual([record["outcome"] for record in structured["records"]], ["excluded_rate_code", "retained", "duplicate", "retained"]) self.assertEqual(structured["records"][1]["res_comment"], "GROUP-A") self.assertEqual(structured["records"][3]["nights"], 0) errors = output_validator.validate(argparse.Namespace(xml=str(xml_path.resolve()), daily=str((output / result["outputs"]["daily_report"]).resolve()), result_json=str((output / "result.json").resolve()), structured_result_json=str((output / "structured-result.json").resolve()), price_reference=str(core.PRICE_REFERENCE.resolve()))) self.assertEqual(errors, []) def test_invalid_whitelist_duplicate_is_not_hidden_by_serialization(self): expected = document(row(), row(company_name="", confirmation_no="BAD-DUPLICATE")) raw = writer.serialize(expected) self.assertEqual(checker.verify(expected, raw)["source_records"], 2) with tempfile.TemporaryDirectory() as root: code, _, _, result, _ = run_processor(raw.decode(), Path(root)) self.assertNotEqual(code, 0) self.assertIn("XML_COMPANY_MISSING", {item["code"] for item in result["errors"]}) if __name__ == "__main__": unittest.main()