0
# Testing Framework
1

2
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.
3

4
## Capabilities
5

6
### Base Test Classes
7

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

10
```python { .api }
11
class BaseCLIDriverTest(unittest.TestCase):
12
    """
13
    Base test class for CLI driver testing.
14
    Provides CLI driver setup and common testing utilities.
15
    """
16
    
17
    def setUp(self): ...
18
    def create_clidriver(self): ...
19
    def run_cmd(self, cmdline, expected_rc=0): ...
20
    def mock_aws_service(self, service_name): ...
21
    def register_command(self, command_name, command_class): ...
22
    def temporary_file(self): ...
23
    def environment_patch(self, env_vars): ...
24

25
class BaseAWSCommandParamsTest(unittest.TestCase):
26
    """
27
    Base test class for AWS command parameter testing.
28
    Provides AWS service mocking and parameter validation.
29
    """
30
    
31
    def setUp(self): ...
32
    def run_cmd(self, cmdline): ...
33
    def assert_params_for_cmd(self, expected_params, ignore_params=None): ...
34
    def before_parameter_build(self, params, model): ...
35
    def after_parameter_build(self, params): ...
36

37
class BaseS3CLICommand(unittest.TestCase):
38
    """
39
    Base test class for S3 command testing.
40
    Provides S3-specific testing utilities and mocking.
41
    """
42
    
43
    def setUp(self): ...
44
    def create_bucket(self, bucket_name): ...
45
    def put_object(self, bucket_name, key, content): ...
46
    def list_objects(self, bucket_name): ...
47
    def delete_object(self, bucket_name, key): ...
48
    def random_bucket_name(self): ...
49

50
class BaseAWSHelpOutputTest(unittest.TestCase):
51
    """
52
    Base test class for AWS help output testing.
53
    Provides help command testing utilities and output validation.
54
    """
55
    
56
    def setUp(self): ...
57
    def run_help_cmd(self, cmdline): ...
58
    def assert_contains(self, contains): ...
59
    def assert_not_contains(self, not_contains): ...
60
    def assert_text_order(self, *args): ...
61
    def get_help_contents(self): ...
62

63
class BaseCLIWireResponseTest(unittest.TestCase):
64
    """
65
    Base test class for CLI wire response testing.
66
    Provides HTTP response mocking and wire-level testing utilities.
67
    """
68
    
69
    def setUp(self): ...
70
    def create_http_response(self, status_code, headers, content): ...
71
    def setup_http_adapter(self): ...
72
    def add_response(self, method, url, **kwargs): ...
73
    def assert_request_made(self, method, url): ...
74
```
75

76
### CLI Testing Functions
77

78
Core functions for executing and testing CLI commands programmatically.
79

80
```python { .api }
81
def create_clidriver():
82
    """
83
    Create CLI driver instance for testing.
84
    
85
    Returns:
86
        CLIDriver: Configured driver for test execution
87
    """
88

89
def aws(*args):
90
    """
91
    Execute AWS CLI commands in test environment.
92
    
93
    Parameters:
94
        *args: CLI command arguments
95
        
96
    Returns:
97
        Test execution result with stdout, stderr, and exit code
98
    """
99

100
def capture_output():
101
    """
102
    Capture CLI output for testing and validation.
103
    
104
    Returns:
105
        Context manager for output capture
106
    """
107
```
108

109
### File and Resource Management
110

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

