import json import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] PACK_ROOT = REPO_ROOT / "hubspot-crm-connector" EXPECTED_PACK_ROOTS = { "gmail-connected-mail-connector", "hubspot-crm-connector", "salesforce-crm-standard", "salesforce-crm-standard-v66", } WORKFLOW_ACTION_CONTACT_FIELDS = [ "email", "salesforceleadid", "salesforcecontactid", "firstname", "lastname", "company", "country_picklist__sfdc_", "hubspot_owner_id", "jobtitle", "leadsource", "lifecyclestage", "no__of_times_mql", "hs_lead_status", "salesforce_industry", "recent_source", "recent_source_details", "lead_source_details", ] HUBSPOT_WEBHOOK_IDEMPOTENCY_COMMON_PARTS = [ ".appId", ".portalId", ".subscriptionId", ".subscriptionType", ".eventId", ".occurredAt", ] def load_json(path: Path): with path.open(encoding="utf-8") as handle: return json.load(handle) def load_text(path: Path): with path.open(encoding="utf-8") as handle: return handle.read() class HubSpotConnectorProfileTest(unittest.TestCase): def test_distribution_contains_only_official_l1_pack_roots(self): pack_roots = { path.name for path in REPO_ROOT.iterdir() if path.is_dir() and (path / "pack.json").exists() } self.assertEqual(pack_roots, EXPECTED_PACK_ROOTS) def test_manifest_declares_outbound_and_inbound_pack_components(self): manifest = load_json(PACK_ROOT / "pack.json") self.assertEqual(manifest["name"], "hubspot-crm-connector") self.assertEqual(manifest["displayName"], "HubSpot CRM Connector") self.assertEqual(manifest["category"], "connector-pack") components = manifest["components"] self.assertEqual( [component["kind"] for component in components], [ "connector_rate_limit", "connector_rate_limit", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "connector_callout_profile", "webhook_secret", "webhook_secret", "connector_capability", "connector_inbound_handler_config", "connector_capability", "connector_inbound_handler_config", "connector_inbound_route", "connector_inbound_route", "connector_inbound_route", "connector_inbound_route", "connector_inbound_route", ], ) self.assertNotIn("named_credential", {component["kind"] for component in components}) self.assertNotIn("connector_flow", {component["kind"] for component in components}) by_key = {(component["kind"], component["businessKey"]): component for component in components} self.assertEqual( by_key[("webhook_secret", "hubspot-crm-connector/webhook")]["path"], "webhook_secrets/webhook.json", ) self.assertEqual( by_key[("webhook_secret", "hubspot-crm-connector/workflow-action")]["path"], "webhook_secrets/workflow_action.json", ) self.assertEqual( by_key[("connector_callout_profile", "hubspot-crm-connector/hubspot-crm-owners")]["path"], "callout/hubspot_crm_owners_profile.json", ) self.assertEqual( by_key[("connector_callout_profile", "hubspot-crm-connector/hubspot-crm-deal-company-write")]["path"], "callout/hubspot_crm_deal_company_write_profile.json", ) for kind in ("signature_verifier", "envelope_parser"): business_key = f"hubspot-crm-connector:{kind}" self.assertEqual( by_key[("connector_capability", business_key)]["payload"], {"namespace": "hubspot-crm-connector", "kind": kind}, ) self.assertEqual( by_key[("connector_inbound_handler_config", business_key)]["path"], f"inbound_handler_configs/{kind}.json", ) routes = { component["businessKey"]: component for component in components if component["kind"] == "connector_inbound_route" } self.assertEqual( set(routes), { "hubspot-crm-connector/contact.creation", "hubspot-crm-connector/contact.associationchange", "hubspot-crm-connector/contact.propertychange", "hubspot-crm-connector/contact.merge", "hubspot-crm-connector/workflow.action.execute", }, ) def test_contact_propertychange_settles_full_record_projection_bursts(self): route = load_json(PACK_ROOT / "routes" / "contact_property_change.json") self.assertEqual(route["eventKind"], "contact.propertychange") self.assertEqual(route["flowDeveloperName"], "HubSpot_Contact_Lead_Upsert") self.assertEqual( route["invocationMode"], "latest_by_source_full_record_preserve_payload_settled", ) self.assertEqual(route["queueClass"], "bulk") def test_profile_binds_to_rate_limit_and_out_of_band_credential(self): rate_limit = load_json(PACK_ROOT / "rate_limits" / "hubspot_crm.json") profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_profile.json") endpoint_class = rate_limit["endpointClass"] self.assertEqual(endpoint_class, "hubspot.crm.objects.read") self.assertEqual(profile["endpointClass"], endpoint_class) credential_ref = "HubSpotPrivateApp" self.assertEqual(profile["credentialRef"], credential_ref) write_rate_limit = load_json(PACK_ROOT / "rate_limits" / "hubspot_crm_write.json") write_profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_contact_write_profile.json") self.assertEqual(write_rate_limit["endpointClass"], "hubspot.crm.objects.write") self.assertEqual(write_profile["endpointClass"], write_rate_limit["endpointClass"]) self.assertEqual(write_profile["credentialRef"], credential_ref) def test_profile_encodes_hubspot_paths_and_retry_contract(self): profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_profile.json") # hubspot-crm is now outbound/shared credential+throttle profile only. self.assertEqual(profile["profileName"], "hubspot-crm") self.assertNotIn("pageSpec", profile) self.assertNotIn("objectPathMap", profile) self.assertNotIn("batchSpec", profile) self.assertEqual(profile["endpointClass"], "hubspot.crm.objects.read") self.assertEqual(profile["credentialRef"], "HubSpotPrivateApp") self.assertEqual(profile["retryPolicy"]["maxAttempts"], 3) self.assertEqual(profile["retryPolicy"]["baseDelay"], 1_000_000_000) self.assertEqual(profile["retryPolicy"]["multiplier"], 2.0) self.assertEqual(profile["retryPolicy"]["maxDelay"], 30_000_000_000) self.assertEqual(profile["retryPolicy"]["retryableStatusCodes"], [429, 500, 502, 503]) # Per-object inbound profiles: each must have exactly 1 objectPathMap entry # and a batchSpec with projectionBodyField set (design #2003). inbound_objects = [ ("contacts", "hubspot_crm_contacts_profile.json"), ("companies", "hubspot_crm_companies_profile.json"), ("deals", "hubspot_crm_deals_profile.json"), ("tickets", "hubspot_crm_tickets_profile.json"), ] for obj_key, filename in inbound_objects: inbound = load_json(PACK_ROOT / "callout" / filename) self.assertEqual(inbound["profileName"], f"hubspot-crm-{obj_key}") self.assertEqual( inbound["objectPathMap"], {obj_key: f"/crm/v3/objects/{obj_key}/batch"}, msg=f"{filename}: objectPathMap must have exactly 1 entry for {obj_key}", ) self.assertEqual(len(inbound["objectPathMap"]), 1, msg=f"{filename}: must have exactly 1 objectPathMap entry") batch_spec = inbound["batchSpec"] self.assertEqual(batch_spec["method"], "POST") self.assertEqual(batch_spec["relativePath"], "/read") self.assertEqual(batch_spec["credentialRef"], "HubSpotPrivateApp") self.assertEqual(batch_spec["endpointClass"], "hubspot.crm.objects.read") self.assertEqual(batch_spec["maxBatchSize"], 100) self.assertEqual(batch_spec["idsRequestField"], "inputs") self.assertEqual(batch_spec["recordsResponsePath"], "results") self.assertEqual(batch_spec["idResponseField"], "id") if obj_key == "contacts": self.assertTrue(batch_spec["allowCanonicalIdAliases"]) else: self.assertNotIn("allowCanonicalIdAliases", batch_spec) self.assertEqual(batch_spec["extraBodyFields"], {"properties": []}) self.assertEqual(batch_spec["projectionBodyField"], "properties") self.assertEqual(batch_spec["recordFlattenKey"], "properties") def test_read_profile_keeps_tenant_property_inventory_out_of_l1(self): # Verified against the contacts inbound profile (representative of all 4 per-object profiles). profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_contacts_profile.json") fetched_properties = set(profile["batchSpec"]["extraBodyFields"]["properties"]) self.assertEqual(fetched_properties, set()) self.assertEqual(profile["batchSpec"]["projectionBodyField"], "properties") def test_contact_write_profile_authorizes_match_search_and_writes(self): manifest = load_json(PACK_ROOT / "pack.json") profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_contact_write_profile.json") profile_component = next( component for component in manifest["components"] if component["businessKey"] == "hubspot-crm-connector/hubspot-crm-contact-write" ) self.assertEqual(profile_component["kind"], "connector_callout_profile") self.assertEqual(profile_component["path"], "callout/hubspot_crm_contact_write_profile.json") self.assertEqual( profile_component["dependsOn"], [ { "kind": "connector_rate_limit", "businessKey": "hubspot-crm-connector/hubspot.crm.objects.write", } ], ) self.assertEqual(profile["profileName"], "hubspot-crm-contact-write") self.assertEqual(profile["retryPolicy"]["maxAttempts"], 1) self.assertNotIn("batchSpec", profile) self.assertNotIn("pageSpec", profile) self.assertEqual( profile["objectPathMap"], { "contacts:create": "/crm/v3/objects/contacts", "contacts:search": "/crm/v3/objects/contacts/search", "contacts:update": "/crm/v3/objects/contacts/{remoteId}", }, ) self.assertEqual( profile["outboundCapabilities"], [ { "operation": "create", "targetObject": "contacts", "method": "POST", "pathTemplate": "/crm/v3/objects/contacts", }, { "operation": "update", "targetObject": "contacts", "method": "PATCH", "pathTemplate": "/crm/v3/objects/contacts/{remoteId}", }, { "operation": "search", "targetObject": "contacts", "method": "POST", "pathTemplate": "/crm/v3/objects/contacts/search", }, ], ) def test_deal_company_write_profile_closes_complete_operation(self): manifest = load_json(PACK_ROOT / "pack.json") profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_deal_company_write_profile.json") component = next( item for item in manifest["components"] if item["businessKey"] == "hubspot-crm-connector/hubspot-crm-deal-company-write" ) self.assertEqual(component["kind"], "connector_callout_profile") self.assertEqual( component["dependsOn"], [ { "kind": "connector_rate_limit", "businessKey": "hubspot-crm-connector/hubspot.crm.objects.write", } ], ) self.assertEqual(profile["profileName"], "hubspot-crm-deal-company-write") self.assertEqual(profile["credentialRef"], "HubSpotPrivateApp") self.assertEqual(profile["endpointClass"], "hubspot.crm.objects.write") self.assertEqual(profile["retryPolicy"]["maxAttempts"], 1) self.assertEqual( profile["objectPathMap"]["companies:search"], "/crm/v3/objects/companies/search", ) self.assertEqual( profile["outboundCapabilities"], [ {"operation": "create", "targetObject": "companies", "method": "POST", "pathTemplate": "/crm/v3/objects/companies"}, {"operation": "update", "targetObject": "companies", "method": "PATCH", "pathTemplate": "/crm/v3/objects/companies/{remoteId}"}, {"operation": "search", "targetObject": "companies", "method": "POST", "pathTemplate": "/crm/v3/objects/companies/search"}, {"operation": "create", "targetObject": "deals", "method": "POST", "pathTemplate": "/crm/v3/objects/deals"}, {"operation": "update", "targetObject": "deals", "method": "PATCH", "pathTemplate": "/crm/v3/objects/deals/{remoteId}"}, { "operation": "associate", "targetObject": "deals", "associationTargetObject": "companies", "method": "PUT", "pathTemplate": "/crm/v3/objects/deals/{fromId}/associations/companies/{toId}/5", }, ], ) def test_owners_profile_reads_hubspot_owner_directory(self): manifest = load_json(PACK_ROOT / "pack.json") profile = load_json(PACK_ROOT / "callout" / "hubspot_crm_owners_profile.json") profile_component = next( component for component in manifest["components"] if component["businessKey"] == "hubspot-crm-connector/hubspot-crm-owners" ) self.assertEqual(profile_component["kind"], "connector_callout_profile") self.assertEqual(profile_component["path"], "callout/hubspot_crm_owners_profile.json") self.assertEqual( profile_component["dependsOn"], [ { "kind": "connector_rate_limit", "businessKey": "hubspot-crm-connector/hubspot.crm.objects.read", } ], ) self.assertEqual(profile["profileName"], "hubspot-crm-owners") self.assertEqual(profile["endpointClass"], "hubspot.crm.objects.read") self.assertEqual(profile["credentialRef"], "HubSpotPrivateApp") self.assertEqual(profile["objectPathMap"], {"owners": "/crm/owners/2026-03"}) self.assertNotIn("batchSpec", profile) self.assertNotIn("pageSpec", profile) self.assertEqual(profile["retryPolicy"]["maxAttempts"], 3) self.assertEqual(profile["retryPolicy"]["retryableStatusCodes"], [429, 500, 502, 503]) def test_readme_discloses_current_runtime_boundaries(self): readme = load_text(PACK_ROOT / "README.md") self.assertIn( "credential definition must be provisioned with `OwnerPack =\n hubspot-crm-connector`", readme, ) self.assertIn( "does not claim greenfield batch/page execution", readme, ) self.assertIn( "batchSpec` for inbound enrichment", readme, ) self.assertIn("downstream app/tenant packs", readme) self.assertIn("projectionBodyField", readme) self.assertNotIn("full contact " + "properties", readme) self.assertIn( "recordFlattenKey", readme, ) self.assertIn( "`latest_by_source_full_record_preserve_payload_settled`", readme, ) self.assertIn( "`github.com/pingcap-inc/nexuscrm-connectors` `v0.1.10` or later", readme, ) self.assertIn( "NexusCRM #3547", readme, ) self.assertIn("hubspot-crm-contact-write", readme) self.assertIn("hubspot-crm-deal-company-write", readme) self.assertIn("outboundCapabilities", readme) self.assertIn( "Do not use `namedcredential.Service.UpsertDefinition`", readme, ) self.assertIn( "runtime throttler cannot read this pack's `connector_rate_limit` row", readme, ) self.assertIn("WEBHOOK_BASE_URL", readme) self.assertIn("contact.creation` -> `HubSpot_Contact_Lead_Upsert", readme) self.assertIn("contact.propertychange` -> `HubSpot_Contact_Lead_Upsert", readme) self.assertNotIn("HubSpot_Company_Account_Upsert", readme) self.assertNotIn("HubSpot_Deal_Opportunity_Upsert", readme) self.assertIn("tenant overlay/app-pack payload outside this repository", readme) self.assertNotIn("companion pack", readme) self.assertIn("{RequestURL}", readme) self.assertIn("composite selector rules", readme) self.assertIn("`subscriptionType=object.creation`", readme) self.assertIn("`subscriptionType=object.propertyChange`", readme) self.assertIn("Deletion events are parsed", readme) self.assertIn("default deletion write route for contacts, companies, or deals", readme) self.assertIn("workflow-action", readme) self.assertIn("workflow.action.execute", readme) self.assertIn("external HubSpot-side adapter", readme) self.assertIn("unique id for the specific execution", readme) self.assertIn("`portalId`", readme) self.assertIn("not a second idempotency-key source", readme) self.assertIn("Verified ingress receipts", readme) self.assertIn("deliveryIdentityCollectionKeyPaths", readme) self.assertIn("deliveryIdentityBodyPath", readme) def test_webhook_secret_definition_stores_no_secret_bytes(self): secret = load_json(PACK_ROOT / "webhook_secrets" / "webhook.json") action_secret = load_json(PACK_ROOT / "webhook_secrets" / "workflow_action.json") self.assertEqual(secret["name"], "webhook") self.assertEqual(secret["algorithm"], "sha256") self.assertEqual(action_secret["name"], "workflow-action") self.assertEqual(action_secret["algorithm"], "hmac-sha256") forbidden = {"secret", "secretValue", "value", "token", "portalId"} self.assertTrue(forbidden.isdisjoint(secret)) self.assertTrue(forbidden.isdisjoint(action_secret)) def test_signature_handler_config_declares_endpoint_specific_hubspot_schemes(self): payload = load_json(PACK_ROOT / "inbound_handler_configs" / "signature_verifier.json") self.assertEqual(payload["namespace"], "hubspot-crm-connector") self.assertEqual(payload["capabilityKind"], "signature_verifier") config_set = payload["config"] self.assertEqual(set(config_set), {"endpointConfigs"}) self.assertEqual(set(config_set["endpointConfigs"]), {"webhook", "workflow-action"}) config = config_set["endpointConfigs"]["webhook"] self.assertEqual(config["scheme"], "sha256-secret-prefix-template") self.assertEqual(config["signingTemplate"], "{RawBody}") self.assertNotIn("timestampHeader", config) self.assertEqual(config["signatureHeader"], "X-HubSpot-Signature") self.assertEqual(config["signatureEncoding"], "hex") self.assertTrue(config["verifiedIngressReceipt"]) self.assertEqual(config["deliveryIdentityCollectionPath"], ".") self.assertEqual( config["deliveryIdentityCollectionKeyPaths"], [ ".portalId", ".appId", ".subscriptionId", ".eventId", ".occurredAt", ".subscriptionType", ".attemptNumber", ], ) self.assertNotIn("deliveryIdentityBodyPath", config) self.assertNotIn("receiptHeaderAllowlist", config) self.assertNotIn("skewWindow", config) self.assertEqual(config["secretName"], "webhook") self.assertNotIn("methodHeader", config) self.assertNotIn("uriHeader", config) self.assertNotIn("deliveryIdentityEventArrayPath", config) self.assertNotIn("deliveryIdentityEventKeyPaths", config) action_config = config_set["endpointConfigs"]["workflow-action"] self.assertEqual(action_config["scheme"], "hmac-sha256-template") self.assertEqual(action_config["signingTemplate"], "{Method}{RequestURL}{RawBody}{Timestamp}") self.assertEqual(action_config["timestampHeader"], "X-HubSpot-Request-Timestamp") self.assertEqual(action_config["signatureHeader"], "X-HubSpot-Signature-v3") self.assertEqual(action_config["signatureEncoding"], "base64") self.assertTrue(action_config["verifiedIngressReceipt"]) self.assertEqual(action_config["deliveryIdentityBodyPath"], ".callbackId") self.assertNotIn("deliveryIdentityCollectionPath", action_config) self.assertNotIn("deliveryIdentityCollectionKeyPaths", action_config) self.assertNotIn("receiptHeaderAllowlist", action_config) self.assertEqual(action_config["skewWindow"], "5m") self.assertEqual(action_config["secretName"], "workflow-action") self.assertNotIn("deliveryIdentityEventArrayPath", action_config) self.assertNotIn("deliveryIdentityEventKeyPaths", action_config) def test_envelope_parser_declares_supported_hubspot_events(self): payload = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") typed_subscription_types = [ "contact.creation", "contact.propertyChange", "contact.deletion", "contact.associationChange", "contact.merge", "contact.restore", "contact.privacyDeletion", "company.creation", "company.propertyChange", "company.deletion", "company.associationChange", "company.merge", "company.restore", "deal.creation", "deal.propertyChange", "deal.deletion", "deal.associationChange", "deal.merge", "deal.restore", "ticket.creation", "ticket.propertyChange", "ticket.deletion", "ticket.associationChange", "ticket.merge", "ticket.restore", ] expected_selectors = [ ((".subscriptionType", "object.creation"), (".objectTypeId", "0-1")), ((".subscriptionType", "object.merge"), (".objectTypeId", "0-1")), ((".subscriptionType", "object.propertyChange"), (".objectTypeId", "0-1")), ((".subscriptionType", "object.merge"), (".objectTypeId", "0-2")), ((".subscriptionType", "object.propertyChange"), (".objectTypeId", "0-2")), ((".subscriptionType", "object.merge"), (".objectTypeId", "0-3")), ((".subscriptionType", "object.propertyChange"), (".objectTypeId", "0-3")), ] + [( (".subscriptionType", value), ) for value in typed_subscription_types] self.assertEqual(payload["namespace"], "hubspot-crm-connector") self.assertEqual(payload["capabilityKind"], "envelope_parser") config_set = payload["config"] self.assertEqual(set(config_set), {"endpointConfigs"}) self.assertEqual(set(config_set["endpointConfigs"]), {"webhook", "workflow-action"}) config = config_set["endpointConfigs"]["webhook"] self.assertEqual(config["eventArrayPath"], ".") selectors = [] for rule in config["rules"]: if "matchAll" in rule: selectors.append(tuple((cond["path"], cond["equals"]) for cond in rule["matchAll"])) else: selectors.append(((rule["matchPath"], rule["matchEquals"]),)) self.assertEqual(selectors, expected_selectors) self.assertEqual(len(config["rules"]), len(expected_selectors)) object_kinds = { "contact": "contacts", "company": "companies", "deal": "deals", "ticket": "tickets", } generic_object_type_rules = { ("object.creation", "0-1"): ("contacts", "contact.creation"), ("object.merge", "0-1"): ("contacts", "contact.merge"), ("object.merge", "0-2"): ("companies", "company.merge"), ("object.merge", "0-3"): ("deals", "deal.merge"), ("object.propertyChange", "0-1"): ("contacts", "contact.propertychange"), ("object.propertyChange", "0-2"): ("companies", "company.propertychange"), ("object.propertyChange", "0-3"): ("deals", "deal.propertychange"), } for rule in config["rules"]: self.assertTrue(rule["namespaceFromCtx"]) self.assertNotIn("idempotencyKeyPath", rule) self.assertIn("idempotencyKeyComposite", rule) if "matchAll" in rule: conditions = {cond["path"]: cond["equals"] for cond in rule["matchAll"]} self.assertIn( conditions[".subscriptionType"], {"object.creation", "object.merge", "object.propertyChange"}, ) object_kind, event_kind = generic_object_type_rules[ (conditions[".subscriptionType"], conditions[".objectTypeId"]) ] self.assertEqual(rule["objectKind"], object_kind) self.assertEqual(rule["eventTypeLiteral"], event_kind) match_equals = conditions[".subscriptionType"] else: self.assertEqual(rule["matchPath"], ".subscriptionType") object_name, _ = rule["matchEquals"].split(".", maxsplit=1) self.assertEqual(rule["objectKind"], object_kinds[object_name]) # CRM object routes use object-specific event kinds; tickets # keep the older generic record.* literals from origin/main. if rule["objectKind"] in ("contacts", "companies", "deals"): self.assertEqual(rule["eventTypeLiteral"], rule["matchEquals"].lower()) match_equals = rule["matchEquals"] self.assertNotIn("eventTypeFromPath", rule) if match_equals.endswith(".associationChange"): self.assertEqual(rule["sourceIdPath"], ".fromObjectId") else: self.assertEqual(rule["sourceIdPath"], ".objectId") projection = {field["name"]: field for field in rule["payloadProjection"]} self.assertEqual(len(projection), len(rule["payloadProjection"])) for field in ( "eventId", "subscriptionId", "portalId", "occurredAt", "objectId", "fromObjectId", "propertyName", "propertyValue", "mergedObjectIds", "newObjectId", "numberOfPropertiesMoved", "privacyDeletion", ): self.assertEqual(projection[field]["path"], f".{field}") for field in ("objectId", "fromObjectId", "toObjectId", "newObjectId"): self.assertEqual(projection[field]["mode"], "scalar_string") self.assertEqual(projection["mergedObjectIds"]["mode"], "scalar_string_array") if match_equals.endswith(".merge"): self.assertEqual(projection["primaryObjectId"]["path"], ".primaryObjectId") self.assertEqual(projection["primaryObjectId"]["mode"], "scalar_string") def test_inbound_config_uses_l0_wire_schema_not_local_shorthand(self): signature = load_json(PACK_ROOT / "inbound_handler_configs" / "signature_verifier.json") envelope = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") self.assertEqual(set(signature), {"namespace", "capabilityKind", "config"}) self.assertEqual(set(envelope), {"namespace", "capabilityKind", "config"}) self.assertEqual(set(signature["config"]["endpointConfigs"]), {"webhook", "workflow-action"}) self.assertEqual(set(envelope["config"]["endpointConfigs"]), {"webhook", "workflow-action"}) for rule in envelope["config"]["endpointConfigs"]["webhook"]["rules"]: forbidden_shorthand = {"subscriptionType", "eventKind", "objectType"} self.assertTrue(forbidden_shorthand.isdisjoint(rule)) has_match_all = "matchAll" in rule has_single_selector = "matchPath" in rule or "matchEquals" in rule self.assertNotEqual(has_match_all, has_single_selector) if has_match_all: self.assertNotIn("matchPath", rule) self.assertNotIn("matchEquals", rule) self.assertEqual( {cond["path"] for cond in rule["matchAll"]}, {".subscriptionType", ".objectTypeId"}, ) else: self.assertIn("matchPath", rule) self.assertIn("matchEquals", rule) self.assertIn("eventTypeLiteral", rule) self.assertNotIn("eventTypeFromPath", rule) self.assertIn("objectKind", rule) self.assertIn("sourceIdPath", rule) self.assertNotIn("idempotencyKeyPath", rule) self.assertIn("idempotencyKeyComposite", rule) self.assertIsInstance(rule["payloadProjection"], list) projection = { field["name"]: field for field in envelope["config"]["endpointConfigs"]["webhook"]["rules"][0]["payloadProjection"] } for field in ( "eventId", "subscriptionId", "portalId", "occurredAt", "objectId", "fromObjectId", "propertyName", "propertyValue", "mergedObjectIds", "newObjectId", "numberOfPropertiesMoved", "privacyDeletion", ): self.assertEqual(projection[field]["path"], f".{field}") for field in ("objectId", "fromObjectId", "toObjectId", "newObjectId"): self.assertEqual(projection[field]["mode"], "scalar_string") self.assertEqual(projection["mergedObjectIds"]["mode"], "scalar_string_array") def test_ordinary_webhook_rules_use_stable_ordered_composite_identity(self): envelope = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") rules = envelope["config"]["endpointConfigs"]["webhook"]["rules"] self.assertEqual(len(rules), 32) for index, rule in enumerate(rules): with self.subTest(index=index, event_type=rule["eventTypeLiteral"]): self.assertNotIn("idempotencyKeyPath", rule) composite = rule["idempotencyKeyComposite"] self.assertEqual(set(composite), {"version", "encoding", "parts"}) self.assertEqual(composite["version"], "v1") self.assertEqual(composite["encoding"], "sha256") part_paths = [part["path"] for part in composite["parts"]] self.assertEqual( part_paths[: len(HUBSPOT_WEBHOOK_IDEMPOTENCY_COMMON_PARTS)], HUBSPOT_WEBHOOK_IDEMPOTENCY_COMMON_PARTS, ) self.assertEqual(len(part_paths), len(set(part_paths))) self.assertNotIn(".attemptNumber", part_paths) if "matchAll" in rule: subscription_type = next( condition["equals"] for condition in rule["matchAll"] if condition["path"] == ".subscriptionType" ) expected_specific = [".objectTypeId"] else: subscription_type = rule["matchEquals"] expected_specific = [] if subscription_type.endswith(".propertyChange"): expected_specific.extend([".objectId", ".propertyName"]) elif subscription_type.endswith(".merge"): expected_specific.extend([".objectId", ".primaryObjectId"]) elif subscription_type.endswith(".associationChange"): expected_specific.extend( [ ".fromObjectId", ".toObjectId", ".associationType", ".associationRemoved", ] ) else: expected_specific.append(".objectId") self.assertEqual( part_paths[len(HUBSPOT_WEBHOOK_IDEMPOTENCY_COMMON_PARTS) :], expected_specific, ) match_paths = { condition["path"] for condition in rule.get("matchAll", []) } projection_by_path = { field["path"]: field for field in rule["payloadProjection"] } for path in part_paths: declared_by_rule = ( path in match_paths or path == rule["sourceIdPath"] or path in projection_by_path ) self.assertTrue(declared_by_rule, f"undeclared identity part {path}") if path in projection_by_path: self.assertNotEqual( projection_by_path[path].get("mode"), "scalar_string_array", f"identity part {path} must be scalar", ) def test_ordinary_webhook_composite_values_ignore_retry_attempt(self): envelope = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") rules = envelope["config"]["endpointConfigs"]["webhook"]["rules"] base_event = { "appId": 1001, "portalId": 2002, "subscriptionId": 3003, "subscriptionType": "contact.creation", "eventId": 4004, "occurredAt": 1723651850844, "attemptNumber": 0, "objectTypeId": "0-1", "objectId": 5005, "propertyName": "leadsource", "primaryObjectId": 6006, "fromObjectId": 7007, "toObjectId": 8008, "associationType": "CONTACT_TO_COMPANY", "associationRemoved": False, } for index, rule in enumerate(rules): part_paths = [ part["path"] for part in rule["idempotencyKeyComposite"]["parts"] ] event = dict(base_event) if "matchAll" in rule: conditions = { condition["path"]: condition["equals"] for condition in rule["matchAll"] } event["subscriptionType"] = conditions[".subscriptionType"] event["objectTypeId"] = conditions[".objectTypeId"] else: event["subscriptionType"] = rule["matchEquals"] def selected_values(payload): return tuple(payload[path.removeprefix(".")] for path in part_paths) original = selected_values(event) retry = dict(event, attemptNumber=9) with self.subTest(index=index, mutation="attemptNumber"): self.assertEqual(selected_values(retry), original) for path in part_paths: field = path.removeprefix(".") changed = dict(event) value = changed[field] changed[field] = not value if isinstance(value, bool) else f"{value}-changed" with self.subTest(index=index, mutation=path): self.assertNotEqual(selected_values(changed), original) def test_workflow_action_execute_endpoint_and_action_definition_contract(self): signature = load_json(PACK_ROOT / "inbound_handler_configs" / "signature_verifier.json") envelope = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") route = load_json(PACK_ROOT / "routes" / "workflow_action_execute.json") action = load_json( PACK_ROOT / "workflow_actions" / "add_to_nexuscrm_campaign.action.json" ) runbook = load_text(PACK_ROOT / "runbooks" / "register_workflow_action.md") signature_config = signature["config"]["endpointConfigs"]["workflow-action"] self.assertEqual(signature_config["secretName"], "workflow-action") self.assertEqual(signature_config["signingTemplate"], "{Method}{RequestURL}{RawBody}{Timestamp}") self.assertTrue(signature_config["verifiedIngressReceipt"]) self.assertEqual(signature_config["deliveryIdentityBodyPath"], ".callbackId") action_config = envelope["config"]["endpointConfigs"]["workflow-action"] self.assertEqual(action_config["eventObjectPath"], ".") self.assertNotIn("eventArrayPath", action_config) self.assertEqual(action_config["inboundProcessingMode"], "payload_only_allowed") self.assertEqual(len(action_config["rules"]), 1) rule = action_config["rules"][0] self.assertEqual(rule["eventTypeLiteral"], "workflow.action.execute") self.assertTrue(rule["namespaceFromCtx"]) self.assertEqual(rule["objectKind"], "workflow-actions") self.assertEqual(rule["sourceIdPath"], ".object.objectId") self.assertNotIn("sourceLaneObjectKind", rule) self.assertNotIn("sourceLaneIdPath", rule) self.assertNotIn("targetMutationLaneObjectKind", rule) self.assertNotIn("targetMutationLaneIdPath", rule) self.assertEqual(rule["idempotencyKeyPath"], ".callbackId") self.assertNotIn("idempotencyKeyTemplate", rule) projection = {field["name"]: field["path"] for field in rule["payloadProjection"]} workflow_context_projection = { "callbackId": ".callbackId", "portalId": ".origin.portalId", "objectId": ".object.objectId", "objectType": ".object.objectType", "enrollmentId": ".origin.actionExecutionIndexIdentifier.enrollmentId", "workflowId": ".context.workflowId", "actionId": ".context.actionId", "actionDefinitionId": ".origin.actionDefinitionId", "campaign_id": ".inputFields.campaign_id", "status": ".inputFields.status", } expected_projection = dict(workflow_context_projection) expected_projection.update( {field: f".inputFields.{field}" for field in WORKFLOW_ACTION_CONTACT_FIELDS} ) self.assertEqual(projection, expected_projection) self.assertEqual(route["namespace"], "hubspot-crm-connector") self.assertEqual(route["eventKind"], "workflow.action.execute") self.assertEqual( route["flowDeveloperName"], "HubSpot_Workflow_Action_CampaignMember_Upsert", ) self.assertEqual(route["queueClass"], "interactive") self.assertEqual( action["actionUrl"], "/webhooks/hubspot-crm-connector/workflow-action", ) self.assertFalse(action["published"]) self.assertEqual(action["objectTypes"], ["CONTACT"]) self.assertEqual( [field["typeDefinition"]["name"] for field in action["inputFields"]], ["campaign_id", "status", *WORKFLOW_ACTION_CONTACT_FIELDS], ) by_name = {field["typeDefinition"]["name"]: field for field in action["inputFields"]} self.assertEqual( by_name["campaign_id"]["typeDefinition"]["optionsUrl"], "/hubspot/workflow-actions/options/campaigns?options_token=", ) self.assertEqual( by_name["status"]["typeDefinition"]["optionsUrl"], "/hubspot/workflow-actions/options/statuses?options_token=", ) for field_name in WORKFLOW_ACTION_CONTACT_FIELDS: with self.subTest(field_name=field_name): self.assertEqual(by_name[field_name]["supportedValueTypes"], ["OBJECT_PROPERTY"]) self.assertEqual(by_name[field_name]["isRequired"], field_name == "email") self.assertIn(field_name, action["labels"]["en"]["inputFieldLabels"]) self.assertIn(field_name, action["labels"]["en"]["inputFieldDescriptions"]) self.assertEqual( action["inputFieldDependencies"], [ { "dependencyType": "SINGLE_FIELD", "controllingFieldName": "campaign_id", "dependentFieldNames": ["status"], } ], ) action_text = json.dumps(action) self.assertNotIn("HUBSPOT_DEVELOPER_API_KEY", action_text) self.assertNotIn("hapikey", action_text) self.assertNotIn("secretValue", action_text) self.assertIn("external HubSpot-side adapter", runbook) self.assertIn("store a second copy of campaign or status facts", runbook) def test_named_credential_runbook_stamps_owner_pack_without_secret(self): runbook = load_text(PACK_ROOT / "runbooks" / "provision_named_credential.sql") self.assertIn("INSERT INTO _System_NamedCredential", runbook) self.assertIn("'HubSpotPrivateApp'", runbook) self.assertIn("'https://api.hubapi.com'", runbook) self.assertIn("'hubspot-crm-connector'", runbook) self.assertIn("@hubspot_named_credential_created_by_id", runbook) self.assertIn("ProfileId = 'system_admin'", runbook) self.assertNotIn("NULL,\n CURRENT_TIMESTAMP", runbook) self.assertIn("OwnerPack = VALUES(OwnerPack)", runbook) self.assertNotIn("secretManagerValue", runbook) def test_routes_match_envelope_event_kinds_and_target_flows(self): manifest = load_json(PACK_ROOT / "pack.json") envelope = load_json(PACK_ROOT / "inbound_handler_configs" / "envelope_parser.json") endpoint_configs = envelope["config"]["endpointConfigs"] event_kinds = { rule["eventTypeLiteral"] for config in endpoint_configs.values() for rule in config["rules"] } expected_routes = { "contact.creation": ("HubSpot_Contact_Lead_Upsert", "contact_creation.json"), "contact.associationchange": ( "HubSpot_Contact_Company_Association", "contact_association_change.json", ), "contact.propertychange": ("HubSpot_Contact_Lead_Upsert", "contact_property_change.json"), "contact.merge": ("HubSpot_Contact_Merge_Reconcile", "contact_merge.json"), "workflow.action.execute": ( "HubSpot_Workflow_Action_CampaignMember_Upsert", "workflow_action_execute.json", ), } for event_kind, (flow_name, path) in expected_routes.items(): self.assertIn(event_kind, event_kinds) route = load_json(PACK_ROOT / "routes" / path) self.assertEqual(route["namespace"], "hubspot-crm-connector") self.assertEqual(route["eventKind"], event_kind) self.assertEqual(route["flowDeveloperName"], flow_name) if event_kind == "contact.propertychange": self.assertEqual( route["invocationMode"], "latest_by_source_full_record_preserve_payload_settled", ) self.assertEqual(route["queueClass"], "bulk") elif event_kind == "workflow.action.execute": self.assertEqual(route["queueClass"], "interactive") else: self.assertNotIn("invocationMode", route) self.assertNotIn("queueClass", route) route_component_keys = { component["businessKey"] for component in manifest["components"] if component["kind"] == "connector_inbound_route" } self.assertEqual( route_component_keys, { "hubspot-crm-connector/contact.creation", "hubspot-crm-connector/contact.associationchange", "hubspot-crm-connector/contact.propertychange", "hubspot-crm-connector/contact.merge", "hubspot-crm-connector/workflow.action.execute", }, ) self.assertFalse( { "hubspot-crm-connector/contact.deletion", "hubspot-crm-connector/company.deletion", "hubspot-crm-connector/deal.deletion", } & route_component_keys ) if __name__ == "__main__": unittest.main()