#!/usr/bin/env python3 """Offline review of one fictional Pipedrive enum -> ERP business-code mapping. Inputs are normalized observations, not a REST or webhook parser. Hashes detect changes; they do not authenticate an approver or infer a label's business meaning. No HTTP, storage, CRM changes, ERP writes or historical reconstruction. """ from dataclasses import dataclass, replace, asdict import hashlib import json import re @dataclass(frozen=True) class Option: id: int label: str @dataclass(frozen=True) class Field: company: str entity: str metadata_id: int key: str name: str kind: str options: tuple[Option, ...] complete: bool = True transport_qualified: bool = True @dataclass(frozen=True) class Approval: company: str entity: str key: str schema_sha256: str version: str rules: tuple[tuple[int, str], ...] def positive_id(value): return type(value) is int and value > 0 def valid_text(value): return isinstance(value, str) and bool(value.strip()) def fingerprint(field): """Return a local content digest, or None for inputs outside this profile.""" if (not valid_text(field.company) or field.entity != "deal" or not positive_id(field.metadata_id) or not isinstance(field.key, str) or re.fullmatch(r"[0-9a-f]{40}", field.key) is None or not valid_text(field.name) or field.kind != "enum" or field.complete is not True or field.transport_qualified is not True or not isinstance(field.options, tuple) or not field.options): return None ids = [] for option in field.options: if not isinstance(option, Option) or not positive_id(option.id) or not valid_text(option.label): return None ids.append(option.id) if len(set(ids)) != len(ids): return None canonical = { "company": field.company, "entity": field.entity, "metadata_id": field.metadata_id, "key": field.key, "name": field.name, "kind": field.kind, "options": [{"id": x.id, "label": x.label} for x in sorted(field.options, key=lambda x: x.id)] } raw = json.dumps(canonical, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() return hashlib.sha256(raw).hexdigest() def bind(field, rules, version, reviewed, erp_codes): digest = fingerprint(field) if (digest is None or reviewed is not True or not valid_text(version) or not isinstance(rules, tuple) or not rules): return None native_ids = {x.id for x in field.options} rule_ids = [] for rule in rules: if (not isinstance(rule, tuple) or len(rule) != 2 or not positive_id(rule[0]) or rule[0] not in native_ids or not valid_text(rule[1]) or rule[1] not in erp_codes): return None rule_ids.append(rule[0]) if len(set(rule_ids)) != len(rule_ids): return None return Approval(field.company, field.entity, field.key, digest, version, rules) def prepare(field, approval, selected_id, current_version, erp_codes): held = lambda reason: {"decision": "held", "reason": reason} digest = fingerprint(field) if digest is None: return held("unqualified_field") if not isinstance(approval, Approval): return held("approval_missing") if (field.company, field.entity, field.key) != (approval.company, approval.entity, approval.key): return held("scope_changed") if digest != approval.schema_sha256: return held("dictionary_changed") if current_version != approval.version: return held("mapping_version_changed") if not positive_id(selected_id): return held("selection_unknown_or_outside_profile") if selected_id not in {x.id for x in field.options}: return held("option_unknown") rules = dict(approval.rules) if selected_id not in rules: return held("option_unmapped") code = rules[selected_id] if code not in erp_codes: return held("erp_code_unavailable") return {"decision": "prepared", "company": field.company, "field_key": field.key, "option_id": selected_id, "erp_value": code, "mapping_version": approval.version, "schema_sha256": digest} def run_checks(): names = [] def check(name, condition): if name in names or not condition: raise AssertionError(name) names.append(name) field = Field("COMPANY-A", "deal", 701, "a" * 40, "Canal commercial", "enum", (Option(8, "Revendeur"), Option(9, "Vente directe"))) erp_codes = frozenset({"B2B-REV", "B2B-DIR", "B2B-PART"}) rules = ((8, "B2B-REV"), (9, "B2B-DIR")) approval = bind(field, rules, "mapping-1", True, erp_codes) plan = lambda f=field, a=approval, v=8, version="mapping-1", codes=erp_codes: prepare(f, a, v, version, codes) initial = plan() renamed = replace(field, options=(Option(8, "Partenaire"), Option(9, "Vente directe"))) renewed = bind(renamed, rules, "mapping-2", True, erp_codes) # Keeping or changing the ERP code are supplied review decisions, not guesses. reclassified = bind(renamed, ((8, "B2B-PART"), (9, "B2B-DIR")), "mapping-3", True, erp_codes) check("baseline_prepares_business_code", initial["erp_value"] == "B2B-REV") check("other_option_has_its_own_code", plan(v=9)["erp_value"] == "B2B-DIR") check("unchanged_snapshot_is_repeatable", plan() == initial) check("option_reordering_keeps_fingerprint", fingerprint(replace(field, options=tuple(reversed(field.options)))) == fingerprint(field)) check("label_rename_changes_fingerprint", fingerprint(renamed) != fingerprint(field)) check("same_id_new_label_holds_old_approval", plan(f=renamed)["reason"] == "dictionary_changed") check("review_can_keep_previous_business_code", plan(f=renamed, a=renewed, version="mapping-2")["erp_value"] == "B2B-REV") check("review_can_choose_other_business_code", plan(f=renamed, a=reclassified, version="mapping-3")["erp_value"] == "B2B-PART") check("old_mapping_version_cannot_use_new_approval", plan(f=renamed, a=renewed)["reason"] == "mapping_version_changed") check("current_version_rejects_old_approval", plan(version="mapping-2")["reason"] == "mapping_version_changed") check("review_false_cannot_bind", bind(field, rules, "mapping-1", False, erp_codes) is None) check("truthy_text_is_not_review", bind(field, rules, "mapping-1", "yes", erp_codes) is None) check("missing_approval_holds", plan(a=None)["reason"] == "approval_missing") check("other_company_isolated", plan(f=replace(field, company="COMPANY-B"))["reason"] == "scope_changed") check("other_key_isolated", plan(f=replace(field, key="b" * 40))["reason"] == "scope_changed") check("other_entity_outside_profile", plan(f=replace(field, entity="person"))["reason"] == "unqualified_field") check("metadata_identity_change_requires_review", plan(f=replace(field, metadata_id=702))["reason"] == "dictionary_changed") check("field_name_change_requires_review", plan(f=replace(field, name="Canal facturable"))["reason"] == "dictionary_changed") check("set_not_silently_reduced_to_enum", plan(f=replace(field, kind="set"))["reason"] == "unqualified_field") check("incomplete_snapshot_holds", plan(f=replace(field, complete=False))["reason"] == "unqualified_field") check("unqualified_transport_holds", plan(f=replace(field, transport_qualified=False))["reason"] == "unqualified_field") check("missing_value_does_not_default", plan(v=None)["reason"] == "selection_unknown_or_outside_profile") check("boolean_is_not_option_id", plan(v=True)["reason"] == "selection_unknown_or_outside_profile") check("string_id_requires_adapter_normalization", plan(v="8")["reason"] == "selection_unknown_or_outside_profile") check("unknown_option_has_no_fallback", plan(v=999)["reason"] == "option_unknown") partial = bind(field, ((8, "B2B-REV"),), "partial", True, erp_codes) check("unmapped_known_option_is_held", plan(a=partial, v=9, version="partial")["reason"] == "option_unmapped") check("erp_code_retirement_is_seen", plan(codes=frozenset({"B2B-DIR"}))["reason"] == "erp_code_unavailable") check("unknown_erp_code_cannot_bind", bind(field, ((8, "NOT-A-CODE"),), "x", True, erp_codes) is None) check("duplicate_mapping_ids_cannot_bind", bind(field, ((8, "B2B-REV"), (8, "B2B-DIR")), "x", True, erp_codes) is None) check("mapping_to_absent_option_cannot_bind", bind(field, ((99, "B2B-REV"),), "x", True, erp_codes) is None) check("empty_mapping_cannot_bind", bind(field, (), "x", True, erp_codes) is None) check("blank_mapping_version_cannot_bind", bind(field, rules, " ", True, erp_codes) is None) duplicate_ids = replace(field, options=(Option(8, "Revendeur"), Option(8, "Partenaire"))) check("duplicate_schema_ids_hold", fingerprint(duplicate_ids) is None) duplicate_labels = replace(field, options=(Option(8, "Même libellé"), Option(9, "Même libellé"))) label_approval = bind(duplicate_labels, rules, "labels", True, erp_codes) check("labels_do_not_replace_option_identity", plan(f=duplicate_labels, a=label_approval, v=9, version="labels")["erp_value"] == "B2B-DIR") added = replace(field, options=field.options + (Option(10, "Autre"),)) check("new_option_requires_field_review", plan(f=added)["reason"] == "dictionary_changed") removed = replace(field, options=(field.options[1],)) check("removed_option_requires_review", plan(f=removed)["reason"] == "dictionary_changed") recreated = replace(field, options=(Option(18, "Revendeur"), Option(9, "Vente directe"))) check("same_label_new_id_requires_review", plan(f=recreated)["reason"] == "dictionary_changed") check("empty_dictionary_is_unqualified", fingerprint(replace(field, options=())) is None) check("blank_label_is_unqualified", fingerprint(replace(field, options=(Option(8, " "),))) is None) check("boolean_metadata_id_is_unqualified", fingerprint(replace(field, metadata_id=True)) is None) check("noncustom_key_is_outside_profile", fingerprint(replace(field, key="label")) is None) check("field_input_not_mutated", field.options[0].label == "Revendeur" and field.name == "Canal commercial") check("approval_input_not_mutated", approval.rules == rules and approval.version == "mapping-1") return {"checks_passed": len(names), "check_names": names, "scope": {"http_calls": 0, "crm_operations": 0, "erp_operations": 0, "storage_writes": 0, "native_transport_verified": False, "business_meaning_inferred": False}, "examples": {"initial": initial, "renamed_without_review": plan(f=renamed), "review_keeps_code": plan(f=renamed, a=renewed, version="mapping-2"), "review_changes_code": plan(f=renamed, a=reclassified, version="mapping-3"), "approved_mapping": asdict(approval)}} if __name__ == "__main__": print(json.dumps(run_checks(), ensure_ascii=False, indent=2))