mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-09-20 18:13:49 +00:00
fcdeb3d98d
`Create Crew Deployment` fires before the API call that creates the deployment, so it counts creation ATTEMPTS and cannot carry the uuid - the call that creates the deployment is the call that returns it. Live effect: `create_deployment` reads 76,015 events with 0 carrying a uuid (0.000%), so deployments cannot be joined to anything. Moving the existing span after the response would fix the uuid and silently redefine the metric, turning attempts into successes; the deployment churn figures are built on attempts. So this adds a second span rather than moving the first. `Crew Deployment Created` fires after `_validate_response`, which raises SystemExit on failure - a failed create therefore still counts as an attempt and reports no creation. Both creation paths, git remote and zip upload, converge on that line and both return the uuid. Emits no `deploy:created` feature count: the attempt span already does, and a second emit would double the deployment count that origin-independent aggregation depends on. Tests cover the emitter (uuid carried, distinct span name, no second feature count, absent-vs-empty uuid) and the call site across both creation paths plus the failure path. The warehouse consumer must be widened BEFORE this merges or it delivers nothing: `mv_span_fanout_forward` filters on a hard-coded 13-name allowlist, and `mv_fanout_deployment_spans`'s `multiIf` ends in a catch-all `remove_crew` arm that would mislabel the new span. Runbook prepared separately. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
912 lines
36 KiB
Python
912 lines
36 KiB
Python
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.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 == []
|
|
|
|
|
|
def test_deployment_page_url_prefers_deployment_id():
|
|
assert (
|
|
deploy_main._deployment_page_url(
|
|
"https://app.crewai.com",
|
|
{"uuid": "crew-uuid", "deployment_id": 128687},
|
|
)
|
|
== "https://app.crewai.com/crewai_plus/deployments/128687"
|
|
)
|
|
|
|
|
|
def test_deployment_page_url_prefers_nested_deployment_id_over_crew_uuid():
|
|
assert (
|
|
deploy_main._deployment_page_url(
|
|
"https://app.crewai.com",
|
|
{"uuid": "crew-uuid", "deployment": {"deployment_id": 128687}},
|
|
)
|
|
== "https://app.crewai.com/crewai_plus/deployments/128687"
|
|
)
|
|
|
|
|
|
def test_deployment_page_url_does_not_use_uuid_as_deployment_id():
|
|
assert (
|
|
deploy_main._deployment_page_url(
|
|
"https://app.crewai.com/",
|
|
{"uuid": "crew-uuid", "deployment": {"uuid": "deployment-uuid"}},
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
class TestDeployCommand(unittest.TestCase):
|
|
@patch("crewai_cli.command.get_auth_token")
|
|
@patch("crewai_cli.deploy.main.get_project_name")
|
|
@patch("crewai_cli.command.PlusAPI")
|
|
def setUp(
|
|
self,
|
|
mock_plus_api,
|
|
mock_get_project_name,
|
|
mock_get_auth_token,
|
|
):
|
|
self.mock_get_auth_token = mock_get_auth_token
|
|
self.mock_get_project_name = mock_get_project_name
|
|
self.mock_plus_api = mock_plus_api
|
|
|
|
self.mock_get_auth_token.return_value = "test_token"
|
|
self.mock_get_project_name.return_value = "test_project"
|
|
|
|
self.deploy_command = deploy_main.DeployCommand()
|
|
self.mock_client = self.deploy_command.plus_api_client
|
|
self.mock_client.base_url = "https://app.crewai.com"
|
|
self.mock_browser_open_patcher = patch(
|
|
"crewai_cli.deploy.main.webbrowser.open"
|
|
)
|
|
self.mock_browser_open = self.mock_browser_open_patcher.start()
|
|
self.addCleanup(self.mock_browser_open_patcher.stop)
|
|
|
|
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):
|
|
deploy_main.DeployCommand()
|
|
|
|
def test_validate_response_successful_response(self):
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.json.return_value = {"message": "Success"}
|
|
mock_response.status_code = 200
|
|
mock_response.is_success = True
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._validate_response(mock_response)
|
|
assert fake_out.getvalue() == ""
|
|
|
|
def test_validate_response_json_decode_error(self):
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.json.side_effect = json.JSONDecodeError("Decode error", "", 0)
|
|
mock_response.status_code = 500
|
|
mock_response.content = b"Invalid JSON"
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
with pytest.raises(SystemExit):
|
|
self.deploy_command._validate_response(mock_response)
|
|
output = fake_out.getvalue()
|
|
assert (
|
|
"Failed to parse response from Enterprise API failed. Details:"
|
|
in output
|
|
)
|
|
assert "Status Code: 500" in output
|
|
assert "Response:\nInvalid JSON" in output
|
|
|
|
def test_validate_response_422_error(self):
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.json.return_value = {
|
|
"field1": ["Error message 1"],
|
|
"field2": ["Error message 2"],
|
|
}
|
|
mock_response.status_code = 422
|
|
mock_response.is_success = False
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
with pytest.raises(SystemExit):
|
|
self.deploy_command._validate_response(mock_response)
|
|
output = fake_out.getvalue()
|
|
assert (
|
|
"Failed to complete operation. Please fix the following errors:"
|
|
in output
|
|
)
|
|
assert "Field1 Error message 1" in output
|
|
assert "Field2 Error message 2" in output
|
|
|
|
def test_validate_response_other_error(self):
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.json.return_value = {"error": "Something went wrong"}
|
|
mock_response.status_code = 500
|
|
mock_response.is_success = False
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
with pytest.raises(SystemExit):
|
|
self.deploy_command._validate_response(mock_response)
|
|
output = fake_out.getvalue()
|
|
assert "Request to Enterprise API failed. Details:" in output
|
|
assert "Details:\nSomething went wrong" in output
|
|
|
|
def test_standard_no_param_error_message(self):
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._standard_no_param_error_message()
|
|
self.assertIn("No UUID provided", fake_out.getvalue())
|
|
|
|
def test_display_deployment_info(self):
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._display_deployment_info(
|
|
{"uuid": "test-uuid", "id": 128687, "status": "deployed"}
|
|
)
|
|
self.assertIn("Deploying the crew...", fake_out.getvalue())
|
|
self.assertIn("test-uuid", fake_out.getvalue())
|
|
self.assertIn("deployed", fake_out.getvalue())
|
|
self.assertIn(
|
|
"https://app.crewai.com/crewai_plus/deployments/128687",
|
|
fake_out.getvalue(),
|
|
)
|
|
self.mock_browser_open.assert_called_once_with(
|
|
"https://app.crewai.com/crewai_plus/deployments/128687"
|
|
)
|
|
|
|
def test_display_deployment_info_warns_when_browser_open_returns_false(self):
|
|
self.mock_browser_open.return_value = False
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._display_deployment_info(
|
|
{"uuid": "test-uuid", "id": 128687, "status": "deployed"}
|
|
)
|
|
self.assertIn(
|
|
"Could not open the deployment page automatically.",
|
|
fake_out.getvalue(),
|
|
)
|
|
|
|
self.mock_browser_open.assert_called_once_with(
|
|
"https://app.crewai.com/crewai_plus/deployments/128687"
|
|
)
|
|
|
|
def test_display_deployment_info_warns_when_browser_open_raises(self):
|
|
self.mock_browser_open.side_effect = RuntimeError("no browser")
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._display_deployment_info(
|
|
{"uuid": "test-uuid", "id": 128687, "status": "deployed"}
|
|
)
|
|
self.assertIn(
|
|
"Could not open the deployment page automatically.",
|
|
fake_out.getvalue(),
|
|
)
|
|
|
|
self.mock_browser_open.assert_called_once_with(
|
|
"https://app.crewai.com/crewai_plus/deployments/128687"
|
|
)
|
|
|
|
def test_display_creation_success_resolves_deployment_page_id_from_status(self):
|
|
status_response = MagicMock()
|
|
status_response.is_success = True
|
|
status_response.json.return_value = {
|
|
"uuid": "new-uuid",
|
|
"id": 128774,
|
|
"status": "created",
|
|
}
|
|
self.mock_client.crew_status_by_uuid.return_value = status_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._display_creation_success(
|
|
{"uuid": "new-uuid", "status": "created"}
|
|
)
|
|
output = fake_out.getvalue()
|
|
|
|
self.assertIn("crewai deploy push --uuid new-uuid", output)
|
|
self.assertIn(
|
|
"https://app.crewai.com/crewai_plus/deployments/128774",
|
|
output,
|
|
)
|
|
self.assertNotIn(
|
|
"https://app.crewai.com/crewai_plus/deployments/new-uuid",
|
|
output,
|
|
)
|
|
self.mock_client.crew_status_by_uuid.assert_called_once_with("new-uuid")
|
|
self.mock_browser_open.assert_called_once_with(
|
|
"https://app.crewai.com/crewai_plus/deployments/128774"
|
|
)
|
|
|
|
def test_open_deployment_page_does_not_open_uuid_url_when_status_lookup_fails(
|
|
self,
|
|
):
|
|
status_response = MagicMock()
|
|
status_response.is_success = False
|
|
self.mock_client.crew_status_by_uuid.return_value = status_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._open_deployment_page({"uuid": "new-uuid"})
|
|
output = fake_out.getvalue()
|
|
|
|
self.mock_client.crew_status_by_uuid.assert_called_once_with("new-uuid")
|
|
self.mock_browser_open.assert_not_called()
|
|
self.assertNotIn("deployments/new-uuid", output)
|
|
|
|
def test_display_logs(self):
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command._display_logs(
|
|
[{"timestamp": "2023-01-01", "level": "INFO", "message": "Test log"}]
|
|
)
|
|
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, 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"}
|
|
self.mock_client.deploy_by_uuid.return_value = mock_response
|
|
|
|
self.deploy_command.deploy(uuid="test-uuid", skip_validate=True)
|
|
|
|
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, 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"}
|
|
self.mock_client.deploy_by_name.return_value = mock_response
|
|
|
|
self.deploy_command.deploy(skip_validate=True)
|
|
|
|
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")
|
|
@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_repository, mock_fetch_env):
|
|
mock_fetch_env.return_value = {"ENV_VAR": "value"}
|
|
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
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
self.assertIn("Deployment created successfully!", fake_out.getvalue())
|
|
self.assertIn("new-uuid", fake_out.getvalue())
|
|
|
|
@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_reports_the_created_uuid_from_the_git_path(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""The attempt span cannot carry the uuid; the post-success span must."""
|
|
mock_fetch_env.return_value = {"ENV_VAR": "value"}
|
|
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
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
telemetry.crew_deployment_created_span.assert_called_once_with(
|
|
uuid="new-uuid", source="cli"
|
|
)
|
|
telemetry.create_crew_deployment_span.assert_called_once_with(source="cli")
|
|
|
|
@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_reports_the_created_uuid_from_the_zip_path(
|
|
self, mock_repository, mock_fetch_env, mock_create_project_zip
|
|
):
|
|
"""The two creation paths converge, so the uuid must arrive from both."""
|
|
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.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
self.deploy_command.create_crew(skip_validate=True, confirm=True)
|
|
|
|
telemetry.crew_deployment_created_span.assert_called_once_with(
|
|
uuid="zip-uuid", source="cli"
|
|
)
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_failed_create_counts_the_attempt_but_reports_no_creation(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""The whole point of two spans: a failed create is an attempt, not a success.
|
|
|
|
Collapsing them into one post-success span would silently convert the
|
|
creation-attempt metric into a creation-success metric, which the churn
|
|
figures are built on.
|
|
"""
|
|
mock_fetch_env.return_value = {"ENV_VAR": "value"}
|
|
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 = 500
|
|
mock_response.is_success = False
|
|
mock_response.json.return_value = {"error": "boom"}
|
|
self.mock_client.create_crew.return_value = mock_response
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
with self.assertRaises(SystemExit):
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
telemetry.create_crew_deployment_span.assert_called_once_with(source="cli")
|
|
telemetry.crew_deployment_created_span.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_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(confirm=True, skip_validate=True)
|
|
output = fake_out.getvalue()
|
|
|
|
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_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(confirm=True, skip_validate=True)
|
|
output = fake_out.getvalue()
|
|
|
|
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):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = [
|
|
{"name": "Crew1", "uuid": "uuid1", "status": "active"},
|
|
{"name": "Crew2", "uuid": "uuid2", "status": "inactive"},
|
|
]
|
|
self.mock_client.list_crews.return_value = mock_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command.list_crews()
|
|
self.assertIn("Crew1 (uuid1) active", fake_out.getvalue())
|
|
self.assertIn("Crew2 (uuid2) inactive", fake_out.getvalue())
|
|
|
|
def test_get_crew_status(self):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"name": "InternalCrew", "status": "active"}
|
|
self.mock_client.crew_status_by_name.return_value = mock_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command.get_crew_status()
|
|
self.assertIn("InternalCrew", fake_out.getvalue())
|
|
self.assertIn("active", fake_out.getvalue())
|
|
|
|
def test_get_crew_logs(self):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = [
|
|
{"timestamp": "2023-01-01", "level": "INFO", "message": "Log1"},
|
|
{"timestamp": "2023-01-02", "level": "ERROR", "message": "Log2"},
|
|
]
|
|
self.mock_client.crew_by_name.return_value = mock_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command.get_crew_logs(None)
|
|
self.assertIn("2023-01-01 - INFO: Log1", fake_out.getvalue())
|
|
self.assertIn("2023-01-02 - ERROR: Log2", fake_out.getvalue())
|
|
|
|
def test_remove_crew(self):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 204
|
|
self.mock_client.delete_crew_by_name.return_value = mock_response
|
|
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
self.deploy_command.remove_crew(None)
|
|
self.assertIn(
|
|
"Crew 'test_project' removed successfully", fake_out.getvalue()
|
|
)
|
|
|
|
@unittest.skipIf(sys.version_info < (3, 11), "Requires Python 3.11+")
|
|
def test_parse_toml_python_311_plus(self):
|
|
toml_content = """
|
|
[tool.poetry]
|
|
name = "test_project"
|
|
version = "0.1.0"
|
|
|
|
[tool.poetry.dependencies]
|
|
python = "^3.11"
|
|
crewai = { extras = ["tools"], version = ">=0.51.0,<1.0.0" }
|
|
"""
|
|
parsed = parse_toml(toml_content)
|
|
self.assertEqual(parsed["tool"]["poetry"]["name"], "test_project")
|
|
|
|
@patch(
|
|
"builtins.open",
|
|
new_callable=unittest.mock.mock_open,
|
|
read_data="""
|
|
[project]
|
|
name = "test_project"
|
|
version = "0.1.0"
|
|
requires-python = ">=3.10,<3.14"
|
|
dependencies = ["crewai"]
|
|
""",
|
|
)
|
|
def test_get_project_name_python_310(self, mock_open):
|
|
from crewai_cli.utils import get_project_name
|
|
|
|
project_name = get_project_name()
|
|
print("project_name", project_name)
|
|
self.assertEqual(project_name, "test_project")
|
|
|
|
@unittest.skipIf(sys.version_info < (3, 11), "Requires Python 3.11+")
|
|
@patch(
|
|
"builtins.open",
|
|
new_callable=unittest.mock.mock_open,
|
|
read_data="""
|
|
[project]
|
|
name = "test_project"
|
|
version = "0.1.0"
|
|
requires-python = ">=3.10,<3.14"
|
|
dependencies = ["crewai"]
|
|
""",
|
|
)
|
|
def test_get_project_name_python_311_plus(self, mock_open):
|
|
from crewai_cli.utils import get_project_name
|
|
|
|
project_name = get_project_name()
|
|
self.assertEqual(project_name, "test_project")
|
|
|
|
def test_get_crewai_version(self):
|
|
from crewai_cli.version import get_crewai_version
|
|
|
|
assert isinstance(get_crewai_version(), str)
|