#!/usr/bin/env python3 """Offline Composite/Collections decision experiment, Python 3.9+. The fixtures are invented, already-decoded Python objects. CRM labels are not valid Salesforce IDs. The plan and its declared independent follow-up are local contracts, not a Salesforce request schema. Only temporary SQLite is written. No HTTP, CRM/ERP write, native rollback, JSON decoder or API version is tested. """ from copy import deepcopy from dataclasses import asdict, dataclass, replace from hashlib import sha256 from pathlib import Path from tempfile import TemporaryDirectory import json import re import sqlite3 class ReviewRequired(ValueError): pass class LedgerConflict(ValueError): pass @dataclass(frozen=True) class Plan: outer_all_or_none: bool inner_all_or_none: bool erp_references: tuple = ("CLI-A", "CLI-B") org_label: str = "recette-A" collection_ref: str = "accounts" following_ref: str = "independent" following_independent: bool = True profile: str = "account-link-v1" def validate_plan(plan): if not isinstance(plan, Plan) or plan.profile != "account-link-v1": raise ReviewRequired("plan_profile") if type(plan.outer_all_or_none) is not bool or type(plan.inner_all_or_none) is not bool: raise ReviewRequired("explicit_boolean_flags") if plan.following_independent is not True: raise ReviewRequired("dependent_followup_outside_profile") if not isinstance(plan.org_label, str) or not plan.org_label.strip(): raise ReviewRequired("org_scope") refs = plan.erp_references if not isinstance(refs, tuple) or len(refs) != 2 or any( not isinstance(ref, str) or not ref.strip() for ref in refs ) or len(set(refs)) != 2: raise ReviewRequired("two_distinct_erp_references") names = (plan.collection_ref, plan.following_ref) if any(not isinstance(ref, str) or not re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9_]{0,79}", ref ) for ref in names) or len(set(names)) != 2: raise ReviewRequired("subrequest_names") def save_result(row): if not isinstance(row, dict) or type(row.get("success")) is not bool: raise ReviewRequired("success_type") errors = row.get("errors") if not isinstance(errors, list) or any( not isinstance(error, dict) or not isinstance(error.get("statusCode"), str) or not error["statusCode"] for error in errors ): raise ReviewRequired("record_errors") ident = row.get("id") if row["success"]: if not isinstance(ident, str) or not ident.strip() or errors: raise ReviewRequired("inconsistent_success") elif not errors or ident is not None: raise ReviewRequired("inconsistent_failure") return row["success"], ident, [error["statusCode"] for error in errors] def decide(plan, reply, complete=True, http_status=200): validate_plan(plan) if complete is not True or type(http_status) is not int or http_status != 200: raise ReviewRequired("transport_not_qualified") if not isinstance(reply, dict) or not isinstance(reply.get("compositeResponse"), list): raise ReviewRequired("response_missing") parts = reply["compositeResponse"] if len(parts) != 2 or any(not isinstance(part, dict) or not isinstance( part.get("referenceId"), str) for part in parts): raise ReviewRequired("subrequest_shape") by_ref = {part["referenceId"]: part for part in parts} if len(by_ref) != 2 or set(by_ref) != {plan.collection_ref, plan.following_ref}: raise ReviewRequired("subrequest_correlation") collection, following = by_ref[plan.collection_ref], by_ref[plan.following_ref] if type(collection.get("httpStatusCode")) is not int or collection["httpStatusCode"] != 200: raise ReviewRequired("collection_not_qualified") body = collection.get("body") if not isinstance(body, list) or len(body) != len(plan.erp_references): raise ReviewRequired("ordered_results_missing") results = [save_result(row) for row in body] record_failure = any(not success for success, _, _ in results) if plan.inner_all_or_none and record_failure and any(success for success, _, _ in results): raise ReviewRequired("contradictory_atomic_collection") status = following.get("httpStatusCode") if type(status) is not int: raise ReviewRequired("following_status") if status == 201: if not save_result(following.get("body"))[0]: raise ReviewRequired("following_not_created") following_failure = False elif status == 400: errors = following.get("body") if not isinstance(errors, list) or not errors or any( not isinstance(error, dict) or not isinstance(error.get("errorCode"), str) or not error["errorCode"] for error in errors ): raise ReviewRequired("following_errors") if not plan.outer_all_or_none and any( error["errorCode"] == "PROCESSING_HALTED" for error in errors ): raise ReviewRequired("halt_outside_declared_scope") following_failure = True else: raise ReviewRequired("following_not_final_in_profile") failure = record_failure or following_failure outer_rollback = plan.outer_all_or_none and failure outcome = ("outer_rollback" if outer_rollback else "collection_rollback" if plan.inner_all_or_none and record_failure else "partial" if failure else "complete") rows = [] for ref, (success, ident, codes) in zip(plan.erp_references, results): state = ("rolled_back" if outer_rollback else "kept") if success else ( "rolled_back" if "ALL_OR_NONE_OPERATION_ROLLED_BACK" in codes else "rejected") rows.append({"erp_reference": ref, "observed_id": ident, "decision": state, "error_codes": codes}) return {"outcome": outcome, "records": rows} def fingerprint(value): # Fingerprint of the local decoded value, NOT of original HTTP bytes. raw = json.dumps(value, sort_keys=True, ensure_ascii=True, allow_nan=False, separators=(",", ":")).encode() return sha256(raw).hexdigest() def connect(path): db = sqlite3.connect(str(path), isolation_level=None) db.execute("PRAGMA foreign_keys=ON") return db def initialize(path): db = connect(path) try: db.executescript(""" CREATE TABLE observations ( observation_id TEXT PRIMARY KEY, plan_hash TEXT NOT NULL, response_hash TEXT NOT NULL, decision_json TEXT NOT NULL ); CREATE TABLE account_links ( org_label TEXT NOT NULL, erp_reference TEXT NOT NULL, crm_id TEXT NOT NULL, source_observation TEXT NOT NULL REFERENCES observations(observation_id), PRIMARY KEY(org_label, erp_reference), UNIQUE(org_label, crm_id) ); """) finally: db.close() def persist(path, observation_id, plan, reply, complete=True, http_status=200): validate_plan(plan) if not isinstance(observation_id, str) or not re.fullmatch(r"[a-z][a-z0-9-]{0,39}", observation_id): raise ReviewRequired("observation_identity") try: decision = decide(plan, reply, complete, http_status) except ReviewRequired as error: decision = {"outcome": "review", "reason": str(error), "records": [ {"erp_reference": ref, "observed_id": None, "decision": "review", "error_codes": []} for ref in plan.erp_references]} ph = fingerprint(asdict(plan)) rh = fingerprint({"http_status": http_status, "complete": complete, "reply": reply}) db = connect(path) try: db.execute("BEGIN IMMEDIATE") old = db.execute("SELECT plan_hash,response_hash,decision_json FROM observations " "WHERE observation_id=?", (observation_id,)).fetchone() if old: if old[:2] != (ph, rh): raise LedgerConflict("observation_changed") db.commit() return json.loads(old[2]) db.execute("INSERT INTO observations VALUES(?,?,?,?)", (observation_id, ph, rh, json.dumps(decision, sort_keys=True))) for row in decision["records"]: if row["decision"] != "kept": continue old_link = db.execute("SELECT crm_id FROM account_links WHERE org_label=? " "AND erp_reference=?", (plan.org_label, row["erp_reference"])).fetchone() if old_link and old_link[0] != row["observed_id"]: raise LedgerConflict("existing_link_conflict") if not old_link: db.execute("INSERT INTO account_links VALUES(?,?,?,?)", (plan.org_label, row["erp_reference"], row["observed_id"], observation_id)) db.commit() return decision except sqlite3.IntegrityError as error: db.rollback() raise LedgerConflict("local_one_to_one_constraint") from error except BaseException: db.rollback() raise finally: db.close() def snapshot(path): db = connect(path) try: return {"observations": db.execute("SELECT * FROM observations ORDER BY observation_id").fetchall(), "links": db.execute("SELECT * FROM account_links ORDER BY org_label,erp_reference").fetchall()} finally: db.close() def good(ident): return {"success": True, "id": ident, "errors": []} def bad(code): return {"success": False, "errors": [{"statusCode": code}]} def reply(records=None, following_error=None): following = ({"httpStatusCode": 400, "body": [{"errorCode": following_error}]} if following_error else {"httpStatusCode": 201, "body": good("contact-local")}) return {"compositeResponse": [ {"referenceId": "accounts", "httpStatusCode": 200, "body": records if records is not None else [good("crm-A"), good("crm-B")]}, dict(following, referenceId="independent")]} def main(): checks, matrix = [], [] def check(name, condition): if not condition: raise AssertionError(name) checks.append(name) def raises(name, cls, operation): try: operation() except cls: checks.append(name) else: raise AssertionError(name) with TemporaryDirectory() as directory: root = Path(directory) for outer, inner, outcome, first, links in [ (False, False, "partial", "kept", 1), (False, True, "collection_rollback", "rolled_back", 0), (True, False, "outer_rollback", "rolled_back", 0), (True, True, "outer_rollback", "rolled_back", 0), ]: plan = Plan(outer, inner) records = [bad("ALL_OR_NONE_OPERATION_ROLLED_BACK") if inner else good("crm-A"), bad("DUPLICATES_DETECTED")] value = reply(records, "PROCESSING_HALTED" if outer else None) path = root / (str(int(outer)) + str(int(inner)) + ".sqlite") initialize(path) observed = persist(path, "matrix", plan, value) check("matrix_decision_" + path.stem, observed["outcome"] == outcome and [r["decision"] for r in observed["records"]] == [first, "rejected"]) check("matrix_persisted_links_" + path.stem, len(snapshot(path)["links"]) == links) matrix.append({"outer": outer, "inner": inner, "outcome": outcome, "first_observed_id": observed["records"][0]["observed_id"], "active_links": links}) for outer, inner in [(False, False), (False, True), (True, False), (True, True)]: d = decide(Plan(outer, inner), reply()) check("all_success_" + str(int(outer)) + str(int(inner)), d["outcome"] == "complete" and all(r["decision"] == "kept" for r in d["records"])) check("independent_later_failure_keeps_accounts_when_outer_false", all(r["decision"] == "kept" for r in decide(Plan(False, True), reply(following_error="REQUIRED_FIELD_MISSING"))["records"])) check("later_failure_rolls_back_successful_accounts_when_outer_true", all(r["decision"] == "rolled_back" for r in decide(Plan(True, False), reply(following_error="REQUIRED_FIELD_MISSING"))["records"])) reversed_parts = reply(); reversed_parts["compositeResponse"].reverse() check("outer_correlation_uses_reference_not_array_position", decide(Plan(False, False), reversed_parts) == decide(Plan(False, False), reply())) path = root / "ledger.sqlite"; initialize(path); plan = Plan(False, False) original = reply(); before_input = deepcopy(original) persist(path, "accepted", plan, original); baseline = snapshot(path) check("accepted_links_survive_reopening", len(baseline["links"]) == 2) check("inputs_are_not_modified", original == before_input) persist(path, "accepted", plan, original) check("identical_observation_replay_has_no_new_effect", snapshot(path) == baseline) raises("changed_observation_plan_is_refused", LedgerConflict, lambda: persist(path, "accepted", replace(plan, outer_all_or_none=True), original)) raises("changed_observation_response_is_refused", LedgerConflict, lambda: persist(path, "accepted", plan, reply(following_error="VALIDATION_ERROR"))) check("changed_observation_does_not_overwrite_ledger", snapshot(path) == baseline) persist(path, "unknown", plan, None) check("unknown_response_is_durably_held_without_deleting_prior_links", snapshot(path)["links"] == baseline["links"] and next(json.loads(row[3])["outcome"] for row in snapshot(path)["observations"] if row[0] == "unknown") == "review") persist(path, "rolled-back", Plan(True, False), reply( [good("crm-other"), bad("DUPLICATES_DETECTED")], "PROCESSING_HALTED")) check("rollback_observation_does_not_remove_previous_mapping", snapshot(path)["links"] == baseline["links"]) check("rolled_back_candidate_id_stays_in_observation_only", any( row[0] == "rolled-back" and json.loads(row[3])["records"][0]["observed_id"] == "crm-other" for row in snapshot(path)["observations"])) conflict_plan = replace(plan, erp_references=("CLI-NEW", "CLI-B")) before_conflict = snapshot(path) raises("late_link_conflict_is_refused", LedgerConflict, lambda: persist(path, "late-conflict", conflict_plan, reply([good("crm-new"), good("crm-conflicting-B")]))) check("late_link_conflict_rolls_back_first_link_and_observation", snapshot(path) == before_conflict) persist(path, "other-org", replace(plan, org_label="recette-B"), original) check("same_references_in_another_declared_org_are_separate", len(snapshot(path)["links"]) == 4) before_duplicate = snapshot(path) raises("local_one_to_one_policy_refuses_two_erp_refs_for_one_crm_id", LedgerConflict, lambda: persist(path, "duplicate-crm", replace(plan, erp_references=("CLI-C", "CLI-D")), reply([good("crm-shared"), good("crm-shared")]))) check("one_to_one_conflict_rolls_back_observation_and_first_link", snapshot(path) == before_duplicate) nullable_failure = dict(bad("VALIDATION_ERROR"), id=None) check("failed_result_accepts_an_explicit_null_id_as_well_as_absence", decide(plan, reply([nullable_failure, good("crm-B")]))["records"][0]["decision"] == "rejected") malformed = [] def variant(name, mutation): value = reply(); mutation(value); malformed.append((name, value)) variant("unknown_subrequest_ref", lambda v: v["compositeResponse"][0].update(referenceId="Accounts")) variant("duplicate_subrequest_ref", lambda v: v["compositeResponse"][1].update(referenceId="accounts")) variant("missing_subrequest", lambda v: v["compositeResponse"].pop()) variant("missing_ordered_record_result", lambda v: v["compositeResponse"][0]["body"].pop()) variant("integer_success_is_not_boolean", lambda v: v["compositeResponse"][0]["body"][0].update(success=1)) variant("empty_success_id", lambda v: v["compositeResponse"][0]["body"][0].update(id="")) variant("success_with_errors", lambda v: v["compositeResponse"][0]["body"][0].update(errors=[{"statusCode":"BAD"}])) variant("failure_without_errors", lambda v: v["compositeResponse"][0]["body"].__setitem__(0, {"success":False,"errors":[]})) variant("failed_result_cannot_publish_an_id", lambda v: v["compositeResponse"][0]["body"].__setitem__(0, dict(bad("BAD"), id="crm-unconfirmed"))) variant("save_result_errorcode_is_not_statuscode", lambda v: v["compositeResponse"][0]["body"].__setitem__(0, {"success":False,"errors":[{"errorCode":"BAD"}]})) variant("subrequest_statuscode_is_not_errorcode", lambda v: v["compositeResponse"][1].update(httpStatusCode=400,body=[{"statusCode":"BAD"}])) variant("unqualified_collection_status", lambda v: v["compositeResponse"][0].update(httpStatusCode=400)) variant("following_not_final", lambda v: v["compositeResponse"][1].update(httpStatusCode=202)) for name, value in malformed: raises(name, ReviewRequired, lambda value=value: decide(plan, value)) raises("contradictory_atomic_collection", ReviewRequired, lambda: decide( Plan(False, True), reply([good("crm-A"), bad("DUPLICATES_DETECTED")]))) raises("incomplete_transport_is_not_success", ReviewRequired, lambda: decide(plan, reply(), complete=False)) raises("http_boolean_is_not_status", ReviewRequired, lambda: decide(plan, reply(), http_status=True)) raises("halt_requires_qualified_outer_scope", ReviewRequired, lambda: decide(plan, reply(following_error="PROCESSING_HALTED"))) for name, changed in [("unknown_profile", replace(plan, profile="v2")), ("non_boolean_flag", replace(plan, outer_all_or_none="true")), ("dependent_followup", replace(plan, following_independent=False)), ("non_text_subrequest_ref", replace(plan, collection_ref=[])), ("duplicate_subrequest_plan_names", replace(plan, following_ref="accounts"))]: raises(name, ReviewRequired, lambda changed=changed: decide(changed, reply())) print(json.dumps({"checks":len(checks), "check_names":checks, "matrix":matrix, "http_calls":0, "native_salesforce_executions":0, "crm_writes":0, "erp_writes":0, "temporary_sqlite_writes":True}, indent=2)) if __name__ == "__main__": main()