200 lines
12 KiB
Python
200 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
from arr_ingestion.contracts import IngestionError
|
||
from arr_ingestion.validation import DeliveryValidator
|
||
from arr_processing.policy import load_processor_policy
|
||
from tests.test_arr_ingestion_validation import build_delivery, artifact_ref
|
||
from tests.test_arr_opera_daily_ingest import (
|
||
PROJECT_ROOT, core, reservation, run_processor, xml_document,
|
||
)
|
||
from tests.test_arr_programmatic import coordinator
|
||
|
||
|
||
CESU8_CROWN = bytes.fromhex("eda0bdedb191")
|
||
|
||
|
||
def source_with_trace(text: str) -> str:
|
||
row = reservation(1).replace(
|
||
"</G_RESERVATION>",
|
||
"<LIST_G_DEPT_ID><G_DEPT_ID><TRACE_TEXT>" + text
|
||
+ "</TRACE_TEXT></G_DEPT_ID></LIST_G_DEPT_ID></G_RESERVATION>",
|
||
)
|
||
return xml_document(row)
|
||
|
||
|
||
class EmojiXMLReadTests(unittest.TestCase):
|
||
def read(self, raw: bytes):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
path = Path(temporary) / "source.xml"
|
||
path.write_bytes(raw)
|
||
cleanup = {}
|
||
business_date, rows = core.read_xml(path, input_cleanup=cleanup)
|
||
self.assertEqual(path.read_bytes(), raw)
|
||
return business_date, rows, cleanup
|
||
|
||
def test_plain_and_composite_emoji_are_removed_as_whole_sequences(self):
|
||
for emoji in ("👑", "😀", "👑️", "❤️", "❤", "👍🏽", "🇨🇳", "👨👩👧👦", "1️⃣", "1⃣", "#️⃣", "🏳️🌈", "🏴\U000e0067\U000e0062\U000e0065\U000e006e\U000e0067\U000e007f"):
|
||
with self.subTest(emoji=emoji):
|
||
_, rows, cleanup = self.read(source_with_trace("前" + emoji + "后").encode())
|
||
self.assertEqual(core.source_record(rows[0], 1)["TRACE_TEXT"], "前后")
|
||
self.assertEqual(cleanup, {"ignored_emoji_count": 1})
|
||
|
||
def test_non_emoji_business_text_and_supplementary_cjk_are_preserved(self):
|
||
text = "中文𠮷 ไทย café 0123456789 # * ¥￥$€£฿ 900.00 2026-07-27 + - / & <"
|
||
_, rows, cleanup = self.read(source_with_trace(text).encode())
|
||
self.assertEqual(core.source_record(rows[0], 1)["TRACE_TEXT"], text.replace("&", "&").replace("<", "<"))
|
||
self.assertEqual(cleanup, {"ignored_emoji_count": 0})
|
||
|
||
def test_cesu8_recovers_only_complete_emoji_in_text_and_cdata(self):
|
||
for text in ("前👑后", "<![CDATA[前👑后]]>", "前👨👩👧👦后"):
|
||
with self.subTest(text=text):
|
||
source = source_with_trace(text)
|
||
# Deliberately reproduce a Java UTF-16/CESU-8 style exporter.
|
||
raw = "".join(
|
||
char if ord(char) <= 0xFFFF else chr(0xD800 + ((ord(char) - 0x10000) >> 10)) + chr(0xDC00 + ((ord(char) - 0x10000) & 1023))
|
||
for char in source
|
||
).encode("utf-8", errors="surrogatepass")
|
||
_, rows, cleanup = self.read(raw)
|
||
self.assertEqual(core.source_record(rows[0], 1)["TRACE_TEXT"], "前后")
|
||
self.assertEqual(cleanup["ignored_emoji_count"], 1)
|
||
|
||
def test_character_references_and_literal_emoji_share_cleanup(self):
|
||
_, rows, cleanup = self.read(source_with_trace("👑 👑 👑 ❤️").encode())
|
||
self.assertEqual(core.source_record(rows[0], 1)["TRACE_TEXT"], "")
|
||
self.assertEqual(cleanup["ignored_emoji_count"], 4)
|
||
|
||
def test_unrelated_invalid_bytes_and_surrogates_still_fail(self):
|
||
base = source_with_trace("BAD").encode()
|
||
for bad in (b"\xff", b"\xc0\xaf", b"\xed\xa0\xbd", b"\xed\xb1\x91", b"\xf0\x9f\x91", b"\xed\xa1\x80\xed\xb0\x80", CESU8_CROWN + b"\xff"):
|
||
with self.subTest(bad=bad.hex()):
|
||
with self.assertRaises(core.ProcessingFailure) as caught:
|
||
self.read(base.replace(b"BAD", bad))
|
||
self.assertEqual(caught.exception.errors[0].code, "XML_PARSE_ERROR")
|
||
|
||
def test_markup_is_not_repaired_or_stripped(self):
|
||
original = source_with_trace("ok").encode()
|
||
sources = (
|
||
original.replace(b"<RES_DETAIL>", b"<RES_DET" + CESU8_CROWN + b"AIL>"),
|
||
original.replace(b"<RES_DETAIL>", b'<RES_DETAIL value="' + CESU8_CROWN + b'">'),
|
||
original.replace(b"<TRACE_TEXT>", b"<TRACE_TEXT>" + CESU8_CROWN + b"<broken>"),
|
||
original + CESU8_CROWN,
|
||
)
|
||
for raw in sources:
|
||
with self.subTest(raw=hashlib.sha256(raw).hexdigest()):
|
||
with self.assertRaises(core.ProcessingFailure) as caught:
|
||
self.read(raw)
|
||
self.assertEqual(caught.exception.errors[0].code, "XML_PARSE_ERROR")
|
||
|
||
def test_utf8_bom_and_existing_utf16_inputs(self):
|
||
source = source_with_trace("👑保留")
|
||
for raw in (b"\xef\xbb\xbf" + source.encode().replace("👑".encode(), CESU8_CROWN), source.replace("UTF-8", "UTF-16").encode("utf-16")):
|
||
with self.subTest(encoding=raw[:3]):
|
||
_, rows, cleanup = self.read(raw)
|
||
self.assertEqual(core.source_record(rows[0], 1)["TRACE_TEXT"], "保留")
|
||
self.assertEqual(cleanup["ignored_emoji_count"], 1)
|
||
|
||
def test_doctype_and_mixed_business_dates_remain_errors(self):
|
||
source = source_with_trace("👑")
|
||
unsafe = source.replace("<RES_DETAIL>", '<!DOCTYPE RES_DETAIL [<!ENTITY e "hello">]><RES_DETAIL>')
|
||
with self.assertRaises(core.ProcessingFailure) as caught:
|
||
self.read(unsafe.encode().replace("👑".encode(), CESU8_CROWN))
|
||
self.assertEqual(caught.exception.errors[0].code, "INPUT_XML_UNSAFE_DECLARATION")
|
||
group = source[source.index("<G_GROUP_BY1>"):source.index("</G_GROUP_BY1>") + len("</G_GROUP_BY1>")]
|
||
mixed = source.replace("</LIST_G_GROUP_BY1>", group.replace("20260727", "20260728").replace("27-07-26", "28-07-26") + "</LIST_G_GROUP_BY1>")
|
||
with self.assertRaises(core.ProcessingFailure) as caught:
|
||
self.read(mixed.encode())
|
||
self.assertEqual(caught.exception.errors[0].code, "XML_MULTIPLE_BUSINESS_DATES")
|
||
|
||
|
||
class EmojiPipelineTests(unittest.TestCase):
|
||
def test_retired_v3_replay_keeps_its_original_emoji_behavior(self):
|
||
source = source_with_trace("👑 VIP").encode()
|
||
for raw, expected in ((source, "success"), (source.replace("👑".encode(), CESU8_CROWN), "failed")):
|
||
with self.subTest(expected=expected), tempfile.TemporaryDirectory() as temporary:
|
||
root = Path(temporary)
|
||
_, _, _, result, structured = run_processor(raw, root, legacy_v3_output=True)
|
||
self.assertEqual(result["status"], expected)
|
||
self.assertNotIn("input_cleanup", result)
|
||
self.assertEqual(structured["processor_version"], "3.0.0")
|
||
if expected == "success":
|
||
self.assertEqual(structured["records"][0]["trace_text"], "👑 VIP")
|
||
else:
|
||
self.assertEqual(structured["errors"][0]["code"], "XML_PARSE_ERROR")
|
||
|
||
def test_upload_preserves_original_identity_and_commits_cleaned_facts(self):
|
||
raw = source_with_trace("👑 VIP 👑 VIP 👑 VIP 👑 VIP").encode().replace("👑".encode(), CESU8_CROWN)
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
upload, repository, objects = coordinator(Path(temporary))
|
||
receipt = upload.submit("ARR.XML", raw)
|
||
self.assertEqual(receipt["status"], "succeeded")
|
||
self.assertEqual(receipt["ignored_emoji_count"], 4)
|
||
self.assertEqual(receipt["source_sha256"], hashlib.sha256(raw).hexdigest())
|
||
self.assertEqual(receipt["source_byte_size"], len(raw))
|
||
source = next(objects.rglob("source.xml"))
|
||
self.assertEqual(source.read_bytes(), raw)
|
||
result = json.loads(next(objects.rglob("result.json")).read_text())
|
||
self.assertEqual(result["input_cleanup"], {"ignored_emoji_count": 4})
|
||
self.assertIn("已忽略 4 处表情符号", result["message"])
|
||
records = repository.version_records(receipt["daily_version_id"])
|
||
self.assertEqual(records[0]["trace_text"], "VIP VIP VIP VIP")
|
||
|
||
def test_emoji_only_required_name_still_fails_business_validation(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
code, _, _, result, _ = run_processor(xml_document(reservation(1, full_name="👑")), Path(temporary))
|
||
self.assertNotEqual(code, 0)
|
||
self.assertEqual(result["status"], "failed")
|
||
self.assertTrue(any("FULL_NAME" in item["message"] for item in result["errors"]))
|
||
self.assertEqual(result["input_cleanup"]["ignored_emoji_count"], 1)
|
||
|
||
def test_missing_price_stays_review_required_and_final_replay_uses_original(self):
|
||
raw = xml_document(reservation(1, rate_amount="1800", full_name="👑SYNTHETIC GUEST")).encode().replace("👑".encode(), CESU8_CROWN)
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
upload, repository, _ = coordinator(Path(temporary))
|
||
receipt = upload.submit("ARR.XML", raw)
|
||
self.assertEqual(receipt["status"], "needs_review")
|
||
self.assertEqual(receipt["ignored_emoji_count"], 1)
|
||
review = upload.get_price_review(receipt["job_id"], 50, 0)
|
||
updated = upload.update_price_review_item(receipt["job_id"], review["items"][0]["item_id"], review["case_id"], review["revision"], "900", "synthetic-operator")
|
||
final = upload.finalize_price_review(receipt["job_id"], review["case_id"], updated["revision"], "synthetic-operator")
|
||
self.assertEqual(final["status"], "succeeded")
|
||
self.assertEqual(final["ignored_emoji_count"], 1)
|
||
self.assertIsNotNone(repository.version_records(final["daily_version_id"]))
|
||
|
||
def test_independent_validator_rejects_tampered_cleanup_metadata(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
raw, store, values = build_delivery(source_with_trace("👑"), Path(temporary))
|
||
validator = DeliveryValidator(store, load_processor_policy(PROJECT_ROOT))
|
||
validator.validate(raw)
|
||
for cleanup in (None, {"ignored_emoji_count": 2}, {"ignored_emoji_count": True}, {"ignored_emoji_count": 1, "unexpected": 1}):
|
||
with self.subTest(cleanup=cleanup):
|
||
envelope = json.loads(raw)
|
||
result = json.loads(values["result_json"])
|
||
if cleanup is None:
|
||
result.pop("input_cleanup")
|
||
else:
|
||
result["input_cleanup"] = cleanup
|
||
edited = (json.dumps(result, ensure_ascii=False) + "\n").encode()
|
||
ref = envelope["artifacts"]["result_json"]
|
||
envelope["artifacts"]["result_json"] = artifact_ref(ref["object_key"], ref["original_filename"], edited, ref["mime_type"])
|
||
structured = json.loads(values["structured_result_json"])
|
||
structured["artifacts"]["result_json"]["sha256"] = hashlib.sha256(edited).hexdigest()
|
||
structured["artifacts"]["result_json"]["byte_size"] = len(edited)
|
||
edited_structured = (json.dumps(structured, ensure_ascii=False) + "\n").encode()
|
||
sref = envelope["artifacts"]["structured_result_json"]
|
||
envelope["artifacts"]["structured_result_json"] = artifact_ref(sref["object_key"], sref["original_filename"], edited_structured, sref["mime_type"])
|
||
store.objects[ref["object_key"]] = edited
|
||
store.objects[sref["object_key"]] = edited_structured
|
||
with self.assertRaises(IngestionError) as caught:
|
||
validator.validate(json.dumps(envelope).encode())
|
||
self.assertIn(caught.exception.code, {"RESULT_CONTRACT_INVALID", "OUTPUT_VALIDATION_FAILED"})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|