import json import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] CRM_PACK = REPO_ROOT / "salesforce-crm-standard" def load_json(path: Path): with path.open(encoding="utf-8") as handle: return json.load(handle) class FlowActionCallHandlerContractsTest(unittest.TestCase): def test_change_owner_handler_is_manifest_owned(self): manifest = load_json(CRM_PACK / "pack.json") components = manifest["components"] by_key = { (component["kind"], component["businessKey"]): component for component in components } handler = by_key[("flow_action_call_handler", "changeowner")] self.assertEqual(handler["manageability"], "locked") self.assertEqual( handler["path"], "flow_action_call_handlers/change_owner.json", ) payload = load_json(CRM_PACK / handler["path"]) self.assertEqual( payload, { "actionType": "changeOwner", "handlerRef": "crm:standard-action/change-owner", "atomicPolicy": "reject", "previewPolicy": "reject", "requiresContext": "standard_action_protected", }, ) def test_connector_flow_action_call_handlers_are_manifest_owned(self): # NCRM#3102 Group A: the connector actionCall verbs move off the # compiled-in PlatformReservedActionTypes() into the descriptor-driven # open registry. flow_action_call_handler is an app-pack-only component # kind, and the verbs must have one global owner that is an ancestor of # every caller. Companion packs can call these descriptors while the # standard CRM pack remains free of tenant-specific connector fields. # Each payload must mirror the NexusCRM runtime # ConnectorFlowActionCallRegistrations descriptor exactly. manifest = load_json(CRM_PACK / "pack.json") by_key = { (component["kind"], component["businessKey"]): component for component in manifest["components"] } expected = { "pushtoconnector": { "path": "flow_action_call_handlers/push_to_connector.json", "payload": { "actionType": "pushToConnector", "handlerRef": "platform:connector/push-to-connector", "atomicPolicy": "reject", "previewPolicy": "reject", "requiresContext": "none", }, }, "applyconnectorfieldmappings": { "path": "flow_action_call_handlers/apply_connector_field_mappings.json", "payload": { "actionType": "applyConnectorFieldMappings", "handlerRef": "platform:connector/apply-field-mappings", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, "registerconnectorinboundremoteauthorship": { "path": "flow_action_call_handlers/register_connector_inbound_remote_authorship.json", "payload": { "actionType": "registerConnectorInboundRemoteAuthorship", "handlerRef": "platform:connector/register-inbound-remote-authorship", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, "finalizeconnectorinboundremoteidentity": { "path": "flow_action_call_handlers/finalize_connector_inbound_remote_identity.json", "payload": { "actionType": "finalizeConnectorInboundRemoteIdentity", "handlerRef": "platform:connector/finalize-inbound-remote-identity", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, "resolveconnectorremoteidentity": { "path": "flow_action_call_handlers/resolve_connector_remote_identity.json", "payload": { "actionType": "resolveConnectorRemoteIdentity", "handlerRef": "platform:connector/resolve-remote-identity", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, "recoverconnectorremoteidentity": { "path": "flow_action_call_handlers/recover_connector_remote_identity.json", "payload": { "actionType": "recoverConnectorRemoteIdentity", "handlerRef": "platform:connector/recover-remote-identity", "atomicPolicy": "reject", "previewPolicy": "reject", "requiresContext": "none", }, }, "transferconnectorremoteidentity": { "path": "flow_action_call_handlers/transfer_connector_remote_identity.json", "payload": { "actionType": "transferConnectorRemoteIdentity", "handlerRef": "platform:connector/transfer-remote-identity", "atomicPolicy": "safe", "previewPolicy": "reject", }, }, "setretriable": { "path": "flow_action_call_handlers/set_retriable.json", "payload": { "actionType": "setRetriable", "handlerRef": "platform:connector/set-retriable", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, "setidentityconvergencedependency": { "path": "flow_action_call_handlers/set_identity_convergence_dependency.json", "payload": { "actionType": "setIdentityConvergenceDependency", "handlerRef": "platform:connector/set-identity-convergence-dependency", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "none", }, }, } for business_key, spec in expected.items(): component = by_key[("flow_action_call_handler", business_key)] self.assertEqual(component["manageability"], "locked") self.assertEqual(component["path"], spec["path"]) payload = load_json(CRM_PACK / spec["path"]) self.assertEqual(payload, spec["payload"]) self.assertEqual(business_key, payload["actionType"].lower()) def test_convert_lead_uses_neutral_prepared_update_commit(self): manifest = load_json(CRM_PACK / "pack.json") by_key = { (component["kind"], component["businessKey"]): component for component in manifest["components"] } convert_flow = by_key[("standard_flow", "Std_ConvertLead")] self.assertNotIn( { "kind": "flow_action_call_handler", "businessKey": "stampconvertedanchors", }, convert_flow["dependsOn"], ) self.assertNotIn( { "kind": "flow_action_call_handler", "businessKey": "transferconnectorremoteidentity", }, convert_flow["dependsOn"], ) handler = by_key[ ("flow_action_call_handler", "transferconnectorremoteidentity") ] self.assertEqual(handler["manageability"], "locked") flow = load_json(CRM_PACK / convert_flow["path"]) actions = {action["name"]: action for action in flow["actionCalls"]} preview = actions["PreviewLeadBeforeSaveUpdate"] preview_parameters = { parameter["name"]: parameter["value"] for parameter in preview["inputParameters"] } self.assertEqual( preview_parameters["internalUpdates"], { "ConvertedAccountId": {"elementReference": "accountId"}, "ConvertedContactId": {"elementReference": "contactId"}, "ConvertedOpportunityId": { "elementReference": "opportunityId" }, }, ) commit = actions["MarkLeadConverted"] self.assertEqual(commit["actionType"], "commitPreparedRecordUpdate") self.assertEqual(commit["actionName"], "commitPreparedRecordUpdate") parameters = { parameter["name"]: parameter["value"] for parameter in commit["inputParameters"] } self.assertEqual(parameters["object"], "Lead") self.assertEqual( parameters["recordId"], {"elementReference": "leadId"} ) self.assertEqual( parameters["extraFields"], {"ConvertedDate": {"elementReference": "$Flow.CurrentDate"}}, ) def test_convert_lead_standard_flow_does_not_embed_connector_remote_id_field(self): flow = load_json(CRM_PACK / "flows" / "Std_ConvertLead.flow.json") lookups = {lookup["name"]: lookup for lookup in flow["recordLookups"]} decisions = {decision["name"]: decision for decision in flow["decisions"]} updates = {update["name"]: update for update in flow["recordUpdates"]} subflows = {subflow["name"]: subflow for subflow in flow["subflows"]} actions = {action["name"]: action for action in flow["actionCalls"]} self.assertEqual( subflows["RelinkLeadChildren"]["connector"]["targetReference"], "PreviewLeadBeforeSaveUpdate", ) self.assertNotIn("GetConvertedContactForRemoteIdentity", lookups) self.assertNotIn("MaybeAlignContactConnectorRemoteId", decisions) self.assertNotIn("AlignContactConnectorRemoteId", updates) self.assertNotIn("TransferLeadConnectorRemoteIdentity", actions) self.assertNotIn("HubSpot_Id__c", json.dumps(flow, sort_keys=True)) def test_convert_lead_declares_complete_stable_campaign_parent_prelocks(self): flow = load_json(CRM_PACK / "flows" / "Std_ConvertLead.flow.json") prelocks = { prelock["name"]: prelock for prelock in flow["atomicParentPrelocks"] } self.assertEqual( prelocks["ExistingAccount"], { "name": "ExistingAccount", "parentObject": "Account", "sourceObject": "Account", "parentReferenceField": "Id", "filters": [ { "field": "Id", "operator": "EqualTo", "value": {"elementReference": "accountId"}, } ], }, ) self.assertEqual( prelocks["LeadCampaignMemberships"], { "name": "LeadCampaignMemberships", "parentObject": "Campaign", "sourceObject": "CampaignMember", "parentReferenceField": "CampaignId", "filters": [ { "field": "LeadId", "operator": "EqualTo", "value": {"elementReference": "leadId"}, } ], }, ) def test_convert_lead_campaign_preview_uses_modeled_relationship_path(self): flow = load_json( CRM_PACK / "flows" / "Std_ConvertLeadCampaignPreview.flow.json" ) actions = load_json(CRM_PACK / "actions" / "invocable_standard_actions.json") object_specs = load_json(CRM_PACK / "objects" / "standard_objects_spec.json") assignments = { assignment["name"]: assignment for assignment in flow["assignments"] } set_campaign_output = assignments["SetCampaignOutput"] self.assertEqual( set_campaign_output["assignmentItems"][1]["value"], {"elementReference": "$cm.Campaign.Name"}, ) convert_action = next( action for action in actions if action["name"] == "convertLead" ) convert_flow = load_json(CRM_PACK / "flows" / "Std_ConvertLead.flow.json") self.assertTrue(convert_flow.get("recordUpdates")) self.assertIn( "previewrecordupdate", { str(action.get("actionType", "")).lower() for action in convert_flow.get("actionCalls", []) }, ) self.assertFalse( convert_action.get("createPreviewSupported", False), "update-bearing Lead Convert cannot advertise L0 create preview support", ) self.assertEqual( convert_action["config"]["leadConvertPreview"]["campaignAttribution"][ "summaryFields" ], ["CampaignId", "Campaign.Name", "Status", "HasResponded"], ) internal_read_scopes = { scope["objectApiName"] for scope in convert_action["internalAccessScopes"] if scope["operation"] == "Read" } self.assertIn("CampaignMember", internal_read_scopes) self.assertIn("Campaign", internal_read_scopes) campaign_member = next( spec for spec in object_specs if spec["apiName"] == "CampaignMember" ) campaign_id = next( field for field in campaign_member["fields"] if field["apiName"] == "CampaignId" ) self.assertEqual(campaign_id["relationshipName"], "Campaign") self.assertIn("Campaign", campaign_id["referenceTo"]) self.assertNotIn( "CampaignId_Name", json.dumps({"flow": flow, "action": convert_action}, sort_keys=True), ) if __name__ == "__main__": unittest.main()