CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-awscli

Universal Command Line Environment for AWS.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

testing-framework.mddocs/

Testing Framework

Extensive testing utilities for developing and validating AWS CLI commands, including test base classes, output capture, and AWS service mocking. This framework enables comprehensive testing of CLI functionality and custom extensions.

Capabilities

Base Test Classes

Foundation test classes providing common CLI testing functionality and AWS service mocking.

class BaseCLIDriverTest(unittest.TestCase):
    """
    Base test class for CLI driver testing.
    Provides CLI driver setup and common testing utilities.
    """
    
    def setUp(self): ...
    def create_clidriver(self): ...
    def run_cmd(self, cmdline, expected_rc=0): ...
    def mock_aws_service(self, service_name): ...
    def register_command(self, command_name, command_class): ...
    def temporary_file(self): ...
    def environment_patch(self, env_vars): ...

class BaseAWSCommandParamsTest(unittest.TestCase):
    """
    Base test class for AWS command parameter testing.
    Provides AWS service mocking and parameter validation.
    """
    
    def setUp(self): ...
    def run_cmd(self, cmdline): ...
    def assert_params_for_cmd(self, expected_params, ignore_params=None): ...
    def before_parameter_build(self, params, model): ...
    def after_parameter_build(self, params): ...

class BaseS3CLICommand(unittest.TestCase):
    """
    Base test class for S3 command testing.
    Provides S3-specific testing utilities and mocking.
    """
    
    def setUp(self): ...
    def create_bucket(self, bucket_name): ...
    def put_object(self, bucket_name, key, content): ...
    def list_objects(self, bucket_name): ...
    def delete_object(self, bucket_name, key): ...
    def random_bucket_name(self): ...

class BaseAWSHelpOutputTest(unittest.TestCase):
    """
    Base test class for AWS help output testing.
    Provides help command testing utilities and output validation.
    """
    
    def setUp(self): ...
    def run_help_cmd(self, cmdline): ...
    def assert_contains(self, contains): ...
    def assert_not_contains(self, not_contains): ...
    def assert_text_order(self, *args): ...
    def get_help_contents(self): ...

class BaseCLIWireResponseTest(unittest.TestCase):
    """
    Base test class for CLI wire response testing.
    Provides HTTP response mocking and wire-level testing utilities.
    """
    
    def setUp(self): ...
    def create_http_response(self, status_code, headers, content): ...
    def setup_http_adapter(self): ...
    def add_response(self, method, url, **kwargs): ...
    def assert_request_made(self, method, url): ...

CLI Testing Functions

Core functions for executing and testing CLI commands programmatically.

def create_clidriver():
    """
    Create CLI driver instance for testing.
    
    Returns:
        CLIDriver: Configured driver for test execution
    """

def aws(*args):
    """
    Execute AWS CLI commands in test environment.
    
    Parameters:
        *args: CLI command arguments
        
    Returns:
        Test execution result with stdout, stderr, and exit code
    """

def capture_output():
    """
    Capture CLI output for testing and validation.
    
    Returns:
        Context manager for output capture
    """

File and Resource Management

Utilities for creating and managing temporary files and resources during testing, including platform-specific testing decorators.

class FileCreator:
    """
    Helper class for creating test files and directories.
    Provides automatic cleanup and resource management.
    """

def temporary_file(mode='w', suffix='', prefix='tmp', dir=None, delete=True):
    """
    Create temporary file for testing.
    
    Parameters:
        mode (str): File mode (default: 'w')
        suffix (str): File suffix (default: '')
        prefix (str): File prefix (default: 'tmp')
        dir (str): Directory for temporary file (default: None)
        delete (bool): Delete file on exit (default: True)
    
    Returns:
        Context manager providing temporary file path
    """

def skip_if_windows(reason='Test not supported on Windows'):
    """
    Skip test if running on Windows platform.
    
    Parameters:
        reason (str): Reason for skipping test
    
    Returns:
        unittest.skip decorator
    """

