mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-08 23:58:34 +00:00
Feat/cli deploy (#1240)
* feat: set basic structure deploy commands * feat: add first iteration of CLI Deploy * feat: some minor refactor * feat: Add api, Deploy command and update cli * feat: Remove test token * feat: add auth0 lib, update cli and improve code * feat: update code and decouple auth * fix: parts of the code * feat: Add token manager to encrypt access token and get and save tokens * feat: add audience to costants * feat: add subsystem saving credentials and remove comment of type hinting * feat: add get crew version to send on header of request * feat: add docstrings * feat: add tests for authentication module * feat: add tests for utils * feat: add unit tests for cl * feat: add tests * feat: add deploy man tests * feat: fix type checking issue * feat: rename tests to pass ci * feat: fix pr issues * feat: fix get crewai versoin * fix: add timeout for tests.yml
This commit is contained in:
Eduardo Chiarotti
committed by
GitHub
GitHub
parent
4736604b4d
commit
f5246039e5
94
tests/cli/authentication/test_auth_main.py
Normal file
94
tests/cli/authentication/test_auth_main.py
Normal file
@@ -0,0 +1,94 @@
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import requests
|
||||
from crewai.cli.authentication.main import AuthenticationCommand
|
||||
|
||||
|
||||
class TestAuthenticationCommand(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.auth_command = AuthenticationCommand()
|
||||
|
||||
@patch("crewai.cli.authentication.main.requests.post")
|
||||
def test_get_device_code(self, mock_post):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
"device_code": "123456",
|
||||
"user_code": "ABCDEF",
|
||||
"verification_uri_complete": "https://example.com",
|
||||
"interval": 5,
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
device_code_data = self.auth_command._get_device_code()
|
||||
|
||||
self.assertEqual(device_code_data["device_code"], "123456")
|
||||
self.assertEqual(device_code_data["user_code"], "ABCDEF")
|
||||
self.assertEqual(
|
||||
device_code_data["verification_uri_complete"], "https://example.com"
|
||||
)
|
||||
self.assertEqual(device_code_data["interval"], 5)
|
||||
|
||||
@patch("crewai.cli.authentication.main.console.print")
|
||||
@patch("crewai.cli.authentication.main.webbrowser.open")
|
||||
def test_display_auth_instructions(self, mock_open, mock_print):
|
||||
device_code_data = {
|
||||
"verification_uri_complete": "https://example.com",
|
||||
"user_code": "ABCDEF",
|
||||
}
|
||||
|
||||
self.auth_command._display_auth_instructions(device_code_data)
|
||||
|
||||
mock_print.assert_any_call("1. Navigate to: ", "https://example.com")
|
||||
mock_print.assert_any_call("2. Enter the following code: ", "ABCDEF")
|
||||
mock_open.assert_called_once_with("https://example.com")
|
||||
|
||||
@patch("crewai.cli.authentication.main.requests.post")
|
||||
@patch("crewai.cli.authentication.main.validate_token")
|
||||
@patch("crewai.cli.authentication.main.console.print")
|
||||
def test_poll_for_token_success(self, mock_print, mock_validate_token, mock_post):
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {
|
||||
"id_token": "TOKEN",
|
||||
"access_token": "ACCESS_TOKEN",
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
self.auth_command._poll_for_token({"device_code": "123456"})
|
||||
|
||||
mock_validate_token.assert_called_once_with("TOKEN")
|
||||
mock_print.assert_called_once_with("\nWelcome to CrewAI+ !!", style="green")
|
||||
|
||||
@patch("crewai.cli.authentication.main.requests.post")
|
||||
@patch("crewai.cli.authentication.main.console.print")
|
||||
def test_poll_for_token_error(self, mock_print, mock_post):
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.json.return_value = {
|
||||
"error": "invalid_request",
|
||||
"error_description": "Invalid request",
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with self.assertRaises(requests.HTTPError):
|
||||
self.auth_command._poll_for_token({"device_code": "123456"})
|
||||
|
||||
mock_print.assert_not_called()
|
||||
|
||||
@patch("crewai.cli.authentication.main.requests.post")
|
||||
@patch("crewai.cli.authentication.main.console.print")
|
||||
def test_poll_for_token_timeout(self, mock_print, mock_post):
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.json.return_value = {
|
||||
"error": "authorization_pending",
|
||||
"error_description": "Authorization pending",
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
self.auth_command._poll_for_token({"device_code": "123456", "interval": 0.01})
|
||||
|
||||
mock_print.assert_called_once_with(
|
||||
"Timeout: Failed to get the token. Please try again.", style="bold red"
|
||||
)
|
||||
147
tests/cli/authentication/test_utils.py
Normal file
147
tests/cli/authentication/test_utils.py
Normal file
@@ -0,0 +1,147 @@
|
||||
import json
|
||||
import unittest
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from crewai.cli.authentication.utils import TokenManager, validate_token
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
|
||||
class TestValidateToken(unittest.TestCase):
|
||||
@patch("crewai.cli.authentication.utils.AsymmetricSignatureVerifier")
|
||||
@patch("crewai.cli.authentication.utils.TokenVerifier")
|
||||
def test_validate_token(self, mock_token_verifier, mock_asymmetric_verifier):
|
||||
mock_verifier_instance = mock_token_verifier.return_value
|
||||
mock_id_token = "mock_id_token"
|
||||
|
||||
validate_token(mock_id_token)
|
||||
|
||||
mock_asymmetric_verifier.assert_called_once_with(
|
||||
"https://dev-jzsr0j8zs0atl5ha.us.auth0.com/.well-known/jwks.json"
|
||||
)
|
||||
mock_token_verifier.assert_called_once_with(
|
||||
signature_verifier=mock_asymmetric_verifier.return_value,
|
||||
issuer="https://dev-jzsr0j8zs0atl5ha.us.auth0.com/",
|
||||
audience="CZtyRHuVW80HbLSjk4ggXNzjg4KAt7Oe",
|
||||
)
|
||||
mock_verifier_instance.verify.assert_called_once_with(mock_id_token)
|
||||
|
||||
|
||||
class TestTokenManager(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.token_manager = TokenManager()
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.read_secure_file")
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.save_secure_file")
|
||||
@patch("crewai.cli.authentication.utils.TokenManager._get_or_create_key")
|
||||
def test_get_or_create_key_existing(self, mock_get_or_create, mock_save, mock_read):
|
||||
mock_key = Fernet.generate_key()
|
||||
mock_get_or_create.return_value = mock_key
|
||||
|
||||
token_manager = TokenManager()
|
||||
result = token_manager.key
|
||||
|
||||
self.assertEqual(result, mock_key)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.Fernet.generate_key")
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.read_secure_file")
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.save_secure_file")
|
||||
def test_get_or_create_key_new(self, mock_save, mock_read, mock_generate):
|
||||
mock_key = b"new_key"
|
||||
mock_read.return_value = None
|
||||
mock_generate.return_value = mock_key
|
||||
|
||||
result = self.token_manager._get_or_create_key()
|
||||
|
||||
self.assertEqual(result, mock_key)
|
||||
mock_read.assert_called_once_with("secret.key")
|
||||
mock_generate.assert_called_once()
|
||||
mock_save.assert_called_once_with("secret.key", mock_key)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.save_secure_file")
|
||||
def test_save_tokens(self, mock_save):
|
||||
access_token = "test_token"
|
||||
expires_in = 3600
|
||||
|
||||
self.token_manager.save_tokens(access_token, expires_in)
|
||||
|
||||
mock_save.assert_called_once()
|
||||
args = mock_save.call_args[0]
|
||||
self.assertEqual(args[0], "tokens.enc")
|
||||
decrypted_data = self.token_manager.fernet.decrypt(args[1])
|
||||
data = json.loads(decrypted_data)
|
||||
self.assertEqual(data["access_token"], access_token)
|
||||
expiration = datetime.fromisoformat(data["expiration"])
|
||||
self.assertAlmostEqual(
|
||||
expiration,
|
||||
datetime.now() + timedelta(seconds=expires_in),
|
||||
delta=timedelta(seconds=1),
|
||||
)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.read_secure_file")
|
||||
def test_get_token_valid(self, mock_read):
|
||||
access_token = "test_token"
|
||||
expiration = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
data = {"access_token": access_token, "expiration": expiration}
|
||||
encrypted_data = self.token_manager.fernet.encrypt(json.dumps(data).encode())
|
||||
mock_read.return_value = encrypted_data
|
||||
|
||||
result = self.token_manager.get_token()
|
||||
|
||||
self.assertEqual(result, access_token)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.read_secure_file")
|
||||
def test_get_token_expired(self, mock_read):
|
||||
access_token = "test_token"
|
||||
expiration = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
data = {"access_token": access_token, "expiration": expiration}
|
||||
encrypted_data = self.token_manager.fernet.encrypt(json.dumps(data).encode())
|
||||
mock_read.return_value = encrypted_data
|
||||
|
||||
result = self.token_manager.get_token()
|
||||
|
||||
self.assertIsNone(result)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.get_secure_storage_path")
|
||||
@patch("builtins.open", new_callable=unittest.mock.mock_open)
|
||||
@patch("crewai.cli.authentication.utils.os.chmod")
|
||||
def test_save_secure_file(self, mock_chmod, mock_open, mock_get_path):
|
||||
mock_path = MagicMock()
|
||||
mock_get_path.return_value = mock_path
|
||||
filename = "test_file.txt"
|
||||
content = b"test_content"
|
||||
|
||||
self.token_manager.save_secure_file(filename, content)
|
||||
|
||||
mock_path.__truediv__.assert_called_once_with(filename)
|
||||
mock_open.assert_called_once_with(mock_path.__truediv__.return_value, "wb")
|
||||
mock_open().write.assert_called_once_with(content)
|
||||
mock_chmod.assert_called_once_with(mock_path.__truediv__.return_value, 0o600)
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.get_secure_storage_path")
|
||||
@patch(
|
||||
"builtins.open", new_callable=unittest.mock.mock_open, read_data=b"test_content"
|
||||
)
|
||||
def test_read_secure_file_exists(self, mock_open, mock_get_path):
|
||||
mock_path = MagicMock()
|
||||
mock_get_path.return_value = mock_path
|
||||
mock_path.__truediv__.return_value.exists.return_value = True
|
||||
filename = "test_file.txt"
|
||||
|
||||
result = self.token_manager.read_secure_file(filename)
|
||||
|
||||
self.assertEqual(result, b"test_content")
|
||||
mock_path.__truediv__.assert_called_once_with(filename)
|
||||
mock_open.assert_called_once_with(mock_path.__truediv__.return_value, "rb")
|
||||
|
||||
@patch("crewai.cli.authentication.utils.TokenManager.get_secure_storage_path")
|
||||
def test_read_secure_file_not_exists(self, mock_get_path):
|
||||
mock_path = MagicMock()
|
||||
mock_get_path.return_value = mock_path
|
||||
mock_path.__truediv__.return_value.exists.return_value = False
|
||||
filename = "test_file.txt"
|
||||
|
||||
result = self.token_manager.read_secure_file(filename)
|
||||
|
||||
self.assertIsNone(result)
|
||||
mock_path.__truediv__.assert_called_once_with(filename)
|
||||
@@ -2,8 +2,19 @@ from unittest import mock
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from crewai.cli.cli import reset_memories, test, train, version
|
||||
from crewai.cli.cli import (
|
||||
deploy_create,
|
||||
deploy_list,
|
||||
deploy_logs,
|
||||
deploy_push,
|
||||
deploy_remove,
|
||||
deply_status,
|
||||
reset_memories,
|
||||
signup,
|
||||
test,
|
||||
train,
|
||||
version,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -163,3 +174,106 @@ def test_test_invalid_string_iterations(evaluate_crew, runner):
|
||||
"Usage: test [OPTIONS]\nTry 'test --help' for help.\n\nError: Invalid value for '-n' / '--n_iterations': 'invalid' is not a valid integer.\n"
|
||||
in result.output
|
||||
)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.AuthenticationCommand")
|
||||
def test_signup(command, runner):
|
||||
mock_auth = command.return_value
|
||||
result = runner.invoke(signup)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_auth.signup.assert_called_once()
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_create(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deploy_create)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.create_crew.assert_called_once()
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_list(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deploy_list)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.list_crews.assert_called_once()
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_push(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
uuid = "test-uuid"
|
||||
result = runner.invoke(deploy_push, ["-u", uuid])
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.deploy.assert_called_once_with(uuid=uuid)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_push_no_uuid(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deploy_push)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.deploy.assert_called_once_with(uuid=None)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_status(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
uuid = "test-uuid"
|
||||
result = runner.invoke(deply_status, ["-u", uuid])
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.get_crew_status.assert_called_once_with(uuid=uuid)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_status_no_uuid(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deply_status)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.get_crew_status.assert_called_once_with(uuid=None)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_logs(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
uuid = "test-uuid"
|
||||
result = runner.invoke(deploy_logs, ["-u", uuid])
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.get_crew_logs.assert_called_once_with(uuid=uuid)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_logs_no_uuid(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deploy_logs)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.get_crew_logs.assert_called_once_with(uuid=None)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_remove(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
uuid = "test-uuid"
|
||||
result = runner.invoke(deploy_remove, ["-u", uuid])
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.remove_crew.assert_called_once_with(uuid=uuid)
|
||||
|
||||
|
||||
@mock.patch("crewai.cli.cli.DeployCommand")
|
||||
def test_deploy_remove_no_uuid(command, runner):
|
||||
mock_deploy = command.return_value
|
||||
result = runner.invoke(deploy_remove)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_deploy.remove_crew.assert_called_once_with(uuid=None)
|
||||
|
||||
102
tests/cli/deploy/test_api.py
Normal file
102
tests/cli/deploy/test_api.py
Normal file
@@ -0,0 +1,102 @@
|
||||
import unittest
|
||||
from os import environ
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from crewai.cli.deploy.api import CrewAPI
|
||||
|
||||
|
||||
class TestCrewAPI(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.api_key = "test_api_key"
|
||||
self.api = CrewAPI(self.api_key)
|
||||
|
||||
def test_init(self):
|
||||
self.assertEqual(self.api.api_key, self.api_key)
|
||||
self.assertEqual(
|
||||
self.api.headers,
|
||||
{
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
)
|
||||
|
||||
@patch("crewai.cli.deploy.api.requests.request")
|
||||
def test_make_request(self, mock_request):
|
||||
mock_response = MagicMock()
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
response = self.api._make_request("GET", "test_endpoint")
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"GET", f"{self.api.base_url}/test_endpoint", headers=self.api.headers
|
||||
)
|
||||
self.assertEqual(response, mock_response)
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_deploy_by_name(self, mock_make_request):
|
||||
self.api.deploy_by_name("test_project")
|
||||
mock_make_request.assert_called_once_with("POST", "by-name/test_project/deploy")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_deploy_by_uuid(self, mock_make_request):
|
||||
self.api.deploy_by_uuid("test_uuid")
|
||||
mock_make_request.assert_called_once_with("POST", "test_uuid/deploy")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_status_by_name(self, mock_make_request):
|
||||
self.api.status_by_name("test_project")
|
||||
mock_make_request.assert_called_once_with("GET", "by-name/test_project/status")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_status_by_uuid(self, mock_make_request):
|
||||
self.api.status_by_uuid("test_uuid")
|
||||
mock_make_request.assert_called_once_with("GET", "test_uuid/status")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_logs_by_name(self, mock_make_request):
|
||||
self.api.logs_by_name("test_project")
|
||||
mock_make_request.assert_called_once_with(
|
||||
"GET", "by-name/test_project/logs/deployment"
|
||||
)
|
||||
|
||||
self.api.logs_by_name("test_project", "custom_log")
|
||||
mock_make_request.assert_called_with(
|
||||
"GET", "by-name/test_project/logs/custom_log"
|
||||
)
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_logs_by_uuid(self, mock_make_request):
|
||||
self.api.logs_by_uuid("test_uuid")
|
||||
mock_make_request.assert_called_once_with("GET", "test_uuid/logs/deployment")
|
||||
|
||||
self.api.logs_by_uuid("test_uuid", "custom_log")
|
||||
mock_make_request.assert_called_with("GET", "test_uuid/logs/custom_log")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_delete_by_name(self, mock_make_request):
|
||||
self.api.delete_by_name("test_project")
|
||||
mock_make_request.assert_called_once_with("DELETE", "by-name/test_project")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_delete_by_uuid(self, mock_make_request):
|
||||
self.api.delete_by_uuid("test_uuid")
|
||||
mock_make_request.assert_called_once_with("DELETE", "test_uuid")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_list_crews(self, mock_make_request):
|
||||
self.api.list_crews()
|
||||
mock_make_request.assert_called_once_with("GET", "")
|
||||
|
||||
@patch("crewai.cli.deploy.api.CrewAPI._make_request")
|
||||
def test_create_crew(self, mock_make_request):
|
||||
payload = {"name": "test_crew"}
|
||||
self.api.create_crew(payload)
|
||||
mock_make_request.assert_called_once_with("POST", "", json=payload)
|
||||
|
||||
@patch.dict(environ, {"CREWAI_BASE_URL": "https://custom-url.com/api"})
|
||||
def test_custom_base_url(self):
|
||||
custom_api = CrewAPI("test_key")
|
||||
self.assertEqual(
|
||||
custom_api.base_url,
|
||||
"https://custom-url.com/api",
|
||||
)
|
||||
153
tests/cli/deploy/test_deploy_main.py
Normal file
153
tests/cli/deploy/test_deploy_main.py
Normal file
@@ -0,0 +1,153 @@
|
||||
import unittest
|
||||
from io import StringIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from crewai.cli.deploy.main import DeployCommand
|
||||
|
||||
|
||||
class TestDeployCommand(unittest.TestCase):
|
||||
@patch("crewai.cli.deploy.main.get_auth_token")
|
||||
@patch("crewai.cli.deploy.main.get_project_name")
|
||||
@patch("crewai.cli.deploy.main.CrewAPI")
|
||||
def setUp(self, mock_crew_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_crew_api = mock_crew_api
|
||||
|
||||
self.mock_get_auth_token.return_value = "test_token"
|
||||
self.mock_get_project_name.return_value = "test_project"
|
||||
|
||||
self.deploy_command = DeployCommand()
|
||||
self.mock_client = self.deploy_command.client
|
||||
|
||||
def test_init_success(self):
|
||||
self.assertEqual(self.deploy_command.project_name, "test_project")
|
||||
self.mock_crew_api.assert_called_once_with(api_key="test_token")
|
||||
|
||||
@patch("crewai.cli.deploy.main.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()
|
||||
|
||||
def test_handle_error(self):
|
||||
with patch("sys.stdout", new=StringIO()) as fake_out:
|
||||
self.deploy_command._handle_error(
|
||||
{"error": "Test error", "message": "Test message"}
|
||||
)
|
||||
self.assertIn("Error: Test error", fake_out.getvalue())
|
||||
self.assertIn("Message: Test message", fake_out.getvalue())
|
||||
|
||||
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", "status": "deployed"}
|
||||
)
|
||||
self.assertIn("Deploying the crew...", fake_out.getvalue())
|
||||
self.assertIn("test-uuid", fake_out.getvalue())
|
||||
self.assertIn("deployed", fake_out.getvalue())
|
||||
|
||||
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.DeployCommand._display_deployment_info")
|
||||
def test_deploy_with_uuid(self, mock_display):
|
||||
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")
|
||||
|
||||
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.DeployCommand._display_deployment_info")
|
||||
def test_deploy_with_project_name(self, mock_display):
|
||||
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()
|
||||
|
||||
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.fetch_and_json_env_file")
|
||||
@patch("crewai.cli.deploy.main.get_git_remote_url")
|
||||
@patch("builtins.input")
|
||||
def test_create_crew(self, mock_input, mock_get_git_remote_url, mock_fetch_env):
|
||||
mock_fetch_env.return_value = {"ENV_VAR": "value"}
|
||||
mock_get_git_remote_url.return_value = "https://github.com/test/repo.git"
|
||||
mock_input.return_value = ""
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 201
|
||||
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()
|
||||
self.assertIn("Deployment created successfully!", fake_out.getvalue())
|
||||
self.assertIn("new-uuid", fake_out.getvalue())
|
||||
|
||||
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": "TestCrew", "status": "active"}
|
||||
self.mock_client.status_by_name.return_value = mock_response
|
||||
|
||||
with patch("sys.stdout", new=StringIO()) as fake_out:
|
||||
self.deploy_command.get_crew_status()
|
||||
self.assertIn("TestCrew", 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.logs_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_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()
|
||||
)
|
||||
Reference in New Issue
Block a user