crewAI/lib/cli/tests/test_plus_api.py

from collections.abc import Callable
import os
import unittest
from unittest.mock import ANY, MagicMock, patch

from crewai_cli.plus_api import PlusAPI
import httpx
import pytest


_RealHttpxClient = httpx.Client


class TestPlusAPI(unittest.TestCase):
    def setUp(self):
        self.api_key = "test_api_key"
        self.api = PlusAPI(self.api_key)
        self.org_uuid = "test-org-uuid"

    def test_init(self):
        self.assertEqual(self.api.api_key, self.api_key)
        self.assertEqual(self.api.headers["Authorization"], f"Bearer {self.api_key}")
        self.assertEqual(self.api.headers["Content-Type"], "application/json")
        self.assertIn("CrewAI-CLI/", self.api.headers["User-Agent"])
        self.assertTrue(self.api.headers["X-Crewai-Version"])

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_create_evaluation(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.create_evaluation("6f31fe1a-20bd-4bfe-a011-25d6b9341f62")

        mock_make_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/tracing/evaluations",
            json={"execution_id": "6f31fe1a-20bd-4bfe-a011-25d6b9341f62"},
            timeout=120.0,  # AMP reads the run's spans inside this request
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_create_evaluation_files_it_under_the_project(self, mock_make_request):
        self.api.create_evaluation("run-1", eval_config="{}", project_id="proj-1")

        mock_make_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/tracing/evaluations",
            json={"execution_id": "run-1", "eval_config": "{}", "project_id": "proj-1"},
            timeout=120.0,
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_create_models_evaluation(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.create_models_evaluation(
            ["openai/gpt-4o-mini", "anthropic/claude-haiku-4-5"],
            project_id="proj-1",
            eval_config="// ours\n{}",
            deployment_id="dep-1",
        )

        mock_make_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/tracing/evaluations",
            json={
                "kind": "models",
                "project_id": "proj-1",
                "models": ["openai/gpt-4o-mini", "anthropic/claude-haiku-4-5"],
                "eval_config": "// ours\n{}",
                "deployment_id": "dep-1",
            },
            timeout=120.0,
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_create_models_evaluation_sends_only_what_it_was_given(
        self, mock_make_request
    ):
        self.api.create_models_evaluation(["openai/gpt-4o-mini"], project_id="proj-1")

        mock_make_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/tracing/evaluations",
            json={
                "kind": "models",
                "project_id": "proj-1",
                "models": ["openai/gpt-4o-mini"],
            },
            timeout=120.0,
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_get_evaluation(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.get_evaluation("ev-1")

        mock_make_request.assert_called_once_with(
            "GET",
            "/crewai_plus/api/v1/tracing/evaluations/ev-1",
            params={"can_export": "1"},
            timeout=30.0,
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_export_evaluation(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.export_evaluation("ev-1")

        mock_make_request.assert_called_once_with(
            "GET",
            "/crewai_plus/api/v1/tracing/evaluations/ev-1/export",
            timeout=httpx.Timeout(30.0, read=120.0),
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_login_to_tool_repository(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.login_to_tool_repository()

        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/tools/login", json={}
        )
        self.assertEqual(response, mock_response)

    def assert_request_with_org_id(
        self, mock_client_instance, method: str, endpoint: str, **kwargs
    ):
        mock_client_instance.request.assert_called_once_with(
            method,
            f"{os.getenv('CREWAI_PLUS_URL')}{endpoint}",
            headers={
                "Authorization": ANY,
                "Content-Type": ANY,
                "User-Agent": ANY,
                "X-Crewai-Version": ANY,
                "X-Crewai-Organization-Id": self.org_uuid,
            },
            **kwargs,
        )

    @patch("crewai_core.plus_api.Settings")
    @patch("crewai_core.plus_api.httpx.Client")
    def test_login_to_tool_repository_with_org_uuid(
        self, mock_client_class, mock_settings_class
    ):
        mock_settings = MagicMock()
        mock_settings.org_uuid = self.org_uuid
        mock_settings.enterprise_base_url = os.getenv('CREWAI_PLUS_URL')
        mock_settings_class.return_value = mock_settings
        self.api = PlusAPI(self.api_key)

        mock_client_instance = MagicMock()
        mock_response = MagicMock()
        mock_client_instance.request.return_value = mock_response
        mock_client_class.return_value.__enter__.return_value = mock_client_instance

        response = self.api.login_to_tool_repository()

        self.assert_request_with_org_id(
            mock_client_instance, "POST", "/crewai_plus/api/v1/tools/login", json={}
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_get_tool(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response

        response = self.api.get_tool("test_tool_handle")
        mock_make_request.assert_called_once_with(
            "GET", "/crewai_plus/api/v1/tools/test_tool_handle"
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.Settings")
    @patch("crewai_core.plus_api.httpx.Client")
    def test_get_tool_with_org_uuid(self, mock_client_class, mock_settings_class):
        mock_settings = MagicMock()
        mock_settings.org_uuid = self.org_uuid
        mock_settings.enterprise_base_url = os.getenv('CREWAI_PLUS_URL')
        mock_settings_class.return_value = mock_settings
        self.api = PlusAPI(self.api_key)

        mock_client_instance = MagicMock()
        mock_response = MagicMock()
        mock_client_instance.request.return_value = mock_response
        mock_client_class.return_value.__enter__.return_value = mock_client_instance

        response = self.api.get_tool("test_tool_handle")

        self.assert_request_with_org_id(
            mock_client_instance, "GET", "/crewai_plus/api/v1/tools/test_tool_handle"
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_publish_tool(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response
        handle = "test_tool_handle"
        public = True
        version = "1.0.0"
        description = "Test tool description"
        encoded_file = "encoded_test_file"

        response = self.api.publish_tool(
            handle, public, version, description, encoded_file
        )

        params = {
            "handle": handle,
            "public": public,
            "version": version,
            "file": encoded_file,
            "description": description,
            "available_exports": None,
            "tools_metadata": None,
        }
        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/tools", json=params
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.Settings")
    @patch("crewai_core.plus_api.httpx.Client")
    def test_publish_tool_with_org_uuid(self, mock_client_class, mock_settings_class):
        mock_settings = MagicMock()
        mock_settings.org_uuid = self.org_uuid
        mock_settings.enterprise_base_url = os.getenv('CREWAI_PLUS_URL')
        mock_settings_class.return_value = mock_settings
        self.api = PlusAPI(self.api_key)

        mock_client_instance = MagicMock()
        mock_response = MagicMock()
        mock_client_instance.request.return_value = mock_response
        mock_client_class.return_value.__enter__.return_value = mock_client_instance

        handle = "test_tool_handle"
        public = True
        version = "1.0.0"
        description = "Test tool description"
        encoded_file = "encoded_test_file"

        response = self.api.publish_tool(
            handle, public, version, description, encoded_file
        )

        expected_params = {
            "handle": handle,
            "public": public,
            "version": version,
            "file": encoded_file,
            "description": description,
            "available_exports": None,
            "tools_metadata": None,
        }

        self.assert_request_with_org_id(
            mock_client_instance, "POST", "/crewai_plus/api/v1/tools", json=expected_params
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_publish_tool_without_description(self, mock_make_request):
        mock_response = MagicMock()
        mock_make_request.return_value = mock_response
        handle = "test_tool_handle"
        public = False
        version = "2.0.0"
        description = None
        encoded_file = "encoded_test_file"

        response = self.api.publish_tool(
            handle, public, version, description, encoded_file
        )

        params = {
            "handle": handle,
            "public": public,
            "version": version,
            "file": encoded_file,
            "description": description,
            "available_exports": None,
            "tools_metadata": None,
        }
        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/tools", json=params
        )
        self.assertEqual(response, mock_response)

    @patch.dict("crewai_core.plus_api._MACHINE_TRANSPORT_ENV", clear=True)
    @patch("crewai_core.plus_api._os_proxies", dict)
    @patch("crewai_core.plus_api.httpx.Client")
    def test_make_request(self, mock_client_class):
        mock_client_instance = MagicMock()
        mock_response = MagicMock()
        mock_client_instance.request.return_value = mock_response
        mock_client_class.return_value.__enter__.return_value = mock_client_instance

        response = self.api._make_request("GET", "test_endpoint")

        mock_client_class.assert_called_once_with(
            verify=True, trust_env=False, mounts={}
        )
        mock_client_instance.request.assert_called_once_with(
            "GET", f"{self.api.base_url}/test_endpoint", headers=self.api.headers
        )
        self.assertEqual(response, mock_response)

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_deploy_by_name(self, mock_make_request):
        self.api.deploy_by_name("test_project")
        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/crews/by-name/test_project/deploy"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_deploy_by_uuid(self, mock_make_request):
        self.api.deploy_by_uuid("test_uuid")
        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/crews/test_uuid/deploy"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_a_forced_push_says_so_and_an_unforced_one_sends_what_it_always_did(self, mock_make_request):
        self.api.deploy_by_uuid("test_uuid", force=True)
        mock_make_request.assert_called_with(
            "POST", "/crewai_plus/api/v1/crews/test_uuid/deploy", json={"force": True}
        )
        self.api.deploy_by_name("test_project", project_id="proj-1", force=True)
        mock_make_request.assert_called_with(
            "POST",
            "/crewai_plus/api/v1/crews/by-name/test_project/deploy",
            json={"project_id": "proj-1", "force": True},
        )
        self.api.deploy_by_uuid("test_uuid", force=False)
        mock_make_request.assert_called_with(
            "POST", "/crewai_plus/api/v1/crews/test_uuid/deploy"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_deploy_sends_the_project_id_when_there_is_one(self, mock_make_request):
        self.api.deploy_by_name("test_project", project_id="proj-1")
        mock_make_request.assert_called_with(
            "POST",
            "/crewai_plus/api/v1/crews/by-name/test_project/deploy",
            json={"project_id": "proj-1"},
        )
        self.api.deploy_by_uuid("test_uuid", project_id="proj-1")
        mock_make_request.assert_called_with(
            "POST",
            "/crewai_plus/api/v1/crews/test_uuid/deploy",
            json={"project_id": "proj-1"},
        )

    @patch("crewai_cli.plus_api.PlusAPI._make_multipart_request")
    def test_zip_deploys_send_the_project_id_when_there_is_one(
        self, mock_make_multipart_request
    ):
        self.api.create_crew_from_zip("/tmp/test.zip", name="c", project_id="proj-1")
        mock_make_multipart_request.assert_called_with(
            "POST",
            "/crewai_plus/api/v1/crews/zip",
            zip_file_path="/tmp/test.zip",
            data={"name": "c", "project_id": "proj-1"},
            timeout=300,
        )
        self.api.update_crew_from_zip("u", "/tmp/test.zip", project_id="proj-1")
        mock_make_multipart_request.assert_called_with(
            "POST",
            "/crewai_plus/api/v1/crews/u/zip_update",
            zip_file_path="/tmp/test.zip",
            data={"project_id": "proj-1"},
            timeout=300,
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_crew_status_by_name(self, mock_make_request):
        self.api.crew_status_by_name("test_project")
        mock_make_request.assert_called_once_with(
            "GET", "/crewai_plus/api/v1/crews/by-name/test_project/status"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_crew_status_by_uuid(self, mock_make_request):
        self.api.crew_status_by_uuid("test_uuid")
        mock_make_request.assert_called_once_with(
            "GET", "/crewai_plus/api/v1/crews/test_uuid/status"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_crew_by_name(self, mock_make_request):
        self.api.crew_by_name("test_project")
        mock_make_request.assert_called_once_with(
            "GET", "/crewai_plus/api/v1/crews/by-name/test_project/logs/deployment"
        )

        self.api.crew_by_name("test_project", "custom_log")
        mock_make_request.assert_called_with(
            "GET", "/crewai_plus/api/v1/crews/by-name/test_project/logs/custom_log"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_crew_by_uuid(self, mock_make_request):
        self.api.crew_by_uuid("test_uuid")
        mock_make_request.assert_called_once_with(
            "GET", "/crewai_plus/api/v1/crews/test_uuid/logs/deployment"
        )

        self.api.crew_by_uuid("test_uuid", "custom_log")
        mock_make_request.assert_called_with(
            "GET", "/crewai_plus/api/v1/crews/test_uuid/logs/custom_log"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_delete_crew_by_name(self, mock_make_request):
        self.api.delete_crew_by_name("test_project")
        mock_make_request.assert_called_once_with(
            "DELETE", "/crewai_plus/api/v1/crews/by-name/test_project"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_delete_crew_by_uuid(self, mock_make_request):
        self.api.delete_crew_by_uuid("test_uuid")
        mock_make_request.assert_called_once_with(
            "DELETE", "/crewai_plus/api/v1/crews/test_uuid"
        )

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_list_crews(self, mock_make_request):
        self.api.list_crews()
        mock_make_request.assert_called_once_with("GET", "/crewai_plus/api/v1/crews")

    @patch("crewai_core.plus_api.PlusAPI._make_request")
    def test_create_crew(self, mock_make_request):
        payload = {"name": "test_crew"}
        self.api.create_crew(payload)
        mock_make_request.assert_called_once_with(
            "POST", "/crewai_plus/api/v1/crews", json=payload
        )

    @patch("crewai_cli.plus_api.PlusAPI._make_multipart_request")
    def test_create_crew_from_zip(self, mock_make_multipart_request):
        self.api.create_crew_from_zip(
            "/tmp/test.zip",
            name="test_crew",
            env={"ENV_VAR": "value"},
        )
        mock_make_multipart_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/crews/zip",
            zip_file_path="/tmp/test.zip",
            data={"name": "test_crew", "env[ENV_VAR]": "value"},
            timeout=300,
        )

    @patch("crewai_cli.plus_api.PlusAPI._make_multipart_request")
    def test_update_crew_from_zip(self, mock_make_multipart_request):
        self.api.update_crew_from_zip(
            "test_uuid",
            "/tmp/test.zip",
            env={"ENV_VAR": "value"},
        )
        mock_make_multipart_request.assert_called_once_with(
            "POST",
            "/crewai_plus/api/v1/crews/test_uuid/zip_update",
            zip_file_path="/tmp/test.zip",
            data={"env[ENV_VAR]": "value"},
            timeout=300,
        )
        self.api.update_crew_from_zip("test_uuid", "/tmp/test.zip", force=True)
        assert mock_make_multipart_request.call_args.kwargs["data"] == {"force": "true"}

    @patch("crewai_core.plus_api.Settings")
    @patch.dict(os.environ, {"CREWAI_PLUS_URL": ""})
    def test_custom_base_url(self, mock_settings_class):
        mock_settings = MagicMock()
        mock_settings.enterprise_base_url = "https://custom-url.com/api"
        mock_settings_class.return_value = mock_settings
        custom_api = PlusAPI("test_key")
        self.assertEqual(
            custom_api.base_url,
            "https://custom-url.com/api",
        )

    @patch.dict(os.environ, {"CREWAI_PLUS_URL": "https://custom-url-from-env.com"})
    def test_custom_base_url_from_env(self):
        custom_api = PlusAPI("test_key")
        self.assertEqual(
            custom_api.base_url,
            "https://custom-url-from-env.com",
        )


@patch("crewai_core.plus_api.PlusAPI._make_request")
def test_get_agent(mock_make_request):
    api = PlusAPI("test_api_key")
    mock_response = MagicMock()
    mock_make_request.return_value = mock_response

    response = api.get_agent("test_agent_handle")

    mock_make_request.assert_called_once_with(
        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
    )
    assert response == mock_response


@patch("crewai_core.plus_api.PlusAPI._make_request")
@patch("crewai_core.plus_api.Settings")
def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
    org_uuid = "test-org-uuid"
    mock_settings = MagicMock()
    mock_settings.org_uuid = org_uuid
    mock_settings.enterprise_base_url = os.getenv("CREWAI_PLUS_URL")
    mock_settings_class.return_value = mock_settings

    api = PlusAPI("test_api_key")

    mock_response = MagicMock()
    mock_make_request.return_value = mock_response

    response = api.get_agent("test_agent_handle")

    mock_make_request.assert_called_once_with(
        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
    )
    assert "X-Crewai-Organization-Id" in api.headers
    assert api.headers["X-Crewai-Organization-Id"] == org_uuid
    assert response == mock_response


def _requests_sent_by(call: Callable[[PlusAPI], object]) -> list[httpx.Request]:
    """Run ``call`` against a PlusAPI whose HTTP client answers locally."""
    sent: list[httpx.Request] = []

    def answer(request: httpx.Request) -> httpx.Response:
        sent.append(request)
        return httpx.Response(200, json={})

    transport = httpx.MockTransport(answer)
    # A machine or OS proxy would mount a real transport ahead of the mock one.
    with (
        patch.dict("crewai_core.plus_api._MACHINE_TRANSPORT_ENV", clear=True),
        patch("crewai_core.plus_api._os_proxies", dict),
        patch(
            "crewai_core.plus_api.httpx.Client",
            side_effect=lambda **kwargs: _RealHttpxClient(transport=transport, **kwargs),
        ),
    ):
        call(PlusAPI("test_api_key"))
    return sent


@pytest.mark.parametrize(
    ("call", "path"),
    [
        (
            lambda api: api.get_triggers(),
            "/crewai_plus/api/v1/integrations/apps",
        ),
        (
            lambda api: api.get_trigger_payload("hubspot", "contact_created"),
            "/crewai_plus/api/v1/integrations/hubspot/contact_created/payload",
        ),
    ],
    ids=["get_triggers", "get_trigger_payload"],
)
def test_trigger_requests_wait_60_seconds_only_for_the_response(call, path):
    (request,) = _requests_sent_by(call)

    assert request.url.path == path
    assert request.extensions["timeout"] == {
        "connect": 5.0,
        "read": 60.0,
        "write": 5.0,
        "pool": 5.0,
    }


def test_requests_without_their_own_timeout_keep_the_httpx_default():
    (request,) = _requests_sent_by(lambda api: api.get_agent("an_agent"))

    assert request.url.path == "/crewai_plus/api/v1/agents/an_agent"
    assert request.extensions["timeout"]["read"] == 5.0


def test_an_export_waits_two_minutes_for_its_large_answer():
    """The export carries every setup's outputs and traces, tens of MB: the
    httpx default of 5 s would cut it off."""
    (request,) = _requests_sent_by(lambda api: api.export_evaluation("ev-1"))

    assert request.method == "GET"
    assert request.url.path == "/crewai_plus/api/v1/tracing/evaluations/ev-1/export"
    assert request.extensions["timeout"] == {
        "connect": 30.0,
        "read": 120.0,
        "write": 30.0,
        "pool": 30.0,
    }


def test_reading_an_evaluation_says_this_cli_can_export_it():
    """`can_export=1` lets the brief name `crewai eval export`: an AMP that does
    not know the parameter drops it, and the brief stays silent about it."""
    (request,) = _requests_sent_by(lambda api: api.get_evaluation("ev-1"))

    assert request.method == "GET"
    assert request.url.path == "/crewai_plus/api/v1/tracing/evaluations/ev-1"
    assert dict(request.url.params) == {"can_export": "1"}


@pytest.mark.block_network(allowed_hosts=["127.0.0.1"])
def test_zip_deploy_uploads_go_through_the_proxy_the_machine_names(
    tmp_path, monkeypatch
):
    # Corporate networks and coding-agent sandboxes only allow traffic through
    # the proxy in HTTP(S)_PROXY; the CLI's own ZIP upload must use it too.
    # The machine's variables are the ones crewai-core pinned at startup.
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    import threading
    import zipfile

    seen = []

    class Proxy(BaseHTTPRequestHandler):
        def do_POST(self):
            body = self.rfile.read(int(self.headers["Content-Length"]))
            seen.append((self.path, self.headers.get("Authorization"), body))
            self.send_response(201)
            self.send_header("Content-Length", "2")
            self.end_headers()
            self.wfile.write(b"{}")

        def log_message(self, format, *args):
            pass

    server = ThreadingHTTPServer(("127.0.0.1", 0), Proxy)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    monkeypatch.setattr(
        "crewai_core.plus_api._MACHINE_TRANSPORT_ENV",
        {"HTTP_PROXY": f"http://127.0.0.1:{server.server_address[1]}"},
    )
    project_zip = tmp_path / "project.zip"
    with zipfile.ZipFile(project_zip, "w") as archive:
        archive.writestr("pyproject.toml", "[project]\nname = 'demo'\n")

    try:
        response = PlusAPI("amp-token", base_url="http://amp.test").create_crew_from_zip(
            project_zip, name="demo"
        )
    finally:
        server.shutdown()
        server.server_close()

    assert response.status_code == 201
    ((target, authorization, body),) = seen
    assert target == "http://amp.test/crewai_plus/api/v1/crews/zip"
    assert authorization == "Bearer amp-token"
    assert b'filename="project.zip"' in body