import copy import json import subprocess import tempfile import unittest from dataclasses import dataclass from pathlib import Path from deployments.offline.runtime.reconcile import ( layout_bundle_phase0_live_inventory as live_inventory, ) from deployments.offline.runtime.reconcile import layout_bundle_phase0_review as review from deployments.offline.runtime.reconcile import ( layout_bundle_phase0_source_inventory as source_inventory, ) DOWNSTREAM_COMMIT = "1" * 40 UPSTREAM_COMMIT = "2" * 40 OFFICIAL_COMMIT = "3" * 40 FINAL_PACK_SHA256 = "4" * 64 COLLECTOR_BYTES = b"phase0-live-collector\n" class SourceCheckoutBindingTest(unittest.TestCase): def test_exact_checkout_rejects_untracked_files(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) subprocess.run(["git", "init", "-q", str(root)], check=True) subprocess.run( ["git", "-C", str(root), "config", "user.email", "test@example.invalid"], check=True, ) subprocess.run( ["git", "-C", str(root), "config", "user.name", "Test"], check=True, ) (root / "tracked.txt").write_text("tracked\n") subprocess.run(["git", "-C", str(root), "add", "tracked.txt"], check=True) subprocess.run(["git", "-C", str(root), "commit", "-qm", "fixture"], check=True) commit = subprocess.run( ["git", "-C", str(root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() self.assertEqual( root.resolve(), source_inventory.require_exact_checkout(root, commit, "fixture"), ) (root / "untracked.txt").write_text("untracked\n") with self.assertRaisesRegex(source_inventory.InventoryError, "not clean"): source_inventory.require_exact_checkout(root, commit, "fixture") def _artifact_bytes(document: object) -> bytes: return ( json.dumps(document, indent=2, sort_keys=True, ensure_ascii=False) + "\n" ).encode("utf-8") def _embed_digest(document: dict, field: str) -> None: unsigned = copy.deepcopy(document) unsigned.pop(field, None) document[field] = "sha256:" + review.sha256_bytes(review.canonical_bytes(unsigned)) def _source_document(payload_raw: bytes) -> dict: payload = json.loads(payload_raw) projection = { "objectApiName": payload["objectApiName"], "layoutName": payload["layoutName"], "type": payload["type"], "isDefault": payload["isDefault"], "legacyLayoutTuple": [ payload["objectApiName"].casefold(), payload["layoutName"].casefold(), ], } component = { "kind": "page_layout", "businessKey": "Account/default", "manageability": "defaults_only", "path": "layouts/account.json", "dependencies": [], "manifestIndex": 0, "normalizedDeclarationSha256": "5" * 64, "payloadRawSha256": review.sha256_bytes(payload_raw), "payloadSizeBytes": len(payload_raw), "legacyLayoutProjection": projection, } comparison = { "kind": "page_layout", "businessKey": "Account/default", "expectedSourceAuthority": "l2", "selectionRule": "l2_precedes_l1_by_component_key", "status": "final_payload_matches_builder_selected_source", "finalPayloadRelation": "byte_identical", "finalDeclarationRelation": "identical", "l1": None, "l2": None, "finalPack": None, } document = { "schemaVersion": source_inventory.SOURCE_INVENTORY_VERSION, "mode": "read_only_source_inventory", "status": "complete", "scope": { "downstream": { "repository": "pingcap-inc/nexuscrm-pingcap-pack", "commit": DOWNSTREAM_COMMIT, }, "upstream": { "repository": "pingcap-inc/NexusCRM", "commit": UPSTREAM_COMMIT, }, "officialAppPacks": { "repository": "pingcap-inc/nexuscrm-app-packs", "commit": OFFICIAL_COMMIT, }, "packBuild": { "stackLockSha256": "6" * 64, "intermediatePackPath": "packs/intermediate", "finalPackPath": "packs/final", "finalPackTreeSha256": FINAL_PACK_SHA256, "finalLineageSha256": "7" * 64, }, }, "sourceInventories": { "l1": {"components": []}, "l2": {"components": []}, "finalPack": {"components": [component]}, }, "componentComparisons": [comparison], "coverage": { "l1Components": 0, "l2Components": 0, "finalPackComponents": 1, "unionComponents": 1, "finalPayloadExact": 1, "finalDeclarationDifferent": 0, "notSelectedByFinalPack": 0, "blockedComparisons": 0, }, "blockers": [], "deferred": [ "live layouts and assignments", "portable bundle and routing identities", "initial revision and import plan", ], } _embed_digest(document, "inventoryDigest") return document def _raw_layout( identity: str, object_api_name: str, layout_name: str, config: dict, *, is_deleted: bool, source: str, ) -> dict: return { "Id": identity, "ObjectApiName": object_api_name, "LayoutName": layout_name, "Type": "Detail", "IsDefault": True, "IsDeleted": is_deleted, "ConfigText": json.dumps(config, separators=(",", ":")), "LayoutVersion": 1, "IsCustomized": source == "admin_created", "Source": source, "GenerationVersion": None, "OwnerPack": None if source == "admin_created" else "salesforce-crm-standard", "CreatedDate": None, "LastModifiedDate": None, } def _raw_assignment( identity: str, *, object_api_name: str, layout_id: str, profile_id: str | None = None, role_id: str | None = None, record_type_id: str | None = None, ) -> dict: specificity = sum( value is not None for value in (profile_id, role_id, record_type_id) ) return { "Id": identity, "ObjectApiName": object_api_name, "LayoutId": layout_id, "RoleId": role_id, "ProfileId": profile_id, "RecordTypeId": record_type_id, "Specificity": specificity, "CreatedDate": None, "LastModifiedDate": None, "CreatedById": None, "RouteSource": "admin_authored", "OwnerPack": None, "TargetLayoutId": layout_id, "TargetLayoutObjectApiName": object_api_name, "TargetLayoutIsDeleted": False, "TargetProfileId": profile_id, "TargetProfileIsActive": True if profile_id is not None else None, "TargetProfileIsDeleted": False if profile_id is not None else None, "TargetRoleId": role_id, "TargetRoleIsDeleted": False if role_id is not None else None, "TargetRecordTypeId": record_type_id, "TargetRecordTypeObjectApiName": ( object_api_name if record_type_id is not None else None ), "TargetRecordTypeIsActive": True if record_type_id is not None else None, "TargetRecordTypeIsDeleted": False if record_type_id is not None else None, "TargetCreatedById": None, "TargetCreatedByIsDeleted": None, } def _receipt(source: dict, inventory: dict, inventory_raw: bytes) -> dict: expected_pack = { "name": "salesforce-crm-standard", "category": "app-pack", "manifestFormatVersion": "1", "payloadHash": "8" * 64, "payloadHashAlgorithm": "sha256", "runtimeRelatedListRulesSha256": "9" * 64, "sourceRevisionPolicy": "exact-current-downstream", } installed_pack = { "name": expected_pack["name"], "category": expected_pack["category"], "manifestFormatVersion": expected_pack["manifestFormatVersion"], "status": "active", "sourceKind": "local_path", "sourceUriSha256": "a" * 64, "sourceRevision": DOWNSTREAM_COMMIT, "desiredPayloadHash": expected_pack["payloadHash"], "appliedPayloadHash": expected_pack["payloadHash"], "payloadHashAlgorithm": expected_pack["payloadHashAlgorithm"], "runtimeRelatedListRulesSha256": expected_pack[ "runtimeRelatedListRulesSha256" ], "lastErrorEmpty": True, } document = { "schemaVersion": live_inventory.LIVE_INVENTORY_RECEIPT_VERSION, "mode": "read_only", "generatedAt": "2026-08-13T00:00:00+00:00", "target": { "environment": "staging", "databaseTargetSha256": "b" * 64, "attestation": { "kind": "trusted-active-runtime-v1", "slot": "blue", "runtimeEnvSha256": "c" * 64, "deploySummarySha256": "d" * 64, "activeSelectorSha256": "e" * 64, "releaseMarkerSha256": None, "releaseMetadataSha256": None, "liveNginxSha256": "f" * 64, "backendPort": "8080", "backendService": "nexuscrm-backend", "candidateBundle": { "archiveSha256": "0" * 64, "manifestSha256": "1" * 64, }, "downstreamCommit": DOWNSTREAM_COMMIT, "upstreamCommit": UPSTREAM_COMMIT, "officialAppPacksCommit": OFFICIAL_COMMIT, "finalPackSha256": FINAL_PACK_SHA256, "expectedInstalledPack": expected_pack, "verifyUsernameSha256": "2" * 64, }, "installedPack": installed_pack, }, "sourceInventoryDigest": source["inventoryDigest"], "inventory": { "schemaVersion": inventory["schemaVersion"], "inventoryDigest": inventory["inventoryDigest"], "fileSha256": review.sha256_bytes(inventory_raw), }, "collectorSha256": review.sha256_bytes(COLLECTOR_BYTES), } _embed_digest(document, "receiptSha256") return document @dataclass class BoundReviewFixture: source_payload: dict source_payload_raw: bytes source: dict inventory: dict inventory_raw: bytes receipt: dict receipt_raw: bytes def _bound_fixture() -> BoundReviewFixture: source_payload = { "objectApiName": "Account", "layoutName": "Account Layout", "type": "Detail", "isDefault": True, "sections": [ {"id": "section/a~b", "label": "Main", "fields": ["Name"]} ], } source_payload_raw = _artifact_bytes(source_payload) source = _source_document(source_payload_raw) selected = source_inventory.final_page_layout_candidates(source) layouts = [ _raw_layout( "layout-account", "Account", "Account Layout", { **source_payload, "sections": [ { "id": "section/a~b", "label": "Main", "fields": ["Phone"], } ], "id": "layout-account", "source": "curated", "layoutVersion": 1, }, is_deleted=False, source="curated", ), _raw_layout( "layout-contact", "Contact", "Contact Layout", { "objectApiName": "Contact", "layoutName": "Contact Layout", "type": "Detail", "isDefault": True, "sections": [], }, is_deleted=False, source="admin_created", ), _raw_layout( "layout-legacy", "Legacy__c", "Legacy Layout", { "objectApiName": "Legacy__c", "layoutName": "Legacy Layout", "type": "Detail", "isDefault": True, "sections": [], }, is_deleted=True, source="admin_created", ), ] assignments = [ _raw_assignment( "assignment-wildcard", object_api_name="Account", layout_id="layout-account", ), _raw_assignment( "assignment-specific", object_api_name="Contact", layout_id="layout-contact", profile_id="profile-1", role_id="role-1", record_type_id="record-type-1", ), ] inventory = live_inventory.compile_live_inventory( source_document=source, selected_final_pack_layouts=selected, layout_rows=layouts, assignment_rows=assignments, ) inventory_raw = _artifact_bytes(inventory) receipt = _receipt(source, inventory, inventory_raw) return BoundReviewFixture( source_payload=source_payload, source_payload_raw=source_payload_raw, source=source, inventory=inventory, inventory_raw=inventory_raw, receipt=receipt, receipt_raw=_artifact_bytes(receipt), ) class ReviewPacketTest(unittest.TestCase): def test_review_items_close_over_relations_assignments_and_tombstones(self) -> None: fixture = _bound_fixture() packet = review.compile_review_packet( inventory=fixture.inventory, receipt=fixture.receipt, source_document=fixture.source, source_payloads={"Account/default": fixture.source_payload}, evidence_files={ "inventoryFileSha256": review.sha256_bytes(fixture.inventory_raw), "receiptFileSha256": review.sha256_bytes(fixture.receipt_raw), }, ) self.assertEqual( len(packet["layoutRelations"]), len(fixture.inventory["layoutCandidateRelations"]), ) self.assertEqual( {item["assignmentId"] for item in packet["assignments"]}, {item["id"] for item in fixture.inventory["assignments"]}, ) tombstone_ids = { layout["id"] for layout in fixture.inventory["layouts"] if layout["isDeleted"] is True } reviewed_tombstones = { item["layoutId"] for relation in packet["layoutRelations"] for item in relation["liveLayouts"] if item["deletionState"] is True } self.assertEqual(reviewed_tombstones, tombstone_ids) self.assertEqual( [item["itemOrdinal"] for item in packet["layoutRelations"]], list(range(len(packet["layoutRelations"]))), ) relation_decisions = [ decision for item in packet["layoutRelations"] for decision in item["requiredDecisions"] ] self.assertEqual( sum( decision["decision"] == "registry_identity_required" for decision in relation_decisions ), len(packet["layoutRelations"]), ) self.assertEqual( { decision["evidenceReference"]["legacyLayoutId"] for decision in relation_decisions if decision["decision"] == "tombstone_disposition_required" }, tombstone_ids, ) specific = next( item for item in packet["assignments"] if item["assignmentId"] == "assignment-specific" ) self.assertEqual( [ decision["identityKind"] for decision in specific["requiredDecisions"] if decision["decision"] == "registry_identity_required" ], [ "target_layout_bundle_slot", "catalog_record_type_key", "security_profile_key", "security_role_key", ], ) self.assertEqual(packet["coverage"]["structuralComparisonCount"], 1) self.assertEqual( packet["coverage"]["unresolvedDecisionCount"], sum( len(item["requiredDecisions"]) for item in [*packet["layoutRelations"], *packet["assignments"]] ), ) self.assertEqual( packet["reviewInputDigest"], "sha256:" + review.sha256_bytes( review.canonical_bytes( { key: value for key, value in packet.items() if key != "reviewInputDigest" } ) ), ) forbidden = { "bundlekey", "layoutbundlerevisionid", "layoutvalidationid", "selectedauthority", "generatedat", } def keys(value: object): if isinstance(value, dict): for key, child in value.items(): yield key.casefold() yield from keys(child) elif isinstance(value, list): for child in value: yield from keys(child) self.assertTrue(forbidden.isdisjoint(set(keys(packet)))) self.assertEqual( packet["claims"], { "authoritySelected": False, "portableIdentityMinted": False, "revisionPlanned": False, "servingBehaviorChanged": False, }, ) def test_structural_diff_is_deterministic_and_uses_rfc6901_paths(self) -> None: source = { "gone": None, "a/b": {"~key": [None, 1, True]}, } live = { "a/b": {"new": None, "~key": [None, 2, 1]}, } expected = [ { "pointer": "/a~1b/new", "kind": "add", "source": {"present": False}, "live": {"present": True, "value": None}, }, { "pointer": "/a~1b/~0key/1", "kind": "replace", "source": {"present": True, "value": 1}, "live": {"present": True, "value": 2}, }, { "pointer": "/a~1b/~0key/2", "kind": "replace", "source": {"present": True, "value": True}, "live": {"present": True, "value": 1}, }, { "pointer": "/gone", "kind": "remove", "source": {"present": True, "value": None}, "live": {"present": False}, }, ] self.assertEqual(review.structural_diff(source, live), expected) self.assertEqual( review.structural_diff( dict(reversed(list(source.items()))), dict(reversed(list(live.items()))), ), expected, ) if __name__ == "__main__": unittest.main()