"""Phase 0 public gateway: Dashboard/BFF is public; Core stays private.""" from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from src.domain.cloud.schemas import ( AuthorizeResponseAllowed, GrantBundle, PolicySnapshot, ) from src.domain.codegen.platform_backend import compact_platform_backend_for_prompt from src.domain.codegen.prompt_wrapper import ( _P2HS_CLOUD_CONTRACT, _P2HS_CORE_API_GUIDANCE, _app_recipes, ) from src.domain.connectors.email_send_execute import enqueue_authorized_email_send from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, MESSAGING_SEND_INVOKE_METHOD, ) from src.domain.connectors.public_gateway import ( FORBIDDEN_CODEGEN_CORE_MARKERS, PUBLIC_BFF_STAGED_CAPABILITIES, PUBLIC_BFF_SURFACES, PUBLIC_INVOCABLE_CAPABILITIES, is_public_bff_proxy_capability, is_public_invocable_capability, invoke_path_has_host, public_capability_invoke_method, public_capability_invoke_path, strip_capability_response, ) def test_phase_0_gateway_is_implemented() -> None: pattern = CAPABILITY_CONVERSION_PATTERN assert pattern["phase_0_status"] == "complete" assert pattern["phase_0a_status"] == "complete" assert pattern["current_program_phase"] == 8 assert "proxying" in pattern["bff_owns"] assert pattern["opened_families"]["automation.workflow.trigger"] == "A" def test_public_bff_includes_existing_connector_and_integration_surfaces() -> None: kinds = {row["kind"] for row in PUBLIC_BFF_SURFACES} assert kinds == { "capabilities", "project_connectors", "connector_catalog", "workspace_connector_defaults", "account_linking", } by_kind = {row["kind"]: row for row in PUBLIC_BFF_SURFACES} assert by_kind["connector_catalog"]["path"] == "/api/connectors" assert by_kind["workspace_connector_defaults"]["path"].startswith("/api/workspaces/") assert by_kind["account_linking"]["path"] == "/api/integrations" assert by_kind["project_connectors"]["path"] == "/api/projects/{project_id}/connectors" assert "builder" in by_kind["project_connectors"]["ui_alias"] for row in PUBLIC_BFF_SURFACES: assert invoke_path_has_host(row["path"]) is False def test_document_verbs_are_public_invocable_codegen_defaults() -> None: for verb in ( "document.create", "document.get", "document.insert_text", "document.append_text", "document.replace_text", ): assert verb not in PUBLIC_BFF_STAGED_CAPABILITIES assert is_public_bff_proxy_capability(verb) is True assert is_public_invocable_capability(verb) is True assert "messaging.sms.send" not in PUBLIC_BFF_STAGED_CAPABILITIES assert is_public_bff_proxy_capability("messaging.sms.send") is False assert is_public_bff_proxy_capability("notification.sms") is False def test_public_invoke_paths_are_host_relative() -> None: for capability in PUBLIC_INVOCABLE_CAPABILITIES: method = public_capability_invoke_method(capability) path = public_capability_invoke_path("proj_1", capability) assert invoke_path_has_host(method) is False assert invoke_path_has_host(path) is False assert method.startswith("POST /api/projects/") assert path == f"/api/projects/proj_1/capabilities/{capability}" assert invoke_path_has_host(MESSAGING_SEND_INVOKE_METHOD) is False def test_strip_keeps_reason_codes_and_drops_plan_internals() -> None: public = strip_capability_response( { "capability": "messaging.send", "status": "rejected", "reason_code": "capability_not_bound", "message": "No eligible messaging.send connection", "action": "slack.post_message", "connection_id": "conn_1", "provider": "slack", "plan": {"action_slug": "slack.post_message"}, } ) assert public["reason_code"] == "capability_not_bound" assert public["message"] == "No eligible messaging.send connection" assert "action" not in public assert "connection_id" not in public assert "provider" not in public assert "plan" not in public sms = strip_capability_response( { "capability": "messaging.sms.send", "status": "queued", "message_id": "job_1", "twilio_sid": "SM123", "from_number": "+15551234567", "messaging_service_sid": "MG1", } ) assert sms == { "capability": "messaging.sms.send", "status": "queued", "message_id": "job_1", } def test_codegen_forbids_core_hosts() -> None: blobs = ( _app_recipes(), _P2HS_CLOUD_CONTRACT, _P2HS_CORE_API_GUIDANCE, ) joined = "\n".join(blobs) for marker in FORBIDDEN_CODEGEN_CORE_MARKERS: # Markers appear only as prohibitions, never as call targets. if marker in joined: assert "Never construct a Core host" in joined or "never a Core host" in joined.lower() assert "beapipie.p2hs.com" not in _app_recipes() assert "P2HS_BACKEND_URL" in _P2HS_CORE_API_GUIDANCE assert "Do not construct a Core host" in _P2HS_CORE_API_GUIDANCE assert "POST /api/projects/{project_id}/capabilities/{capability}" in _P2HS_CORE_API_GUIDANCE compact = compact_platform_backend_for_prompt({ "rules": [ "Never construct a Core host from P2HS_BACKEND_URL or beapipie.", ] }) assert any("Never construct a Core host" in rule for rule in compact["rules"]) @pytest.mark.asyncio async def test_email_send_capability_enqueues_cloud_job_without_rewriting_policy() -> None: allowed = AuthorizeResponseAllowed( authorization_id="auth_1", grant=GrantBundle(jwt="jwt", expires_at="2030-01-01T00:00:00Z"), policy_snapshot=PolicySnapshot(capability="email.send", provider="resend"), ) cloud_svc = SimpleNamespace(authorize=AsyncMock(return_value=allowed)) cloud_client = SimpleNamespace( enqueue_control_job=AsyncMock( return_value={"ok": True, "job_id": "job_email_1", "status": "queued"} ) ) project = SimpleNamespace(id=uuid4(), project_ref=None) result = await enqueue_authorized_email_send( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=uuid4(), payload={"to": "a@example.com", "subject": "Hi", "text": "Hello"}, idempotency_key="idem-email-1", ) assert result == {"ok": True, "job_id": "job_email_1", "status": "queued"} auth_req = cloud_svc.authorize.await_args.args[0] assert auth_req.request_type == "email.send" assert auth_req.handshake_id == "core_capability_invoke" enqueue = cloud_client.enqueue_control_job.await_args.kwargs assert enqueue["request_type"] == "email.send" assert enqueue["payload"]["subject"] == "Hi" assert "action" not in enqueue["payload"] assert "connection_id" not in enqueue["payload"] @pytest.mark.asyncio async def test_bff_to_core_to_cloud_enqueue_shape_for_messaging() -> None: """Public path has no Core host; enqueue stays integration.execute after Core resolve.""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute allowed = AuthorizeResponseAllowed( authorization_id="auth_msg", grant=GrantBundle(jwt="jwt", expires_at="2030-01-01T00:00:00Z"), policy_snapshot=PolicySnapshot( capability="integration.execute", provider="slack", action="slack.post_message", ), ) cloud_svc = SimpleNamespace(authorize=AsyncMock(return_value=allowed)) cloud_client = SimpleNamespace( enqueue_control_job=AsyncMock( return_value={"ok": True, "job_id": "job_msg_1", "status": "queued"} ) ) project = SimpleNamespace(id=uuid4(), project_ref=None) connection_id = uuid4() result = await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=uuid4(), connection_id=connection_id, action="slack.post_message", input={"text": "hello"}, provider_hint="slack", handshake_id="core_capability_invoke", idempotency_key="idem-msg-1", ) assert result["ok"] is True public_path = public_capability_invoke_path(str(project.id), "messaging.send") assert invoke_path_has_host(public_path) is False public = strip_capability_response( { "capability": "messaging.send", "status": result["status"], "message_id": result["job_id"], "action": "slack.post_message", "connection_id": str(connection_id), } ) assert public == { "capability": "messaging.send", "status": "queued", "message_id": "job_msg_1", } queued = cloud_client.enqueue_control_job.await_args.kwargs assert queued["request_type"] == "integration.execute" @pytest.mark.asyncio async def test_bff_to_core_to_cloud_enqueue_shape_for_make_workflow() -> None: """Phase 1B: capability invoke enqueues existing Make integration.execute.""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.automation_workflow import to_public_automation_workflow_result from src.domain.connectors.automation_workflow import AutomationWorkflowResolution allowed = AuthorizeResponseAllowed( authorization_id="auth_make", grant=GrantBundle(jwt="jwt", expires_at="2030-01-01T00:00:00Z"), policy_snapshot=PolicySnapshot( capability="integration.execute", provider="make_com", action="make_com.trigger_scenario", ), ) cloud_svc = SimpleNamespace(authorize=AsyncMock(return_value=allowed)) cloud_client = SimpleNamespace( enqueue_control_job=AsyncMock( return_value={"ok": True, "job_id": "job_make_1", "status": "queued"} ) ) project = SimpleNamespace(id=uuid4(), project_ref=None) connection_id = uuid4() result = await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=uuid4(), connection_id=connection_id, action="make_com.trigger_scenario", input={"payload": {"name": "Jane"}}, provider_hint="make_com", handshake_id="core_capability_invoke", idempotency_key="idem-make-1", ) assert result["ok"] is True public_path = public_capability_invoke_path(str(project.id), "automation.workflow.trigger") assert invoke_path_has_host(public_path) is False queued = cloud_client.enqueue_control_job.await_args.kwargs assert queued["request_type"] == "integration.execute" assert queued["payload"]["action"] == "make_com.trigger_scenario" assert queued["payload"]["input"] == {"payload": {"name": "Jane"}} public = strip_capability_response( to_public_automation_workflow_result( AutomationWorkflowResolution(ok=True), enqueue=result, ) ) assert public == { "capability": "automation.workflow.trigger", "status": "queued", "execution_ref": "job_make_1", } assert "webhook_url" not in public assert "scenario_id" not in public assert "run_id" not in public assert "action" not in public