import re import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] REGISTRY_MACRO = ( ROOT / "macros/digital_bi_dbt/core/customer_identity_key_registry.sql" ).read_text() ORCHESTRATION_MACRO = ( ROOT / "macros/digital_bi_dbt/core/customer_identity_daily_orchestration.sql" ).read_text() def macro_body(source: str, name: str) -> str: match = re.search( rf"\{{% macro {re.escape(name)}\b.*?%\}}(.*?)\{{% endmacro %\}}", source, flags=re.DOTALL, ) if not match: raise AssertionError(f"macro {name} not found") return match.group(1) class CustomerIdentityDailyOrchestrationContractTest(unittest.TestCase): def test_fingerprint_contract_is_canonical_lowercase_hex(self) -> None: fingerprint_expression = macro_body( REGISTRY_MACRO, "_customer_identity_registration_payload_fingerprint_expression", ) self.assertIn("LOWER(TO_HEX(SHA256(COALESCE(", fingerprint_expression) normalizer = macro_body( REGISTRY_MACRO, "_normalize_customer_identity_registration_payload_fingerprint", ) self.assertIn("| string | trim | lower", normalizer) self.assertIn("^[a-f0-9]{64}$", normalizer) self.assertNotIn("| upper", normalizer) fingerprint_pattern = re.compile(r"^[a-f0-9]{64}$") lowercase_digest = "8c9eabac325193f8765d339a061ba480b4e76a248b9ac6c619e2a8e98e1c9726" uppercase_digest = lowercase_digest.upper() self.assertRegex(lowercase_digest.strip().lower(), fingerprint_pattern) self.assertRegex(uppercase_digest.strip().lower(), fingerprint_pattern) self.assertEqual(lowercase_digest, uppercase_digest.lower()) def test_malformed_fingerprints_remain_rejected(self) -> None: fingerprint_pattern = re.compile(r"^[a-f0-9]{64}$") malformed = ( "", "abc", "g" * 64, "0" * 63, "0" * 65, "0" * 63 + "-", ) for value in malformed: with self.subTest(value=value): self.assertIsNone(fingerprint_pattern.fullmatch(value.strip().lower())) def test_dry_run_fingerprint_round_trips_through_both_apply_gates(self) -> None: audit = macro_body( ORCHESTRATION_MACRO, "_customer_identity_daily_audit_query" ) entrypoint = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) writer = macro_body( REGISTRY_MACRO, "_reconcile_customer_identity_key_registry" ) public_writer = macro_body( REGISTRY_MACRO, "register_customer_identity_key_candidates" ) fingerprint_helper = ( "_customer_identity_registration_payload_fingerprint_expression()" ) normalizer = ( "_normalize_customer_identity_registration_payload_fingerprint(" ) self.assertGreaterEqual(audit.count(fingerprint_helper), 2) self.assertIn(normalizer, entrypoint) self.assertIn(normalizer, public_writer) self.assertIn(normalizer, writer) self.assertIn( "audit['registration_payload_fingerprint'] != approved_fingerprint", entrypoint, ) self.assertIn( "ASSERT current_registration_payload_fingerprint = '{{ approved_fingerprint }}'", writer, ) def test_payload_set_change_still_blocks_before_registry_dml(self) -> None: writer = macro_body( REGISTRY_MACRO, "_reconcile_customer_identity_key_registry" ) entrypoint = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) pre_apply_fingerprint_gate = entrypoint.index( "audit['registration_payload_fingerprint'] != approved_fingerprint" ) snapshot_or_registration = entrypoint.index( "snapshot_customer_identity_key_registry" ) transactional_fingerprint_gate = writer.index( "ASSERT current_registration_payload_fingerprint = '{{ approved_fingerprint }}'" ) merge = writer.index("MERGE {{ registry_relation }}") self.assertLess(pre_apply_fingerprint_gate, snapshot_or_registration) self.assertLess(transactional_fingerprint_gate, merge) def test_entrypoint_defaults_to_read_only_dry_run(self) -> None: signature = re.search( r"macro orchestrate_customer_identity_registry_daily\((.*?)\) %}", ORCHESTRATION_MACRO, flags=re.DOTALL, ) self.assertIsNotNone(signature) self.assertIn("mode='dry_run'", signature.group(1)) body = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) dry_run_branch, apply_branch = body.rsplit("{% else %}", maxsplit=1) self.assertIn("registry_writes=0", dry_run_branch) self.assertNotIn("snapshot_customer_identity_key_registry", dry_run_branch) self.assertNotIn("register_customer_identity_key_candidates", dry_run_branch) self.assertIn("snapshot_customer_identity_key_registry", apply_branch) self.assertIn("register_customer_identity_key_candidates", apply_branch) def test_apply_requires_run_id_and_exact_confirmation(self) -> None: body = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) self.assertIn("apply mode requires an explicit run_id", body) self.assertIn( "apply mode requires an approved non-negative expected_insert_count", body ) self.assertIn( "apply mode requires the approved 64-character registration payload fingerprint", body, ) self.assertIn("confirm_apply != 'APPLY_APPEND_ONLY'", body) self.assertIn("DBT_CLOUD_RUN_ID", ORCHESTRATION_MACRO) def test_apply_binds_pre_audit_and_insert_to_approved_count(self) -> None: body = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) self.assertIn( "audit['expected_insert_count'] != approved_expected_insert_count", body ) self.assertIn( "inserted_row_count != approved_expected_insert_count", body ) self.assertIn( "audit['registration_payload_fingerprint'] != approved_fingerprint", body, ) writer = macro_body( REGISTRY_MACRO, "_reconcile_customer_identity_key_registry" ) approved_assert = writer.index( "ASSERT expected_insert_count = {{ approved_expected_insert_count | int }}" ) fingerprint_assert = writer.index( "ASSERT current_registration_payload_fingerprint = '{{ approved_fingerprint }}'" ) merge = writer.index("MERGE {{ registry_relation }}") self.assertLess(approved_assert, merge) self.assertLess(fingerprint_assert, merge) fingerprint_helper = "_customer_identity_registration_payload_fingerprint_expression()" self.assertIn(fingerprint_helper, ORCHESTRATION_MACRO) self.assertIn(fingerprint_helper, writer) def test_registry_dml_is_insert_only(self) -> None: body = macro_body( REGISTRY_MACRO, "_reconcile_customer_identity_key_registry" ) self.assertIn("BEGIN TRANSACTION", body) self.assertIn("WHEN NOT MATCHED THEN", body) self.assertIn("ASSERT @@row_count = expected_insert_count", body) self.assertIn("approved_expected_insert_count", body) self.assertIsNone(re.search(r"\bWHEN\s+MATCHED\b", body, re.IGNORECASE)) self.assertIsNone(re.search(r"\bUPDATE\b", body, re.IGNORECASE)) self.assertIsNone(re.search(r"\bDELETE\b", body, re.IGNORECASE)) public_writer = macro_body( REGISTRY_MACRO, "register_customer_identity_key_candidates" ) self.assertIn( "requires an approved non-negative expected_insert_count", public_writer ) self.assertIn( "requires an approved registration payload fingerprint", public_writer ) def test_snapshot_is_not_silently_recreated_and_precedes_registration(self) -> None: snapshot = macro_body( REGISTRY_MACRO, "snapshot_customer_identity_key_registry" ) self.assertIn("CREATE SNAPSHOT TABLE {{ snapshot_relation }}", snapshot) self.assertNotIn("IF NOT EXISTS", snapshot) entrypoint = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) snapshot_position = entrypoint.index( "snapshot_customer_identity_key_registry" ) registration_position = entrypoint.index( "register_customer_identity_key_candidates" ) self.assertLess(snapshot_position, registration_position) def test_used_run_id_requires_original_snapshot(self) -> None: body = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) self.assertIn( "audit['registry_rows_for_run_id'] > 0 and not audit['snapshot_exists']", body, ) self.assertIn( "run_id already owns registry rows but its original preimage snapshot is missing or expired", body, ) writer = macro_body( REGISTRY_MACRO, "_reconcile_customer_identity_key_registry" ) preimage_assert = writer.index("approved registry preimage snapshot is missing") merge = writer.index("MERGE {{ registry_relation }}") self.assertLess(preimage_assert, merge) def test_exact_apply_command_is_idempotent_only_for_original_payload(self) -> None: body = macro_body( ORCHESTRATION_MACRO, "orchestrate_customer_identity_registry_daily" ) self.assertIn( "audit['registry_rows_for_run_id'] != approved_expected_insert_count", body, ) self.assertIn( "audit['registry_run_payload_fingerprint'] != approved_fingerprint", body, ) self.assertIn("{% set inserted_row_count = 0 %}", body) self.assertIn("idempotent customer identity retry confirmed", body) def test_audit_exposes_required_zero_mutation_contract(self) -> None: audit = macro_body( ORCHESTRATION_MACRO, "_customer_identity_daily_audit_query" ) for field in ( "missing_account_count", "missing_tenant_count", "expected_insert_count", "registration_payload_fingerprint", "registry_run_payload_fingerprint", "drift_total_count", "invalid_item_count", "planned_update_count", "planned_delete_count", "planned_rebinding_count", "snapshot_exists", ): self.assertIn(field, audit) if __name__ == "__main__": unittest.main()