"""6P CRM end-to-end application tests. No new CRM semantics.""" from __future__ import annotations from uuid import uuid4 from src.domain.app_blocks.crm_composition import ( INBOUND_CRM_EVENT_COMPOSITION, LOCAL_LEAD_STAGES, capability_readiness, ) from src.domain.app_blocks.crm_e2e import ( DURABLE_OBJECT_REF_DECISION, PORTABLE_SALES_CAPABILITIES, contact_form_submit, crm_snapshot, fixture_bare_crm_prompt, fixture_contact_form_crm_sync, fixture_contact_form_only, fixture_local_lead_only, fixture_portable_sales_app, invoke_path_is_public, local_p_records, pack_is_selectable, public_bff_body, snapshot_leaks, ) from src.domain.app_blocks.registry import load_default_registry from src.domain.codegen.crm_codegen import select_crm_generation from src.domain.connectors.crm_association import resolve_crm_association from src.domain.connectors.crm_contact import ( resolve_crm_contact, to_public_crm_contact_result, ) from src.domain.connectors.crm_deal import ( map_deal_capability_input, resolve_crm_deal, to_public_crm_deal_result, ) from src.domain.connectors.crm_events import ( FROZEN_CRM_EVENT_TYPES, CRMAttributedConnection, InMemoryCRMEventDedupe, InMemoryCRMOutboundFingerprints, ingest_verified_crm_inbound, record_crm_outbound_mutation, ) from src.domain.connectors.crm_resolve import ( CRMPlan, DispatchAuthorization, dispatch_crm_plan, ) from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, CRM_END_TO_END_PHASE_6P, CRM_FAMILY_PLAN, CRM_PUBLIC_INVOKE_VERBS, ) from src.domain.connectors.public_gateway import PHASE_6_STATUS from src.domain.connectors.router import capability_router from src.domain.feature_packs.catalog_v2 import CATALOG_OVERLAYS from src.main import app _CAPS = {"integrations.execute"} _HS_STAGE = { "stage_map": {"qualified": "appointmentscheduled", "won": "closedwon"}, "pipeline_map": {"sales": "default"}, } _SF_STAGE = {"stage_map": {"qualified": "Qualification", "won": "Closed Won"}} _FIELDS = { "email": "ada@example.com", "first_name": "Ada", "last_name": "Lovelace", "phone": "1", } def test_6p_closes_crm_program() -> 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 CAPABILITY_CONVERSION_PATTERN["closed_families"]["crm.*"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["next_candidates"] == ("email.templates",) assert CAPABILITY_CONVERSION_PATTERN["opened_families"]["forms.*"] == "composition_integration" assert CRM_FAMILY_PLAN["status"] == "6p_complete" assert CRM_FAMILY_PLAN["next_slice"] == "forms_first_class_product_domain" assert CRM_END_TO_END_PHASE_6P["status"] == "complete" assert CRM_END_TO_END_PHASE_6P["durable_object_ref"] == DURABLE_OBJECT_REF_DECISION assert CRM_END_TO_END_PHASE_6P["inbound_block_composition"] == "DEFER" assert len(CRM_PUBLIC_INVOKE_VERBS) == 14 assert DURABLE_OBJECT_REF_DECISION == "DEFERRED" assert INBOUND_CRM_EVENT_COMPOSITION == "DEFER" def test_layer1_contact_form_without_crm() -> None: app_a = fixture_contact_form_only() ids = [block.id for block in app_a["plan"].blocks] assert "forms.contact" in ids assert "crm.lead" not in ids assert app_a["composition"]["optional_capability"] is None assert "crm.contact" not in app_a["source"] assert snapshot_leaks(app_a["source"]) == () result = contact_form_submit(fields=_FIELDS, sync_enabled=False) assert result["local"]["persisted"] is True assert result["crm"] is None assert result["ux"] == "success" def test_layer1_contact_form_plus_upsert_same_for_both_providers() -> None: app_b = fixture_contact_form_crm_sync() ids = [block.id for block in app_b["plan"].blocks] assert "forms.contact" in ids assert "contact-capture" in app_b["plan"].required_feature_packs assert app_b["composition"]["optional_capability"] == "crm.contact.upsert" assert app_b["codegen"].capability == "crm.contact.upsert" hs = select_crm_generation( "Sync form submissions to CRM by email", connected_provider="hubspot" ) sf = select_crm_generation( "Sync form submissions to CRM by email", connected_provider="salesforce" ) assert hs.capability == sf.capability == "crm.contact.upsert" assert hs.invoke_path == sf.invoke_path assert snapshot_leaks(app_b["source"]) == () assert "crm.contact.upsert" in app_b["source"] assert pack_is_selectable("crm-pipeline") is False def test_layer1_local_lead_and_bare_crm_do_not_select_legacy_pipeline() -> None: lead = fixture_local_lead_only() ids = [block.id for block in lead["plan"].blocks] assert "crm.lead" in ids assert "lead-lifecycle" in lead["plan"].required_feature_packs assert lead["composition"]["rewritten_to_crm_contact"] is False assert "crm.contact.create" not in lead["source"] assert "crm.deal.create" not in lead["source"] assert LOCAL_LEAD_STAGES[0] == "new" bare = fixture_bare_crm_prompt() bare_ids = [block.id for block in bare["plan"].blocks] assert "crm.lead" not in bare_ids assert "crm-pipeline" not in bare["plan"].required_feature_packs overlay = CATALOG_OVERLAYS["crm-pipeline"] assert overlay["legacy_aggregate"] is True def test_layer1_no_email_keeps_local_form() -> None: result = contact_form_submit( fields={"first_name": "Ada", "message": "hi"}, sync_enabled=True, ) assert result["local"]["persisted"] is True assert result["crm"] is None assert result["crm_unavailable"] is True assert result["local_rolled_back"] is False def test_layer2_portable_sales_graph_and_escape_hatch() -> None: portable = fixture_portable_sales_app() assert portable["codegen_graph"] == set(PORTABLE_SALES_CAPABILITIES) assert snapshot_leaks(portable["source"]) == () for capability in PORTABLE_SALES_CAPABILITIES: assert invoke_path_is_public(capability) assert capability in portable["source"] soql = select_crm_generation( "Run this Salesforce SOQL query SELECT Id FROM Contact" ) assert soql.surface == "provider_action" assert soql.provider_action == "salesforce.query_soql" label = select_crm_generation("Use HubSpot association label X") assert label.surface == "provider_action" def test_layer2_new_composition_avoids_legacy_sync_function() -> None: source = fixture_contact_form_crm_sync()["source"] assert "contact-capture-crm-sync" not in source assert "CRM_CONNECTION_ID" not in source assert "hubspot.upsert_contact" not in source registry = load_default_registry() block = registry.require("forms.contact") assert any(item.capability == "crm.contact.upsert" for item in block.provider_bindings) def test_layer3_best_effort_failure_isolation() -> None: for reason in ( "crm_unbound", "reauthorization_required", "crm_rate_limited", "crm_provider_unavailable", ): result = contact_form_submit( fields=_FIELDS, sync_enabled=True, crm_enqueue={"ok": False, "reason_code": reason}, ) assert result["local"]["persisted"] is True assert result["local_rolled_back"] is False assert result["ux"] == "success" assert result["crm"]["reason_code"] == reason assert result["crm"]["external_id"] is None def test_layer3_queued_does_not_invent_external_id() -> None: result = contact_form_submit( fields=_FIELDS, sync_enabled=True, crm_enqueue={"ok": True, "status": "queued"}, ) assert result["crm"]["status"] == "queued" assert result["crm"]["external_id"] is None hs = crm_snapshot("hubspot") resolved = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[hs], enabled_capabilities=_CAPS, ) public = to_public_crm_contact_result( resolved, capability="crm.contact.upsert", enqueue={"ok": True} ) assert public.status == "queued" assert public.contact is None def test_layer3_unbound_ambiguous_revoked_and_binding() -> None: unbound = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[], enabled_capabilities=_CAPS, ) assert unbound.reason_code == "crm_unbound" two = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[crm_snapshot("hubspot"), crm_snapshot("hubspot")], enabled_capabilities=_CAPS, ) mixed = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[crm_snapshot("hubspot"), crm_snapshot("salesforce")], enabled_capabilities=_CAPS, ) assert two.reason_code == mixed.reason_code == "binding_ambiguous" revoked = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[crm_snapshot("hubspot", status="inactive")], enabled_capabilities=_CAPS, ) assert revoked.reason_code == "crm_unbound" bound = crm_snapshot("salesforce") recovered = resolve_crm_contact( capability="crm.contact.upsert", payload={"contact": dict(_FIELDS)}, connections=[bound], enabled_capabilities=_CAPS, binding_connection_id=bound.connection_id, ) assert recovered.ok is True assert recovered.plan is not None assert recovered.plan.provider == "salesforce" assert recovered.plan.action == "salesforce.upsert_contact" assert fixture_contact_form_crm_sync()["source"] == fixture_contact_form_crm_sync()["source"] assert capability_readiness("crm.contact.upsert", bound=False) == ( "available_unconfigured" ) def test_layer3_cross_project_connection_isolation() -> None: project_a = crm_snapshot("hubspot") project_b = crm_snapshot("salesforce") resolved = resolve_crm_contact( capability="crm.contact.get", payload={"contact": {"object_type": "contact", "external_id": "x"}}, connections=[project_a], enabled_capabilities=_CAPS, binding_connection_id=project_b.connection_id, ) assert resolved.ok is False plan = CRMPlan( capability="crm.contact.upsert", connection_id=project_a.connection_id, provider="hubspot", action="hubspot.upsert_contact", resolved_input={}, ) denied = dispatch_crm_plan( plan, DispatchAuthorization(allowed=False, reason_code="permission_denied"), ) assert denied.ok is False assert denied.reason_code == "crm_permission_denied" def test_layer3_p_records_are_not_c_results() -> None: local = local_p_records() hs = crm_snapshot("hubspot") search = resolve_crm_contact( capability="crm.contact.search", payload={"email": "local@p2hs.test"}, connections=[hs], enabled_capabilities=_CAPS, ) assert search.ok is True public = to_public_crm_contact_result( search, capability="crm.contact.search", enqueue={ "ok": True, "result": {"results": [], "paging": {"next": {"after": "2"}}}, }, ) dumped = str(public.items) for rows in local.values(): for row in rows: assert row["id"] not in dumped assert public.next_cursor assert "after" not in (public.next_cursor or "") def test_layer3_deal_stage_alias_and_unknown_omitted() -> None: hs_in = map_deal_capability_input( "crm.deal.create", "hubspot", {"deal": {"name": "A", "stage": "qualified"}}, _HS_STAGE, ) sf_in = map_deal_capability_input( "crm.deal.create", "salesforce", {"deal": {"name": "A", "stage": "qualified"}}, _SF_STAGE, ) assert hs_in["properties"]["dealstage"] == "appointmentscheduled" assert sf_in["fields"]["StageName"] == "Qualification" hs = crm_snapshot("hubspot", config=_HS_STAGE) bound = resolve_crm_deal( capability="crm.deal.get", payload={"deal": {"object_type": "deal", "external_id": "1"}}, connections=[hs], enabled_capabilities=_CAPS, ) won = to_public_crm_deal_result( bound, capability="crm.deal.get", enqueue={ "ok": True, "result": { "id": "1", "properties": {"dealname": "A", "dealstage": "closedwon"}, }, }, ) assert won.deal["stage"] == "won" unknown = to_public_crm_deal_result( bound, capability="crm.deal.get", enqueue={ "ok": True, "result": { "id": "1", "properties": {"dealname": "A", "dealstage": "legalreview"}, }, }, ) assert "stage" not in unknown.deal assert "legalreview" not in str(unknown.deal) sf_pipe = map_deal_capability_input( "crm.deal.create", "salesforce", {"deal": {"name": "A", "pipeline": "sales"}}, _SF_STAGE, ) assert sf_pipe == "crm_pipeline_unsupported" def test_layer3_associations_and_unsupported_pair() -> None: payload_cc = { "association": { "from": {"object_type": "contact", "external_id": "c1"}, "to": {"object_type": "company", "external_id": "co1"}, } } payload_dc = { "association": { "from": {"object_type": "deal", "external_id": "d1"}, "to": {"object_type": "company", "external_id": "co1"}, } } for provider in ("hubspot", "salesforce"): row = crm_snapshot(provider) contact_company = resolve_crm_association( capability="crm.association.create", payload=payload_cc, connections=[row], enabled_capabilities=_CAPS, ) deal_company = resolve_crm_association( capability="crm.association.create", payload=payload_dc, connections=[row], enabled_capabilities=_CAPS, ) assert contact_company.ok and deal_company.ok bad = 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_snapshot("hubspot")], enabled_capabilities=_CAPS, ) assert bad.reason_code == "crm_association_unsupported" gap = select_crm_generation("Associate this contact with the deal") assert gap.surface == "gap" def test_layer3_error_normalization_and_secret_strip() -> None: dirty = { "status": "rejected", "reason_code": "crm_rate_limited", "provider": "hubspot", "connection_id": str(uuid4()), "access_token": "secret", "portal_id": "111", "dealstage": "closedwon", "AccountId": "001", "instance_url": "https://x.my.salesforce.com", } public = public_bff_body(dirty) assert public["reason_code"] == "crm_rate_limited" assert "access_token" not in public assert "portal_id" not in public assert "provider" not in public assert "connection_id" not in public paths = {getattr(route, "path", "") for route in capability_router.routes} app_paths = {getattr(route, "path", "") for route in app.routes} for verb in CRM_PUBLIC_INVOKE_VERBS: expected = f"/api/projects/{{project_id}}/capabilities/{verb}" assert expected in paths assert expected in app_paths def test_layer3_inbound_event_plane_not_block_wired() -> None: assert set(FROZEN_CRM_EVENT_TYPES) == { "crm.contact.created", "crm.contact.updated", "crm.company.created", "crm.company.updated", "crm.deal.created", "crm.deal.updated", } overlay = CATALOG_OVERLAYS["contact-capture"] assert not overlay.get("webhook_bindings") hs = CRMAttributedConnection( project_id=uuid4(), workspace_id=uuid4(), connection=crm_snapshot( "hubspot", config={"portal_id": "111", "app_id": "99"} ), ) prints = InMemoryCRMOutboundFingerprints() record_crm_outbound_mutation( connection_id=hs.connection.connection_id, object_type="contact", external_id="9", fields={"email"}, fingerprints=prints, ) echo = ingest_verified_crm_inbound( provider="hubspot", verified=True, payload=[ { "eventId": 1, "subscriptionType": "contact.propertyChange", "objectId": 9, "portalId": 111, "propertyName": "email", "propertyValue": "a@b.com", } ], connections=[hs], dedupe=InMemoryCRMEventDedupe(), fingerprints=prints, ) assert echo.suppressed other = ingest_verified_crm_inbound( provider="hubspot", verified=True, payload=[ { "eventId": 2, "subscriptionType": "contact.propertyChange", "objectId": 9, "portalId": 111, "propertyName": "phone", "propertyValue": "555", } ], connections=[hs], dedupe=InMemoryCRMEventDedupe(), fingerprints=prints, ) assert len(other.accepted) == 1 assert other.accepted[0].event.event_type == "crm.contact.updated" def test_layer3_local_id_is_not_object_ref() -> None: result = contact_form_submit( fields=_FIELDS, sync_enabled=True, crm_enqueue={"ok": True, "result": {"id": "hs_99"}}, ) assert result["local"]["id"] != result["crm"]["external_id"] assert result["ids_distinct"] is True lead = fixture_local_lead_only()["composition"] assert lead["local_entity"] == "lead" assert lead["optional_capability"] is None