Files
crewAI/lib/cli/tests/test_plus_api.py
T
João MouraandClaude Opus 5.5 1410318b2e
CodeQL Advanced / Analyze (actions) (push) Canceled after 0s
CodeQL Advanced / Analyze (python) (push) Canceled after 0s
Vulnerability Scan / Detect changes (push) Canceled after 0s
Vulnerability Scan / pip-audit (push) Canceled after 0s
feat(cli): crewai eval export downloads an evaluation as files (#8187)
* crewai eval export downloads an evaluation's full data as files

A coding agent that is asked to act on an evaluation needs more than the
brief: the report, every setup's outputs, every check with the judge's
words, and the traces. AMP serves them as one export; this writes them
into ./crewai-eval-<id>/ with a .gitignore of `*`, so customer data is
never committed by accident. `crewai eval` becomes a group so the
subcommand can live under it; every existing invocation is unchanged.

The paths come from whichever AMP answered, so only plain relative paths
that resolve inside the folder are written; the rest are skipped and
counted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* A .gitignore link already in the export folder is replaced, never followed

Writing through it would have put `*` into whatever file it points at,
outside the folder. The tests also read the export as UTF-8, as it is
written, so a Windows locale cannot misread its dashes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* The export writes through folder handles and refuses Windows device names

A name Windows keeps for a device (CON, PRN, AUX, NUL, COM0-9, LPT0-9, with
any extension) is skipped like any other unsafe path, on every platform: on
Windows `NUL.md` is the null device, and the export would report a file
written that nobody can read.

On Linux and macOS each folder of a path is opened inside the one before it,
never through a link, and the file inside the last: no path is checked and
then opened again, so a link another process swaps in while the export runs
is never gone through. A link where a folder goes skips that file; a link at
a file's own place, symbolic or hard, is removed and the file made anew; the
export folder itself being a link stops the export with one sentence.
Windows keeps the resolved-path check.

Reading an evaluation sends `can_export=1`, so the brief names
`crewai eval export` only when the CLI, AMP and the evaluation service all
serve it. An AMP without the export route (a 404 that is not AMP's
`not_found`) says it needs an update; AMP's `export_unsupported` and
`export_too_large` answers are printed in AMP's own words.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* A folder where an exported file goes is skipped; the path writer replaces hard links and refuses a junction

- A folder already standing where an exported file goes is skipped and
  counted, as a file where a folder goes already was, instead of
  stopping the export.
- On the path writer (Windows), whatever is at a file's place is removed
  and the file created anew. A hard link there is no longer written
  through.
- The export folder must resolve to itself. A Windows junction, which
  `is_symlink` does not see, stops the export like a link.
- The file handle is closed if wrapping it fails.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Say why a missing file is no error before the export makes it

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* A folder where the export's .gitignore goes stops the export before any data

The data never goes into the folder without the .gitignore that keeps it
out of git. The Windows path writer's comment now states the race it
leaves open, and why that loss is bounded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 22:46:13 -07:00

614 lines
23 KiB
Python

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("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(trust_env=False, verify=True)
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)
with 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"}