import hashlib import json import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] CRM_PACK = ROOT / "salesforce-crm-standard" def load_json(path: Path): return json.loads(path.read_text()) class SendRecordEmailContractsTest(unittest.TestCase): def test_contact_and_lead_email_opt_out_default_to_false(self): objects = load_json(CRM_PACK / "objects" / "standard_objects_spec.json") by_name = {item["apiName"]: item for item in objects} for object_api_name in ("Contact", "Lead"): fields = { item["apiName"]: item for item in by_name[object_api_name]["fields"] } opt_out = fields["HasOptedOutOfEmail"] self.assertEqual(opt_out["type"], "boolean", object_api_name) self.assertEqual(opt_out.get("default"), "0", object_api_name) def test_gmail_provider_definition_is_an_independent_secret_free_connector_pack(self): connector_root = ROOT / "gmail-connected-mail-connector" manifest = load_json(connector_root / "pack.json") self.assertEqual(manifest["name"], "gmail-connected-mail-connector") self.assertEqual(manifest["category"], "connector-pack") self.assertEqual(manifest["version"], "0.1.0") self.assertEqual(manifest["dependencies"], []) provider_binding = manifest["extensions"]["connectedMailProviderDefinitions"] self.assertEqual(provider_binding, [{ "connectorKey": "gmail-connected-mail", "path": "integrations/connected_mail_gmail.json", "sha256": "d476f596fe85ea78baefe290e5ea1b3ad1f092962f5ade30750a58961ef48c28", }]) self.assertEqual(manifest["components"], [{ "kind": "connector_callout_profile", "businessKey": "gmail-connected-mail-connector/gmail-delivery", "manageability": "locked", "path": "callout/gmail_delivery_profile.json", }]) profile = load_json(connector_root / "callout" / "gmail_delivery_profile.json") self.assertEqual(profile["profileName"], "gmail-delivery") self.assertEqual(profile["credentialRef"], "ConnectedMailOAuthClient") self.assertEqual(profile["endpointClass"], "mail.send") self.assertEqual(profile["retryPolicy"]["maxAttempts"], 1) self.assertEqual(profile["objectPathMap"], {"message:send": "/mail/send"}) provider_path = connector_root / provider_binding[0]["path"] self.assertEqual(hashlib.sha256(provider_path.read_bytes()).hexdigest(), provider_binding[0]["sha256"]) provider = load_json(provider_path) self.assertEqual(provider["schema"], "nexuscrm.connected-mail-http-provider/v1") self.assertEqual(provider["connectorKey"], "gmail-connected-mail") self.assertEqual(provider["publicPath"], "/api/connected-mailboxes/gmail") self.assertEqual(provider["binding"], { "ownerPack": "gmail-connected-mail-connector", "businessKey": "gmail-connected-mail-connector/gmail-delivery", "definitionVersion": 1, "clientCredentialRef": "ConnectedMailOAuthClient", }) self.assertEqual(provider["mailRequestField"], "raw") self.assertEqual(provider["mailResponseIdField"], "id") self.assertEqual(provider["authorizationParameters"], { "access_type": "offline", "prompt": "consent", }) self.assertNotIn("clientId", provider) self.assertNotIn("clientSecret", provider) def test_action_is_bounded_to_contact_lead_and_sdr_permission(self): actions = load_json(CRM_PACK / "actions" / "invocable_standard_actions.json") action = next(item for item in actions if item["name"] == "sendRecordEmail") self.assertEqual(action["flowDeveloperName"], "Std_SendRecordEmail") self.assertEqual(action["category"], "ActivityTimeline") self.assertEqual(action["rendererId"], "app-pack:crm/action-renderer/email-compose") self.assertEqual(action["targetObjects"], ["Contact", "Lead"]) self.assertEqual(action["legacyNames"], ["taskEmail"]) self.assertEqual(action["requiredPermissions"], ["SendRecordEmail"]) self.assertTrue(action["isAtomic"]) self.assertFalse(action["headerFallbackEligible"]) self.assertEqual( [item["apiName"] for item in action["inputs"]], [ "requestId", "recordId", "objectApiName", "subject", "body", "cc", "bcc", "htmlBody", "attachments", ], ) self.assertTrue(all(item["isRequired"] for item in action["inputs"][:5])) self.assertTrue(all(not item["isRequired"] for item in action["inputs"][5:])) self.assertEqual( [item["apiName"] for item in action["outputs"]], ["intentId", "deliveryState", "failureCode"], ) outbound_mail = action["config"]["outboundMail"] self.assertEqual(outbound_mail["connectorOwnerKey"], "salesforce-crm-standard") self.assertEqual(outbound_mail["flowActionName"], "EnqueueMail") self.assertEqual(outbound_mail["actionSlotId"], "send-mail") self.assertEqual(outbound_mail["destinationSlotId"], "primary-mailbox") self.assertEqual( outbound_mail["calloutProfileOwnerPack"], "gmail-connected-mail-connector", ) self.assertEqual( outbound_mail["calloutProfileBusinessKey"], "gmail-connected-mail-connector/gmail-delivery", ) completion = outbound_mail["mailCompletion"] self.assertEqual(completion["recipientAddressField"], "Email") self.assertEqual(completion["recipientOptOutField"], "HasOptedOutOfEmail") self.assertEqual(completion["activity"]["objectApiName"], "Task") self.assertEqual(completion["message"]["objectApiName"], "EmailMessage") self.assertEqual(completion["message"]["ccAddressField"], "CcAddress") self.assertEqual(completion["message"]["bccAddressField"], "BccAddress") self.assertEqual(completion["message"]["htmlBodyField"], "HtmlBody") self.assertEqual(completion["message"]["attachmentIdsField"], "AttachmentIds") attachments_input = next(item for item in action["inputs"] if item["apiName"] == "attachments") self.assertTrue(attachments_input["isCollection"]) # The registry checks requiredPermissions for every invocation. Runtime # owns source-record/FLS/opt-out checks and the fixed completion write. self.assertEqual(action["inputGuards"], []) def test_flow_atomically_enqueues_one_provider_neutral_mail_intent(self): flow = load_json(CRM_PACK / "flows" / "Std_SendRecordEmail.flow.json") self.assertTrue(flow["runAtomically"]) self.assertEqual([item["name"] for item in flow["variables"]], [ "requestId", "recordId", "objectApiName", "subject", "body", "cc", "bcc", "htmlBody", "attachments", "intentId", "deliveryState", "failureCode", ]) self.assertEqual(len(flow["actionCalls"]), 1) action_call = flow["actionCalls"][0] self.assertEqual(action_call["name"], "EnqueueMail") self.assertEqual(action_call["actionType"], "enqueueOutboundMailV3") self.assertEqual(action_call["outputParameters"], [ {"name": "intentId", "value": "intentId"}, {"name": "deliveryState", "value": "deliveryState"}, {"name": "failureCode", "value": "failureCode"}, ]) handler = load_json(CRM_PACK / "flow_action_call_handlers" / "send_record_email.json") self.assertEqual(handler, { "actionType": "enqueueOutboundMailV3", "handlerRef": "platform:outbound-mail/enqueue/v3", "atomicPolicy": "safe", "previewPolicy": "reject", "requiresContext": "standard_action_protected", "sendsEmail": True, }) def test_manifest_orders_handler_flow_and_action(self): manifest = load_json(CRM_PACK / "pack.json") self.assertEqual(manifest["dependencies"], [{ "name": "gmail-connected-mail-connector", "versionConstraint": ">=0.1.0 <0.2.0", }]) components = manifest["components"] by_key = {(item["kind"], item["businessKey"]): item for item in components} flow = by_key[("standard_flow", "Std_SendRecordEmail")] action = by_key[("invocable_action", "bootstrap_standard_actions")] self.assertIn( {"kind": "flow_action_call_handler", "businessKey": "enqueueoutboundmailv3"}, flow["dependsOn"], ) self.assertIn( {"kind": "standard_flow", "businessKey": "Std_SendRecordEmail"}, action["dependsOn"], ) self.assertIn( {"kind": "object_spec", "businessKey": "standard_objects_spec:crm"}, action["dependsOn"], ) def test_email_message_declares_the_minimum_installable_activity_log_shape(self): objects = load_json(CRM_PACK / "objects" / "standard_objects_spec.json") email_message = next(item for item in objects if item["apiName"] == "EmailMessage") self.assertEqual(email_message["keyPrefix"], "02s") self.assertEqual(email_message["sharingModel"], "Private") fields = {item["apiName"]: item for item in email_message["fields"]} self.assertEqual( set(fields), { "ParentId", "ActivityId", "FromAddress", "ToAddress", "CcAddress", "BccAddress", "Subject", "TextBody", "HtmlBody", "AttachmentIds", "HasAttachment", "Incoming", "Status", "MessageDate", }, ) self.assertEqual(fields["ParentId"]["referenceTo"], ["Case"]) self.assertFalse(fields["ParentId"]["updateable"]) self.assertEqual(fields["ActivityId"]["referenceTo"], ["Task"]) self.assertEqual(fields["AttachmentIds"]["type"], "textarea") self.assertEqual(fields["HasAttachment"]["formula"], "NOT(ISBLANK(AttachmentIds))") related_lists = load_json( CRM_PACK / "related_lists" / "standard_related_list_rules.json" ) case_email = next( item for item in related_lists["rules"] if item["key"] == "case-email-messages" ) self.assertEqual(fields[case_email["lookupField"]]["referenceTo"], ["Case"]) self.assertEqual( [item["value"] for item in fields["Status"]["options"]], ["0", "1", "2", "3", "4", "5"], ) def test_email_message_record_detail_has_an_authored_default_layout(self): components = load_json(CRM_PACK / "pack.json")["components"] layout_component = next( item for item in components if item["kind"] == "page_layout" and item["businessKey"] == "EmailMessage/default" ) self.assertEqual( layout_component["path"], "layouts/email_message_page_layout.json" ) layout = load_json(CRM_PACK / layout_component["path"]) self.assertEqual(layout["objectApiName"], "EmailMessage") self.assertTrue(layout["isDefault"]) rendered_fields = { field for section in layout["sections"] for field in section["fields"] } self.assertEqual( rendered_fields, { "ActivityId", "FromAddress", "CcAddress", "BccAddress", "Incoming", "MessageDate", "ParentId", "Status", "Subject", "TextBody", "HtmlBody", "HasAttachment", "AttachmentIds", "ToAddress", }, ) if __name__ == "__main__": unittest.main()