113
```python { .api }
114
class FileCreator:
115
    """
116
    Helper class for creating test files and directories.
117
    Provides automatic cleanup and resource management.
118
    """
119

120
def temporary_file(mode='w', suffix='', prefix='tmp', dir=None, delete=True):
121
    """
122
    Create temporary file for testing.
123
    
124
    Parameters:
125
        mode (str): File mode (default: 'w')
126
        suffix (str): File suffix (default: '')
127
        prefix (str): File prefix (default: 'tmp')
128
        dir (str): Directory for temporary file (default: None)
129
        delete (bool): Delete file on exit (default: True)
130
    
131
    Returns:
132
        Context manager providing temporary file path
133
    """
134

135
def skip_if_windows(reason='Test not supported on Windows'):
136
    """
137
    Skip test if running on Windows platform.
138
    
139
    Parameters:
140
        reason (str): Reason for skipping test
141
    
142
    Returns:
143
        unittest.skip decorator
144
    """
145

146
def skip_if_not_windows(reason='Test only supported on Windows'):
147
    """
148
    Skip test if not running on Windows platform.
149
    
150
    Parameters:
151
        reason (str): Reason for skipping test
152
    
153
    Returns:
154
        unittest.skip decorator
155
    """
156

157
def random_bucket_name(prefix='test-bucket', suffix_length=8):
158
    """
159
    Generate random S3 bucket name for testing.
160
    
161
    Parameters:
162
        prefix (str): Bucket name prefix (default: 'test-bucket')
163
        suffix_length (int): Length of random suffix (default: 8)
164
    
165
    Returns:
166
        str: Random bucket name following S3 naming conventions
167
    """
168

169
def create_bucket_notification_event(bucket_name, key, event_name='s3:ObjectCreated:*'):
170
    """
171
    Create S3 bucket notification event for testing.
172
    
173
    Parameters:
174
        bucket_name (str): S3 bucket name
175
        key (str): Object key
176
        event_name (str): Event type (default: 's3:ObjectCreated:*')
177
    
178
    Returns:
179
        dict: S3 event notification structure
180
    """
181

182
def create_multipart_upload_id(bucket_name, key):
183
    """
184
    Create multipart upload ID for S3 testing.
185
    
186
    Parameters:
187
        bucket_name (str): S3 bucket name  
188
        key (str): Object key
189
    
190
    Returns:
191
        str: Multipart upload ID
192
    """
193
```
194

195
### HTTP Response Testing
196

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

199
```python { .api }
200
def mock_make_request(status_code=200, headers=None, content=b''):
201
    """
202
    Create mock HTTP request for testing.
203
    
204
    Parameters:
205
        status_code (int): HTTP status code (default: 200)
206
        headers (dict): Response headers (default: None)
207
        content (bytes): Response content (default: b'')
208
    
209
    Returns:
210
        Mock HTTP response object
211
    """
212

213
def assert_request_headers(request, expected_headers):
214
    """
215
    Assert that request contains expected headers.
216
    
217
    Parameters:
218
        request: HTTP request object
219
        expected_headers (dict): Expected headers to validate
220
    """
221

222
def create_streaming_response(chunks, status_code=200):
223
    """
224
    Create streaming HTTP response for testing.
225
    
226
    Parameters:
227
        chunks (list): List of content chunks
228
        status_code (int): HTTP status code (default: 200)
229
    
230
    Returns:
231
        Streaming response object
232
    """
233
```
234

235
### Platform Testing Utilities
236

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

239
```python { .api }
240
def platform_test(platforms):
241
    """
242
    Decorator to run test only on specified platforms.
243
    
244
    Parameters:
245
        platforms (list): List of platform names ('windows', 'darwin', 'linux')
246
    
247
    Returns:
248
        Test decorator
249
    """
250

251
def requires_binary(binary_name):
252
    """
253
    Skip test if required binary is not available.
254
    
255
    Parameters:
256
        binary_name (str): Name of required binary
257
    
258
    Returns:
259
        unittest.skipUnless decorator
260
    """
261

262
def mock_platform(platform_name):
263
    """
264
    Mock platform.system() for testing.
265
    
266
    Parameters:
267
        platform_name (str): Platform name to mock
268
    
269
    Returns:
270
        Context manager for platform mocking
271
    """
272
```
273

274
**Usage Example:**
275
```python
276
import unittest
277
from awscli.testutils import (
278
    BaseCLIDriverTest, BaseAWSHelpOutputTest, BaseCLIWireResponseTest,
279
    capture_output, temporary_file, skip_if_windows, platform_test
280
)
281

282
class TestCustomCommand(BaseCLIDriverTest):
283
    def test_custom_command_execution(self):
284
        """Test custom command executes successfully."""
285
        with capture_output() as captured:
286
            exit_code = self.run_cmd(['custom-command', '--help'])
287
        
288
        self.assertEqual(exit_code, 0)
289
        self.assertIn('Custom command help', captured.stdout)
290
    
291
    def test_custom_command_with_args(self):
292
        """Test custom command with arguments."""
293
        result = self.run_cmd([
294
            'custom-command',
295
            '--input', 'test-input',
296
            '--format', 'json'
297
        ])
298
        
299
        self.assertEqual(result.rc, 0)
300
        self.assertIn('test-input', result.stdout)
301

302
class TestHelpOutput(BaseAWSHelpOutputTest):
303
    def test_service_help_output(self):
304
        """Test service help output format."""
305
        self.run_help_cmd(['ec2', 'help'])
306
        
307
        # Validate help content
308
        self.assert_contains('EC2')
309
        self.assert_contains('describe-instances')
310
        self.assert_text_order('DESCRIPTION', 'SYNOPSIS', 'OPTIONS')
311
    
312
    def test_operation_help_output(self):
313
        """Test operation help output."""
314
        self.run_help_cmd(['ec2', 'describe-instances', 'help'])
315
        
316
        self.assert_contains('describe-instances')
317
        self.assert_not_contains('invalid-content')
318

319
class TestWireResponse(BaseCLIWireResponseTest):
320
    def test_http_response_handling(self):
321
        """Test HTTP response processing."""
322
        # Set up mock response
323
        response = self.create_http_response(
324
            status_code=200,
325
            headers={'Content-Type': 'application/json'},
326
            content=b'{"instances": []}'
327
        )
328
        
329
        self.add_response('POST', 'https://ec2.amazonaws.com/', response)
330
        
331
        # Execute command
332
        result = self.run_cmd(['ec2', 'describe-instances'])
333
        
334
        # Verify request was made
335
        self.assert_request_made('POST', 'https://ec2.amazonaws.com/')
336
        self.assertEqual(result.rc, 0)
337
```
338

