mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-09-20 18:13:49 +00:00
5990ead5ec
* feat(cli): record why a deployment create failed `crewai deploy create` counts every attempt (`Create Crew Deployment`) and every success (`Crew Deployment Created`), but the gap between them carried no cause: among clients able to emit the success span, the CLI succeeds 96.7% of the time and the run TUI 36.4%, and nothing said why. A third span, `Crew Deployment Failed`, now fires for every failure after the attempt is counted, with a closed vocabulary `reason` (api_4xx, api_5xx, invalid_response, network_error, zip_error, user_declined, unexpected), the HTTP `status_code` when the API answered, and the existing `source`. Never the error message. The request path is factored into `_request_crew_creation`; every exception is classified, reported and re-raised unchanged, so CLI and TUI behaviour is the same as before. HTTP failures are classified before `_validate_response`, which still prints and exits as it did. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(cli): classify deploy create failures by status and by stage Review fixes on the failure span. Check the HTTP class before the body so a gateway's HTML page counts as api_4xx / api_5xx with its code. Treat a 2xx whose body is not a JSON object carrying uuid and status as invalid_response and exit cleanly, instead of emitting a success span and crashing in the display step. Recognise archive failures by a dedicated ArchiveError (a ValueError) raised from create_project_zip, so the git helpers' own ValueErrors no longer read as zip_error; a failed ZIP write is wrapped and its partial file removed. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(cli): treat staging and temp-file failures as archive errors `create_project_zip` only wrapped the ZIP write, so an `OSError` while staging files or creating the temporary archive escaped as a bare `OSError` and the deploy command recorded it as `unexpected` instead of `zip_error`. The archive boundary now covers staging, temp-file creation and the write; the staging directory is removed on every path, and no partial archive is left behind. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(cli): tell a non-JSON 2xx apart from a non-creation 2xx A proxy's 200 HTML page and a JSON body missing the creation fields were both recorded as `invalid_response`. They are different failures, one in the network path and one in the API contract, so the deploy failure span now records `invalid_json` for the first and `invalid_creation_response` for the second. Requested in review. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com> Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com>
1319 lines
54 KiB
Python
1319 lines
54 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
|
|
from crewai_cli.deploy.archive import ArchiveError
|
|
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 _status_response(
|
|
self,
|
|
*,
|
|
uuid: str = "test-uuid",
|
|
zip_deployment: bool | None = None,
|
|
is_success: bool = True,
|
|
) -> MagicMock:
|
|
response = MagicMock()
|
|
response.is_success = is_success
|
|
payload: dict = {"uuid": uuid}
|
|
if zip_deployment is not None:
|
|
payload["zip_deployment"] = zip_deployment
|
|
response.json.return_value = payload
|
|
return response
|
|
|
|
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
|
|
)
|
|
self.mock_client.crew_status_by_uuid.return_value = self._status_response(
|
|
zip_deployment=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.crew_status_by_uuid.assert_called_once_with("test-uuid")
|
|
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
|
|
)
|
|
self.mock_client.crew_status_by_name.return_value = self._status_response(
|
|
zip_deployment=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.crew_status_by_name.assert_called_once_with("test_project")
|
|
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
|
|
self.mock_client.crew_status_by_name.return_value = self._status_response(
|
|
zip_deployment=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"
|
|
)
|
|
self.mock_client.crew_status_by_name.return_value = self._status_response(
|
|
zip_deployment=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.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")
|
|
self.mock_client.crew_status_by_uuid.return_value = self._status_response(
|
|
zip_deployment=True
|
|
)
|
|
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.crew_status_by_uuid.assert_called_once_with("test-uuid")
|
|
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")
|
|
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 = self._status_response(
|
|
zip_deployment=True
|
|
)
|
|
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.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_zip_amp_source_uploads_even_when_origin_exists(
|
|
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 = (
|
|
"https://github.com/test/repo.git"
|
|
)
|
|
mock_repository.return_value.create_initial_commit_if_needed.return_value = (
|
|
False
|
|
)
|
|
mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
|
|
self.mock_client.crew_status_by_uuid.return_value = self._status_response(
|
|
zip_deployment=True
|
|
)
|
|
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.git.Repository")
|
|
@patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
|
|
def test_deploy_falls_back_to_origin_when_zip_deployment_is_missing(
|
|
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
|
|
)
|
|
self.mock_client.crew_status_by_uuid.return_value = self._status_response()
|
|
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")
|
|
self.mock_client.update_crew_from_zip.assert_not_called()
|
|
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_falls_back_to_origin_when_status_request_fails(
|
|
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
|
|
)
|
|
self.mock_client.crew_status_by_uuid.return_value = self._status_response(
|
|
zip_deployment=True, is_success=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")
|
|
self.mock_client.update_crew_from_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_falls_back_to_zip_when_status_fails_and_origin_is_missing(
|
|
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")
|
|
self.mock_client.crew_status_by_uuid.side_effect = httpx.ConnectError(
|
|
"offline"
|
|
)
|
|
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.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()
|
|
|
|
# --- why a create failed (the Crew Deployment Failed span) ---------------
|
|
|
|
def _git_path(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 = ""
|
|
|
|
@staticmethod
|
|
def _api_response(status_code: int, body: object) -> MagicMock:
|
|
response = MagicMock()
|
|
response.status_code = status_code
|
|
response.is_success = 200 <= status_code < 300
|
|
if isinstance(body, Exception):
|
|
response.json.side_effect = body
|
|
else:
|
|
response.json.return_value = body
|
|
return response
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_rejected_create_reports_the_api_class_and_status(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""A 4xx names the class and carries the exact code; the message never leaves."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
422, {"name": ["has already been taken"]}
|
|
)
|
|
|
|
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.crew_deployment_failed_span.assert_called_once_with(
|
|
"api_4xx", source="cli", status_code=422
|
|
)
|
|
telemetry.crew_deployment_created_span.assert_not_called()
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_server_error_reports_api_5xx(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
503, {"error": "upstream unavailable"}
|
|
)
|
|
|
|
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.crew_deployment_failed_span.assert_called_once_with(
|
|
"api_5xx", source="cli", status_code=503
|
|
)
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_non_json_success_body_reports_invalid_json(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""A proxy's 200 HTML page: _validate_response rejects it, so it is a
|
|
failure with a 2xx attached, and it is not the same failure as a JSON
|
|
body that is missing the creation fields."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
200, ValueError("not json")
|
|
)
|
|
|
|
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.crew_deployment_failed_span.assert_called_once_with(
|
|
"invalid_json", source="cli", status_code=200
|
|
)
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_transport_failure_reports_network_error_and_propagates(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""No response, so no status; the exception reaches the caller unchanged."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.side_effect = httpx.ConnectError("refused")
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
with pytest.raises(httpx.ConnectError, match="refused"):
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"network_error", 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_a_failed_archive_reports_zip_error(
|
|
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.side_effect = ArchiveError(
|
|
"No deployable project files were found."
|
|
)
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
with pytest.raises(ArchiveError, match="No deployable project files"):
|
|
self.deploy_command.create_crew(skip_validate=True, confirm=True)
|
|
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"zip_error", source="cli"
|
|
)
|
|
self.mock_client.create_crew_from_zip.assert_not_called()
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_git_helper_error_is_not_a_zip_error(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""The git helpers raise plain ValueError; only ArchiveError is a ZIP problem."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
mock_repository.return_value.origin_url.side_effect = ValueError(
|
|
"Git remote lookup failed"
|
|
)
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
with pytest.raises(ValueError, match="Git remote lookup failed"):
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"unexpected", 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_non_json_error_page_keeps_its_api_class(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""A gateway's HTML 502 is an api_5xx, not an invalid response."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
502, ValueError("not json")
|
|
)
|
|
|
|
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.crew_deployment_failed_span.assert_called_once_with(
|
|
"api_5xx", source="cli", status_code=502
|
|
)
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_success_body_that_is_not_a_creation_reports_invalid_creation_response(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""A 2xx JSON body without a deployment uuid is a broken creation
|
|
contract, not a success, and must not crash."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
for body in ([{"uuid": "in-a-list"}], {"status": "created"}, {}):
|
|
with self.subTest(body=body):
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
200, body
|
|
)
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()) as fake_out:
|
|
with self.assertRaises(SystemExit) as exit_info:
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
assert exit_info.exception.code == 1
|
|
assert "no deployment uuid was returned" in fake_out.getvalue()
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"invalid_creation_response", source="cli", status_code=200
|
|
)
|
|
telemetry.crew_deployment_created_span.assert_not_called()
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_an_abort_at_the_prompt_reports_user_declined(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
mock_input.side_effect = KeyboardInterrupt
|
|
|
|
with patch.object(self.deploy_command, "_telemetry") as telemetry:
|
|
with patch("sys.stdout", new=StringIO()):
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
self.deploy_command.create_crew(skip_validate=True)
|
|
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"user_declined", source="cli"
|
|
)
|
|
self.mock_client.create_crew.assert_not_called()
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_failure_from_the_run_tui_keeps_its_source(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
"""The TUI succeeds far less often than the CLI; the split must survive."""
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
403, {"error": "forbidden"}
|
|
)
|
|
|
|
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, confirm=True, source="tui"
|
|
)
|
|
|
|
telemetry.crew_deployment_failed_span.assert_called_once_with(
|
|
"api_4xx", source="tui", status_code=403
|
|
)
|
|
|
|
@patch("crewai_cli.deploy.main.fetch_and_json_env_file")
|
|
@patch("crewai_cli.deploy.main.git.Repository")
|
|
@patch("builtins.input")
|
|
def test_a_successful_create_reports_no_failure(
|
|
self, mock_input, mock_repository, mock_fetch_env
|
|
):
|
|
self._git_path(mock_input, mock_repository, mock_fetch_env)
|
|
self.mock_client.create_crew.return_value = self._api_response(
|
|
201, {"uuid": "new-uuid", "status": "created"}
|
|
)
|
|
|
|
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_failed_span.assert_not_called()
|
|
telemetry.crew_deployment_created_span.assert_called_once_with(
|
|
uuid="new-uuid", 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_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)
|