def skip_if_not_windows(reason='Test only supported on Windows'):
    """
    Skip test if not running on Windows platform.
    
    Parameters:
        reason (str): Reason for skipping test
    
    Returns:
        unittest.skip decorator
    """

def random_bucket_name(prefix='test-bucket', suffix_length=8):
    """
    Generate random S3 bucket name for testing.
    
    Parameters:
        prefix (str): Bucket name prefix (default: 'test-bucket')
        suffix_length (int): Length of random suffix (default: 8)
    
    Returns:
        str: Random bucket name following S3 naming conventions
    """

def create_bucket_notification_event(bucket_name, key, event_name='s3:ObjectCreated:*'):
    """
    Create S3 bucket notification event for testing.
    
    Parameters:
        bucket_name (str): S3 bucket name
        key (str): Object key
        event_name (str): Event type (default: 's3:ObjectCreated:*')
    
    Returns:
        dict: S3 event notification structure
    """

def create_multipart_upload_id(bucket_name, key):
    """
    Create multipart upload ID for S3 testing.
    
    Parameters:
        bucket_name (str): S3 bucket name  
        key (str): Object key
    
    Returns:
        str: Multipart upload ID
    """

HTTP Response Testing

Utilities for testing HTTP interactions and wire-level protocol behavior.

def mock_make_request(status_code=200, headers=None, content=b''):
    """
    Create mock HTTP request for testing.
    
    Parameters:
        status_code (int): HTTP status code (default: 200)
        headers (dict): Response headers (default: None)
        content (bytes): Response content (default: b'')
    
    Returns:
        Mock HTTP response object
    """

def assert_request_headers(request, expected_headers):
    """
    Assert that request contains expected headers.
    
    Parameters:
        request: HTTP request object
        expected_headers (dict): Expected headers to validate
    """

def create_streaming_response(chunks, status_code=200):
    """
    Create streaming HTTP response for testing.
    
    Parameters:
        chunks (list): List of content chunks
        status_code (int): HTTP status code (default: 200)
    
    Returns:
        Streaming response object
    """

Platform Testing Utilities

Cross-platform testing utilities and decorators for handling OS-specific behavior.

def platform_test(platforms):
    """
    Decorator to run test only on specified platforms.
    
    Parameters:
        platforms (list): List of platform names ('windows', 'darwin', 'linux')
    
    Returns:
        Test decorator
    """

def requires_binary(binary_name):
    """
    Skip test if required binary is not available.
    
    Parameters:
        binary_name (str): Name of required binary
    
    Returns:
        unittest.skipUnless decorator
    """

def mock_platform(platform_name):
    """
    Mock platform.system() for testing.
    
    Parameters:
        platform_name (str): Platform name to mock
    
    Returns:
        Context manager for platform mocking
    """

Usage Example:

import unittest
from awscli.testutils import (
    BaseCLIDriverTest, BaseAWSHelpOutputTest, BaseCLIWireResponseTest,
    capture_output, temporary_file, skip_if_windows, platform_test
)

class TestCustomCommand(BaseCLIDriverTest):
    def test_custom_command_execution(self):
        """Test custom command executes successfully."""
        with capture_output() as captured:
            exit_code = self.run_cmd(['custom-command', '--help'])
        
        self.assertEqual(exit_code, 0)
        self.assertIn('Custom command help', captured.stdout)
    
    def test_custom_command_with_args(self):
        """Test custom command with arguments."""
        result = self.run_cmd([
            'custom-command',
            '--input', 'test-input',
            '--format', 'json'
        ])
        
        self.assertEqual(result.rc, 0)
        self.assertIn('test-input', result.stdout)