339
## Test Class Implementations
340

341
### CLI Driver Testing
342

343
```python
344
class TestCLIDriver(BaseCLIDriverTest):
345
    """Test CLI driver functionality."""
346
    
347
    def setUp(self):
348
        super().setUp()
349
        # Additional setup for CLI driver testing
350
        self.driver = self.create_clidriver()
351
    
352
    def test_service_command_execution(self):
353
        """Test service command execution."""
354
        # Mock AWS service
355
        with self.mock_aws_service('ec2'):
356
            result = self.run_cmd(['ec2', 'describe-instances'])
357
            self.assertEqual(result.rc, 0)
358
    
359
    def test_argument_parsing(self):
360
        """Test argument parsing functionality."""
361
        result = self.run_cmd([
362
            's3', 'ls', 's3://my-bucket',
363
            '--recursive',
364
            '--output', 'json'
365
        ])
366
        
367
        self.assertEqual(result.rc, 0)
368
        self.validate_json_output(result.stdout)
369
```
370

371
### AWS Command Parameter Testing
372

373
```python
374
class TestEC2Commands(BaseAWSCommandParamsTest):
375
    """Test EC2 command parameter handling."""
376
    
377
    def setUp(self):
378
        super().setUp()
379
        self.service_name = 'ec2'
380
    
381
    def test_describe_instances_parameters(self):
382
        """Test describe-instances parameter processing."""
383
        # Set up expected parameters
384
        expected_params = {
385
            'InstanceIds': ['i-1234567890abcdef0'],
386
            'Filters': [
387
                {'Name': 'instance-state-name', 'Values': ['running']}
388
            ]
389
        }
390
        
391
        # Execute command and verify parameters
392
        self.run_cmd([
393
            'ec2', 'describe-instances',
394
            '--instance-ids', 'i-1234567890abcdef0',
395
            '--filters', 'Name=instance-state-name,Values=running'
396
        ])
397
        
398
        # Verify parameters were processed correctly
399
        self.assert_params_for_cmd(expected_params)
400
    
401
    def test_parameter_validation(self):
402
        """Test parameter validation and error handling."""
403
        result = self.run_cmd([
404
            'ec2', 'describe-instances',
405
            '--invalid-parameter', 'value'
406
        ])
407
        
408
        self.assertNotEqual(result.rc, 0)
409
        self.assertIn('Unknown parameter', result.stderr)
410
```
411

412
### S3 Command Testing
413

414
```python
415
class TestS3Commands(BaseS3CLICommand):
416
    """Test S3-specific command functionality."""
417
    
418
    def setUp(self):
419
        super().setUp()
420
        self.bucket_name = self.random_bucket_name()
421
        self.create_bucket(self.bucket_name)
422
    
423
    def test_s3_ls_command(self):
424
        """Test S3 ls command."""
425
        # Create test objects
426
        self.put_object(self.bucket_name, 'test-file.txt', b'test content')
427
        
428
        # Execute ls command
429
        result = self.run_cmd(['s3', 'ls', f's3://{self.bucket_name}'])
430
        
431
        self.assertEqual(result.rc, 0)
432
        self.assertIn('test-file.txt', result.stdout)
433
    
434
    def test_s3_cp_command(self):
435
        """Test S3 cp command."""
436
        with self.temporary_file() as temp_file:
437
            # Write test content
438
            with open(temp_file, 'w') as f:
439
                f.write('test content')
440
            
441
            # Copy to S3
442
            result = self.run_cmd([
443
                's3', 'cp', temp_file, f's3://{self.bucket_name}/uploaded.txt'
444
            ])
445
            
446
            self.assertEqual(result.rc, 0)
447
            
448
            # Verify object exists
449
            objects = self.list_objects(self.bucket_name)
450
            self.assertIn('uploaded.txt', [obj['Key'] for obj in objects])
451
```
452

