mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-05-01 07:13:00 +00:00
test: add CLI tests to crewai-cli package
Move and adapt all CLI tests from lib/crewai/tests/cli/ to lib/cli/tests/, updating import paths from crewai.cli.* to crewai_cli.* and adjusting mock targets accordingly.
This commit is contained in:
170
lib/cli/tests/triggers/test_main.py
Normal file
170
lib/cli/tests/triggers/test_main.py
Normal file
@@ -0,0 +1,170 @@
|
||||
import json
|
||||
import subprocess
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import httpx
|
||||
from crewai_cli.triggers.main import TriggersCommand
|
||||
|
||||
|
||||
class TestTriggersCommand(unittest.TestCase):
|
||||
@patch("crewai_cli.command.get_auth_token")
|
||||
@patch("crewai_cli.command.PlusAPI")
|
||||
def setUp(self, mock_plus_api, mock_get_auth_token):
|
||||
self.mock_get_auth_token = mock_get_auth_token
|
||||
self.mock_plus_api = mock_plus_api
|
||||
|
||||
self.mock_get_auth_token.return_value = "test_token"
|
||||
|
||||
self.triggers_command = TriggersCommand()
|
||||
self.mock_client = self.triggers_command.plus_api_client
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_list_triggers_success(self, mock_console_print):
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {
|
||||
"apps": [
|
||||
{
|
||||
"name": "Test App",
|
||||
"slug": "test-app",
|
||||
"description": "A test application",
|
||||
"is_connected": True,
|
||||
"triggers": [
|
||||
{
|
||||
"name": "Test Trigger",
|
||||
"slug": "test-trigger",
|
||||
"description": "A test trigger"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
self.mock_client.get_triggers.return_value = mock_response
|
||||
|
||||
self.triggers_command.list_triggers()
|
||||
|
||||
self.mock_client.get_triggers.assert_called_once()
|
||||
mock_console_print.assert_any_call("[bold blue]Fetching available triggers...[/bold blue]")
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_list_triggers_no_apps(self, mock_console_print):
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"apps": []}
|
||||
self.mock_client.get_triggers.return_value = mock_response
|
||||
|
||||
self.triggers_command.list_triggers()
|
||||
|
||||
mock_console_print.assert_any_call("[yellow]No triggers found.[/yellow]")
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_list_triggers_api_error(self, mock_console_print):
|
||||
self.mock_client.get_triggers.side_effect = Exception("API Error")
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command.list_triggers()
|
||||
|
||||
mock_console_print.assert_any_call("[bold red]Error fetching triggers: API Error[/bold red]")
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_execute_with_trigger_invalid_format(self, mock_console_print):
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command.execute_with_trigger("invalid-format")
|
||||
|
||||
mock_console_print.assert_called_with(
|
||||
"[bold red]Error: Trigger must be in format 'app_slug/trigger_slug'[/bold red]"
|
||||
)
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
@patch.object(TriggersCommand, "_run_crew_with_payload")
|
||||
def test_execute_with_trigger_success(self, mock_run_crew, mock_console_print):
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {
|
||||
"sample_payload": {"key": "value", "data": "test"}
|
||||
}
|
||||
self.mock_client.get_trigger_payload.return_value = mock_response
|
||||
|
||||
self.triggers_command.execute_with_trigger("test-app/test-trigger")
|
||||
|
||||
self.mock_client.get_trigger_payload.assert_called_once_with("test-app", "test-trigger")
|
||||
mock_run_crew.assert_called_once_with({"key": "value", "data": "test"})
|
||||
mock_console_print.assert_any_call(
|
||||
"[bold blue]Fetching trigger payload for test-app/test-trigger...[/bold blue]"
|
||||
)
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_execute_with_trigger_not_found(self, mock_console_print):
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 404
|
||||
mock_response.json.return_value = {"error": "Trigger not found"}
|
||||
self.mock_client.get_trigger_payload.return_value = mock_response
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command.execute_with_trigger("test-app/nonexistent-trigger")
|
||||
|
||||
mock_console_print.assert_any_call("[bold red]Error: Trigger not found[/bold red]")
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_execute_with_trigger_api_error(self, mock_console_print):
|
||||
self.mock_client.get_trigger_payload.side_effect = Exception("API Error")
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command.execute_with_trigger("test-app/test-trigger")
|
||||
|
||||
mock_console_print.assert_any_call(
|
||||
"[bold red]Error executing crew with trigger: API Error[/bold red]"
|
||||
)
|
||||
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_run_crew_with_payload_success(self, mock_subprocess):
|
||||
payload = {"key": "value", "data": "test"}
|
||||
mock_subprocess.return_value = None
|
||||
|
||||
self.triggers_command._run_crew_with_payload(payload)
|
||||
|
||||
mock_subprocess.assert_called_once_with(
|
||||
["uv", "run", "run_with_trigger", json.dumps(payload)],
|
||||
capture_output=False,
|
||||
text=True,
|
||||
check=True
|
||||
)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_run_crew_with_payload_failure(self, mock_subprocess):
|
||||
payload = {"key": "value"}
|
||||
mock_subprocess.side_effect = subprocess.CalledProcessError(1, "uv")
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command._run_crew_with_payload(payload)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_run_crew_with_payload_empty_payload(self, mock_subprocess):
|
||||
payload = {}
|
||||
mock_subprocess.return_value = None
|
||||
|
||||
self.triggers_command._run_crew_with_payload(payload)
|
||||
|
||||
mock_subprocess.assert_called_once_with(
|
||||
["uv", "run", "run_with_trigger", "{}"],
|
||||
capture_output=False,
|
||||
text=True,
|
||||
check=True
|
||||
)
|
||||
|
||||
@patch("crewai_cli.triggers.main.console.print")
|
||||
def test_execute_with_trigger_with_default_error_message(self, mock_console_print):
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 404
|
||||
mock_response.json.return_value = {}
|
||||
self.mock_client.get_trigger_payload.return_value = mock_response
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
self.triggers_command.execute_with_trigger("test-app/test-trigger")
|
||||
|
||||
mock_console_print.assert_any_call("[bold red]Error: Trigger not found[/bold red]")
|
||||
Reference in New Issue
Block a user