forked from cleveragents/cleveragents-core
834 lines
30 KiB
Python
834 lines
30 KiB
Python
"""
|
|
Unit tests for cli.py module
|
|
|
|
Tests the command-line interface for CleverAgents.
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
from click.testing import CliRunner
|
|
import tempfile
|
|
|
|
from cleveragents.cli import (
|
|
main,
|
|
run,
|
|
interactive,
|
|
generate_examples,
|
|
visualize,
|
|
_normalize_targets,
|
|
_generate_mermaid_diagram,
|
|
_generate_dot_diagram,
|
|
_generate_ascii_diagram,
|
|
_validate_config_files,
|
|
)
|
|
from cleveragents.core.exceptions import CleverAgentsException, UnsafeConfigurationError
|
|
from cleveragents.reactive.config_parser import ReactiveConfig
|
|
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
"""Fixture for CLI runner."""
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_config_file():
|
|
"""Fixture for a temporary config file."""
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', suffix='.yaml', delete=False) as f:
|
|
f.write("""
|
|
agents:
|
|
test_agent:
|
|
type: llm
|
|
config:
|
|
provider: openai
|
|
model: gpt-4
|
|
|
|
streams:
|
|
test_stream:
|
|
type: cold
|
|
operators:
|
|
- type: map
|
|
|
|
merges:
|
|
- sources: [__input__]
|
|
target: test_stream
|
|
|
|
splits:
|
|
- source: test_stream
|
|
targets:
|
|
output: {}
|
|
""")
|
|
temp_path = Path(f.name)
|
|
yield temp_path
|
|
temp_path.unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_config_file():
|
|
"""Fixture for an empty config file."""
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', suffix='.yaml', delete=False) as f:
|
|
temp_path = Path(f.name)
|
|
yield temp_path
|
|
temp_path.unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_reactive_config():
|
|
"""Fixture for a mock ReactiveConfig."""
|
|
config = Mock(spec=ReactiveConfig)
|
|
config.agents = {
|
|
'agent1': Mock(type='llm'),
|
|
'agent2': Mock(type='tool'),
|
|
}
|
|
|
|
# Mock routes
|
|
route1 = Mock()
|
|
route1.type.value = 'stream'
|
|
route1.stream_type.value = 'cold'
|
|
route1.agents = ['agent1']
|
|
route1.operators = [Mock()]
|
|
|
|
route2 = Mock()
|
|
route2.type.value = 'graph'
|
|
route2.nodes = [Mock(), Mock()]
|
|
route2.edges = [Mock()]
|
|
|
|
config.routes = {
|
|
'stream1': route1,
|
|
'graph1': route2,
|
|
}
|
|
|
|
config.merges = [
|
|
{'sources': ['__input__'], 'target': 'stream1'}
|
|
]
|
|
|
|
config.splits = [
|
|
{'source': 'stream1', 'targets': {'output1': {}, 'output2': {}}}
|
|
]
|
|
|
|
return config
|
|
|
|
|
|
class TestMainCommand:
|
|
"""Test suite for the main click group."""
|
|
|
|
def test_main_command_exists(self, runner):
|
|
"""Test that main command can be invoked."""
|
|
result = runner.invoke(main, ['--help'])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'Reactive CleverAgents' in result.output
|
|
|
|
def test_main_version_option(self, runner):
|
|
"""Test the version option."""
|
|
result = runner.invoke(main, ['--version'])
|
|
|
|
# Version should be displayed
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestRunCommand:
|
|
"""Test suite for the run command."""
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
@patch('cleveragents.cli.asyncio.run')
|
|
def test_run_command_basic(self, mock_asyncio_run, mock_app_class, runner, temp_config_file):
|
|
"""Test basic run command execution."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
mock_asyncio_run.return_value = "Test output"
|
|
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test prompt'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'Test output' in result.output
|
|
mock_app_class.assert_called_once()
|
|
mock_asyncio_run.assert_called_once()
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
@patch('cleveragents.cli.asyncio.run')
|
|
def test_run_command_with_output_file(self, mock_asyncio_run, mock_app_class, runner, temp_config_file):
|
|
"""Test run command with output file."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
mock_asyncio_run.return_value = "Test result"
|
|
|
|
# Generate path to a non-existent file to avoid overwrite confirmation prompt
|
|
output_path = Path(tempfile.gettempdir()) / f'cleveragent_test_{os.getpid()}_{id(self)}.txt'
|
|
|
|
try:
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test',
|
|
'--output', str(output_path)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert f'Output written to {output_path}' in result.output
|
|
|
|
# Verify file was written
|
|
with open(output_path, 'r') as f:
|
|
assert f.read() == "Test result"
|
|
finally:
|
|
output_path.unlink(missing_ok=True)
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_with_unsafe_flag(self, mock_app_class, runner, temp_config_file):
|
|
"""Test run command with unsafe flag."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
|
|
with patch('cleveragents.cli.asyncio.run', return_value="output"):
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test',
|
|
'--unsafe'
|
|
])
|
|
|
|
# Check that unsafe=True was passed
|
|
call_args = mock_app_class.call_args
|
|
assert call_args[0][2] is True # unsafe parameter
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_with_verbose_flag(self, mock_app_class, runner, temp_config_file):
|
|
"""Test run command with verbose flag."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
|
|
with patch('cleveragents.cli.asyncio.run', return_value="output"):
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test',
|
|
'--verbose'
|
|
])
|
|
|
|
# Check that verbose=True was passed
|
|
call_args = mock_app_class.call_args
|
|
assert call_args[0][1] is True # verbose parameter
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_unsafe_configuration_error(self, mock_app_class, runner, temp_config_file):
|
|
"""Test run command handles UnsafeConfigurationError."""
|
|
mock_app_class.side_effect = UnsafeConfigurationError("Unsafe config detected")
|
|
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Unsafe config detected' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_cleveragents_exception(self, mock_app_class, runner, temp_config_file):
|
|
"""Test run command handles CleverAgentsException."""
|
|
mock_app_class.side_effect = CleverAgentsException("Test error")
|
|
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Test error' in result.output
|
|
|
|
def test_run_command_file_not_found_error(self, runner):
|
|
"""Test run command handles FileNotFoundError."""
|
|
result = runner.invoke(run, [
|
|
'--config', '/nonexistent/config.yaml',
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
# Click should catch the non-existent file before our handler
|
|
assert result.exit_code != 0
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_generic_exception(self, mock_app_class, runner, temp_config_file):
|
|
"""Test run command handles generic exceptions."""
|
|
mock_app_class.side_effect = RuntimeError("Generic error")
|
|
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Generic error' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
@patch('cleveragents.cli.asyncio.run')
|
|
def test_run_command_multiple_config_files(self, mock_asyncio_run, mock_app_class, runner, temp_config_file):
|
|
"""Test run command with multiple config files."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
mock_asyncio_run.return_value = "output"
|
|
|
|
# Create second config file
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', suffix='.yaml', delete=False) as f:
|
|
f.write("agents: {}")
|
|
temp_config2 = Path(f.name)
|
|
|
|
try:
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--config', str(temp_config2),
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
# Verify multiple configs were passed
|
|
call_args = mock_app_class.call_args
|
|
assert len(call_args[0][0]) == 2
|
|
finally:
|
|
temp_config2.unlink(missing_ok=True)
|
|
|
|
@patch('cleveragents.cli._validate_config_files')
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_run_command_file_not_found_after_validation(self, mock_app_class, mock_validate, runner, temp_config_file):
|
|
"""Test run command handles FileNotFoundError after validation."""
|
|
# Validation passes but app initialization raises FileNotFoundError
|
|
mock_validate.return_value = None
|
|
mock_app_class.side_effect = FileNotFoundError("Config disappeared")
|
|
|
|
result = runner.invoke(run, [
|
|
'--config', str(temp_config_file),
|
|
'--prompt', 'test'
|
|
])
|
|
|
|
assert result.exit_code == 2
|
|
assert 'Error: Configuration file not found' in result.output
|
|
|
|
|
|
class TestInteractiveCommand:
|
|
"""Test suite for the interactive command."""
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
@patch('cleveragents.cli.asyncio.run')
|
|
def test_interactive_command_basic(self, mock_asyncio_run, mock_app_class, runner, temp_config_file):
|
|
"""Test basic interactive command execution."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
mock_asyncio_run.return_value = None
|
|
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
mock_app_class.assert_called_once()
|
|
mock_asyncio_run.assert_called_once()
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
@patch('cleveragents.cli.asyncio.run')
|
|
def test_interactive_command_with_history(self, mock_asyncio_run, mock_app_class, runner, temp_config_file):
|
|
"""Test interactive command with history file."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
mock_asyncio_run.return_value = None
|
|
|
|
history_file = Path('/tmp/test_history.txt')
|
|
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file),
|
|
'--history', str(history_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
# Verify asyncio.run was called with app.start_interactive_session
|
|
mock_asyncio_run.assert_called_once()
|
|
# The call should be asyncio.run(app.start_interactive_session(history_file=history))
|
|
call_args = mock_asyncio_run.call_args[0]
|
|
# Just verify it was called, the coroutine object doesn't preserve kwargs easily
|
|
assert len(call_args) > 0
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_interactive_command_with_verbose(self, mock_app_class, runner, temp_config_file):
|
|
"""Test interactive command with verbose flag."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
|
|
with patch('cleveragents.cli.asyncio.run'):
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file),
|
|
'--verbose'
|
|
])
|
|
|
|
# Check verbose=True was passed
|
|
call_args = mock_app_class.call_args
|
|
assert call_args[0][1] is True
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_interactive_command_with_unsafe(self, mock_app_class, runner, temp_config_file):
|
|
"""Test interactive command with unsafe flag."""
|
|
mock_app = Mock()
|
|
mock_app_class.return_value = mock_app
|
|
|
|
with patch('cleveragents.cli.asyncio.run'):
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file),
|
|
'--unsafe'
|
|
])
|
|
|
|
# Check unsafe=True was passed
|
|
call_args = mock_app_class.call_args
|
|
assert call_args[0][2] is True
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_interactive_command_unsafe_error(self, mock_app_class, runner, temp_config_file):
|
|
"""Test interactive command handles UnsafeConfigurationError."""
|
|
mock_app_class.side_effect = UnsafeConfigurationError("Unsafe config")
|
|
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file)
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Unsafe config' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_interactive_command_cleveragents_error(self, mock_app_class, runner, temp_config_file):
|
|
"""Test interactive command handles CleverAgentsException."""
|
|
mock_app_class.side_effect = CleverAgentsException("Test error")
|
|
|
|
result = runner.invoke(interactive, [
|
|
'--config', str(temp_config_file)
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Test error' in result.output
|
|
|
|
|
|
class TestGenerateExamplesCommand:
|
|
"""Test suite for the generate_examples command."""
|
|
|
|
def test_generate_examples_default_output(self, runner):
|
|
"""Test generating examples with default output directory."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_path = Path(temp_dir) / 'examples'
|
|
|
|
result = runner.invoke(generate_examples, [
|
|
'--output', str(output_path)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'Generated reactive example configurations' in result.output
|
|
assert output_path.exists()
|
|
|
|
# Check that all three files were created
|
|
assert (output_path / 'basic_reactive.yaml').exists()
|
|
assert (output_path / 'advanced_reactive.yaml').exists()
|
|
assert (output_path / 'collaboration_reactive.yaml').exists()
|
|
|
|
def test_generate_examples_custom_output(self, runner):
|
|
"""Test generating examples with custom output directory."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
custom_path = Path(temp_dir) / 'my_examples'
|
|
|
|
result = runner.invoke(generate_examples, [
|
|
'--output', str(custom_path)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert custom_path.exists()
|
|
|
|
def test_generate_examples_creates_directory(self, runner):
|
|
"""Test that generate_examples creates the directory if it doesn't exist."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
nested_path = Path(temp_dir) / 'nested' / 'path' / 'examples'
|
|
|
|
result = runner.invoke(generate_examples, [
|
|
'--output', str(nested_path)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert nested_path.exists()
|
|
|
|
def test_generate_examples_basic_config_content(self, runner):
|
|
"""Test that basic config has expected content."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_path = Path(temp_dir)
|
|
|
|
runner.invoke(generate_examples, ['--output', str(output_path)])
|
|
|
|
basic_file = output_path / 'basic_reactive.yaml'
|
|
with open(basic_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
assert 'agents:' in content
|
|
assert 'streams:' in content
|
|
assert 'llm_agent' in content
|
|
assert 'input_processor' in content
|
|
|
|
def test_generate_examples_advanced_config_content(self, runner):
|
|
"""Test that advanced config has expected content."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_path = Path(temp_dir)
|
|
|
|
runner.invoke(generate_examples, ['--output', str(output_path)])
|
|
|
|
advanced_file = output_path / 'advanced_reactive.yaml'
|
|
with open(advanced_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
assert 'classifier' in content
|
|
assert 'question_handler' in content
|
|
assert 'command_handler' in content
|
|
|
|
def test_generate_examples_collaboration_config_content(self, runner):
|
|
"""Test that collaboration config has expected content."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_path = Path(temp_dir)
|
|
|
|
runner.invoke(generate_examples, ['--output', str(output_path)])
|
|
|
|
collab_file = output_path / 'collaboration_reactive.yaml'
|
|
with open(collab_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
assert 'researcher' in content
|
|
assert 'analyzer' in content
|
|
assert 'writer' in content
|
|
assert 'editor' in content
|
|
|
|
|
|
class TestVisualizeCommand:
|
|
"""Test suite for the visualize command."""
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_mermaid_format(self, mock_app_class, runner, temp_config_file, mock_reactive_config):
|
|
"""Test visualize command with mermaid format."""
|
|
mock_app = Mock()
|
|
mock_app.config = mock_reactive_config
|
|
mock_app_class.return_value = mock_app
|
|
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file),
|
|
'--format', 'mermaid'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'graph TD' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_dot_format(self, mock_app_class, runner, temp_config_file, mock_reactive_config):
|
|
"""Test visualize command with dot format."""
|
|
mock_app = Mock()
|
|
mock_app.config = mock_reactive_config
|
|
mock_app_class.return_value = mock_app
|
|
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file),
|
|
'--format', 'dot'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'digraph StreamNetwork' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_ascii_format(self, mock_app_class, runner, temp_config_file, mock_reactive_config):
|
|
"""Test visualize command with ascii format."""
|
|
mock_app = Mock()
|
|
mock_app.config = mock_reactive_config
|
|
mock_app_class.return_value = mock_app
|
|
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file),
|
|
'--format', 'ascii'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert 'Reactive Stream Network' in result.output
|
|
assert 'Agents:' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_with_output_file(self, mock_app_class, runner, temp_config_file, mock_reactive_config):
|
|
"""Test visualize command with output file."""
|
|
mock_app = Mock()
|
|
mock_app.config = mock_reactive_config
|
|
mock_app_class.return_value = mock_app
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', delete=False) as output_file:
|
|
output_path = Path(output_file.name)
|
|
|
|
try:
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file),
|
|
'--format', 'mermaid',
|
|
'--output', str(output_path)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert f'Stream diagram written to {output_path}' in result.output
|
|
|
|
# Verify file was written
|
|
with open(output_path, 'r') as f:
|
|
content = f.read()
|
|
assert 'graph TD' in content
|
|
finally:
|
|
output_path.unlink(missing_ok=True)
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_no_config(self, mock_app_class, runner, temp_config_file):
|
|
"""Test visualize command when no config is loaded."""
|
|
mock_app = Mock()
|
|
mock_app.config = None
|
|
mock_app_class.return_value = mock_app
|
|
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file)
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: No configuration loaded' in result.output
|
|
|
|
@patch('cleveragents.cli.ReactiveCleverAgentsApp')
|
|
def test_visualize_command_error_handling(self, mock_app_class, runner, temp_config_file):
|
|
"""Test visualize command error handling."""
|
|
mock_app_class.side_effect = CleverAgentsException("Visualization error")
|
|
|
|
result = runner.invoke(visualize, [
|
|
'--config', str(temp_config_file)
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert 'Error: Visualization error' in result.output
|
|
|
|
|
|
class TestHelperFunctions:
|
|
"""Test suite for helper functions."""
|
|
|
|
def test_normalize_targets_with_dict(self):
|
|
"""Test normalizing targets from dict format."""
|
|
targets_data = {'target1': {}, 'target2': {'condition': 'test'}}
|
|
result = _normalize_targets(targets_data)
|
|
|
|
assert result == ['target1', 'target2']
|
|
|
|
def test_normalize_targets_with_string(self):
|
|
"""Test normalizing targets from string format."""
|
|
targets_data = 'single_target'
|
|
result = _normalize_targets(targets_data)
|
|
|
|
assert result == ['single_target']
|
|
|
|
def test_normalize_targets_with_list(self):
|
|
"""Test normalizing targets from list format."""
|
|
targets_data = ['target1', 'target2', 'target3']
|
|
result = _normalize_targets(targets_data)
|
|
|
|
assert result == ['target1', 'target2', 'target3']
|
|
|
|
def test_normalize_targets_with_invalid_type(self):
|
|
"""Test normalizing targets with invalid type returns empty list."""
|
|
targets_data = 12345
|
|
result = _normalize_targets(targets_data)
|
|
|
|
assert result == []
|
|
|
|
def test_normalize_targets_with_none(self):
|
|
"""Test normalizing targets with None returns empty list."""
|
|
targets_data = None
|
|
result = _normalize_targets(targets_data)
|
|
|
|
assert result == []
|
|
|
|
def test_generate_mermaid_diagram(self, mock_reactive_config):
|
|
"""Test generating mermaid diagram."""
|
|
diagram = _generate_mermaid_diagram(mock_reactive_config)
|
|
|
|
assert 'graph TD' in diagram
|
|
assert 'agent1' in diagram
|
|
assert 'agent2' in diagram
|
|
assert 'stream1' in diagram
|
|
assert 'graph1' in diagram
|
|
|
|
def test_generate_mermaid_diagram_with_hot_stream(self):
|
|
"""Test generating mermaid diagram with hot stream."""
|
|
config = Mock()
|
|
config.agents = {}
|
|
|
|
route = Mock()
|
|
route.type.value = 'stream'
|
|
route.stream_type.value = 'hot'
|
|
route.agents = []
|
|
|
|
config.routes = {'hot_stream': route}
|
|
config.merges = []
|
|
config.splits = []
|
|
|
|
diagram = _generate_mermaid_diagram(config)
|
|
|
|
assert 'hot_stream(hot_stream)' in diagram
|
|
|
|
def test_generate_dot_diagram(self, mock_reactive_config):
|
|
"""Test generating dot diagram."""
|
|
diagram = _generate_dot_diagram(mock_reactive_config)
|
|
|
|
assert 'digraph StreamNetwork' in diagram
|
|
assert 'agent1' in diagram
|
|
assert 'stream1' in diagram
|
|
assert 'shape=box' in diagram or 'shape=ellipse' in diagram
|
|
|
|
def test_generate_dot_diagram_with_graph_route(self, mock_reactive_config):
|
|
"""Test generating dot diagram with graph route."""
|
|
diagram = _generate_dot_diagram(mock_reactive_config)
|
|
|
|
assert 'graph1' in diagram
|
|
assert 'shape=hexagon' in diagram
|
|
|
|
def test_generate_ascii_diagram(self, mock_reactive_config):
|
|
"""Test generating ASCII diagram."""
|
|
diagram = _generate_ascii_diagram(mock_reactive_config)
|
|
|
|
assert 'Reactive Stream Network' in diagram
|
|
assert 'Agents:' in diagram
|
|
assert 'Routes:' in diagram
|
|
assert 'agent1' in diagram
|
|
assert 'stream1' in diagram
|
|
|
|
def test_generate_ascii_diagram_with_merges_and_splits(self, mock_reactive_config):
|
|
"""Test ASCII diagram includes merges and splits."""
|
|
diagram = _generate_ascii_diagram(mock_reactive_config)
|
|
|
|
assert 'Merges:' in diagram
|
|
assert 'Splits:' in diagram
|
|
|
|
def test_validate_config_files_valid(self, temp_config_file):
|
|
"""Test validating a valid config file."""
|
|
# Should not raise
|
|
_validate_config_files([temp_config_file])
|
|
|
|
def test_validate_config_files_no_files(self):
|
|
"""Test validating with no config files raises error."""
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([])
|
|
|
|
assert 'No configuration files provided' in str(exc_info.value)
|
|
|
|
def test_validate_config_files_nonexistent(self):
|
|
"""Test validating nonexistent file raises error."""
|
|
nonexistent = Path('/nonexistent/file.yaml')
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
_validate_config_files([nonexistent])
|
|
|
|
def test_validate_config_files_empty_file(self, empty_config_file):
|
|
"""Test validating empty file raises error."""
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([empty_config_file])
|
|
|
|
assert 'is empty' in str(exc_info.value)
|
|
|
|
def test_validate_config_files_whitespace_only(self):
|
|
"""Test validating file with only whitespace raises error."""
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', suffix='.yaml', delete=False) as f:
|
|
f.write(' \n\n \t\t \n')
|
|
whitespace_file = Path(f.name)
|
|
|
|
try:
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([whitespace_file])
|
|
|
|
assert 'empty or contains only whitespace' in str(exc_info.value)
|
|
finally:
|
|
whitespace_file.unlink(missing_ok=True)
|
|
|
|
def test_validate_config_files_dev_null(self):
|
|
"""Test validating /dev/null raises error."""
|
|
if Path('/dev/null').exists():
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([Path('/dev/null')])
|
|
|
|
# /dev/null appears as empty file, so error message mentions "empty"
|
|
assert 'empty' in str(exc_info.value).lower() or 'not a valid configuration file' in str(exc_info.value)
|
|
|
|
def test_validate_config_files_multiple_valid(self, temp_config_file):
|
|
"""Test validating multiple valid files."""
|
|
with tempfile.NamedTemporaryFile(mode='w', prefix='cleveragent_', suffix='.yaml', delete=False) as f:
|
|
f.write('agents: {}')
|
|
temp_config2 = Path(f.name)
|
|
|
|
try:
|
|
# Should not raise
|
|
_validate_config_files([temp_config_file, temp_config2])
|
|
finally:
|
|
temp_config2.unlink(missing_ok=True)
|
|
|
|
def test_validate_config_files_dev_zero(self):
|
|
"""Test validating /dev/zero raises error."""
|
|
if Path('/dev/zero').exists():
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([Path('/dev/zero')])
|
|
|
|
# /dev/zero might be caught as empty file first
|
|
assert 'empty' in str(exc_info.value).lower() or 'not a valid configuration file' in str(exc_info.value)
|
|
|
|
def test_validate_config_files_nul_name(self):
|
|
"""Test validating file named 'null' or 'NUL' raises error."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
null_file = Path(temp_dir) / 'null'
|
|
with open(null_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([null_file])
|
|
|
|
assert 'not a valid configuration file' in str(exc_info.value)
|
|
|
|
@patch('builtins.open')
|
|
def test_validate_config_files_os_error(self, mock_open, temp_config_file):
|
|
"""Test validating file that raises OSError."""
|
|
mock_open.side_effect = OSError("Permission denied")
|
|
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([temp_config_file])
|
|
|
|
assert 'Cannot read configuration file' in str(exc_info.value)
|
|
|
|
@patch('builtins.open')
|
|
def test_validate_config_files_io_error(self, mock_open, temp_config_file):
|
|
"""Test validating file that raises IOError."""
|
|
mock_open.side_effect = IOError("I/O error")
|
|
|
|
with pytest.raises(CleverAgentsException) as exc_info:
|
|
_validate_config_files([temp_config_file])
|
|
|
|
assert 'Cannot read configuration file' in str(exc_info.value)
|
|
|
|
|
|
class TestMainModuleExecution:
|
|
"""Test suite for module execution."""
|
|
|
|
def test_main_module_can_be_executed(self):
|
|
"""Test that the module's main function can be called."""
|
|
# Import the module to ensure it's loadable
|
|
import cleveragents.cli
|
|
|
|
assert hasattr(cleveragents.cli, 'main')
|
|
assert callable(cleveragents.cli.main)
|
|
|
|
def test_cli_module_name_attribute(self):
|
|
"""Test that CLI module has expected attributes."""
|
|
import cleveragents.cli as cli_module
|
|
|
|
assert cli_module.__name__ == 'cleveragents.cli'
|
|
|
|
def test_all_commands_registered(self, runner):
|
|
"""Test that all commands are registered with the main group."""
|
|
result = runner.invoke(main, ['--help'])
|
|
|
|
assert 'run' in result.output
|
|
assert 'interactive' in result.output
|
|
assert 'generate-examples' in result.output
|
|
assert 'visualize' in result.output
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|
|
|