453
## Advanced Testing Patterns
454

455
### Mock Service Integration
456

457
```python
458
class TestWithMockServices(BaseCLIDriverTest):
459
    """Test with comprehensive service mocking."""
460
    
461
    def setUp(self):
462
        super().setUp()
463
        # Set up service mocks
464
        self.ec2_mock = self.mock_aws_service('ec2')
465
        self.s3_mock = self.mock_aws_service('s3')
466
    
467
    def test_cross_service_operation(self):
468
        """Test operation involving multiple services."""
469
        # Configure EC2 mock
470
        self.ec2_mock.describe_instances.return_value = {
471
            'Reservations': [{
472
                'Instances': [{
473
                    'InstanceId': 'i-1234567890abcdef0',
474
                    'State': {'Name': 'running'}
475
                }]
476
            }]
477
        }
478
        
479
        # Configure S3 mock
480
        self.s3_mock.list_objects_v2.return_value = {
481
            'Contents': [{'Key': 'config.json', 'Size': 1024}]
482
        }
483
        
484
        # Execute commands and verify interactions
485
        ec2_result = self.run_cmd(['ec2', 'describe-instances'])
486
        s3_result = self.run_cmd(['s3', 'ls', 's3://config-bucket'])
487
        
488
        self.assertEqual(ec2_result.rc, 0)
489
        self.assertEqual(s3_result.rc, 0)
490
        
491
        # Verify service calls
492
        self.ec2_mock.describe_instances.assert_called_once()
493
        self.s3_mock.list_objects_v2.assert_called_once()
494
```
495

496
### Custom Command Testing
497

498
```python
499
class TestCustomCommands(BaseCLIDriverTest):
500
    """Test custom command implementations."""
501
    
502
    def setUp(self):
503
        super().setUp()
504
        # Register custom commands
505
        from mypackage.commands import MyCustomCommand
506
        self.register_command('my-custom', MyCustomCommand)
507
    
508
    def test_custom_command_registration(self):
509
        """Test custom command is properly registered."""
510
        result = self.run_cmd(['help'])
511
        self.assertIn('my-custom', result.stdout)
512
    
513
    def test_custom_command_execution(self):
514
        """Test custom command execution."""
515
        result = self.run_cmd([
516
            'my-custom',
517
            '--parameter', 'value',
518
            '--flag'
519
        ])
520
        
521
        self.assertEqual(result.rc, 0)
522
        self.assertIn('Expected output', result.stdout)
523
    
524
    def test_custom_command_error_handling(self):
525
        """Test custom command error handling."""
526
        result = self.run_cmd([
527
            'my-custom',
528
            '--invalid-parameter', 'value'
529
        ])
530
        
531
        self.assertNotEqual(result.rc, 0)
532
        self.assertIn('Invalid parameter', result.stderr)
533
```
534

535
### Output Format Testing
536

537
```python
538
class TestOutputFormats(BaseCLIDriverTest):
539
    """Test different output format handling."""
540
    
541
    def test_json_output(self):
542
        """Test JSON output format."""
543
        result = self.run_cmd([
544
            'ec2', 'describe-instances',
545
            '--output', 'json'
546
        ])
547
        
548
        self.assertEqual(result.rc, 0)
549
        
550
        # Validate JSON structure
551
        import json
552
        data = json.loads(result.stdout)
553
        self.assertIn('Reservations', data)
554
    
555
    def test_table_output(self):
556
        """Test table output format."""
557
        result = self.run_cmd([
558
            'ec2', 'describe-instances',
559
            '--output', 'table'
560
        ])
561
        
562
        self.assertEqual(result.rc, 0)
563
        
564
        # Validate table structure
565
        lines = result.stdout.strip().split('\n')
566
        self.assertGreater(len(lines), 2)  # Header + data
567
        self.assertIn('|', result.stdout)  # Table formatting
568
    
569
    def test_text_output(self):
570
        """Test text output format."""
571
        result = self.run_cmd([
572
            'ec2', 'describe-instances',
573
            '--output', 'text'
574
        ])
575
        
576
        self.assertEqual(result.rc, 0)
577
        self.assertIn('RESERVATIONS', result.stdout)
578
```
579

