andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-08 00:15:12 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Support ZIP deployment fallback and JSON crew project env runs (#6166)

Browse Source 
* Update crewAI CLI with various enhancements and fixes

- Updated `create_json_crew.py` to require `crewai[tools]>=1.14.7`.
- Enhanced `git.py` with improved repository initialization, including automatic initial commit creation and exclusion patterns for initial commits.
- Modified `install_crew.py` to allow error handling during installation with an optional `raise_on_error` parameter.
- Expanded `plus_api.py` to include methods for creating and updating crews from ZIP files.
- Introduced a new `archive.py` for creating deployable ZIP archives of CrewAI projects, ensuring local artifacts are excluded.
- Updated `run_crew.py` to manage JSON crew dependencies and run crews in the project's environment.
- Enhanced deployment logic in `main.py` to handle ZIP uploads and improve user feedback during deployment processes.
- Added tests for new functionalities and ensured existing tests reflect recent changes in behavior and requirements.

* fix(cli): address deploy zip review feedback

* fix(cli): sync missing lockfile before deploy

* fix(cli): preserve remote deploy on git setup warnings

* test(cli): use single deploy main import style

* fix(cli): skip project install for json crew sync

* fix(cli): load json runner from source checkout

* fix(cli): skip json crew sync when locked

* fix(cli): address deploy zip review feedback

* fix(cli): pass env on zip redeploy

* fix(cli): harden json run and zip fallback

* fix(cli): validate before deploy lock install

* fix(cli): respect poetry lock for json runs

* fix(cli): align json zip wrapper detection

* fix(deps): bump starlette audit floor

* fix(cli): avoid auth retry for deploy exits

* fix(cli): update json zip script entrypoints
This commit is contained in:
João Moura
2026-06-15 18:46:54 -03:00
committed by GitHub
parent a5cc6f6d0e
commit 53c2284484
20 changed files with 2449 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
lib/cli/tests/deploy/test_archive.py Normal file
View File

@@ -0,0 +1,270 @@
from pathlib import Path
import subprocess
import zipfile
import pytest
from crewai_cli.deploy.archive import create_project_zip
def test_create_project_zip_excludes_local_artifacts(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "uv.lock").write_text("# lock\n")
(tmp_path / "src").mkdir()
(tmp_path / "src" / "main.py").write_text("print('hello')\n")
(tmp_path / ".env").write_text("OPENAI_API_KEY=secret\n")
(tmp_path / ".env.example").write_text("OPENAI_API_KEY=\n")
(tmp_path / "__pycache__").mkdir()
(tmp_path / "__pycache__" / "main.pyc").write_bytes(b"compiled")
(tmp_path / ".git").mkdir()
(tmp_path / ".git" / "config").write_text("[core]\n")
archive_path = create_project_zip("demo", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
finally:
archive_path.unlink(missing_ok=True)
assert names == {
"pyproject.toml",
"uv.lock",
"src/main.py",
".env.example",
}
def test_create_project_zip_uses_repository_file_list(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "uv.lock").write_text("# lock\n")
(tmp_path / "ignored.txt").write_text("ignored\n")
class RepositoryStub:
def deployable_files(self) -> list[str]:
return ["pyproject.toml", "uv.lock"]
archive_path = create_project_zip(
"demo",
project_dir=tmp_path,
repository=RepositoryStub(), # type: ignore[arg-type]
)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
finally:
archive_path.unlink(missing_ok=True)
assert names == {"pyproject.toml", "uv.lock"}
def test_create_project_zip_without_repository_uses_git_ignore_rules(
tmp_path: Path,
):
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / ".gitignore").write_text("node_modules/\nsecret.txt\n")
(tmp_path / "src").mkdir()
(tmp_path / "src" / "main.py").write_text("print('hello')\n")
(tmp_path / "node_modules").mkdir()
(tmp_path / "node_modules" / "package.json").write_text("{}\n")
(tmp_path / "secret.txt").write_text("secret\n")
try:
subprocess.run(
["git", "init"],
cwd=tmp_path,
capture_output=True,
check=True,
text=True,
)
except (FileNotFoundError, subprocess.CalledProcessError) as exc:
pytest.skip(f"git is not available in this environment: {exc}")
archive_path = create_project_zip("demo", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
finally:
archive_path.unlink(missing_ok=True)
assert names == {
".gitignore",
"pyproject.toml",
"src/main.py",
}
def test_create_project_zip_does_not_fallback_when_repository_listing_fails(
tmp_path: Path,
):
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
class RepositoryStub:
def deployable_files(self) -> list[str]:
raise RuntimeError("git listing failed")
with pytest.raises(RuntimeError, match="git listing failed"):
create_project_zip(
"demo",
project_dir=tmp_path,
repository=RepositoryStub(), # type: ignore[arg-type]
)
def test_create_project_zip_excludes_symlinked_files(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
outside_file = tmp_path.parent / f"{tmp_path.name}-secret.txt"
outside_file.write_text("secret\n")
archive_path: Path | None = None
try:
try:
(tmp_path / "external-secret.txt").symlink_to(outside_file)
except OSError as exc:
pytest.skip(f"symlinks are not supported in this environment: {exc}")
archive_path = create_project_zip("demo", project_dir=tmp_path)
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
finally:
if archive_path is not None:
archive_path.unlink(missing_ok=True)
outside_file.unlink(missing_ok=True)
assert names == {"pyproject.toml"}
def test_create_project_zip_adds_json_project_wrapper(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "json_crew"
version = "0.1.0"
dependencies = ["crewai[tools]>=1.15"]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.crewai]
type = "crew"
""".strip()
+ "\n"
)
(tmp_path / "agents").mkdir()
(tmp_path / "agents" / "researcher.jsonc").write_text("{}\n")
(tmp_path / "crew.jsonc").write_text("{}\n")
archive_path = create_project_zip("json_crew", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
crew_py = archive.read("src/json_crew/crew.py").decode()
main_py = archive.read("src/json_crew/main.py").decode()
pyproject = archive.read("pyproject.toml").decode()
finally:
archive_path.unlink(missing_ok=True)
assert "uv.lock" not in names
assert "crew.jsonc" in names
assert "agents/researcher.jsonc" in names
assert "src/json_crew/__init__.py" in names
assert "src/json_crew/crew.py" in names
assert "src/json_crew/main.py" in names
assert "src/json_crew/config/agents.yaml" in names
assert "src/json_crew/config/tasks.yaml" in names
assert "load_crew(_crew_path())" in crew_py
assert "JsonCrew" in crew_py
assert "from json_crew.crew import JsonCrew" in main_py
assert "run_crew = \"json_crew.main:run\"" in pyproject
def test_create_project_zip_updates_existing_json_project_scripts(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "json_crew"
version = "0.1.0"
[project.scripts]
json_crew = "old.module:run"
run_crew = "old.module:run"
custom = "custom.module:main"
[tool.crewai]
type = "crew"
""".strip()
+ "\n"
)
(tmp_path / "crew.jsonc").write_text("{}\n")
archive_path = create_project_zip("json_crew", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
pyproject = archive.read("pyproject.toml").decode()
finally:
archive_path.unlink(missing_ok=True)
assert 'json_crew = "json_crew.main:run"' in pyproject
assert 'run_crew = "json_crew.main:run"' in pyproject
assert 'train = "json_crew.main:train"' in pyproject
assert 'replay = "json_crew.main:replay"' in pyproject
assert 'test = "json_crew.main:test"' in pyproject
assert 'run_with_trigger = "json_crew.main:run_with_trigger"' in pyproject
assert 'custom = "custom.module:main"' in pyproject
assert "old.module:run" not in pyproject
assert "[tool.crewai]" in pyproject
@pytest.mark.parametrize(
"tool_config",
[
'tool = "invalid"\n',
'[tool]\ncrewai = "invalid"\n',
],
)
def test_create_project_zip_adds_json_wrapper_for_malformed_tool_config(
tmp_path: Path, tool_config: str
):
(tmp_path / "pyproject.toml").write_text(
f"""
[project]
name = "json_crew"
version = "0.1.0"
{tool_config}
""".strip()
+ "\n"
)
(tmp_path / "crew.jsonc").write_text("{}\n")
archive_path = create_project_zip("json_crew", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
pyproject = archive.read("pyproject.toml").decode()
finally:
archive_path.unlink(missing_ok=True)
assert "src/json_crew/crew.py" in names
assert "src/json_crew/main.py" in names
assert "run_crew = \"json_crew.main:run\"" in pyproject
def test_create_project_zip_rejects_empty_normalized_package_name(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "!!!"
version = "0.1.0"
[tool.crewai]
type = "crew"
""".strip()
+ "\n"
)
(tmp_path / "crew.jsonc").write_text("{}\n")
with pytest.raises(
ValueError,
match=r"Could not derive a valid Python package name",
):
create_project_zip("invalid", project_dir=tmp_path)

430
lib/cli/tests/deploy/test_deploy_main.py
View File

@@ -1,16 +1,172 @@
import sys
import unittest
from io import StringIO
from pathlib import Path
import subprocess
from unittest.mock import MagicMock, Mock, patch
import pytest
import json
import crewai_cli.deploy.main as deploy_main
import httpx
from crewai_cli.deploy.main import DeployCommand
from crewai_cli.deploy.validate import Severity, ValidationResult
from crewai_cli.utils import parse_toml
def test_ensure_lockfile_for_deploy_runs_install_when_lock_missing(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
calls = []
def fake_install_crew(proxy_options, *, raise_on_error=False):
calls.append((proxy_options, raise_on_error))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
deploy_main._ensure_lockfile_for_deploy()
assert calls == [([], True)]
def test_ensure_lockfile_for_deploy_skips_when_lock_exists(monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "uv.lock").write_text("# lock\n")
calls = []
def fake_install_crew(proxy_options, *, raise_on_error=False):
calls.append((proxy_options, raise_on_error))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
deploy_main._ensure_lockfile_for_deploy()
assert calls == []
def test_ensure_lockfile_for_deploy_skips_without_pyproject(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
calls = []
def fake_install_crew(proxy_options, *, raise_on_error=False):
calls.append((proxy_options, raise_on_error))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
deploy_main._ensure_lockfile_for_deploy()
assert calls == []
def test_ensure_lockfile_for_deploy_failure_exits_nonzero(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
def fake_install_crew(proxy_options, *, raise_on_error=False):
raise subprocess.CalledProcessError(42, ["uv", "sync"])
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
with pytest.raises(SystemExit) as exc_info:
deploy_main._ensure_lockfile_for_deploy()
assert exc_info.value.code == 42
class _FakeDeployValidator:
def __init__(self, results: list[ValidationResult]):
self.results = results
@property
def errors(self) -> list[ValidationResult]:
return [
result
for result in self.results
if result.severity is Severity.ERROR
]
def run(self) -> list[ValidationResult]:
return self.results
def test_prepare_project_for_deploy_blocks_install_when_validation_fails(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
install_calls = []
rendered_results = []
missing_lockfile = ValidationResult(
Severity.ERROR,
"missing_lockfile",
"Expected to find a lockfile",
)
invalid_config = ValidationResult(
Severity.ERROR,
"invalid_crew_json",
"crew.jsonc has invalid configuration",
)
monkeypatch.setattr(
deploy_main,
"DeployValidator",
lambda: _FakeDeployValidator([missing_lockfile, invalid_config]),
)
monkeypatch.setattr(deploy_main, "render_report", rendered_results.append)
def fake_install_crew(proxy_options, *, raise_on_error=False):
install_calls.append((proxy_options, raise_on_error))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
assert deploy_main._prepare_project_for_deploy(skip_validate=False) is False
assert install_calls == []
assert [[result.code for result in results] for results in rendered_results] == [
["invalid_crew_json"]
]
def test_prepare_project_for_deploy_creates_missing_lock_after_validation(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
install_calls = []
missing_lockfile = ValidationResult(
Severity.ERROR,
"missing_lockfile",
"Expected to find a lockfile",
)
validators = [
_FakeDeployValidator([missing_lockfile]),
_FakeDeployValidator([]),
]
def fake_validator():
return validators.pop(0)
def fake_install_crew(proxy_options, *, raise_on_error=False):
install_calls.append((proxy_options, raise_on_error))
(tmp_path / "uv.lock").write_text("# lock\n")
monkeypatch.setattr(deploy_main, "DeployValidator", fake_validator)
monkeypatch.setattr(deploy_main, "render_report", lambda results: None)
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
assert deploy_main._prepare_project_for_deploy(skip_validate=False) is True
assert install_calls == [([], True)]
assert validators == []
class TestDeployCommand(unittest.TestCase):
@patch("crewai_cli.command.get_auth_token")
@patch("crewai_cli.deploy.main.get_project_name")
@@ -28,19 +184,25 @@ class TestDeployCommand(unittest.TestCase):
self.mock_get_auth_token.return_value = "test_token"
self.mock_get_project_name.return_value = "test_project"
self.deploy_command = DeployCommand()
self.deploy_command = deploy_main.DeployCommand()
self.mock_client = self.deploy_command.plus_api_client
def test_init_success(self):
self.assertEqual(self.deploy_command.project_name, "test_project")
self.mock_plus_api.assert_called_once_with(api_key="test_token")
@patch("builtins.input")
def test_confirm_zip_input_only_confirms_env_vars(self, mock_input):
self.deploy_command._confirm_zip_input({"MODEL": "openai/gpt-5"}, False)
mock_input.assert_called_once_with("Press Enter to continue with 1 env vars: MODEL")
@patch("crewai_cli.command.get_auth_token")
def test_init_failure(self, mock_get_auth_token):
mock_get_auth_token.side_effect = Exception("Auth failed")
with self.assertRaises(SystemExit):
DeployCommand()
deploy_main.DeployCommand()
def test_validate_response_successful_response(self):
mock_response = Mock(spec=httpx.Response)
@@ -123,8 +285,15 @@ class TestDeployCommand(unittest.TestCase):
)
self.assertIn("2023-01-01 - INFO: Test log", fake_out.getvalue())
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_uuid(self, mock_display):
def test_deploy_with_uuid(self, mock_display, mock_repository):
mock_repository.return_value.origin_url.return_value = (
"https://github.com/test/repo.git"
)
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"uuid": "test-uuid"}
@@ -135,8 +304,15 @@ class TestDeployCommand(unittest.TestCase):
self.mock_client.deploy_by_uuid.assert_called_once_with("test-uuid")
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_project_name(self, mock_display):
def test_deploy_with_project_name(self, mock_display, mock_repository):
mock_repository.return_value.origin_url.return_value = (
"https://github.com/test/repo.git"
)
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"uuid": "test-uuid"}
@@ -147,17 +323,142 @@ class TestDeployCommand(unittest.TestCase):
self.mock_client.deploy_by_name.assert_called_once_with("test_project")
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_remote_keeps_remote_path_when_fetch_fails(
self, mock_display, mock_repository, mock_create_project_zip
):
repository = mock_repository.return_value
repository.origin_url.return_value = "https://github.com/test/repo.git"
repository.fetch.side_effect = ValueError("fetch failed")
repository.create_initial_commit_if_needed.return_value = False
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "test-uuid"}
self.mock_client.deploy_by_name.return_value = mock_response
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command.deploy(skip_validate=True)
output = fake_out.getvalue()
mock_repository.assert_called_once_with(fetch=False)
repository.fetch.assert_called_once_with()
self.assertIn("Continuing with remote deployment", output)
self.mock_client.deploy_by_name.assert_called_once_with("test_project")
self.mock_client.update_crew_from_zip.assert_not_called()
mock_create_project_zip.assert_not_called()
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_remote_keeps_remote_path_when_initial_commit_fails(
self, mock_display, mock_repository, mock_create_project_zip
):
repository = mock_repository.return_value
repository.origin_url.return_value = "https://github.com/test/repo.git"
repository.create_initial_commit_if_needed.side_effect = RuntimeError(
"commit failed"
)
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "test-uuid"}
self.mock_client.deploy_by_name.return_value = mock_response
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command.deploy(skip_validate=True)
output = fake_out.getvalue()
mock_repository.assert_called_once_with(fetch=False)
repository.fetch.assert_called_once_with()
self.assertIn("Continuing with remote deployment", output)
self.mock_client.deploy_by_name.assert_called_once_with("test_project")
self.mock_client.update_crew_from_zip.assert_not_called()
mock_create_project_zip.assert_not_called()
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository.origin_url")
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_uuid_without_remote_updates_from_zip(
self, mock_display, mock_repository, mock_fetch_env, mock_create_project_zip
):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
mock_repository.return_value.origin_url.return_value = None
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"uuid": "test-uuid"}
self.mock_client.update_crew_from_zip.return_value = mock_response
self.deploy_command.deploy(uuid="test-uuid", skip_validate=True)
self.mock_client.update_crew_from_zip.assert_called_once_with(
"test-uuid",
Path("/tmp/test_project.zip"),
env={"ENV_VAR": "value"},
)
self.mock_client.deploy_by_uuid.assert_not_called()
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository")
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
def test_deploy_with_project_name_without_remote_updates_from_zip(
self, mock_display, mock_repository, mock_fetch_env, mock_create_project_zip
):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
mock_repository.return_value.origin_url.return_value = None
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
status_response = MagicMock()
status_response.status_code = 200
status_response.is_success = True
status_response.json.return_value = {"uuid": "test-uuid"}
update_response = MagicMock()
update_response.status_code = 200
update_response.json.return_value = {"uuid": "test-uuid"}
self.mock_client.crew_status_by_name.return_value = status_response
self.mock_client.update_crew_from_zip.return_value = update_response
self.deploy_command.deploy(skip_validate=True)
self.mock_client.crew_status_by_name.assert_called_once_with("test_project")
self.mock_client.update_crew_from_zip.assert_called_once_with(
"test-uuid",
Path("/tmp/test_project.zip"),
env={"ENV_VAR": "value"},
)
self.mock_client.deploy_by_name.assert_not_called()
mock_display.assert_called_once_with({"uuid": "test-uuid"})
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository")
@patch("builtins.input")
@pytest.mark.timeout(180)
def test_create_crew(self, mock_input, mock_git_origin_url, mock_fetch_env):
def test_create_crew(self, mock_input, mock_repository, mock_fetch_env):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
mock_git_origin_url.return_value = "https://github.com/test/repo.git"
mock_repository.return_value.origin_url.return_value = (
"https://github.com/test/repo.git"
)
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_input.return_value = ""
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "new-uuid", "status": "created"}
self.mock_client.create_crew.return_value = mock_response
@@ -166,38 +467,127 @@ class TestDeployCommand(unittest.TestCase):
self.assertIn("Deployment created successfully!", fake_out.getvalue())
self.assertIn("new-uuid", fake_out.getvalue())
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository")
def test_create_crew_without_git_repo_shows_setup_help(
self, mock_repository, mock_fetch_env
def test_create_crew_without_git_repo_initializes_and_uses_zip(
self, mock_repository, mock_fetch_env, mock_create_project_zip
):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
mock_repository.side_effect = ValueError("not a Git repository")
initialized_repository = MagicMock()
initialized_repository.origin_url.return_value = None
mock_repository.initialize.return_value = initialized_repository
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "zip-uuid", "status": "created"}
self.mock_client.create_crew_from_zip.return_value = mock_response
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command.create_crew(skip_validate=True)
self.deploy_command.create_crew(confirm=True, skip_validate=True)
output = fake_out.getvalue()
self.assertIn("Deployment requires a Git repository", output)
self.assertIn("git init", output)
self.assertIn("git remote add origin <your-repo-url>", output)
self.assertIn("Initialized a local Git repository", output)
self.assertIn("Deploying from a ZIP upload", output)
mock_repository.initialize.assert_called_once_with()
mock_create_project_zip.assert_called_once_with(
"test_project", repository=initialized_repository
)
self.mock_client.create_crew_from_zip.assert_called_once_with(
Path("/tmp/test_project.zip"),
name="test_project",
env={"ENV_VAR": "value"},
)
self.mock_client.create_crew.assert_not_called()
@patch("crewai_cli.deploy.main.git.Repository")
def test_prepare_git_repository_returns_repo_when_init_commit_fails(
self, mock_repository
):
recovered_repository = MagicMock()
mock_repository.side_effect = [
ValueError("not a Git repository"),
recovered_repository,
]
mock_repository.initialize.side_effect = RuntimeError("commit failed")
with patch("sys.stdout", new=StringIO()) as fake_out:
repository = self.deploy_command._prepare_git_repository()
self.assertIs(repository, recovered_repository)
self.assertIn("Git auto-setup did not complete", fake_out.getvalue())
mock_repository.initialize.assert_called_once_with()
self.assertEqual(mock_repository.call_count, 2)
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository")
def test_create_crew_without_remote_shows_remote_help(
self, mock_repository, mock_fetch_env
def test_create_crew_without_remote_uses_zip(
self, mock_repository, mock_fetch_env, mock_create_project_zip
):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
mock_repository.return_value.origin_url.return_value = None
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
False
)
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "zip-uuid", "status": "created"}
self.mock_client.create_crew_from_zip.return_value = mock_response
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command.create_crew(skip_validate=True)
self.deploy_command.create_crew(confirm=True, skip_validate=True)
output = fake_out.getvalue()
self.assertIn("No remote repository URL found.", output)
self.assertIn("git remote add origin <your-repo-url>", output)
self.assertIn("git push -u origin HEAD", output)
self.assertIn("No origin remote found.", output)
self.assertIn("Deploying from a ZIP upload", output)
mock_create_project_zip.assert_called_once_with(
"test_project", repository=mock_repository.return_value
)
self.mock_client.create_crew_from_zip.assert_called_once_with(
Path("/tmp/test_project.zip"),
name="test_project",
env={"ENV_VAR": "value"},
)
self.mock_client.create_crew.assert_not_called()
@patch("crewai_cli.deploy.main.create_project_zip")
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
@patch("crewai_cli.deploy.main.git.Repository")
def test_create_crew_without_remote_uses_git_file_list_when_commit_fails(
self, mock_repository, mock_fetch_env, mock_create_project_zip
):
mock_fetch_env.return_value = {"ENV_VAR": "value"}
repository = mock_repository.return_value
repository.origin_url.return_value = None
repository.create_initial_commit_if_needed.side_effect = RuntimeError(
"commit failed"
)
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json.return_value = {"uuid": "zip-uuid", "status": "created"}
self.mock_client.create_crew_from_zip.return_value = mock_response
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command.create_crew(confirm=True, skip_validate=True)
output = fake_out.getvalue()
self.assertIn("Continuing with ZIP deployment using Git", output)
self.assertIn("file listing", output)
mock_create_project_zip.assert_called_once_with(
"test_project", repository=repository
)
self.mock_client.create_crew_from_zip.assert_called_once_with(
Path("/tmp/test_project.zip"),
name="test_project",
env={"ENV_VAR": "value"},
)
self.mock_client.create_crew.assert_not_called()
def test_list_crews(self):

8
lib/cli/tests/deploy/test_validate.py
View File

@@ -481,7 +481,7 @@ def test_modern_crewai_pin_does_not_warn(tmp_path: Path) -> None:
def test_create_crew_aborts_on_validation_error(tmp_path: Path) -> None:
"""`crewai deploy create` must not contact the API when validation fails."""
from unittest.mock import MagicMock, patch as mock_patch
from unittest.mock import patch as mock_patch
from crewai_cli.deploy.main import DeployCommand
@@ -490,10 +490,10 @@ def test_create_crew_aborts_on_validation_error(tmp_path: Path) -> None:
mock_patch("crewai_cli.deploy.main.get_project_name", return_value="p"),
mock_patch("crewai_cli.command.PlusAPI") as mock_api,
mock_patch(
"crewai_cli.deploy.main.validate_project"
) as mock_validate,
"crewai_cli.deploy.main._prepare_project_for_deploy",
return_value=False,
),
):
mock_validate.return_value = MagicMock(ok=False)
cmd = DeployCommand()
cmd.create_crew()
assert not cmd.plus_api_client.create_crew.called

33
lib/cli/tests/test_crew_run_tui.py
View File

@@ -19,6 +19,7 @@ from crewai.events.types.tool_usage_events import (
ToolUsageFinishedEvent,
ToolUsageStartedEvent,
)
from crewai_cli.command import AuthenticationRequiredError
from crewai_cli import run_crew
from crewai_cli.crew_run_tui import CrewRunApp
@@ -68,7 +69,7 @@ def test_chain_deploy_skips_validation_after_auth_retry(monkeypatch) -> None:
create_calls.append(kwargs)
FakeDeployCommand.attempts += 1
if FakeDeployCommand.attempts == 1:
raise SystemExit(1)
raise AuthenticationRequiredError
class FakeAuthenticationCommand:
def login(self) -> None:
@@ -83,12 +84,38 @@ def test_chain_deploy_skips_validation_after_auth_retry(monkeypatch) -> None:
run_crew._chain_deploy()
assert create_calls == [
{"confirm": False, "skip_validate": True},
{"confirm": False, "skip_validate": True},
{"confirm": True, "skip_validate": True},
{"confirm": True, "skip_validate": True},
]
assert login_calls == [True]
def test_chain_deploy_does_not_login_for_deploy_exit(monkeypatch, capsys) -> None:
create_calls: list[dict[str, object]] = []
login_calls: list[bool] = []
class FakeDeployCommand:
def create_crew(self, **kwargs) -> None:
create_calls.append(kwargs)
raise SystemExit(42)
class FakeAuthenticationCommand:
def login(self) -> None:
login_calls.append(True)
monkeypatch.setattr("crewai_cli.deploy.main.DeployCommand", FakeDeployCommand)
monkeypatch.setattr(
"crewai_cli.authentication.main.AuthenticationCommand",
FakeAuthenticationCommand,
)
run_crew._chain_deploy()
assert create_calls == [{"confirm": True, "skip_validate": True}]
assert login_calls == []
assert "Deploy failed with exit code 42" in capsys.readouterr().out
def test_plan_step_status_updates_only_the_explicit_step() -> None:
app = _app_with_plan()

54
lib/cli/tests/test_git.py
View File

@@ -31,6 +31,18 @@ def test_is_git_not_installed(fp):
Repository(path=".")
def test_fetch_failure_raises_value_error(fp):
fp.register(["git", "--version"], stdout="git version 2.30.0\n")
fp.register(["git", "rev-parse", "--is-inside-work-tree"], stdout="true\n")
fp.register(["git", "fetch"], returncode=128, stderr="remote unavailable\n")
with pytest.raises(
ValueError,
match=r"Git fetch failed with exit code 128 for command \['git', 'fetch'\]: remote unavailable",
):
Repository(path=".")
def test_status(fp, repository):
fp.register(
["git", "status", "--branch", "--porcelain"],
@@ -99,3 +111,45 @@ def test_origin_url(fp, repository):
stdout="https://github.com/user/repo.git\n",
)
assert repository.origin_url() == "https://github.com/user/repo.git"
def test_initialize_creates_initial_commit(fp, tmp_path):
fp.register(["git", "--version"], stdout="git version 2.30.0\n")
fp.register(["git", "init"], stdout="")
fp.register(["git", "--version"], stdout="git version 2.30.0\n")
fp.register(["git", "rev-parse", "--is-inside-work-tree"], stdout="true\n")
fp.register(["git", "rev-parse", "--verify", "HEAD"], returncode=1)
fp.register(["git", "add", "."], stdout="")
fp.register(
[
"git",
"-c",
"user.name=CrewAI",
"-c",
"user.email=deploy@crewai.com",
"commit",
"--allow-empty",
"-m",
"Initial crew",
],
stdout="",
)
repo = Repository.initialize(path=str(tmp_path))
assert repo.path == str(tmp_path)
exclude_file = tmp_path / ".git" / "info" / "exclude"
exclude_text = exclude_file.read_text()
assert ".env" in exclude_text
assert "!.env.example" in exclude_text
assert "!.env.sample" in exclude_text
assert "__pycache__/" in exclude_text
def test_deployable_files_uses_git_excludes(fp, repository):
fp.register(
["git", "ls-files", "--cached", "--others", "--exclude-standard"],
stdout="pyproject.toml\nsrc/main.py\n",
)
assert repository.deployable_files() == ["pyproject.toml", "src/main.py"]

102
lib/cli/tests/test_install_crew.py Normal file
View File

@@ -0,0 +1,102 @@
from pathlib import Path
import subprocess
import pytest
import crewai_cli.install_crew as install_crew_module
@pytest.fixture(autouse=True)
def _tool_credentials(monkeypatch):
monkeypatch.setattr(
install_crew_module,
"build_env_with_all_tool_credentials",
lambda: {"CREWAI_TEST": "1"},
)
def test_install_crew_json_project_skips_project_install(
fp, monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "json_crew"
[tool.crewai]
type = "crew"
""".strip()
)
(tmp_path / "crew.jsonc").write_text("{}\n")
fp.register(["uv", "sync", "--no-install-project"], stdout="")
install_crew_module.install_crew([])
def test_install_crew_json_project_with_python_package_installs_project(
fp, monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "hybrid-crew"
[tool.crewai]
type = "crew"
""".strip()
)
(tmp_path / "crew.jsonc").write_text("{}\n")
package_dir = tmp_path / "src" / "hybrid_crew"
package_dir.mkdir(parents=True)
(package_dir / "crew.py").write_text("class HybridCrew: ...\n")
fp.register(["uv", "sync"], stdout="")
install_crew_module.install_crew([])
def test_install_crew_flow_project_installs_project(fp, monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text(
"""
[project]
name = "flow_project"
[tool.crewai]
type = "flow"
""".strip()
)
(tmp_path / "crew.jsonc").write_text("{}\n")
fp.register(["uv", "sync"], stdout="")
install_crew_module.install_crew([])
def test_install_crew_classic_project_installs_project(
fp, monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'classic'\n")
fp.register(["uv", "sync"], stdout="")
install_crew_module.install_crew([])
def test_install_crew_install_project_false_adds_no_install_project(fp):
fp.register(["uv", "sync", "--no-install-project", "--frozen"], stdout="")
install_crew_module.install_crew(["--frozen"], install_project=False)
def test_install_crew_reraises_sync_failure_when_requested(fp):
fp.register(["uv", "sync"], returncode=1, stderr="sync failed\n")
with pytest.raises(subprocess.CalledProcessError):
install_crew_module.install_crew([], raise_on_error=True)
def test_install_crew_swallows_sync_failure_by_default(fp):
fp.register(["uv", "sync"], returncode=1, stderr="sync failed\n")
install_crew_module.install_crew([])

30
lib/cli/tests/test_plus_api.py
View File

@@ -292,6 +292,36 @@ class TestPlusAPI(unittest.TestCase):
"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,
)
@patch("crewai_core.plus_api.Settings")
@patch.dict(os.environ, {"CREWAI_PLUS_URL": ""})
def test_custom_base_url(self, mock_settings_class):

330
lib/cli/tests/test_run_crew.py
View File

@@ -2,6 +2,8 @@
import os
from pathlib import Path
import subprocess
import sys
import pytest
from crewai_core.constants import CREWAI_TRAINED_AGENTS_FILE_ENV
@@ -14,16 +16,340 @@ def test_run_crew_forwards_trained_agents_file_to_json_crews(monkeypatch):
monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: True)
called: dict = {}
def fake_run_json_crew(trained_agents_file=None):
def fake_run_json_crew_in_project_env(trained_agents_file=None):
called["trained_agents_file"] = trained_agents_file
monkeypatch.setattr(run_crew_module, "_run_json_crew", fake_run_json_crew)
monkeypatch.setattr(
run_crew_module,
"_run_json_crew_in_project_env",
fake_run_json_crew_in_project_env,
)
run_crew_module.run_crew(trained_agents_file="some.pkl")
assert called == {"trained_agents_file": "some.pkl"}
def test_json_run_uses_project_env_when_pyproject_exists(monkeypatch, tmp_path: Path):
"""JSON crew runs should execute inside the project uv environment."""
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
install_calls = []
subprocess_calls = []
monkeypatch.setattr(
run_crew_module,
"_install_json_crew_dependencies_if_needed",
lambda: install_calls.append(True),
)
monkeypatch.setattr(
run_crew_module,
"build_env_with_all_tool_credentials",
lambda: {"EXISTING": "value"},
)
def fake_subprocess_run(command, **kwargs):
subprocess_calls.append((command, kwargs))
monkeypatch.setattr(run_crew_module.subprocess, "run", fake_subprocess_run)
run_crew_module._run_json_crew_in_project_env(
trained_agents_file="trained.pkl"
)
expected_env = {
"EXISTING": "value",
run_crew_module._CREWAI_CLI_RUNNER_PACKAGE_DIR_ENV: str(
Path(run_crew_module.__file__).resolve().parent
),
CREWAI_TRAINED_AGENTS_FILE_ENV: "trained.pkl",
}
if local_crewai_source_dir := run_crew_module._find_local_crewai_source_dir():
expected_env[run_crew_module._CREWAI_RUNNER_SOURCE_DIR_ENV] = str(
local_crewai_source_dir
)
assert install_calls == [True]
assert subprocess_calls == [
(
[
"uv",
"run",
"--no-sync",
"python",
"-c",
run_crew_module._JSON_CREW_RUNNER_CODE,
],
{
"capture_output": False,
"text": True,
"check": True,
"env": expected_env,
},
)
]
def test_json_run_uses_poetry_run_for_poetry_lock_without_uv_lock(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "poetry.lock").write_text("# lock\n")
monkeypatch.setattr(
run_crew_module,
"_install_json_crew_dependencies_if_needed",
lambda: None,
)
monkeypatch.setattr(
run_crew_module,
"build_env_with_all_tool_credentials",
lambda: {},
)
subprocess_calls = []
def fake_subprocess_run(command, **kwargs):
subprocess_calls.append((command, kwargs))
monkeypatch.setattr(run_crew_module.subprocess, "run", fake_subprocess_run)
run_crew_module._run_json_crew_in_project_env()
expected_env = {
run_crew_module._CREWAI_CLI_RUNNER_PACKAGE_DIR_ENV: str(
Path(run_crew_module.__file__).resolve().parent
),
}
if local_crewai_source_dir := run_crew_module._find_local_crewai_source_dir():
expected_env[run_crew_module._CREWAI_RUNNER_SOURCE_DIR_ENV] = str(
local_crewai_source_dir
)
assert subprocess_calls == [
(
[
"poetry",
"run",
"python",
"-c",
run_crew_module._JSON_CREW_RUNNER_CODE,
],
{
"capture_output": False,
"text": True,
"check": True,
"env": expected_env,
},
)
]
def test_json_runner_code_loads_current_cli_package_over_project_env(tmp_path: Path):
old_parent = tmp_path / "old"
old_pkg = old_parent / "crewai_cli"
old_pkg.mkdir(parents=True)
(old_pkg / "__init__.py").write_text("")
(old_pkg / "run_crew.py").write_text("raise ImportError('old package used')\n")
old_crewai_project = old_parent / "crewai" / "project"
old_crewai_project.mkdir(parents=True)
(old_parent / "crewai" / "__init__.py").write_text("")
(old_crewai_project / "__init__.py").write_text("")
(old_crewai_project / "json_loader.py").write_text(
"raise ImportError('old crewai used')\n"
)
current_pkg = tmp_path / "current" / "crewai_cli"
current_pkg.mkdir(parents=True)
marker = tmp_path / "marker.txt"
(current_pkg / "__init__.py").write_text("")
(current_pkg / "run_crew.py").write_text(
"from pathlib import Path\n"
"from crewai.project.json_loader import SOURCE\n"
"def _run_json_crew(trained_agents_file=None):\n"
f" Path({str(marker)!r}).write_text(SOURCE + ':' + (trained_agents_file or ''))\n"
)
current_crewai_project = tmp_path / "current_crewai_src" / "crewai" / "project"
current_crewai_project.mkdir(parents=True)
(tmp_path / "current_crewai_src" / "crewai" / "__init__.py").write_text("")
(current_crewai_project / "__init__.py").write_text("")
(current_crewai_project / "json_loader.py").write_text("SOURCE = 'current'\n")
env = os.environ.copy()
env["PYTHONPATH"] = str(old_parent)
env[run_crew_module._CREWAI_CLI_RUNNER_PACKAGE_DIR_ENV] = str(current_pkg)
env[run_crew_module._CREWAI_RUNNER_SOURCE_DIR_ENV] = str(
tmp_path / "current_crewai_src"
)
env[CREWAI_TRAINED_AGENTS_FILE_ENV] = "trained.pkl"
subprocess.run(
[sys.executable, "-c", run_crew_module._JSON_CREW_RUNNER_CODE],
check=True,
env=env,
cwd=tmp_path,
)
assert marker.read_text() == "current:trained.pkl"
def test_json_run_without_pyproject_runs_in_process(monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
called: dict = {}
def fake_run_json_crew(trained_agents_file=None):
called["trained_agents_file"] = trained_agents_file
return "result"
monkeypatch.setattr(run_crew_module, "_run_json_crew", fake_run_json_crew)
assert (
run_crew_module._run_json_crew_in_project_env(
trained_agents_file="trained.pkl"
)
== "result"
)
assert called == {"trained_agents_file": "trained.pkl"}
def test_json_project_env_run_failure_exits_nonzero(monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(
run_crew_module, "_install_json_crew_dependencies_if_needed", lambda: None
)
monkeypatch.setattr(
run_crew_module, "build_env_with_all_tool_credentials", lambda: {}
)
def fake_subprocess_run(command, **kwargs):
raise subprocess.CalledProcessError(7, command)
monkeypatch.setattr(run_crew_module.subprocess, "run", fake_subprocess_run)
with pytest.raises(SystemExit) as exc_info:
run_crew_module._run_json_crew_in_project_env()
assert exc_info.value.code == 7
def test_json_run_installs_dependencies_when_pyproject_has_no_lockfile(
monkeypatch, tmp_path: Path
):
"""JSON crew runs should lock/sync project dependencies only when needed."""
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
calls = []
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
calls.append((proxy_options, raise_on_error, install_project))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
run_crew_module._install_json_crew_dependencies_if_needed()
assert calls == [([], True, None)]
def test_json_run_syncs_frozen_when_uv_lock_exists_without_venv(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "uv.lock").write_text("# lock\n")
calls = []
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
calls.append((proxy_options, raise_on_error, install_project))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
run_crew_module._install_json_crew_dependencies_if_needed()
assert calls == [(["--frozen"], True, None)]
def test_json_run_skips_uv_sync_when_only_poetry_lock_exists_without_venv(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / "poetry.lock").write_text("# lock\n")
calls = []
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
calls.append((proxy_options, raise_on_error, install_project))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
run_crew_module._install_json_crew_dependencies_if_needed()
assert calls == []
@pytest.mark.parametrize("lockfile", ["uv.lock", "poetry.lock"])
def test_json_run_skips_dependency_install_when_lockfile_and_venv_exist(
monkeypatch, tmp_path: Path, lockfile: str
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
(tmp_path / lockfile).write_text("# lock\n")
(tmp_path / ".venv").mkdir()
calls = []
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
calls.append((proxy_options, raise_on_error, install_project))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
run_crew_module._install_json_crew_dependencies_if_needed()
assert calls == []
def test_json_run_skips_dependency_install_without_pyproject(
monkeypatch, tmp_path: Path
):
monkeypatch.chdir(tmp_path)
calls = []
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
calls.append((proxy_options, raise_on_error))
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
run_crew_module._install_json_crew_dependencies_if_needed()
assert calls == []
def test_json_run_install_failure_exits_nonzero(monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
def fake_install_crew(
proxy_options, *, raise_on_error=False, install_project=None
):
raise subprocess.CalledProcessError(42, ["uv", "sync"])
monkeypatch.setattr("crewai_cli.install_crew.install_crew", fake_install_crew)
with pytest.raises(SystemExit) as exc_info:
run_crew_module._install_json_crew_dependencies_if_needed()
assert exc_info.value.code == 42
def test_run_json_crew_exports_trained_agents_env(monkeypatch, tmp_path: Path):
"""JSON crews run in-process, so the pickle path must land in the env var."""
monkeypatch.chdir(tmp_path)