class TestHelpOutput(BaseAWSHelpOutputTest):
    def test_service_help_output(self):
        """Test service help output format."""
        self.run_help_cmd(['ec2', 'help'])
        
        # Validate help content
        self.assert_contains('EC2')
        self.assert_contains('describe-instances')
        self.assert_text_order('DESCRIPTION', 'SYNOPSIS', 'OPTIONS')
    
    def test_operation_help_output(self):
        """Test operation help output."""
        self.run_help_cmd(['ec2', 'describe-instances', 'help'])
        
        self.assert_contains('describe-instances')
        self.assert_not_contains('invalid-content')

class TestWireResponse(BaseCLIWireResponseTest):
    def test_http_response_handling(self):
        """Test HTTP response processing."""
        # Set up mock response
        response = self.create_http_response(
            status_code=200,
            headers={'Content-Type': 'application/json'},
            content=b'{"instances": []}'
        )
        
        self.add_response('POST', 'https://ec2.amazonaws.com/', response)
        
        # Execute command
        result = self.run_cmd(['ec2', 'describe-instances'])
        
        # Verify request was made
        self.assert_request_made('POST', 'https://ec2.amazonaws.com/')
        self.assertEqual(result.rc, 0)

Test Class Implementations

CLI Driver Testing

class TestCLIDriver(BaseCLIDriverTest):
    """Test CLI driver functionality."""
    
    def setUp(self):
        super().setUp()
        # Additional setup for CLI driver testing
        self.driver = self.create_clidriver()
    
    def test_service_command_execution(self):
        """Test service command execution."""
        # Mock AWS service
        with self.mock_aws_service('ec2'):
            result = self.run_cmd(['ec2', 'describe-instances'])
            self.assertEqual(result.rc, 0)
    
    def test_argument_parsing(self):
        """Test argument parsing functionality."""
        result = self.run_cmd([
            's3', 'ls', 's3://my-bucket',
            '--recursive',
            '--output', 'json'
        ])
        
        self.assertEqual(result.rc, 0)
        self.validate_json_output(result.stdout)

AWS Command Parameter Testing

class TestEC2Commands(BaseAWSCommandParamsTest):
    """Test EC2 command parameter handling."""
    
    def setUp(self):
        super().setUp()
        self.service_name = 'ec2'
    
    def test_describe_instances_parameters(self):
        """Test describe-instances parameter processing."""
        # Set up expected parameters
        expected_params = {
            'InstanceIds': ['i-1234567890abcdef0'],
            'Filters': [
                {'Name': 'instance-state-name', 'Values': ['running']}
            ]
        }
        
        # Execute command and verify parameters
        self.run_cmd([
            'ec2', 'describe-instances',
            '--instance-ids', 'i-1234567890abcdef0',
            '--filters', 'Name=instance-state-name,Values=running'
        ])
        
        # Verify parameters were processed correctly
        self.assert_params_for_cmd(expected_params)
    
    def test_parameter_validation(self):
        """Test parameter validation and error handling."""
        result = self.run_cmd([
            'ec2', 'describe-instances',
            '--invalid-parameter', 'value'
        ])
        
        self.assertNotEqual(result.rc, 0)
        self.assertIn('Unknown parameter', result.stderr)

S3 Command Testing

class TestS3Commands(BaseS3CLICommand):
    """Test S3-specific command functionality."""
    
    def setUp(self):
        super().setUp()
        self.bucket_name = self.random_bucket_name()
        self.create_bucket(self.bucket_name)
    
    def test_s3_ls_command(self):
        """Test S3 ls command."""
        # Create test objects
        self.put_object(self.bucket_name, 'test-file.txt', b'test content')
        
        # Execute ls command
        result = self.run_cmd(['s3', 'ls', f's3://{self.bucket_name}'])
        
        self.assertEqual(result.rc, 0)
        self.assertIn('test-file.txt', result.stdout)
    
    def test_s3_cp_command(self):
        """Test S3 cp command."""
        with self.temporary_file() as temp_file:
            # Write test content
            with open(temp_file, 'w') as f:
                f.write('test content')
            
            # Copy to S3
            result = self.run_cmd([
                's3', 'cp', temp_file, f's3://{self.bucket_name}/uploaded.txt'
            ])
            
            self.assertEqual(result.rc, 0)
            
            # Verify object exists
            objects = self.list_objects(self.bucket_name)
            self.assertIn('uploaded.txt', [obj['Key'] for obj in objects])

