88 lines
4.3 KiB
Python
88 lines
4.3 KiB
Python
"""Independently compare processing XML to explicit rows, never API acceptance.
|
|
|
|
This module does not import/call the serializer or infer values from its output.
|
|
It validates every child, scalar, row position and note/trace occurrence.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
import hashlib
|
|
import io
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from .arr_xml_contract import MAX_BYTES, MAX_ELEMENTS, SCALARS, VERSION, SourceDocument, validate_document
|
|
from .collect_arr_source import CollectionError, require
|
|
|
|
|
|
def _structure(node, names):
|
|
require(not node.attrib and not (node.text or "").strip() and not (node.tail or "").strip(),
|
|
"arr_xml_unexpected_structure")
|
|
require(Counter(child.tag for child in node) == Counter(names), "arr_xml_children_mismatch")
|
|
|
|
|
|
def _leaf(node, expected):
|
|
require(not node.attrib and not len(node) and not (node.tail or "").strip(), "arr_xml_leaf_structure")
|
|
require((node.text or "") == expected, "arr_xml_value_mismatch")
|
|
|
|
|
|
def verify(document: SourceDocument, payload: bytes) -> dict:
|
|
validate_document(document)
|
|
require(type(payload) is bytes and 0 < len(payload) <= MAX_BYTES, "arr_xml_payload_budget")
|
|
try:
|
|
text = payload.decode("utf-8-sig")
|
|
except UnicodeError:
|
|
raise CollectionError("arr_xml_utf8_required") from None
|
|
require("\x00" not in text and not re.search(r"<!\s*(?:DOCTYPE|ENTITY)\b", text, re.I), "arr_xml_unsafe_declaration")
|
|
declaration = re.match(r"<\?xml\s+[^?]*\?>", text)
|
|
if declaration:
|
|
encoding = re.search(r"encoding\s*=\s*['\"]([^'\"]+)['\"]", declaration[0])
|
|
require(encoding is None or encoding[1].lower() in {"utf-8", "utf8"}, "arr_xml_utf8_required")
|
|
# Count before building the final semantic checks; cap nesting and reject PIs
|
|
# or comments even where a normal tree parser would silently discard them.
|
|
try:
|
|
depth = count = 0
|
|
root = None
|
|
for event, node in ET.iterparse(io.BytesIO(payload), events=("start", "end", "comment", "pi")):
|
|
require(event not in {"comment", "pi"}, "arr_xml_unexpected_annotation")
|
|
if event == "start":
|
|
depth += 1
|
|
count += 1
|
|
require(depth <= 10 and count <= MAX_ELEMENTS, "arr_xml_structure_budget")
|
|
if root is None:
|
|
root = node
|
|
else:
|
|
depth -= 1
|
|
except (ET.ParseError, ValueError, RecursionError):
|
|
raise CollectionError("arr_xml_parse_failed") from None
|
|
require(root is not None and root.tag == "RES_DETAIL", "arr_xml_root_mismatch")
|
|
_structure(root, ["LIST_G_GROUP_BY1"])
|
|
groups = root[0]
|
|
_structure(groups, ["G_GROUP_BY1"])
|
|
group = groups[0]
|
|
_structure(group, ["GROUPBY1_SORT_COL", "GROUPBY1_COL", "LIST_G_RESERVATION"])
|
|
day = document.report_date
|
|
_leaf(group.find("GROUPBY1_SORT_COL"), f"{day.year:04d}{day.month:02d}{day.day:02d}")
|
|
_leaf(group.find("GROUPBY1_COL"), day.isoformat())
|
|
rows = group.find("LIST_G_RESERVATION")
|
|
_structure(rows, ["G_RESERVATION"] * len(document.rows))
|
|
for actual, expected in zip(rows, document.rows):
|
|
_structure(actual, ["RESORT", "RESV_NAME_ID", *(tag for _, tag in SCALARS),
|
|
"LIST_G_COMMENT_RESV_NAME_ID", "LIST_G_DEPT_ID"])
|
|
_leaf(actual.find("RESORT"), document.hotel_id)
|
|
_leaf(actual.find("RESV_NAME_ID"), expected.reservation_id)
|
|
for attribute, tag in SCALARS:
|
|
_leaf(actual.find(tag), getattr(expected, attribute))
|
|
for outer_tag, inner_tag, value_tag, values in (
|
|
("LIST_G_COMMENT_RESV_NAME_ID", "G_COMMENT_RESV_NAME_ID", "RES_COMMENT", expected.notes),
|
|
("LIST_G_DEPT_ID", "G_DEPT_ID", "TRACE_TEXT", expected.traces),
|
|
):
|
|
container = actual.find(outer_tag)
|
|
_structure(container, [inner_tag] * len(values))
|
|
for entry, value in zip(container, values):
|
|
_structure(entry, [value_tag])
|
|
_leaf(entry[0], value)
|
|
return {"version": VERSION, "serialization_verified": True, "source_records": len(document.rows),
|
|
"source_sha256": hashlib.sha256(payload).hexdigest(), "source_bytes": len(payload),
|
|
"source_mapping_verified": False, "report_equivalence_verified": False, "finance_ready": False}
|