"""6H crm.company/deal get + exact-name search. Generic crm.search stays rejected.""" from __future__ import annotations from uuid import uuid4 from src.domain.connectors.catalog import slugs_implementing from src.domain.connectors.crm_company import ( map_company_capability_input, resolve_crm_company, to_public_crm_company_result, ) from src.domain.connectors.crm_contact import ( decode_crm_cursor, encode_crm_cursor, 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_resolve import CRMConnectionSnapshot, DispatchAuthorization from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, CRM_FAMILY_PLAN, CRM_HYPOTHESIZED_VERBS_NOT_FROZEN, CRM_SEARCH_READ_PHASE_6H, INVOKE_CAPABILITIES, is_implemented, ) from src.domain.connectors.public_gateway import ( PHASE_6_STATUS, PUBLIC_INVOCABLE_CAPABILITIES, strip_capability_response, ) from src.domain.connectors.router import capability_router _CAPS = {"integrations.execute"} def _crm(provider: str, **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, ) data.update(overrides) return CRMConnectionSnapshot(**data) def test_6h_acceptance_matrix() -> None: assert PHASE_6_STATUS == "6p_complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert CRM_FAMILY_PLAN["company_search"] == "name_exact" assert CRM_FAMILY_PLAN["company_search_domain"] == "REJECT" assert CRM_FAMILY_PLAN["deal_search"] == "name_or_stage_exact" assert CRM_FAMILY_PLAN["deal_search_pipeline"] == "REJECT" assert CRM_FAMILY_PLAN["generic_crm_search"] == "REJECT" assert CRM_SEARCH_READ_PHASE_6H["matrix"]["crm.company.get"] == "FREEZE" assert CRM_SEARCH_READ_PHASE_6H["matrix"]["crm.deal.search(name)"] == "FREEZE" assert CRM_SEARCH_READ_PHASE_6H["matrix"]["crm.search"] == "REJECT" assert "crm.search" in CRM_HYPOTHESIZED_VERBS_NOT_FROZEN assert "crm.search" not in INVOKE_CAPABILITIES for verb in ( "crm.company.get", "crm.company.search", "crm.deal.get", "crm.deal.search", ): spec = INVOKE_CAPABILITIES[verb] assert spec["status"] == "implemented" assert spec["invocable"] is True assert spec["method"] == f"POST /api/projects/{{project_id}}/capabilities/{verb}" assert is_implemented(verb) is True slugs = slugs_implementing(verb) assert any(s.startswith("hubspot.") for s in slugs) assert any(s.startswith("salesforce.") for s in slugs) assert verb in PUBLIC_INVOCABLE_CAPABILITIES assert verb not in CRM_HYPOTHESIZED_VERBS_NOT_FROZEN paths = {getattr(route, "path", "") for route in capability_router.routes} assert "/api/projects/{project_id}/capabilities/crm.company.get" in paths assert "/api/projects/{project_id}/capabilities/crm.deal.search" in paths def test_company_get_both_providers_and_object_type() -> None: hs = map_company_capability_input( "crm.company.get", "hubspot", {"company": {"object_type": "company", "external_id": "99"}}, ) assert hs == { "company_id": "99", "properties": ["name", "domain", "phone"], } sf = map_company_capability_input( "crm.company.get", "salesforce", {"company": {"object_type": "company", "external_id": "001xx"}}, ) assert sf == { "account_id": "001xx", "fields": ["Id", "Name", "Phone"], } assert "Website" not in sf["fields"] bad = resolve_crm_company( capability="crm.company.get", payload={"company": {"object_type": "contact", "external_id": "99"}}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ) assert bad.reason_code == "crm_validation_failed" def test_deal_get_both_providers() -> None: hs = map_deal_capability_input( "crm.deal.get", "hubspot", {"deal": {"object_type": "deal", "external_id": "d1"}}, ) assert isinstance(hs, dict) assert hs["deal_id"] == "d1" sf = map_deal_capability_input( "crm.deal.get", "salesforce", {"deal": {"object_type": "deal", "external_id": "006xx"}}, ) assert isinstance(sf, dict) assert sf["opportunity_id"] == "006xx" assert "RecordTypeId" not in sf["fields"] def test_get_not_found_and_search_zero_one_many() -> None: get_hs = resolve_crm_company( capability="crm.company.get", payload={"company": {"object_type": "company", "external_id": "missing"}}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ) missing = to_public_crm_company_result( get_hs, capability="crm.company.get", enqueue={"ok": True, "result": {"properties": {}}} ) assert missing.reason_code == "crm_company_not_found" deal_missing = to_public_crm_deal_result( resolve_crm_deal( capability="crm.deal.get", payload={"deal": {"object_type": "deal", "external_id": "missing"}}, connections=[_crm("salesforce")], enabled_capabilities=_CAPS, ), capability="crm.deal.get", enqueue={"ok": True, "result": {"Name": "Gone"}}, ) assert deal_missing.reason_code == "crm_deal_not_found" empty = to_public_crm_company_result( resolve_crm_company( capability="crm.company.search", payload={"name": "Acme"}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ), capability="crm.company.search", enqueue={"ok": True, "result": {"results": []}}, ) assert empty.status == "succeeded" assert empty.items == [] many = to_public_crm_deal_result( resolve_crm_deal( capability="crm.deal.search", payload={"name": "Renewal"}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ), capability="crm.deal.search", enqueue={ "ok": True, "result": { "results": [ {"id": "1", "properties": {"dealname": "Renewal"}}, {"id": "2", "properties": {"dealname": "Renewal"}}, ] }, }, ) assert many.items is not None assert len(many.items) == 2 def test_company_search_is_exact_name_and_rejects_domain() -> None: hs = map_company_capability_input( "crm.company.search", "hubspot", {"name": "Acme"} ) assert isinstance(hs, dict) assert hs["filter_groups"][0]["filters"][0] == { "propertyName": "name", "operator": "EQ", "value": "Acme", } sf = map_company_capability_input( "crm.company.search", "salesforce", {"name": "Acme"} ) assert sf == { "filters": {"Name": "Acme"}, "fields": ["Id", "Name", "Phone"], "limit": 10, } assert map_company_capability_input( "crm.company.search", "hubspot", {"name": "Acme", "domain": "acme.com"} ) == "crm_field_unsupported" leaked = resolve_crm_company( capability="crm.company.search", payload={"name": "Acme", "filter_groups": []}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ) assert leaked.reason_code == "provider_fields_forbidden" soql = resolve_crm_company( capability="crm.company.search", payload={"name": "Acme", "query": "SELECT Id FROM Account"}, connections=[_crm("salesforce")], enabled_capabilities=_CAPS, ) assert soql.reason_code == "provider_fields_forbidden" def test_deal_search_rejects_pipeline_and_wraps_cursor() -> None: assert ( map_deal_capability_input( "crm.deal.search", "salesforce", {"name": "Renewal", "pipeline": "default"}, ) == "crm_pipeline_unsupported" ) planned = resolve_crm_deal( capability="crm.deal.search", payload={"name": "Renewal"}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ) public = to_public_crm_deal_result( planned, capability="crm.deal.search", enqueue={ "ok": True, "result": { "results": [{"id": "1", "properties": {"dealname": "Renewal"}}], "paging": {"next": {"after": "hs_after_token"}}, }, }, ) assert public.next_cursor != "hs_after_token" assert decode_crm_cursor("hubspot", public.next_cursor or "") == "hs_after_token" stripped = strip_capability_response( {**public.as_dict(), "after": "hs_after_token", "nextRecordsUrl": "/query"} ) assert "after" not in stripped assert "nextRecordsUrl" not in stripped replay = map_deal_capability_input( "crm.deal.search", "hubspot", {"name": "Renewal", "cursor": encode_crm_cursor("hubspot", "hs_after_token")}, ) assert isinstance(replay, dict) assert replay["after"] == "hs_after_token" assert ( map_deal_capability_input( "crm.deal.search", "salesforce", {"name": "Renewal", "cursor": "x"} ) == "crm_validation_failed" ) def test_shared_resolver_and_authorization_on_reads() -> None: two = resolve_crm_company( capability="crm.company.get", payload={"company": {"object_type": "company", "external_id": "1"}}, connections=[_crm("hubspot"), _crm("hubspot")], enabled_capabilities=_CAPS, provider_hint="hubspot", ) assert two.reason_code == "binding_ambiguous" ok = resolve_crm_deal( capability="crm.deal.get", payload={"deal": {"object_type": "deal", "external_id": "1"}}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ) assert ok.ok and ok.plan is not None from src.domain.connectors.crm_deal import dispatch_crm_deal_plan dispatched = dispatch_crm_deal_plan( ok.plan, DispatchAuthorization( allowed=False, reason_code="crm_permission_denied", message="execution not authorized", ), diagnostics=ok.diagnostics, ) assert dispatched.reason_code == "crm_permission_denied" def test_contact_and_queued_regressions() -> None: queued = to_public_crm_company_result( resolve_crm_company( capability="crm.company.get", payload={"company": {"object_type": "company", "external_id": "1"}}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ), capability="crm.company.get", enqueue={"ok": True, "job_id": "job_1", "status": "queued"}, ).as_dict() assert queued == {"capability": "crm.company.get", "status": "queued"} contact = to_public_crm_contact_result( resolve_crm_contact( capability="crm.contact.search", payload={"email": "ada@example.com"}, connections=[_crm("hubspot")], enabled_capabilities=_CAPS, ), capability="crm.contact.search", enqueue={"ok": True, "result": {"results": []}}, ) assert contact.status == "succeeded" assert contact.items == []