Advanced Testing Patterns

Mock Service Integration

class TestWithMockServices(BaseCLIDriverTest):
    """Test with comprehensive service mocking."""
    
    def setUp(self):
        super().setUp()
        # Set up service mocks
        self.ec2_mock = self.mock_aws_service('ec2')
        self.s3_mock = self.mock_aws_service('s3')
    
    def test_cross_service_operation(self):
        """Test operation involving multiple services."""
        # Configure EC2 mock
        self.ec2_mock.describe_instances.return_value = {
            'Reservations': [{
                'Instances': [{
                    'InstanceId': 'i-1234567890abcdef0',
                    'State': {'Name': 'running'}
                }]
            }]
        }
        
        # Configure S3 mock
        self.s3_mock.list_objects_v2.return_value = {
            'Contents': [{'Key': 'config.json', 'Size': 1024}]
        }
        
        # Execute commands and verify interactions
        ec2_result = self.run_cmd(['ec2', 'describe-instances'])
        s3_result = self.run_cmd(['s3', 'ls', 's3://config-bucket'])
        
        self.assertEqual(ec2_result.rc, 0)
        self.assertEqual(s3_result.rc, 0)
        
        # Verify service calls
        self.ec2_mock.describe_instances.assert_called_once()
        self.s3_mock.list_objects_v2.assert_called_once()

Custom Command Testing

class TestCustomCommands(BaseCLIDriverTest):
    """Test custom command implementations."""
    
    def setUp(self):
        super().setUp()
        # Register custom commands
        from mypackage.commands import MyCustomCommand
        self.register_command('my-custom', MyCustomCommand)
    
    def test_custom_command_registration(self):
        """Test custom command is properly registered."""
        result = self.run_cmd(['help'])
        self.assertIn('my-custom', result.stdout)
    
    def test_custom_command_execution(self):
        """Test custom command execution."""
        result = self.run_cmd([
            'my-custom',
            '--parameter', 'value',
            '--flag'
        ])
        
        self.assertEqual(result.rc, 0)
        self.assertIn('Expected output', result.stdout)
    
    def test_custom_command_error_handling(self):
        """Test custom command error handling."""
        result = self.run_cmd([
            'my-custom',
            '--invalid-parameter', 'value'
        ])
        
        self.assertNotEqual(result.rc, 0)
        self.assertIn('Invalid parameter', result.stderr)

Output Format Testing

class TestOutputFormats(BaseCLIDriverTest):
    """Test different output format handling."""
    
    def test_json_output(self):
        """Test JSON output format."""
        result = self.run_cmd([
            'ec2', 'describe-instances',
            '--output', 'json'
        ])
        
        self.assertEqual(result.rc, 0)
        
        # Validate JSON structure
        import json
        data = json.loads(result.stdout)
        self.assertIn('Reservations', data)
    
    def test_table_output(self):
        """Test table output format."""
        result = self.run_cmd([
            'ec2', 'describe-instances',
            '--output', 'table'
        ])
        
        self.assertEqual(result.rc, 0)
        
        # Validate table structure
        lines = result.stdout.strip().split('\n')
        self.assertGreater(len(lines), 2)  # Header + data
        self.assertIn('|', result.stdout)  # Table formatting
    
    def test_text_output(self):
        """Test text output format."""
        result = self.run_cmd([
            'ec2', 'describe-instances',
            '--output', 'text'
        ])
        
        self.assertEqual(result.rc, 0)
        self.assertIn('RESERVATIONS', result.stdout)

Test Utilities and Helpers

File Management