580
## Test Utilities and Helpers
581

582
### File Management
583

584
```python
585
class TestFileOperations(BaseCLIDriverTest):
586
    """Test file-based operations."""
587
    
588
    def test_with_temporary_files(self):
589
        """Test operations with temporary files."""
590
        with self.temporary_file() as temp_path:
591
            # Write test data
592
            with open(temp_path, 'w') as f:
593
                f.write('{"test": "data"}')
594
            
595
            # Use file in command
596
            result = self.run_cmd([
597
                'ec2', 'run-instances',
598
                '--cli-input-json', f'file://{temp_path}'
599
            ])
600
            
601
            # File is automatically cleaned up
602
```
603

604
### Environment Configuration
605

606
```python
607
def test_with_environment_config(self):
608
    """Test with specific environment configuration."""
609
    import os
610
    
611
    # Set test environment
612
    test_env = {
613
        'AWS_DEFAULT_REGION': 'us-west-2',
614
        'AWS_DEFAULT_OUTPUT': 'json'
615
    }
616
    
617
    with self.environment_patch(test_env):
618
        result = self.run_cmd(['configure', 'list'])
619
        self.assertIn('us-west-2', result.stdout)
620
        self.assertIn('json', result.stdout)
621

622
### Platform-Specific Testing
623

624
```python
625
class TestPlatformBehavior(BaseCLIDriverTest):
626
    """Test platform-specific command behavior."""
627
    
628
    @skip_if_windows('Unix-specific path handling')
629
    def test_unix_path_handling(self):
630
        """Test Unix path handling."""
631
        result = self.run_cmd([
632
            's3', 'cp', '/tmp/test.txt', 's3://bucket/test.txt'
633
        ])
634
        self.assertEqual(result.rc, 0)
635
    
636
    @skip_if_not_windows('Windows-specific drive handling')
637
    def test_windows_drive_handling(self):
638
        """Test Windows drive handling."""
639
        result = self.run_cmd([
640
            's3', 'cp', 'C:\\temp\\test.txt', 's3://bucket/test.txt'
641
        ])
642
        self.assertEqual(result.rc, 0)
643
    
644
    @platform_test(['linux', 'darwin'])
645
    def test_unix_only_feature(self):
646
        """Test feature available only on Unix systems."""
647
        result = self.run_cmd(['configure', 'set', 'cli_pager', 'less'])
648
        self.assertEqual(result.rc, 0)
649
    
650
    @requires_binary('git')
651
    def test_git_integration(self):
652
        """Test Git integration functionality."""
653
        with temporary_file(suffix='.git') as git_file:
654
            result = self.run_cmd(['codecommit', 'credential-helper', git_file])
655
            self.assertEqual(result.rc, 0)
656

657
### HTTP Wire Testing
658

659
```python
660
class TestHTTPProtocol(BaseCLIWireResponseTest):
661
    """Test HTTP protocol interactions."""
662
    
663
    def setUp(self):
664
        super().setUp()
665
        self.setup_http_adapter()
666
    
667
    def test_retry_behavior(self):
668
        """Test retry behavior on HTTP errors."""
669
        # First call fails with 500
670
        self.add_response(
671
            'POST', 'https://ec2.amazonaws.com/',
672
            status_code=500,
673
            content=b'Internal Server Error'
674
        )
675
        
676
        # Second call succeeds
677
        self.add_response(
678
            'POST', 'https://ec2.amazonaws.com/',
679
            status_code=200,
680
            content=b'{"Reservations": []}'
681
        )
682
        
683
        result = self.run_cmd(['ec2', 'describe-instances'])
684
        
685
        # Should succeed after retry
686
        self.assertEqual(result.rc, 0)
687
        self.assert_request_made('POST', 'https://ec2.amazonaws.com/')
688
    
689
    def test_streaming_response(self):
690
        """Test streaming response handling."""
691
        chunks = [b'{"', b'Reservations', b'": []}']
692
        response = self.create_streaming_response(chunks)
693
        
694
        self.add_response('POST', 'https://ec2.amazonaws.com/', response)
695
        
696
        result = self.run_cmd(['ec2', 'describe-instances'])
697
        self.assertEqual(result.rc, 0)
698
        
699
        # Verify complete JSON was processed
700
        import json
701
        data = json.loads(result.stdout)
702
        self.assertIn('Reservations', data)
703
```