from __future__ import annotations import json import tempfile import unittest from pathlib import Path import yaml from scripts import apply_deploy_seed as seed_dispatcher from scripts import connected_mail_oauth from scripts import render_serviceification_test_stack as stack_renderer ROOT = Path(__file__).resolve().parents[1] L2_PACK = ROOT / "packs" / "Salesforce-PingCAP-L2" FINAL_PACK = ROOT / "packs" / "NOT-EDIT-final-merged-packs" MAIL_COMPANION = ROOT / "packs" / "PingCAP-Gmail-Companion" MAILBOX_CAPABILITY = "runtime.connected-mailboxes.authorize.v1" def mail_completion() -> dict[str, object]: return { "recipientAddressField": "Email", "recipientOptOutField": "HasOptedOutOfEmail", "activity": { "objectApiName": "Task", "sourceRecordField": "WhoId", "subjectField": "Subject", "activityDateField": "ActivityDate", "statusField": "Status", "priorityField": "Priority", "bodyField": "Description", "ownerField": "OwnerId", "typeField": "Type", "subtypeField": "TaskSubtype", "fixedSubject": "Email", "fixedStatus": "Completed", "fixedPriority": "Normal", "fixedType": "Email", "fixedSubtype": "Email", }, "message": { "objectApiName": "EmailMessage", "activityIdField": "ActivityId", "fromAddressField": "FromAddress", "toAddressField": "ToAddress", "ccAddressField": "CcAddress", "bccAddressField": "BccAddress", "subjectField": "Subject", "textBodyField": "TextBody", "htmlBodyField": "HtmlBody", "attachmentIdsField": "AttachmentIds", "incomingField": "Incoming", "statusField": "Status", "messageDateField": "MessageDate", "fixedIncoming": False, "fixedStatus": "3", }, } def mail_action() -> dict[str, object]: return { "name": "sendRecordEmail", "flowDeveloperName": "Std_SendRecordEmail", "requiredPermissions": ["SendRecordEmail"], "targetObjects": ["Contact", "Lead"], "isAtomic": True, "config": { "outboundMail": { "connectorOwnerKey": "salesforce-crm-standard", "flowActionName": "EnqueueMail", "actionSlotId": "send-mail", "destinationSlotId": "primary-mailbox", "calloutProfileOwnerPack": "gmail-connected-mail-connector", "calloutProfileBusinessKey": ( "gmail-connected-mail-connector/gmail-delivery" ), "mailCompletion": mail_completion(), } }, } def callout_binding() -> dict[str, object]: return { "authorizedWorkloadId": "integration-service", "kind": "CONNECTOR_CALLOUT_PROFILE", "ownerPack": "gmail-connected-mail-connector", "packInstallationId": "install-gmail", "packPayloadDigest": "nexus-pack-sha256-v1:" + "1" * 64, "businessKey": "gmail-connected-mail-connector/gmail-delivery", "definitionVersion": 1, "credentialRefs": ["ConnectedMailOAuthClient"], } class OutboundMailCatalogRouteTest(unittest.TestCase): def render(self, action: dict[str, object]) -> list[dict[str, object]]: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "actions.json" path.write_text(json.dumps([action]), encoding="utf-8") return stack_renderer.outbound_mail_routes_from_actions( path, [callout_binding()] ) def test_expands_the_l1_declaration_to_exact_contact_and_lead_routes(self) -> None: routes = self.render(mail_action()) self.assertEqual( [route["sourceObject"] for route in routes], ["Contact", "Lead"] ) self.assertTrue( all(route["authorizedWorkloadId"] == "integration-service" for route in routes) ) self.assertEqual( { "ccAddressField": "CcAddress", "bccAddressField": "BccAddress", "htmlBodyField": "HtmlBody", }, { field: routes[0]["mailCompletion"]["message"][field] for field in ("ccAddressField", "bccAddressField", "htmlBodyField") }, ) encoded = json.dumps(routes, sort_keys=True).lower() for forbidden in ("clientsecret", "refreshtoken", "oauth2.googleapis.com"): self.assertNotIn(forbidden, encoded) def test_accepts_the_pingcap_api_enabled_mail_audience(self) -> None: action = mail_action() action["requiredPermissions"] = ["ApiEnabled"] self.assertEqual(self.render(mail_action()), self.render(action)) def test_rejects_missing_or_unknown_mail_permission_policies(self) -> None: for required in (None, [], ["UnknownPermission"], ["ApiEnabled", "SendRecordEmail"]): with self.subTest(required=required): action = mail_action() action["requiredPermissions"] = required with self.assertRaisesRegex(stack_renderer.ValidationFailure, "must require only"): self.render(action) def test_preserves_the_complete_mail_v3_completion_shape(self) -> None: action = mail_action() routes = self.render(action) for route in routes: message = route["mailCompletion"]["message"] self.assertEqual("CcAddress", message["ccAddressField"]) self.assertEqual("BccAddress", message["bccAddressField"]) self.assertEqual("HtmlBody", message["htmlBodyField"]) self.assertEqual("AttachmentIds", message["attachmentIdsField"]) def test_accepts_the_previous_complete_mail_v2_shape(self) -> None: action = mail_action() action["config"]["outboundMail"]["mailCompletion"]["message"].pop( "attachmentIdsField" ) routes = self.render(action) for route in routes: self.assertNotIn( "attachmentIdsField", route["mailCompletion"]["message"] ) def test_rejects_a_partial_mail_v2_completion_shape(self) -> None: action = mail_action() message = action["config"]["outboundMail"]["mailCompletion"]["message"] message.pop("bccAddressField") message.pop("htmlBodyField") with self.assertRaisesRegex(stack_renderer.ValidationFailure, "unknown shape"): self.render(action) def test_rejects_an_attachment_field_without_the_mail_v2_fields(self) -> None: action = mail_action() message = action["config"]["outboundMail"]["mailCompletion"]["message"] for field in ("ccAddressField", "bccAddressField", "htmlBodyField"): message.pop(field) with self.assertRaisesRegex(stack_renderer.ValidationFailure, "unknown shape"): self.render(action) def test_checked_in_final_pack_renders_mail_v3_routes(self) -> None: routes = stack_renderer.outbound_mail_routes_from_actions( FINAL_PACK / "actions" / "invocable_standard_actions.json", [callout_binding()], ) self.assertEqual(["Contact", "Lead"], [route["sourceObject"] for route in routes]) for route in routes: message = route["mailCompletion"]["message"] self.assertEqual("CcAddress", message["ccAddressField"]) self.assertEqual("BccAddress", message["bccAddressField"]) self.assertEqual("HtmlBody", message["htmlBodyField"]) self.assertEqual("AttachmentIds", message["attachmentIdsField"]) def test_rejects_a_non_atomic_or_secret_extended_route_declaration(self) -> None: non_atomic = mail_action() non_atomic["isAtomic"] = False with self.assertRaisesRegex(stack_renderer.ValidationFailure, "must be atomic"): self.render(non_atomic) secret_extended = mail_action() secret_extended["config"]["outboundMail"]["clientSecret"] = "not-authority" with self.assertRaisesRegex(stack_renderer.ValidationFailure, "unknown shape"): self.render(secret_extended) def test_rejects_a_symlinked_action_authority(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) target = root / "target.json" target.write_text(json.dumps([mail_action()]), encoding="utf-8") authority = root / "actions.json" authority.symlink_to(target) with self.assertRaisesRegex( stack_renderer.ValidationFailure, "regular non-symlink", ): stack_renderer.outbound_mail_routes_from_actions( authority, [callout_binding()], ) def test_mailbox_authorization_uses_the_record_email_system_permission(self) -> None: grants = [{"authorized_workload_id": "integration-service", "capability_id": MAILBOX_CAPABILITY}] capability = stack_renderer.d3_workload_capabilities(grants, "2s")[0] actions = json.loads((FINAL_PACK / "actions/invocable_standard_actions.json").read_text()) action = next(item for item in actions if item["name"] == "sendRecordEmail") self.assertEqual(["ApiEnabled"], action["requiredPermissions"]) self.assertEqual(action["requiredPermissions"], [capability.get("required_system_permission")]) self.assertNotIn("required_custom_permission", capability) def test_mailbox_workload_grant_is_delegable_by_api_sessions(self) -> None: installed = ["runtime.records.query.v1"] grants = [ { "authorized_workload_id": "integration-service", "capability_id": MAILBOX_CAPABILITY, } ] self.assertEqual( stack_renderer.d3_session_scope_capabilities(installed, grants), [*installed, MAILBOX_CAPABILITY], ) class L2MailActivationContractTest(unittest.TestCase): def test_contact_and_lead_email_opt_out_default_to_false(self) -> None: final_objects = json.loads( (FINAL_PACK / "objects" / "standard_objects_spec.json").read_text() ) final_by_name = {item["apiName"]: item for item in final_objects} for object_api_name in ("Contact", "Lead"): l2_payload = json.loads( (L2_PACK / "objects" / f"{object_api_name}.json").read_text() ) if isinstance(l2_payload, list): l2_object = next( item for item in l2_payload if item["apiName"] == object_api_name ) else: l2_object = l2_payload for source, object_spec in ( ("L2", l2_object), ("final", final_by_name[object_api_name]), ): fields = { item["apiName"]: item for item in object_spec["fields"] } opt_out = fields["HasOptedOutOfEmail"] self.assertEqual(opt_out["type"], "boolean", (source, object_api_name)) self.assertEqual( opt_out.get("default"), "0", (source, object_api_name) ) def test_connected_mail_ingress_targets_the_integration_service(self) -> None: template = ( ROOT / "deployments/offline/nginx/nexuscrm-env.conf.template" ).read_text() self.assertIn("__CONNECTED_MAIL_INGRESS_BLOCK__", template) self.assertNotIn("location /api/connected-mailboxes/ {", template) def test_l2_overrides_mail_access_and_installs_its_connector(self) -> None: l2_actions = json.loads( (L2_PACK / "actions" / "invocable_standard_actions.json").read_text() ) overrides = [action for action in l2_actions if action["name"] == "sendRecordEmail"] self.assertEqual(1, len(overrides)) self.assertEqual(["ApiEnabled"], overrides[0]["requiredPermissions"]) final_actions = json.loads( (FINAL_PACK / "actions" / "invocable_standard_actions.json").read_text() ) final_action = next(action for action in final_actions if action["name"] == "sendRecordEmail") self.assertEqual(overrides[0], final_action) stack = json.loads((ROOT / "stack.lock.json").read_text()) connector = "official_repo:gmail-connected-mail-connector" final_pack = "packs/NOT-EDIT-final-merged-packs" companion = "packs/PingCAP-Gmail-Companion" self.assertIn(connector, stack["packs"]) self.assertLess(stack["packs"].index(connector), stack["packs"].index(final_pack)) self.assertLess(stack["packs"].index(final_pack), stack["packs"].index(companion)) def test_final_pack_carries_l1_mail_and_l2_permission_authority(self) -> None: manifest = json.loads((FINAL_PACK / "pack.json").read_text()) components = { (component["kind"], component["businessKey"]): component for component in manifest["components"] } for identity in ( ("custom_permission", "SendRecordEmail"), ("standard_flow", "Std_SendRecordEmail"), ("flow_action_call_handler", "enqueueoutboundmailv3"), ): self.assertIn(identity, components) self.assertNotIn( ("permission_set_definition", "PingCAP_SDR_Send_Record_Email"), components, ) action_component = components[("invocable_action", "bootstrap_standard_actions")] dependencies = { (dependency["kind"], dependency["businessKey"]) for dependency in action_component["dependsOn"] } self.assertIn(("standard_flow", "Std_SendRecordEmail"), dependencies) self.assertIn(("custom_permission", "SendRecordEmail"), dependencies) actions = json.loads((FINAL_PACK / action_component["path"]).read_text()) send_actions = [action for action in actions if action.get("name") == "sendRecordEmail"] self.assertEqual(1, len(send_actions)) self.assertTrue(send_actions[0]["isAtomic"]) self.assertEqual( ["intentId", "deliveryState", "failureCode"], [output["apiName"] for output in send_actions[0]["outputs"]], ) self.assertEqual( "EnqueueMail", send_actions[0]["config"]["outboundMail"]["flowActionName"], ) companion_manifest = json.loads((MAIL_COMPANION / "pack.json").read_text()) companion_components = { (component["kind"], component["businessKey"]): component for component in companion_manifest["components"] } permission_component = companion_components[ ("permission_set_definition", "PingCAP_SDR_Send_Record_Email") ] self.assertIn( ( "permission_set_assignment_policy", "PingCAP_SDR_Send_Record_Email_SDR", ), companion_components, ) permission_set = json.loads( (MAIL_COMPANION / permission_component["path"]).read_text() ) self.assertEqual([], permission_set["systemPermissions"]) self.assertEqual(["SendRecordEmail"], permission_set["customPermissions"]) assignment_component = companion_components[ ( "permission_set_assignment_policy", "PingCAP_SDR_Send_Record_Email_SDR", ) ] assignment_policy = json.loads( (MAIL_COMPANION / assignment_component["path"]).read_text() ) # Record email is a general sales affordance, not an SDR-only one: every # profile a person signs in with gets it. The list deliberately omits the # integration/API-only, guest and community profiles, which have no # connected mailbox and must not be able to create EmailMessage rows. self.assertEqual( [ "Alliance & Channel", "CEG - Growth", "Contract Manager", "DM", "Demand Gen", "Demand Gen Manager", "Education", "Finance", "GGC", "IT User", "Legal", "Management", "Management API-Enabled", "Marketing", "Marketing India", "Marketing User", "Minimum Access - Salesforce", "NA&EMEA SDR Lead", "Outsourcing SDR", "PM", "PoC Profile", "Pre-Sales", "Read Only", "Read Only-IT", "SDR", "Sales", "Sales Ops", "Solution Manager", "Standard User", "System Administrator", "System Administrator (SF imported)", "Tech Support", ], assignment_policy["profiles_any_of"], ) self.assertLessEqual( len(assignment_policy["profiles_any_of"]), 64, "assignment policies accept at most 64 profile selectors", ) def test_contact_and_lead_email_affordance_uses_only_the_gmail_composer(self) -> None: actions = json.loads( (FINAL_PACK / "actions" / "invocable_standard_actions.json").read_text() ) for object_api_name in ("Contact", "Lead"): email_actions = [ action for action in actions if action.get("category") == "ActivityTimeline" and action.get("label") == "Email" and object_api_name in (action.get("targetObjects") or []) ] self.assertEqual( ["sendRecordEmail"], [action.get("name") for action in email_actions], object_api_name, ) self.assertEqual(["ApiEnabled"], email_actions[0]["requiredPermissions"], object_api_name) self.assertEqual( "app-pack:crm/action-renderer/email-compose", email_actions[0].get("rendererId"), object_api_name, ) lead_layout = json.loads( (FINAL_PACK / "layouts" / "lead_page_layout.json").read_text() ) lead_quick_actions = [ action.get("name") for action in lead_layout.get("quickActions", []) ] self.assertEqual( ["taskCall", "sendRecordEmail", "taskLeadWeeklyUpdate"], lead_quick_actions, ) def test_l2_task_picklists_accept_the_mail_completion_fixed_values(self) -> None: task_objects = json.loads((L2_PACK / "objects" / "Task.json").read_text()) task = next(obj for obj in task_objects if obj["apiName"] == "Task") fields = {field["apiName"]: field for field in task["fields"]} actions = json.loads( (FINAL_PACK / "actions" / "invocable_standard_actions.json").read_text() ) action = next(item for item in actions if item.get("name") == "sendRecordEmail") activity = action["config"]["outboundMail"]["mailCompletion"]["activity"] for field_key, value_key in ( ("subjectField", "fixedSubject"), ("statusField", "fixedStatus"), ("priorityField", "fixedPriority"), ("typeField", "fixedType"), ("subtypeField", "fixedSubtype"), ): field_name = activity[field_key] allowed_values = { option["value"] for option in fields[field_name]["options"] } self.assertIn(activity[value_key], allowed_values, field_name) def test_sdr_profile_is_a_dispatcher_owned_environment_baseline(self) -> None: seed = yaml.safe_load( ( ROOT / "deployments/offline/runtime/seeds/deploy-baseline.seed.yaml" ).read_text() ) profiles = [ record for record in seed["baseline_records"] if record["object"] == "Profile" and record["match"] == {"Name": "SDR"} ] self.assertEqual(1, len(profiles)) self.assertEqual("pingcap_sdr_profile", profiles[0]["key"]) catalog = json.loads( ( ROOT / "deployments/offline/runtime/seeds/deploy-seed-catalog.json" ).read_text() ) self.assertEqual( {"baseline_records": 25, "objects": 6}, catalog["modules"]["deploy_baseline"]["declared_stats"], ) expected_object_order = tuple( dict.fromkeys(record["object"] for record in seed["baseline_records"]) ) self.assertEqual( expected_object_order, seed_dispatcher.ADAPTERS["deploy_baseline_v1"]["objects"], ) def test_deployment_oauth_credentials_fail_closed_as_one_pair(self) -> None: client_id, client_secret = connected_mail_oauth.CONNECTED_MAIL_OAUTH_ENV_KEYS self.assertEqual({}, connected_mail_oauth.connected_mail_oauth_env({})) with self.assertRaisesRegex(ValueError, "supplied together"): connected_mail_oauth.connected_mail_oauth_env({client_id: "only-one-side"}) with self.assertRaisesRegex(ValueError, "non-canonical"): connected_mail_oauth.connected_mail_oauth_env( {client_id: "client-id", client_secret: "secret\nvalue"} ) self.assertEqual( {client_id: "client-id", client_secret: "client-secret"}, connected_mail_oauth.connected_mail_oauth_env( {client_id: "client-id", client_secret: "client-secret"} ), ) def test_release_workflows_supply_connected_mail_oauth_to_prepare(self) -> None: expected = { "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_ID": ( "${{ secrets.NEXUSCRM_GOOGLE_CLIENT_ID }}" ), "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_SECRET": ( "${{ secrets.NEXUSCRM_GOOGLE_CLIENT_SECRET }}" ), } for relative_path in ( ".github/workflows/deploy-staging-blue-green-environment.yml", ".github/workflows/deploy-multi-prod-d3-environment.yml", ): workflow = (ROOT / relative_path).read_text() for key, authority in expected.items(): self.assertIn(f"{key}: {authority}", workflow, relative_path) def test_every_environment_validates_oauth_before_prepare_side_effects(self) -> None: prepare_sources = { "deployments/offline/runtime/prepare_ci_runtime_env.py": ( "resolved_refs = resolve_prepare_refs(" ), "deployments/offline/runtime/prepare_exp_runtime_env.py": ( "resolved_refs = resolve_prepare_refs(" ), "deployments/offline/runtime/prepare_staging_runtime_env.py": ( "active_slot = infer_active_d3_slot_from_live_conf(env_name)" ), "deploy/online/runtime/prepare_multi_prod_runtime_env.py": ( "outer_env_root = runtime_env_file(args.env).parents[1]" ), } validation = "connected_mail_oauth_env(" for relative_path, first_side_effect in prepare_sources.items(): source = (ROOT / relative_path).read_text() self.assertIn(validation, source, relative_path) validation_block = source.split(validation, 1)[1].split(")", 1)[0] self.assertIn("os.environ", validation_block, relative_path) self.assertLess( source.index(validation), source.index(first_side_effect), relative_path, ) if __name__ == "__main__": unittest.main()