class TestFileOperations(BaseCLIDriverTest):
    """Test file-based operations."""
    
    def test_with_temporary_files(self):
        """Test operations with temporary files."""
        with self.temporary_file() as temp_path:
            # Write test data
            with open(temp_path, 'w') as f:
                f.write('{"test": "data"}')
            
            # Use file in command
            result = self.run_cmd([
                'ec2', 'run-instances',
                '--cli-input-json', f'file://{temp_path}'
            ])
            
            # File is automatically cleaned up

Environment Configuration

def test_with_environment_config(self):
    """Test with specific environment configuration."""
    import os
    
    # Set test environment
    test_env = {
        'AWS_DEFAULT_REGION': 'us-west-2',
        'AWS_DEFAULT_OUTPUT': 'json'
    }
    
    with self.environment_patch(test_env):
        result = self.run_cmd(['configure', 'list'])
        self.assertIn('us-west-2', result.stdout)
        self.assertIn('json', result.stdout)

### Platform-Specific Testing

```python
class TestPlatformBehavior(BaseCLIDriverTest):
    """Test platform-specific command behavior."""
    
    @skip_if_windows('Unix-specific path handling')
    def test_unix_path_handling(self):
        """Test Unix path handling."""
        result = self.run_cmd([
            's3', 'cp', '/tmp/test.txt', 's3://bucket/test.txt'
        ])
        self.assertEqual(result.rc, 0)
    
    @skip_if_not_windows('Windows-specific drive handling')
    def test_windows_drive_handling(self):
        """Test Windows drive handling."""
        result = self.run_cmd([
            's3', 'cp', 'C:\\temp\\test.txt', 's3://bucket/test.txt'
        ])
        self.assertEqual(result.rc, 0)
    
    @platform_test(['linux', 'darwin'])
    def test_unix_only_feature(self):
        """Test feature available only on Unix systems."""
        result = self.run_cmd(['configure', 'set', 'cli_pager', 'less'])
        self.assertEqual(result.rc, 0)
    
    @requires_binary('git')
    def test_git_integration(self):
        """Test Git integration functionality."""
        with temporary_file(suffix='.git') as git_file:
            result = self.run_cmd(['codecommit', 'credential-helper', git_file])
            self.assertEqual(result.rc, 0)

### HTTP Wire Testing

```python
class TestHTTPProtocol(BaseCLIWireResponseTest):
    """Test HTTP protocol interactions."""
    
    def setUp(self):
        super().setUp()
        self.setup_http_adapter()
    
    def test_retry_behavior(self):
        """Test retry behavior on HTTP errors."""
        # First call fails with 500
        self.add_response(
            'POST', 'https://ec2.amazonaws.com/',
            status_code=500,
            content=b'Internal Server Error'
        )
        
        # Second call succeeds
        self.add_response(
            'POST', 'https://ec2.amazonaws.com/',
            status_code=200,
            content=b'{"Reservations": []}'
        )
        
        result = self.run_cmd(['ec2', 'describe-instances'])
        
        # Should succeed after retry
        self.assertEqual(result.rc, 0)
        self.assert_request_made('POST', 'https://ec2.amazonaws.com/')
    
    def test_streaming_response(self):
        """Test streaming response handling."""
        chunks = [b'{"', b'Reservations', b'": []}']
        response = self.create_streaming_response(chunks)
        
        self.add_response('POST', 'https://ec2.amazonaws.com/', response)
        
        result = self.run_cmd(['ec2', 'describe-instances'])
        self.assertEqual(result.rc, 0)
        
        # Verify complete JSON was processed
        import json
        data = json.loads(result.stdout)
        self.assertIn('Reservations', data)

Install with Tessl CLI

npx tessl i tessl/pypi-awscli

docs

argument-processing.md

command-system.md

core-driver.md

custom-commands.md

error-handling.md

help-system.md

index.md

output-formatting.md

plugin-system.md

testing-framework.md

utilities.md

tile.json