Your Name
264 lines
9.6 KiB
Python
264 lines
9.6 KiB
Python
"""Step definitions for CleverClaude CLI features."""
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from click.testing import CliRunner
|
|
from hypothesis import given as hypothesis_given
|
|
from hypothesis import strategies as st
|
|
|
|
from cleverclaude.cli.main import app
|
|
|
|
|
|
@given("the CleverClaude CLI is available")
|
|
def step_cli_available(_context):
|
|
"""Ensure CleverClaude CLI is importable."""
|
|
_context.runner = CliRunner()
|
|
assert _context.runner is not None
|
|
|
|
|
|
@given("I have a test environment")
|
|
def step_test_environment(_context):
|
|
"""Set up test environment."""
|
|
# Use the test context from environment.py
|
|
assert hasattr(_context, "test_context")
|
|
|
|
|
|
@given("I have an empty directory")
|
|
def step_empty_directory(_context):
|
|
"""Create an empty test directory."""
|
|
_context.test_dir = Path(tempfile.mkdtemp(prefix="cleverclaude_cli_test_"))
|
|
os.chdir(_context.test_dir)
|
|
|
|
|
|
@given("I have a target directory {dirname}")
|
|
def step_target_directory(_context, dirname):
|
|
"""Create a target directory."""
|
|
_context.test_dir = Path(tempfile.mkdtemp(prefix="cleverclaude_cli_test_"))
|
|
_context.target_dir = _context.test_dir / dirname
|
|
os.chdir(_context.test_dir)
|
|
|
|
|
|
@given("I have a directory with existing files")
|
|
def step_directory_with_files(_context):
|
|
"""Create a directory with existing files."""
|
|
_context.test_dir = Path(tempfile.mkdtemp(prefix="cleverclaude_cli_test_"))
|
|
os.chdir(_context.test_dir)
|
|
|
|
# Create some existing files
|
|
(_context.test_dir / "existing_file.txt").write_text("This file already exists")
|
|
(_context.test_dir / "README.md").write_text("# Existing Project")
|
|
|
|
|
|
@given("I have an initialized CleverClaude project")
|
|
def step_initialized_project(_context):
|
|
"""Create an initialized CleverClaude project."""
|
|
_context.test_dir = Path(tempfile.mkdtemp(prefix="cleverclaude_cli_test_"))
|
|
os.chdir(_context.test_dir)
|
|
|
|
# Run init command to set up project
|
|
result = _context.runner.invoke(app, ["init"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
@given("CleverClaude is running")
|
|
def step_cleverclaude_running(_context):
|
|
"""Ensure CleverClaude system is running for testing."""
|
|
# This would start a test instance of CleverClaude
|
|
# For now, we'll mock this
|
|
_context.cleverclaude_running = True
|
|
|
|
|
|
@when('I run "{command}"')
|
|
def step_run_command(context, command):
|
|
"""Execute a CLI command."""
|
|
parts = command.split()
|
|
|
|
# Handle different command formats
|
|
if parts[0] == "cleverclaude":
|
|
args = parts[1:] # Remove "cleverclaude"
|
|
context.result = context.runner.invoke(app, args)
|
|
elif len(parts) >= 3 and parts[0] == "python" and parts[1] == "-m" and parts[2] == "cleverclaude":
|
|
args = parts[3:] # Remove "python -m cleverclaude"
|
|
context.result = context.runner.invoke(app, args)
|
|
else:
|
|
# Direct subprocess call for integration testing
|
|
try:
|
|
result = subprocess.run(
|
|
parts, capture_output=True, text=True, timeout=30, cwd=getattr(context, "test_dir", None)
|
|
)
|
|
|
|
# Create a mock result object
|
|
class MockResult:
|
|
def __init__(self, returncode, stdout, stderr):
|
|
self.exit_code = returncode
|
|
self.output = stdout + stderr
|
|
|
|
context.result = MockResult(result.returncode, result.stdout, result.stderr)
|
|
except subprocess.TimeoutExpired:
|
|
|
|
class MockResult:
|
|
def __init__(self):
|
|
self.exit_code = 124 # Timeout exit code
|
|
self.output = "Command timed out"
|
|
|
|
context.result = MockResult()
|
|
|
|
|
|
@when("I start the orchestration system in test mode")
|
|
def step_start_orchestration(_context):
|
|
"""Start CleverClaude orchestration in test mode."""
|
|
# This would involve starting the system asynchronously
|
|
# For testing, we'll simulate this
|
|
_context.orchestration_started = True
|
|
_context.result = type("MockResult", (), {"exit_code": 0, "output": "System started successfully"})()
|
|
|
|
|
|
@then("the exit code should be {code:d}")
|
|
def step_check_exit_code(context, code):
|
|
"""Verify exit code."""
|
|
assert context.result.exit_code == code, f"Expected exit code {code}, got {context.result.exit_code}"
|
|
|
|
|
|
@then('the output should contain "{text}"')
|
|
def step_output_contains(context, text):
|
|
"""Check if output contains text."""
|
|
assert text in context.result.output, f"Output does not contain '{text}'. Output was: {context.result.output}"
|
|
|
|
|
|
@then('the directory "{dirname}" should exist')
|
|
def step_directory_exists(context, dirname):
|
|
"""Check if directory exists."""
|
|
test_dir = getattr(context, "test_dir", Path.cwd())
|
|
dir_path = test_dir / dirname
|
|
assert dir_path.exists() and dir_path.is_dir(), f"Directory '{dirname}' does not exist at {test_dir}"
|
|
|
|
|
|
@then('the file "{filename}" should exist')
|
|
def step_file_exists(context, filename):
|
|
"""Check if file exists."""
|
|
test_dir = getattr(context, "test_dir", Path.cwd())
|
|
file_path = test_dir / filename
|
|
assert file_path.exists() and file_path.is_file(), f"File '{filename}' does not exist at {test_dir}"
|
|
|
|
|
|
@then('the file "{filename}" should contain "{text}"')
|
|
def step_file_contains(context, filename, text):
|
|
"""Check if file contains specific text."""
|
|
test_dir = getattr(context, "test_dir", Path.cwd())
|
|
file_path = test_dir / filename
|
|
assert file_path.exists(), f"File '{filename}' does not exist"
|
|
|
|
content = file_path.read_text()
|
|
assert text in content, f"File '{filename}' does not contain '{text}'"
|
|
|
|
|
|
@then("the system should initialize successfully")
|
|
def step_system_initializes(_context):
|
|
"""Verify system initialization."""
|
|
assert getattr(_context, "orchestration_started", False), "System did not start"
|
|
|
|
|
|
@then("the agent manager should be running")
|
|
def step_agent_manager_running(_context):
|
|
"""Verify agent manager is running."""
|
|
# This would check if the agent manager is actually running
|
|
# For testing, we'll assume success if orchestration started
|
|
assert getattr(_context, "orchestration_started", False), "Agent manager not running"
|
|
|
|
|
|
@then("the API server should be accessible")
|
|
def step_api_server_accessible(_context):
|
|
"""Verify API server is accessible."""
|
|
# This would check if the API server is responding
|
|
# For testing, we'll simulate this
|
|
assert getattr(_context, "orchestration_started", False), "API server not accessible"
|
|
|
|
|
|
@then("the output should contain system health information")
|
|
def step_output_contains_health_info(context):
|
|
"""Check if output contains system health information."""
|
|
health_indicators = ["status", "health", "running", "active"]
|
|
output_lower = context.result.output.lower()
|
|
assert any(indicator in output_lower for indicator in health_indicators), "No health information found in output"
|
|
|
|
|
|
@then("the output should contain agent count")
|
|
def step_output_contains_agent_count(context):
|
|
"""Check if output contains agent count information."""
|
|
output_lower = context.result.output.lower()
|
|
agent_indicators = ["agent", "count", "total", "active"]
|
|
assert any(indicator in output_lower for indicator in agent_indicators), (
|
|
"No agent count information found in output"
|
|
)
|
|
|
|
|
|
@then("the output should contain memory usage")
|
|
def step_output_contains_memory_usage(context):
|
|
"""Check if output contains memory usage information."""
|
|
output_lower = context.result.output.lower()
|
|
memory_indicators = ["memory", "usage", "ram", "heap"]
|
|
assert any(indicator in output_lower for indicator in memory_indicators), (
|
|
"No memory usage information found in output"
|
|
)
|
|
|
|
|
|
@then("the output should contain command-specific help")
|
|
def step_output_contains_help(context):
|
|
"""Check if output contains command-specific help."""
|
|
help_indicators = ["help", "usage", "options", "commands"]
|
|
output_lower = context.result.output.lower()
|
|
assert any(indicator in output_lower for indicator in help_indicators), "No help information found in output"
|
|
|
|
|
|
@when("I fuzz test the CLI with random invalid arguments")
|
|
def step_fuzz_cli_invalid(context):
|
|
"""Fuzz test the CLI with random invalid arguments."""
|
|
runner = context.runner
|
|
results = []
|
|
|
|
@hypothesis_given(
|
|
st.lists(
|
|
st.one_of(
|
|
st.text(min_size=1, max_size=50), st.integers(), st.floats(allow_nan=False, allow_infinity=False)
|
|
),
|
|
min_size=1,
|
|
max_size=10,
|
|
)
|
|
)
|
|
def test_random_invalid_args(args):
|
|
# Convert all args to strings
|
|
str_args = [str(arg) for arg in args]
|
|
|
|
try:
|
|
result = runner.invoke(app, str_args)
|
|
results.append(result)
|
|
|
|
# Should either succeed (exit code 0) or fail gracefully (non-zero but not crash)
|
|
assert result.exit_code in [0, 1, 2], f"Unexpected exit code: {result.exit_code}"
|
|
|
|
except Exception as e:
|
|
# Should not raise unhandled exceptions
|
|
raise AssertionError(f"CLI crashed with unhandled exception: {e}") from e
|
|
|
|
# Run the hypothesis test
|
|
test_random_invalid_args()
|
|
context.fuzz_results = results
|
|
|
|
|
|
@then("all invocations should either succeed or fail gracefully")
|
|
def step_all_succeed_or_fail_gracefully(context):
|
|
"""Verify all fuzz test invocations succeeded or failed gracefully."""
|
|
assert hasattr(context, "fuzz_results"), "No fuzz test results found"
|
|
# If we get here, hypothesis didn't raise any assertion errors
|
|
|
|
|
|
@then("no invocation should crash the system")
|
|
def step_no_crashes(_context):
|
|
"""Verify no invocations crashed the system."""
|
|
# This is verified by the fuzz test above - if we reach here, no crashes occurred
|
|
pass
|