shiyuyun/ARR-2.0-0918
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recognize corroborated empty Oracle associations

Browse Source
This commit is contained in:
Wyndham ARR committed 2026-10-08 18:58:18 +08:00
1 parent 4559d5a642
commit de9eb532f2
9 files changed
+735 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_arr_data_review.py
+281
View File
@@ -44,6 +44,8 @@ from tests import test_arr_download_postgres_integration as pg_helpers
PROJECT = Path(__file__).resolve().parents[1]
ACTOR = "operator@example.test"
SOURCE_MANIFEST = "c" * 64
REANALYSIS_MANIFEST = "e" * 64
REANALYSIS_POLICY = "oracle-optional-association/v1"
CONTEXT = {"hotel_id": HOTEL, "source_kind": "test_transport"}
@@ -77,6 +79,23 @@ def gap(document, field, sequence=1, state="missing"):
document["status"], document["input_complete"] = "collected_with_gaps", False
def optional_omission_source(document):
result = copy.deepcopy(document)
for row in result["records"]:
for field, reason, empty_value in (("BLOCK_CODE", "missing_reservation_block", ""),
("PRODUCTS", "missing_reservation_packages", [])):
observation = row["fields"][field]
if observation.get("state") == "missing" and observation.get("reason") == reason:
row["fields"][field] = {"state": "empty", "value": empty_value}
unresolved = any(value["state"] != "available"
and not (field in review_module.OPTIONAL and value["state"] == "empty")
for row in result["records"] for field, value in row["fields"].items())
result["input_complete"] = not unresolved
result["status"] = "collected_with_gaps" if unresolved else "collected"
result["optional_omission_policy"] = REANALYSIS_POLICY
return result
class DataFieldReviewTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
@@ -103,6 +122,247 @@ class DataFieldReviewTests(unittest.TestCase):
self.assertEqual(raised.exception.code, "ARR_DATA_REVIEW_" + code)
self.assertEqual(raised.exception.status, status)
def omission_gaps(self, document=None, sequence=1):
document = self.document if document is None else document
for field, reason in (("BLOCK_CODE", "missing_reservation_block"),
("PRODUCTS", "missing_reservation_packages")):
gap(document, field, sequence=sequence)
document["records"][sequence - 1]["fields"][field]["reason"] = reason
def reanalyze(self, review, document=None, **kwargs):
return self.service.apply_source_reanalysis(REQUEST,
raw(optional_omission_source(self.document) if document is None else document),
REANALYSIS_MANIFEST, REANALYSIS_POLICY, expected_revision=review["revision"], **kwargs)
def test_source_reanalysis_removes_only_25_optional_gaps_without_confirming_a_room(self):
template = self.document["records"][0]
self.document["records"] = []
for sequence in range(1, 19):
row = copy.deepcopy(template)
row["source_sequence"], row["reservation_id"] = sequence, f"synthetic-{sequence}"
row["fields"]["CONFIRMATION_NO"]["value"] = f"SYNTHETIC-{sequence:03d}"
row["fields"]["DISP_ROOM_NO"]["value"] = f"ROOM-{sequence:03d}"
self.document["records"].append(row)
gap(self.document, "BLOCK_CODE", sequence)
row["fields"]["BLOCK_CODE"]["reason"] = "missing_reservation_block"
if sequence <= 7:
gap(self.document, "PRODUCTS", sequence)
row["fields"]["PRODUCTS"]["reason"] = "missing_reservation_packages"
gap(self.document, "DISP_ROOM_NO")
original = raw(self.document)
review = self.prepare()
self.assertEqual(review["pending_count"], 26)
revised = self.reanalyze(review)
self.assertEqual((revised["revision"], revised["pending_count"], revised["total_count"]), (1, 1, 1))
self.assertEqual([item["field"] for item in revised["items"]], ["DISP_ROOM_NO"])
self.assertFalse(revised["can_finalize"])
self.assertIsNone(self.service.payload(REQUEST))
self.assertEqual(self.service.original(REQUEST), (original, SOURCE_MANIFEST))
directory = self.root / REQUEST
_, _, state = self.service._read(directory)
self.assertEqual(state["decisions"], {})
self.assertEqual([event["action"] for event in state["events"]], ["source_reanalysis"])
self.assertNotIn("actor", state["events"][0])
receipt = json.loads((directory / state["source_reanalysis"]["receipt_file"]).read_bytes())
self.assertEqual(len(receipt["changes"]), 25)
self.assertEqual(receipt["parent_original_sha256"], hashlib.sha256(original).hexdigest())
for path in directory.iterdir():
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
def test_reanalysis_preserves_staff_decisions_and_final_manifest_has_both_source_bindings(self):
self.omission_gaps()
gap(self.document, "DISP_ROOM_NO")
before = raw(self.document)
first = self.prepare()
staff = self.update(first, "BLOCK_CODE", "STAFF-VERIFIED-BLOCK")
payload = raw(optional_omission_source(self.document))
revised = self.reanalyze(staff)
confirmed = next(item for item in revised["items"] if item["field"] == "BLOCK_CODE")
self.assertEqual((confirmed["value"], confirmed["source_state"], confirmed["confirmed"]),
("STAFF-VERIFIED-BLOCK", "available", True))
self.assertEqual(revised["pending_count"], 1)
restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
self.assertEqual(restored.get(REQUEST), revised)
revised = restored.update(REQUEST, "1:DISP_ROOM_NO", revised["revision"], "STAFF-ROOM", ACTOR)
# A lost maintenance response can be retried even after a later staff save.
self.assertEqual(restored.apply_source_reanalysis(REQUEST, payload, REANALYSIS_MANIFEST,
REANALYSIS_POLICY, expected_revision=staff["revision"]), revised)
frozen = restored.finalize(REQUEST, revised["revision"], ACTOR)
result, binding = restored.payload(REQUEST)
self.assertEqual(restored.original(REQUEST), (before, SOURCE_MANIFEST))
derived = json.loads(result)
fields = derived["records"][0]["fields"]
self.assertEqual(fields["BLOCK_CODE"]["value"], "STAFF-VERIFIED-BLOCK")
self.assertEqual(fields["PRODUCTS"], {"state": "empty", "value": []})
audit = derived["manual_data_review"]["manifest"]
self.assertEqual(audit["original_sha256"], hashlib.sha256(before).hexdigest())
self.assertEqual(audit["source_manifest_sha256"], SOURCE_MANIFEST)
self.assertEqual(audit["source_reanalysis"]["policy_id"], REANALYSIS_POLICY)
self.assertEqual(audit["source_reanalysis"]["reanalyzed_sha256"], hashlib.sha256(payload).hexdigest())
self.assertEqual(audit["source_reanalysis"]["reanalysis_manifest_sha256"], REANALYSIS_MANIFEST)
self.assertEqual([event["action"] for event in audit["events"]],
["set_field", "source_reanalysis", "set_field", "finalize"])
self.assertEqual({change["actor"] for change in audit["changes"]}, {ACTOR})
self.assertRegex(binding, r"^[0-9a-f]{64}$")
self.assertEqual(restored.finalize(REQUEST, revised["revision"], ACTOR), frozen)
self.assert_error("FROZEN", lambda: restored.apply_source_reanalysis(REQUEST, payload,
REANALYSIS_MANIFEST, REANALYSIS_POLICY, expected_revision=frozen["revision"]), 409)
def test_reanalysis_rejects_other_source_changes_and_leaves_revision_unchanged(self):
self.document = source_document(2)
self.omission_gaps()
review = self.prepare()
valid = optional_omission_source(self.document)
mutations = [
lambda data: data.update(hotel_id="OTHER"),
lambda data: data.update(report_date="2026-09-16"),
lambda data: data.update(source_kind="OTHER"),
lambda data: data.update(collection_complete=False),
lambda data: data.update(input_complete=False),
lambda data: data.update(status="collected_with_gaps"),
lambda data: data.update(optional_omission_policy="OTHER"),
lambda data: data.update(extra="not allowed"),
lambda data: data["records"].reverse(),
lambda data: data["records"][0].update(reservation_id="OTHER"),
lambda data: data["records"][0]["related"].update(extra="not allowed"),
lambda data: data["records"][0]["fields"]["FULL_NAME"].update(value="OTHER"),
lambda data: data["records"][0]["fields"]["BLOCK_CODE"].update(reason="not allowed"),
]
for change in mutations:
with self.subTest(change=change):
document = copy.deepcopy(valid)
change(document)
with self.assertRaises(CollectionError):
self.reanalyze(review, document)
self.assertEqual(self.service.get(REQUEST), review)
self.assertEqual(list((self.root / REQUEST).glob("*reanalysis*")), [])
def test_reanalysis_never_clears_failed_ambiguous_available_or_unrecognized_observations(self):
for index, (state, reason) in enumerate((("failed", "missing_reservation_packages"),
("ambiguous", "missing_reservation_packages"), ("available", "missing_reservation_packages"),
("missing", "synthetic_field_gap")), 1):
with self.subTest(state=state, reason=reason):
document = source_document()
self.omission_gaps(document)
observation = document["records"][0]["fields"]["PRODUCTS"]
observation["state"], observation["reason"] = state, reason
if state == "available":
observation["value"] = [{"package": {"packageCode": "EXISTING"}}]
request_id = f"{index:032x}"
review = self.prepare(document, request_id)
proposed = optional_omission_source(document)
proposed["records"][0]["fields"]["PRODUCTS"] = {"state": "empty", "value": []}
proposed["input_complete"], proposed["status"] = True, "collected"
with self.assertRaisesRegex(CollectionError, "data_review_reanalysis_observation_invalid"):
self.service.apply_source_reanalysis(request_id, raw(proposed), REANALYSIS_MANIFEST,
REANALYSIS_POLICY, expected_revision=review["revision"])
self.assertEqual(self.service.get(request_id), review)
def test_reanalysis_is_revision_guarded_and_rejects_other_policy_or_evidence(self):
self.omission_gaps()
review = self.prepare()
payload = raw(optional_omission_source(self.document))
staff = self.update(review, "BLOCK_CODE", "STAFF")
self.assert_error("CONFLICT", lambda: self.reanalyze(review), 409)
with self.assertRaisesRegex(CollectionError, "data_review_reanalysis_policy_invalid"):
self.service.apply_source_reanalysis(REQUEST, payload, REANALYSIS_MANIFEST, "OTHER",
expected_revision=staff["revision"])
with self.assertRaisesRegex(CollectionError, "data_review_reanalysis_manifest_invalid"):
self.service.apply_source_reanalysis(REQUEST, payload, "invalid", REANALYSIS_POLICY,
expected_revision=staff["revision"])
current = self.reanalyze(staff)
self.assertEqual(self.reanalyze(staff), current)
self.assert_error("CONFLICT", lambda: self.service.apply_source_reanalysis(REQUEST, payload,
"f" * 64, REANALYSIS_POLICY, expected_revision=current["revision"]), 409)
self.assertEqual(self.service.get(REQUEST), current)
def test_reanalysis_publication_recovers_after_crash_without_publishing_partial_source(self):
self.omission_gaps()
review = self.prepare()
real_atomic = review_module.atomic_json
def interrupted(path, value, **kwargs):
if path.name == "state.json":
raise OSError("synthetic reanalysis pointer interruption")
return real_atomic(path, value, **kwargs)
with patch.object(review_module, "atomic_json", side_effect=interrupted):
with self.assertRaisesRegex(OSError, "synthetic reanalysis pointer interruption"):
self.reanalyze(review)
self.assertEqual(self.service.get(REQUEST), review)
staff = self.update(review, "BLOCK_CODE", "PRESERVED-AFTER-INTERRUPTION")
restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
result = restored.apply_source_reanalysis(REQUEST, raw(optional_omission_source(self.document)),
REANALYSIS_MANIFEST, REANALYSIS_POLICY, expected_revision=staff["revision"])
self.assertEqual(result["revision"], staff["revision"] + 1)
self.assertEqual(result["pending_count"], 0)
self.assertEqual(result["items"][0]["value"], "PRESERVED-AFTER-INTERRUPTION")
_, _, state = restored._read(self.root / REQUEST)
self.assertEqual([event["action"] for event in state["events"]], ["set_field", "source_reanalysis"])
def test_reanalysis_with_no_pending_items_still_requires_finalize_and_recovers_its_intent(self):
self.omission_gaps()
original = raw(self.document)
revised = self.reanalyze(self.prepare())
self.assertTrue(revised["can_finalize"])
self.assertEqual(revised["items"], [])
self.assertEqual(self.prepare(), revised, "executor cannot bypass an unfinalized reanalysis")
self.assertIsNone(self.service.payload(REQUEST))
real_atomic = review_module.atomic_json
def interrupted(path, value, **kwargs):
if path.name == "state.json":
raise OSError("synthetic finalize pointer interruption")
return real_atomic(path, value, **kwargs)
with patch.object(review_module, "atomic_json", side_effect=interrupted):
with self.assertRaisesRegex(OSError, "synthetic finalize pointer interruption"):
self.service.finalize(REQUEST, revised["revision"], ACTOR)
frozen_bytes = (self.root / REQUEST / "reviewed-source.json").read_bytes()
self.assertEqual(self.service.get(REQUEST), revised)
self.assertIsNone(self.service.payload(REQUEST))
self.assert_error("FROZEN", lambda: self.reanalyze(revised), 409)
restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
final = restored.finalize(REQUEST, revised["revision"], "different-operator@example.test")
self.assertEqual(final["status"], "finalized")
self.assertEqual(restored.payload(REQUEST)[0], frozen_bytes)
self.assertEqual(restored.original(REQUEST), (original, SOURCE_MANIFEST))
audit = json.loads(frozen_bytes)["manual_data_review"]["manifest"]
self.assertEqual(audit["changes"], [])
self.assertEqual([event["action"] for event in audit["events"]], ["source_reanalysis", "finalize"])
self.assertEqual(audit["events"][-1]["actor"], ACTOR)
def test_reanalysis_and_staff_save_share_one_revision_winner(self):
self.omission_gaps()
gap(self.document, "DISP_ROOM_NO")
review = self.prepare()
start = threading.Barrier(2)
def mutate(reanalysis):
start.wait(timeout=3)
try:
return self.reanalyze(review) if reanalysis else self.update(review, "DISP_ROOM_NO", "STAFF-ROOM")
except PortalError as error:
return error
with ThreadPoolExecutor(max_workers=2) as pool:
results = list(pool.map(mutate, [True, False]))
winners = [result for result in results if isinstance(result, dict)]
conflicts = [result for result in results if isinstance(result, PortalError)]
self.assertEqual((len(winners), len(conflicts)), (1, 1))
self.assertEqual((conflicts[0].code, conflicts[0].status), ("ARR_DATA_REVIEW_CONFLICT", 409))
self.assertEqual(self.service.get(REQUEST), winners[0])
def test_reanalysis_payload_and_receipt_tampering_are_rejected(self):
self.omission_gaps()
revised = self.reanalyze(self.prepare())
directory = self.root / REQUEST
_, _, state = self.service._read(directory)
reference = state["source_reanalysis"]
for name in (reference["file"], reference["receipt_file"]):
with self.subTest(name=name):
path = directory / name
before = path.read_bytes()
path.write_bytes(before + b" ")
with self.assertRaisesRegex(CollectionError, "data_review_reanalysis_changed"):
self.service.get(REQUEST)
path.write_bytes(before)
self.assertEqual(self.service.get(REQUEST), revised)
def test_complete_source_and_gaps_only_on_excluded_rows_need_no_review(self):
self.assertIsNone(self.prepare())
excluded = source_document()
@@ -134,6 +394,27 @@ class DataFieldReviewTests(unittest.TestCase):
self.assertEqual({item["field"] for item in review["items"]}, set(invalid))
self.assertEqual(review["pending_count"], len(invalid))
def test_review_exposes_source_failure_separately_from_optional_absence(self):
gap(self.document, "DISP_ROOM_NO")
gap(self.document, "BLOCK_CODE")
gap(self.document, "PRODUCTS", state="failed")
gap(self.document, "RES_COMMENT", state="ambiguous")
before = raw(self.document)
review = self.prepare()
items = {item["field"]: item for item in review["items"]}
self.assertEqual({field: (item["source_state"], item["can_be_empty"])
for field, item in items.items()}, {
"DISP_ROOM_NO": ("missing", False), "BLOCK_CODE": ("missing", True),
"PRODUCTS": ("failed", True), "RES_COMMENT": ("ambiguous", True),
})
self.assertEqual(review["pending_count"], 4)
self.assertFalse(review["can_finalize"])
self.assertEqual(self.service.original(REQUEST)[0], before)
confirmed = self.update(review, "BLOCK_CODE", "")
item = next(item for item in confirmed["items"] if item["field"] == "BLOCK_CODE")
self.assertEqual((item["source_state"], item["confirmed"]), ("empty", True))
self.assertEqual(confirmed["pending_count"], 3)
def test_departure_before_arrival_is_an_editable_issue(self):
self.document["records"][0]["fields"]["DEPARTURE"]["value"] = "2026-09-14"
review = self.prepare()