510 lines
13 KiB
Python
510 lines
13 KiB
Python
"""
|
|
Python Test Helper Functions
|
|
|
|
Utility functions for CLI testing with pytest and Click.testing.CliRunner
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Callable
|
|
from click.testing import CliRunner, Result
|
|
|
|
|
|
class CLITestHarness:
|
|
"""Test harness for CLI testing with helpful assertion methods"""
|
|
|
|
def __init__(self, cli_app):
|
|
"""
|
|
Initialize test harness
|
|
|
|
Args:
|
|
cli_app: Click CLI application to test
|
|
"""
|
|
self.cli = cli_app
|
|
self.runner = CliRunner()
|
|
|
|
def run(
|
|
self,
|
|
args: List[str],
|
|
input_data: Optional[str] = None,
|
|
env: Optional[Dict[str, str]] = None
|
|
) -> Result:
|
|
"""
|
|
Run CLI command
|
|
|
|
Args:
|
|
args: Command arguments
|
|
input_data: Input for interactive prompts
|
|
env: Environment variables
|
|
|
|
Returns:
|
|
Click Result object
|
|
"""
|
|
return self.runner.invoke(self.cli, args, input=input_data, env=env)
|
|
|
|
def assert_success(
|
|
self,
|
|
args: List[str],
|
|
expected_in_output: Optional[str] = None
|
|
) -> Result:
|
|
"""
|
|
Run command and assert successful execution
|
|
|
|
Args:
|
|
args: Command arguments
|
|
expected_in_output: Optional string expected in output
|
|
|
|
Returns:
|
|
Click Result object
|
|
|
|
Raises:
|
|
AssertionError: If command fails or output doesn't match
|
|
"""
|
|
result = self.run(args)
|
|
assert result.exit_code == 0, f"Command failed: {result.output}"
|
|
|
|
if expected_in_output:
|
|
assert expected_in_output in result.output, \
|
|
f"Expected '{expected_in_output}' in output: {result.output}"
|
|
|
|
return result
|
|
|
|
def assert_failure(
|
|
self,
|
|
args: List[str],
|
|
expected_in_output: Optional[str] = None
|
|
) -> Result:
|
|
"""
|
|
Run command and assert it fails
|
|
|
|
Args:
|
|
args: Command arguments
|
|
expected_in_output: Optional string expected in output
|
|
|
|
Returns:
|
|
Click Result object
|
|
|
|
Raises:
|
|
AssertionError: If command succeeds or output doesn't match
|
|
"""
|
|
result = self.run(args)
|
|
assert result.exit_code != 0, f"Command should have failed: {result.output}"
|
|
|
|
if expected_in_output:
|
|
assert expected_in_output in result.output, \
|
|
f"Expected '{expected_in_output}' in output: {result.output}"
|
|
|
|
return result
|
|
|
|
def assert_exit_code(self, args: List[str], expected_code: int) -> Result:
|
|
"""
|
|
Run command and assert specific exit code
|
|
|
|
Args:
|
|
args: Command arguments
|
|
expected_code: Expected exit code
|
|
|
|
Returns:
|
|
Click Result object
|
|
|
|
Raises:
|
|
AssertionError: If exit code doesn't match
|
|
"""
|
|
result = self.run(args)
|
|
assert result.exit_code == expected_code, \
|
|
f"Expected exit code {expected_code}, got {result.exit_code}"
|
|
return result
|
|
|
|
def run_json(self, args: List[str]) -> Dict[str, Any]:
|
|
"""
|
|
Run command and parse JSON output
|
|
|
|
Args:
|
|
args: Command arguments
|
|
|
|
Returns:
|
|
Parsed JSON object
|
|
|
|
Raises:
|
|
AssertionError: If command fails
|
|
json.JSONDecodeError: If output is not valid JSON
|
|
"""
|
|
result = self.assert_success(args)
|
|
return json.loads(result.output)
|
|
|
|
|
|
def create_temp_workspace() -> Path:
|
|
"""
|
|
Create temporary workspace directory
|
|
|
|
Returns:
|
|
Path to temporary workspace
|
|
"""
|
|
temp_dir = Path(tempfile.mkdtemp(prefix='cli-test-'))
|
|
return temp_dir
|
|
|
|
|
|
def cleanup_workspace(workspace: Path) -> None:
|
|
"""
|
|
Clean up temporary workspace
|
|
|
|
Args:
|
|
workspace: Path to workspace to remove
|
|
"""
|
|
if workspace.exists():
|
|
shutil.rmtree(workspace)
|
|
|
|
|
|
def create_temp_file(content: str, suffix: str = '.txt') -> Path:
|
|
"""
|
|
Create temporary file with content
|
|
|
|
Args:
|
|
content: File content
|
|
suffix: File extension
|
|
|
|
Returns:
|
|
Path to created file
|
|
"""
|
|
fd, path = tempfile.mkstemp(suffix=suffix)
|
|
with os.fdopen(fd, 'w') as f:
|
|
f.write(content)
|
|
return Path(path)
|
|
|
|
|
|
def assert_file_exists(filepath: Path, message: Optional[str] = None) -> None:
|
|
"""
|
|
Assert file exists
|
|
|
|
Args:
|
|
filepath: Path to file
|
|
message: Optional custom error message
|
|
"""
|
|
assert filepath.exists(), message or f"File does not exist: {filepath}"
|
|
|
|
|
|
def assert_file_contains(filepath: Path, expected: str) -> None:
|
|
"""
|
|
Assert file contains expected text
|
|
|
|
Args:
|
|
filepath: Path to file
|
|
expected: Expected text
|
|
"""
|
|
content = filepath.read_text()
|
|
assert expected in content, \
|
|
f"Expected '{expected}' in file {filepath}\nActual content: {content}"
|
|
|
|
|
|
def assert_json_output(result: Result, schema: Dict[str, type]) -> Dict[str, Any]:
|
|
"""
|
|
Assert output is valid JSON matching schema
|
|
|
|
Args:
|
|
result: Click Result object
|
|
schema: Expected schema as dict of {key: expected_type}
|
|
|
|
Returns:
|
|
Parsed JSON object
|
|
|
|
Raises:
|
|
AssertionError: If JSON is invalid or doesn't match schema
|
|
"""
|
|
try:
|
|
data = json.loads(result.output)
|
|
except json.JSONDecodeError as e:
|
|
raise AssertionError(f"Invalid JSON output: {e}\nOutput: {result.output}")
|
|
|
|
for key, expected_type in schema.items():
|
|
assert key in data, f"Missing key in JSON output: {key}"
|
|
assert isinstance(data[key], expected_type), \
|
|
f"Expected type {expected_type} for key {key}, got {type(data[key])}"
|
|
|
|
return data
|
|
|
|
|
|
def mock_env_vars(vars_dict: Dict[str, str]) -> Callable[[], None]:
|
|
"""
|
|
Mock environment variables
|
|
|
|
Args:
|
|
vars_dict: Dictionary of environment variables to set
|
|
|
|
Returns:
|
|
Function to restore original environment
|
|
|
|
Example:
|
|
restore = mock_env_vars({'API_KEY': 'test_key'})
|
|
# ... run tests ...
|
|
restore()
|
|
"""
|
|
original = {}
|
|
|
|
for key, value in vars_dict.items():
|
|
original[key] = os.environ.get(key)
|
|
os.environ[key] = value
|
|
|
|
def restore():
|
|
for key, value in original.items():
|
|
if value is None:
|
|
os.environ.pop(key, None)
|
|
else:
|
|
os.environ[key] = value
|
|
|
|
return restore
|
|
|
|
|
|
def compare_output_lines(result: Result, expected_lines: List[str]) -> None:
|
|
"""
|
|
Compare output with expected lines
|
|
|
|
Args:
|
|
result: Click Result object
|
|
expected_lines: List of expected lines in output
|
|
|
|
Raises:
|
|
AssertionError: If any expected line is missing
|
|
"""
|
|
output = result.output
|
|
for expected in expected_lines:
|
|
assert expected in output, \
|
|
f"Expected line '{expected}' not found in output:\n{output}"
|
|
|
|
|
|
def parse_table_output(result: Result) -> List[Dict[str, str]]:
|
|
"""
|
|
Parse table output into list of dictionaries
|
|
|
|
Args:
|
|
result: Click Result object with table output
|
|
|
|
Returns:
|
|
List of row dictionaries
|
|
|
|
Note:
|
|
Expects table with headers and │ separators
|
|
"""
|
|
lines = result.output.strip().split('\n')
|
|
|
|
# Find header line
|
|
header_line = None
|
|
for i, line in enumerate(lines):
|
|
if '│' in line and i > 0:
|
|
header_line = i
|
|
break
|
|
|
|
if header_line is None:
|
|
raise ValueError("Could not find table header")
|
|
|
|
# Parse headers
|
|
headers = [h.strip() for h in lines[header_line].split('│') if h.strip()]
|
|
|
|
# Parse rows
|
|
rows = []
|
|
for line in lines[header_line + 2:]: # Skip separator
|
|
if '│' in line:
|
|
values = [v.strip() for v in line.split('│') if v.strip()]
|
|
if len(values) == len(headers):
|
|
rows.append(dict(zip(headers, values)))
|
|
|
|
return rows
|
|
|
|
|
|
class SnapshotTester:
|
|
"""Helper for snapshot testing CLI output"""
|
|
|
|
def __init__(self, snapshot_dir: Path):
|
|
"""
|
|
Initialize snapshot tester
|
|
|
|
Args:
|
|
snapshot_dir: Directory to store snapshots
|
|
"""
|
|
self.snapshot_dir = snapshot_dir
|
|
self.snapshot_dir.mkdir(exist_ok=True)
|
|
|
|
def assert_matches(
|
|
self,
|
|
result: Result,
|
|
snapshot_name: str,
|
|
update: bool = False
|
|
) -> None:
|
|
"""
|
|
Assert output matches snapshot
|
|
|
|
Args:
|
|
result: Click Result object
|
|
snapshot_name: Name of snapshot file
|
|
update: Whether to update snapshot
|
|
|
|
Raises:
|
|
AssertionError: If output doesn't match snapshot
|
|
"""
|
|
snapshot_file = self.snapshot_dir / f'{snapshot_name}.txt'
|
|
|
|
if update or not snapshot_file.exists():
|
|
snapshot_file.write_text(result.output)
|
|
return
|
|
|
|
expected = snapshot_file.read_text()
|
|
assert result.output == expected, \
|
|
f"Output doesn't match snapshot {snapshot_name}\n" \
|
|
f"Expected:\n{expected}\n\nActual:\n{result.output}"
|
|
|
|
|
|
class MockConfig:
|
|
"""Mock configuration file for testing"""
|
|
|
|
def __init__(self, workspace: Path, filename: str = '.myclirc'):
|
|
"""
|
|
Initialize mock config
|
|
|
|
Args:
|
|
workspace: Workspace directory
|
|
filename: Config filename
|
|
"""
|
|
self.config_path = workspace / filename
|
|
self.data = {}
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
"""Set configuration value"""
|
|
self.data[key] = value
|
|
self.save()
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Get configuration value"""
|
|
return self.data.get(key, default)
|
|
|
|
def save(self) -> None:
|
|
"""Save configuration to file"""
|
|
import yaml
|
|
with open(self.config_path, 'w') as f:
|
|
yaml.dump(self.data, f)
|
|
|
|
def load(self) -> None:
|
|
"""Load configuration from file"""
|
|
if self.config_path.exists():
|
|
import yaml
|
|
with open(self.config_path, 'r') as f:
|
|
self.data = yaml.safe_load(f) or {}
|
|
|
|
|
|
def wait_for_file(filepath: Path, timeout: float = 5.0) -> None:
|
|
"""
|
|
Wait for file to exist
|
|
|
|
Args:
|
|
filepath: Path to file
|
|
timeout: Timeout in seconds
|
|
|
|
Raises:
|
|
TimeoutError: If file doesn't exist within timeout
|
|
"""
|
|
import time
|
|
start = time.time()
|
|
|
|
while not filepath.exists():
|
|
if time.time() - start > timeout:
|
|
raise TimeoutError(f"Timeout waiting for file: {filepath}")
|
|
time.sleep(0.1)
|
|
|
|
|
|
def capture_output(func: Callable) -> Dict[str, str]:
|
|
"""
|
|
Capture stdout and stderr during function execution
|
|
|
|
Args:
|
|
func: Function to execute
|
|
|
|
Returns:
|
|
Dictionary with 'stdout' and 'stderr' keys
|
|
"""
|
|
import sys
|
|
from io import StringIO
|
|
|
|
old_stdout = sys.stdout
|
|
old_stderr = sys.stderr
|
|
|
|
stdout_capture = StringIO()
|
|
stderr_capture = StringIO()
|
|
|
|
sys.stdout = stdout_capture
|
|
sys.stderr = stderr_capture
|
|
|
|
try:
|
|
func()
|
|
finally:
|
|
sys.stdout = old_stdout
|
|
sys.stderr = old_stderr
|
|
|
|
return {
|
|
'stdout': stdout_capture.getvalue(),
|
|
'stderr': stderr_capture.getvalue()
|
|
}
|
|
|
|
|
|
class IntegrationTestHelper:
|
|
"""Helper for integration testing with state management"""
|
|
|
|
def __init__(self, cli_app, workspace: Optional[Path] = None):
|
|
"""
|
|
Initialize integration test helper
|
|
|
|
Args:
|
|
cli_app: Click CLI application
|
|
workspace: Optional workspace directory
|
|
"""
|
|
self.harness = CLITestHarness(cli_app)
|
|
self.workspace = workspace or create_temp_workspace()
|
|
self.original_cwd = Path.cwd()
|
|
|
|
def __enter__(self):
|
|
"""Enter context - change to workspace"""
|
|
os.chdir(self.workspace)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Exit context - restore cwd and cleanup"""
|
|
os.chdir(self.original_cwd)
|
|
cleanup_workspace(self.workspace)
|
|
|
|
def run_workflow(self, commands: List[List[str]]) -> List[Result]:
|
|
"""
|
|
Run multiple commands in sequence
|
|
|
|
Args:
|
|
commands: List of command argument lists
|
|
|
|
Returns:
|
|
List of Result objects
|
|
"""
|
|
results = []
|
|
for cmd in commands:
|
|
result = self.harness.run(cmd)
|
|
results.append(result)
|
|
if result.exit_code != 0:
|
|
break
|
|
return results
|
|
|
|
def assert_workflow_success(self, commands: List[List[str]]) -> List[Result]:
|
|
"""
|
|
Run workflow and assert all commands succeed
|
|
|
|
Args:
|
|
commands: List of command argument lists
|
|
|
|
Returns:
|
|
List of Result objects
|
|
|
|
Raises:
|
|
AssertionError: If any command fails
|
|
"""
|
|
results = []
|
|
for i, cmd in enumerate(commands):
|
|
result = self.harness.assert_success(cmd)
|
|
results.append(result)
|
|
return results
|