"""6K HubSpot implementation completeness. Not a new canonical CRM phase.""" from __future__ import annotations from uuid import uuid4 import pytest from src.domain.connectors.catalog import get_connector, slugs_implementing from src.domain.connectors.crm_association import ( HUBSPOT_DEFAULT_TYPE_ID, map_association_input, resolve_crm_association, ) from src.domain.connectors.crm_canonical import CRMAssociation from src.domain.connectors.crm_company import map_company_capability_input from src.domain.connectors.crm_contact import map_contact_capability_input from src.domain.connectors.crm_deal import map_deal_capability_input from src.domain.connectors.crm_events import ( CRMAttributedConnection, CRMEventError, InMemoryCRMEventDedupe, ingest_verified_crm_inbound, ) from src.domain.connectors.crm_resolve import ( HUBSPOT_NULL_CLEAR, CRMConnectionSnapshot, map_enqueue_error, ) from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, CRM_CONTACT_FROZEN_VERBS, CRM_CONTACT_PHASE_6C, CRM_FAMILY_PLAN, CRM_HUBSPOT_COMPLETENESS_PHASE_6K, INVOKE_CAPABILITIES, is_implemented, ) from src.domain.connectors.public_gateway import PHASE_6_STATUS, PUBLIC_INVOCABLE_CAPABILITIES def _crm(provider: str = "hubspot", **overrides) -> CRMConnectionSnapshot: data = dict( connection_id=uuid4(), connector_key=provider, status="active", has_secrets=True, permission_mode="allow", execution_domain="C", requires_secrets=True, client_side_only=False, config={"portal_id": "111"}, ) data.update(overrides) return CRMConnectionSnapshot(**data) def test_6k_closes_hubspot_without_public_truth() -> None: assert PHASE_6_STATUS == "6p_complete" assert CAPABILITY_CONVERSION_PATTERN["phase_6_status"] == "6p_complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert CRM_FAMILY_PLAN["status"] == "6p_complete" assert CRM_FAMILY_PLAN["next_slice"] == "forms_first_class_product_domain" assert CRM_HUBSPOT_COMPLETENESS_PHASE_6K["not_a_new_canonical_phase"] is True assert CRM_HUBSPOT_COMPLETENESS_PHASE_6K["null_clear"] == "empty_string_crm_v3" assert CRM_HUBSPOT_COMPLETENESS_PHASE_6K["catalog_invocable"] is False assert CRM_HUBSPOT_COMPLETENESS_PHASE_6K["implements_attached"] is False assert CRM_CONTACT_PHASE_6C["hubspot_null_clear"] == "empty_string_crm_v3" assert HUBSPOT_NULL_CLEAR == "empty_string_crm_v3" for verb in CRM_CONTACT_FROZEN_VERBS: spec = INVOKE_CAPABILITIES[verb] assert spec["invocable"] is True assert spec["method"] == f"POST /api/projects/{{project_id}}/capabilities/{verb}" assert is_implemented(verb) is True assert verb in PUBLIC_INVOCABLE_CAPABILITIES slugs = slugs_implementing(verb) assert any(s.startswith("hubspot.") for s in slugs) assert any(s.startswith("salesforce.") for s in slugs) def test_hubspot_create_contact_email_is_optional_in_catalog() -> None: connector = get_connector("hubspot") assert connector is not None assert connector.version == "01.06.00" action = next(row for row in connector.actions if row.key == "create_contact") assert "email" not in action.input_schema.get("required", []) assert action.implements == ("crm.contact.create",) def test_canonical_create_without_email_stays_create() -> None: mapped = map_contact_capability_input( "crm.contact.create", "hubspot", {"contact": {"first_name": "Ada", "properties": {"vip": True}}}, ) assert mapped == { "properties": {"firstname": "Ada", "vip": True}, } def test_null_clear_survives_canonical_mapping() -> None: contact = map_contact_capability_input( "crm.contact.update", "hubspot", { "contact": {"object_type": "contact", "external_id": "1"}, "changes": {"phone": None, "properties": {"custom_note": None}}, }, ) assert isinstance(contact, dict) assert contact["properties"]["phone"] is None assert contact["properties"]["custom_note"] is None company = map_company_capability_input( "crm.company.update", "hubspot", { "company": {"object_type": "company", "external_id": "2"}, "changes": {"phone": None, "properties": {"industry": None}}, }, ) assert isinstance(company, dict) assert company["properties"]["phone"] is None assert company["properties"]["industry"] is None deal = map_deal_capability_input( "crm.deal.update", "hubspot", { "deal": {"object_type": "deal", "external_id": "3"}, "changes": {"amount": None, "properties": {"description": None}}, }, config={}, ) assert isinstance(deal, dict) assert deal["properties"]["amount"] is None assert deal["properties"]["description"] is None def test_association_defaults_and_contact_deal_stay_provider_specific() -> None: assert HUBSPOT_DEFAULT_TYPE_ID == {"contact_company": 1, "deal_company": 5} mapped = map_association_input( "contact_company", "hubspot", CRMAssociation.model_validate( { "from": {"object_type": "contact", "external_id": "c1"}, "to": {"object_type": "company", "external_id": "co1"}, } ), ) assert isinstance(mapped, dict) assert mapped["association_type_id"] == 1 deal_company = map_association_input( "deal_company", "hubspot", CRMAssociation.model_validate( { "from": {"object_type": "deal", "external_id": "d1"}, "to": {"object_type": "company", "external_id": "co1"}, } ), ) assert isinstance(deal_company, dict) assert deal_company["association_type_id"] == 5 rejected = resolve_crm_association( capability="crm.association.create", payload={ "association": { "from": {"object_type": "contact", "external_id": "c1"}, "to": {"object_type": "deal", "external_id": "d1"}, } }, connections=[_crm()], enabled_capabilities={"integrations.execute"}, ) assert rejected.ok is False assert rejected.reason_code == "crm_association_unsupported" def test_hubspot_errors_normalize_for_canonical_callers() -> None: assert map_enqueue_error("scope_missing") == "reauthorization_required" assert map_enqueue_error("crm_not_found", capability="crm.contact.get") == ( "crm_contact_not_found" ) assert map_enqueue_error("crm_not_found", capability="crm.company.get") == ( "crm_company_not_found" ) assert map_enqueue_error("crm_not_found", capability="crm.deal.get") == ( "crm_deal_not_found" ) assert map_enqueue_error("connection_inactive") == "reauthorization_required" def test_inactive_hubspot_portal_is_identified_but_not_dispatched() -> None: row = CRMAttributedConnection( project_id=uuid4(), workspace_id=uuid4(), connection=_crm(status="inactive"), ) with pytest.raises(CRMEventError) as err: ingest_verified_crm_inbound( provider="hubspot", verified=True, payload=[ { "eventId": 9, "subscriptionType": "contact.creation", "objectId": 1, "portalId": 111, } ], connections=[row], dedupe=InMemoryCRMEventDedupe(), ) assert err.value.code == "crm_connection_inactive"