0
# Testing Support
1
2
Built-in stubbing capabilities for testing AWS service interactions with mock responses, error simulation, and comprehensive parameter validation. The testing framework allows you to write unit tests without making actual AWS API calls.
3
4
## Core Imports
5
6
```python
7
from botocore.stub import Stubber, ANY
8
from botocore.exceptions import (
9
StubResponseError,
10
StubAssertionError,
11
UnStubbedResponseError
12
)
13
```
14
15
## Capabilities
16
17
### Stubber Class
18
19
Primary class for stubbing AWS service client responses in tests.
20
21
```python { .api }
22
class Stubber:
23
def __init__(self, client: BaseClient):
24
"""
25
Initialize stubber for AWS service client.
26
27
Args:
28
client: AWS service client to stub
29
"""
30
31
def activate(self) -> None:
32
"""
33
Activate response stubbing on the client.
34
Registers event handlers to intercept API calls.
35
"""
36
37
def deactivate(self) -> None:
38
"""
39
Deactivate response stubbing on the client.
40
Unregisters event handlers and restores normal operation.
41
"""
42
43
def add_response(
44
self,
45
method: str,
46
service_response: dict,
47
expected_params: dict = None
48
) -> None:
49
"""
50
Add mock response for service method.
51
52
Args:
53
method: Client method name to stub
54
service_response: Response data to return
55
expected_params: Expected parameters for validation
56
"""
57
58
def add_client_error(
59
self,
60
method: str,
61
service_error_code: str = '',
62
service_message: str = '',
63
http_status_code: int = 400,
64
service_error_meta: dict = None,
65
expected_params: dict = None,
66
response_meta: dict = None,
67
modeled_fields: dict = None
68
) -> None:
69
"""
70
Add ClientError response for service method.
71
72
Args:
73
method: Client method name to stub
74
service_error_code: AWS error code (e.g., 'NoSuchBucket')
75
service_message: Human-readable error message
76
http_status_code: HTTP status code for error
77
service_error_meta: Additional error metadata
78
expected_params: Expected parameters for validation
79
response_meta: Additional response metadata
80
modeled_fields: Validated fields based on error shape
81
"""
82
83
def assert_no_pending_responses(self) -> None:
84
"""
85
Assert that all stubbed responses were consumed.
86
Raises AssertionError if unused responses remain.
87
"""
88
89
def __enter__(self) -> 'Stubber':
90
"""Context manager entry - activates stubber."""
91
92
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
93
"""Context manager exit - deactivates stubber."""
94
```
95
96
### Parameter Matching
97
98
Special constants and utilities for flexible parameter validation.
99
100
```python { .api }
101
ANY: object
102
"""
103
Wildcard constant that matches any parameter value.
104
Use for parameters with unpredictable values like timestamps or UUIDs.
105
"""
106
```
107
108
### Testing Exceptions
109
110
Specific exceptions for stubbing-related errors.
111
112
```python { .api }
113
class StubResponseError(BotoCoreError):
114
"""Base exception for stubber response errors."""
115
pass
116
117
class StubAssertionError(StubResponseError, AssertionError):
118
"""Parameter validation failed during stubbed call."""
119
pass
120
121
class UnStubbedResponseError(StubResponseError):
122
"""API call made without corresponding stubbed response."""
123
pass
124
```
125
126
## Basic Usage
127
128
### Simple Response Stubbing
129
130
Basic example of stubbing a successful service response.
131
132
```python
133
import botocore.session
134
from botocore.stub import Stubber
135
136
# Create client and stubber
137
session = botocore.session.get_session()
138
s3_client = session.create_client('s3', region_name='us-east-1')
139
stubber = Stubber(s3_client)
140
141
# Define expected response
142
response = {
143
'Buckets': [
144
{
145
'Name': 'test-bucket',
146
'CreationDate': datetime.datetime(2020, 1, 1)
147
}
148
],
149
'ResponseMetadata': {
150
'RequestId': 'abc123',
151
'HTTPStatusCode': 200
152
}
153
}
154
155
# Add stubbed response
156
stubber.add_response('list_buckets', response)
157
158
# Activate stubber and make call
159
stubber.activate()
160
result = s3_client.list_buckets()
161
stubber.deactivate()
162
163
assert result == response
164
```
165
166
### Context Manager Usage
167
168
Using stubber as context manager for automatic activation/deactivation.
169
170
```python
171
import botocore.session
172
from botocore.stub import Stubber
173
174
session = botocore.session.get_session()
175
ec2_client = session.create_client('ec2', region_name='us-west-2')
176
177
response = {
178
'Instances': [
179
{
180
'InstanceId': 'i-1234567890abcdef0',
181
'State': {'Name': 'running'},
182
'InstanceType': 't2.micro'
183
}
184
]
185
}
186
187
with Stubber(ec2_client) as stubber:
188
stubber.add_response('describe_instances', response)
189
result = ec2_client.describe_instances()
190
191
assert result == response
192
```
193
194
## Parameter Validation
195
196
### Expected Parameters
197
198
Validate that methods are called with expected parameters.
199
200
```python
201
import botocore.session
202
from botocore.stub import Stubber
203
204
session = botocore.session.get_session()
205
s3_client = session.create_client('s3', region_name='us-east-1')
206
207
response = {
208
'Contents': [
209
{
210
'Key': 'test-file.txt',
211
'Size': 1024,
212
'LastModified': datetime.datetime(2020, 1, 1)
213
}
214
]
215
}
216
217
expected_params = {
218
'Bucket': 'my-test-bucket',
219
'Prefix': 'uploads/',
220
'MaxKeys': 100
221
}
222
223
with Stubber(s3_client) as stubber:
224
stubber.add_response('list_objects_v2', response, expected_params)
225
226
# This call matches expected parameters
227
result = s3_client.list_objects_v2(
228
Bucket='my-test-bucket',
229
Prefix='uploads/',
230
MaxKeys=100
231
)
232
```
233
234
### Using ANY for Flexible Matching
235
236
Ignore specific parameter values that are unpredictable.
237
238
```python
239
import botocore.session
240
from botocore.stub import Stubber, ANY
241
242
session = botocore.session.get_session()
243
dynamodb_client = session.create_client('dynamodb', region_name='us-east-1')
244
245
response = {
246
'Item': {
247
'id': {'S': 'test-id'},
248
'name': {'S': 'Test Item'}
249
}
250
}
251
252
# Use ANY for unpredictable parameters
253
expected_params = {
254
'TableName': 'my-table',
255
'Key': {'id': {'S': 'test-id'}},
256
'ConsistentRead': ANY # Don't care about this parameter
257
}
258
259
with Stubber(dynamodb_client) as stubber:
260
stubber.add_response('get_item', response, expected_params)
261
262
# ConsistentRead can be any value
263
result = dynamodb_client.get_item(
264
TableName='my-table',
265
Key={'id': {'S': 'test-id'}},
266
ConsistentRead=True # or False, doesn't matter
267
)
268
```
269
270
## Error Simulation
271
272
### Client Errors
273
274
Simulate AWS service errors for error handling tests.
275
276
```python
277
import botocore.session
278
from botocore.stub import Stubber
279
from botocore.exceptions import ClientError
280
import pytest
281
282
session = botocore.session.get_session()
283
s3_client = session.create_client('s3', region_name='us-east-1')
284
285
with Stubber(s3_client) as stubber:
286
# Add error response
287
stubber.add_client_error(
288
'get_object',
289
service_error_code='NoSuchKey',
290
service_message='The specified key does not exist.',
291
http_status_code=404
292
)
293
294
# Test error handling
295
with pytest.raises(ClientError) as exc_info:
296
s3_client.get_object(Bucket='test-bucket', Key='nonexistent.txt')
297
298
error = exc_info.value
299
assert error.response['Error']['Code'] == 'NoSuchKey'
300
assert error.response['ResponseMetadata']['HTTPStatusCode'] == 404
301
```
302
303
### Service Errors with Metadata
304
305
Add additional error metadata and response fields.
306
307
```python
308
import botocore.session
309
from botocore.stub import Stubber
310
from botocore.exceptions import ClientError
311
312
session = botocore.session.get_session()
313
s3_client = session.create_client('s3', region_name='us-east-1')
314
315
with Stubber(s3_client) as stubber:
316
stubber.add_client_error(
317
'restore_object',
318
service_error_code='InvalidObjectState',
319
service_message='Object is in invalid state',
320
http_status_code=403,
321
service_error_meta={
322
'StorageClass': 'GLACIER',
323
'ActualObjectState': 'Archived'
324
},
325
response_meta={
326
'RequestId': 'error-123',
327
'HostId': 'host-error-456'
328
}
329
)
330
331
with pytest.raises(ClientError) as exc_info:
332
s3_client.restore_object(
333
Bucket='test-bucket',
334
Key='archived-file.txt',
335
RestoreRequest={'Days': 7}
336
)
337
338
error = exc_info.value
339
assert error.response['Error']['StorageClass'] == 'GLACIER'
340
assert error.response['ResponseMetadata']['RequestId'] == 'error-123'
341
```
342
343
## Testing Frameworks Integration
344
345
### pytest Integration
346
347
Complete example using pytest for AWS service testing.
348
349
```python
350
import pytest
351
import botocore.session
352
from botocore.stub import Stubber
353
from botocore.exceptions import ClientError
354
355
@pytest.fixture
356
def s3_client():
357
"""Create S3 client for testing."""
358
session = botocore.session.get_session()
359
return session.create_client('s3', region_name='us-east-1')
360
361
@pytest.fixture
362
def s3_stubber(s3_client):
363
"""Create activated stubber for S3 client."""
364
with Stubber(s3_client) as stubber:
365
yield stubber
366
367
def test_successful_bucket_creation(s3_client, s3_stubber):
368
"""Test successful bucket creation."""
369
expected_params = {'Bucket': 'test-bucket'}
370
response = {
371
'Location': '/test-bucket',
372
'ResponseMetadata': {'HTTPStatusCode': 200}
373
}
374
375
s3_stubber.add_response('create_bucket', response, expected_params)
376
377
result = s3_client.create_bucket(Bucket='test-bucket')
378
assert result['Location'] == '/test-bucket'
379
380
def test_bucket_already_exists_error(s3_client, s3_stubber):
381
"""Test bucket creation when bucket already exists."""
382
s3_stubber.add_client_error(
383
'create_bucket',
384
service_error_code='BucketAlreadyExists',
385
service_message='The requested bucket name is not available.',
386
http_status_code=409
387
)
388
389
with pytest.raises(ClientError) as exc_info:
390
s3_client.create_bucket(Bucket='existing-bucket')
391
392
assert exc_info.value.response['Error']['Code'] == 'BucketAlreadyExists'
393
394
def test_multiple_operations(s3_client, s3_stubber):
395
"""Test multiple stubbed operations in sequence."""
396
# First operation: create bucket
397
s3_stubber.add_response(
398
'create_bucket',
399
{'Location': '/test-bucket'},
400
{'Bucket': 'test-bucket'}
401
)
402
403
# Second operation: put object
404
s3_stubber.add_response(
405
'put_object',
406
{'ETag': '"abc123"'},
407
{
408
'Bucket': 'test-bucket',
409
'Key': 'test-file.txt',
410
'Body': b'Hello, World!'
411
}
412
)
413
414
# Third operation: list objects
415
s3_stubber.add_response(
416
'list_objects_v2',
417
{
418
'Contents': [
419
{'Key': 'test-file.txt', 'Size': 13}
420
]
421
},
422
{'Bucket': 'test-bucket'}
423
)
424
425
# Execute operations in order
426
s3_client.create_bucket(Bucket='test-bucket')
427
s3_client.put_object(
428
Bucket='test-bucket',
429
Key='test-file.txt',
430
Body=b'Hello, World!'
431
)
432
result = s3_client.list_objects_v2(Bucket='test-bucket')
433
434
assert len(result['Contents']) == 1
435
assert result['Contents'][0]['Key'] == 'test-file.txt'
436
```
437
438
### unittest Integration
439
440
Using stubber with Python's built-in unittest framework.
441
442
```python
443
import unittest
444
import botocore.session
445
from botocore.stub import Stubber
446
from botocore.exceptions import ClientError
447
448
class TestS3Operations(unittest.TestCase):
449
450
def setUp(self):
451
"""Set up test fixtures before each test method."""
452
session = botocore.session.get_session()
453
self.s3_client = session.create_client('s3', region_name='us-east-1')
454
self.stubber = Stubber(self.s3_client)
455
456
def tearDown(self):
457
"""Clean up after each test method."""
458
# Ensure stubber is deactivated
459
try:
460
self.stubber.deactivate()
461
except:
462
pass # Already deactivated
463
464
def test_upload_file_success(self):
465
"""Test successful file upload."""
466
expected_params = {
467
'Bucket': 'uploads',
468
'Key': 'document.pdf',
469
'Body': unittest.mock.ANY # File content can vary
470
}
471
472
response = {
473
'ETag': '"d41d8cd98f00b204e9800998ecf8427e"',
474
'ResponseMetadata': {'HTTPStatusCode': 200}
475
}
476
477
self.stubber.add_response('put_object', response, expected_params)
478
self.stubber.activate()
479
480
result = self.s3_client.put_object(
481
Bucket='uploads',
482
Key='document.pdf',
483
Body=b'PDF content here'
484
)
485
486
self.assertEqual(result['ResponseMetadata']['HTTPStatusCode'], 200)
487
self.stubber.assert_no_pending_responses()
488
489
def test_file_not_found_error(self):
490
"""Test handling of file not found error."""
491
self.stubber.add_client_error(
492
'get_object',
493
service_error_code='NoSuchKey',
494
service_message='The specified key does not exist.'
495
)
496
self.stubber.activate()
497
498
with self.assertRaises(ClientError) as context:
499
self.s3_client.get_object(Bucket='test', Key='missing.txt')
500
501
self.assertEqual(
502
context.exception.response['Error']['Code'],
503
'NoSuchKey'
504
)
505
506
if __name__ == '__main__':
507
unittest.main()
508
```
509
510
## Complex Workflows
511
512
### Multi-Service Testing
513
514
Test workflows involving multiple AWS services.
515
516
```python
517
import botocore.session
518
from botocore.stub import Stubber
519
import pytest
520
521
class AWSWorkflowTest:
522
"""Test complex workflow across multiple AWS services."""
523
524
def setup_method(self):
525
"""Set up clients and stubbers for each test."""
526
session = botocore.session.get_session()
527
528
# Create clients
529
self.s3_client = session.create_client('s3', region_name='us-east-1')
530
self.lambda_client = session.create_client('lambda', region_name='us-east-1')
531
self.sns_client = session.create_client('sns', region_name='us-east-1')
532
533
# Create stubbers
534
self.s3_stubber = Stubber(self.s3_client)
535
self.lambda_stubber = Stubber(self.lambda_client)
536
self.sns_stubber = Stubber(self.sns_client)
537
538
def test_file_processing_workflow(self):
539
"""Test complete file processing workflow."""
540
# 1. Upload file to S3
541
self.s3_stubber.add_response(
542
'put_object',
543
{
544
'ETag': '"abc123"',
545
'ResponseMetadata': {'HTTPStatusCode': 200}
546
},
547
{
548
'Bucket': 'processing-bucket',
549
'Key': 'input/data.csv',
550
'Body': ANY
551
}
552
)
553
554
# 2. Invoke Lambda function
555
self.lambda_stubber.add_response(
556
'invoke',
557
{
558
'StatusCode': 200,
559
'Payload': b'{"status": "success", "records_processed": 1000}'
560
},
561
{
562
'FunctionName': 'process-data-function',
563
'Payload': ANY
564
}
565
)
566
567
# 3. Send notification
568
self.sns_stubber.add_response(
569
'publish',
570
{
571
'MessageId': 'msg-123',
572
'ResponseMetadata': {'HTTPStatusCode': 200}
573
},
574
{
575
'TopicArn': 'arn:aws:sns:us-east-1:123456789012:processing-complete',
576
'Message': ANY
577
}
578
)
579
580
# Activate all stubbers
581
with self.s3_stubber, self.lambda_stubber, self.sns_stubber:
582
# Execute workflow
583
self.s3_client.put_object(
584
Bucket='processing-bucket',
585
Key='input/data.csv',
586
Body=b'name,age\nJohn,30\nJane,25'
587
)
588
589
lambda_response = self.lambda_client.invoke(
590
FunctionName='process-data-function',
591
Payload='{"source": "s3://processing-bucket/input/data.csv"}'
592
)
593
594
self.sns_client.publish(
595
TopicArn='arn:aws:sns:us-east-1:123456789012:processing-complete',
596
Message='Data processing completed successfully'
597
)
598
599
# Verify Lambda response
600
assert lambda_response['StatusCode'] == 200
601
```
602
603
### Pagination Testing
604
605
Test paginated operations with multiple page responses.
606
607
```python
608
import botocore.session
609
from botocore.stub import Stubber
610
611
def test_paginated_list_objects():
612
"""Test paginated S3 list operations."""
613
session = botocore.session.get_session()
614
s3_client = session.create_client('s3', region_name='us-east-1')
615
616
with Stubber(s3_client) as stubber:
617
# First page
618
stubber.add_response(
619
'list_objects_v2',
620
{
621
'Contents': [
622
{'Key': f'file-{i}.txt', 'Size': 100}
623
for i in range(1000)
624
],
625
'IsTruncated': True,
626
'NextContinuationToken': 'token-123'
627
},
628
{'Bucket': 'large-bucket', 'MaxKeys': 1000}
629
)
630
631
# Second page
632
stubber.add_response(
633
'list_objects_v2',
634
{
635
'Contents': [
636
{'Key': f'file-{i}.txt', 'Size': 100}
637
for i in range(1000, 1500)
638
],
639
'IsTruncated': False
640
},
641
{
642
'Bucket': 'large-bucket',
643
'MaxKeys': 1000,
644
'ContinuationToken': 'token-123'
645
}
646
)
647
648
# Test pagination manually
649
response1 = s3_client.list_objects_v2(
650
Bucket='large-bucket',
651
MaxKeys=1000
652
)
653
assert len(response1['Contents']) == 1000
654
assert response1['IsTruncated'] is True
655
656
response2 = s3_client.list_objects_v2(
657
Bucket='large-bucket',
658
MaxKeys=1000,
659
ContinuationToken=response1['NextContinuationToken']
660
)
661
assert len(response2['Contents']) == 500
662
assert response2['IsTruncated'] is False
663
```
664
665
## Best Practices
666
667
### Test Organization
668
669
Structure your tests for maintainability and clarity.
670
671
```python
672
import botocore.session
673
from botocore.stub import Stubber
674
import pytest
675
676
class TestDataProcessing:
677
"""Group related tests in classes."""
678
679
@pytest.fixture(autouse=True)
680
def setup_clients(self):
681
"""Automatically set up clients for all tests."""
682
session = botocore.session.get_session()
683
self.s3_client = session.create_client('s3', region_name='us-east-1')
684
self.dynamodb_client = session.create_client('dynamodb', region_name='us-east-1')
685
686
def test_valid_data_processing(self):
687
"""Test processing with valid data."""
688
with Stubber(self.s3_client) as s3_stubber:
689
# Add S3 stubs for valid data scenario
690
s3_stubber.add_response('get_object', self._valid_data_response())
691
692
# Test implementation here
693
pass
694
695
def test_invalid_data_handling(self):
696
"""Test processing with invalid data."""
697
with Stubber(self.s3_client) as s3_stubber:
698
# Add S3 stubs for invalid data scenario
699
s3_stubber.add_client_error(
700
'get_object',
701
service_error_code='NoSuchKey',
702
service_message='File not found'
703
)
704
705
# Test error handling here
706
pass
707
708
def _valid_data_response(self):
709
"""Helper method for consistent test data."""
710
return {
711
'Body': MockStreamingBody(b'valid,csv,data'),
712
'ContentLength': 15,
713
'ResponseMetadata': {'HTTPStatusCode': 200}
714
}
715
716
class MockStreamingBody:
717
"""Mock streaming body for S3 responses."""
718
719
def __init__(self, content):
720
self._content = content
721
722
def read(self, amt=None):
723
return self._content
724
725
def close(self):
726
pass
727
```
728
729
### Stub Data Management
730
731
Create reusable stub data for consistent testing.
732
733
```python
734
# test_data.py - Centralized test data
735
class S3TestData:
736
"""Centralized S3 test response data."""
737
738
@staticmethod
739
def list_buckets_response():
740
return {
741
'Buckets': [
742
{'Name': 'bucket-1', 'CreationDate': datetime.datetime(2020, 1, 1)},
743
{'Name': 'bucket-2', 'CreationDate': datetime.datetime(2020, 2, 1)}
744
],
745
'ResponseMetadata': {'HTTPStatusCode': 200}
746
}
747
748
@staticmethod
749
def empty_bucket_response():
750
return {
751
'Contents': [],
752
'ResponseMetadata': {'HTTPStatusCode': 200}
753
}
754
755
@staticmethod
756
def access_denied_error():
757
return {
758
'service_error_code': 'AccessDenied',
759
'service_message': 'Access Denied',
760
'http_status_code': 403
761
}
762
763
# test_s3_operations.py - Using centralized data
764
from test_data import S3TestData
765
766
def test_bucket_listing(s3_client):
767
"""Test using centralized test data."""
768
with Stubber(s3_client) as stubber:
769
stubber.add_response('list_buckets', S3TestData.list_buckets_response())
770
771
result = s3_client.list_buckets()
772
assert len(result['Buckets']) == 2
773
```
774
775
### Error Testing Patterns
776
777
Comprehensive error condition testing.
778
779
```python
780
import pytest
781
from botocore.exceptions import ClientError
782
783
class TestErrorScenarios:
784
"""Test various AWS error conditions."""
785
786
@pytest.mark.parametrize("error_code,http_status,expected_action", [
787
('NoSuchBucket', 404, 'create_bucket'),
788
('AccessDenied', 403, 'check_permissions'),
789
('InternalError', 500, 'retry_operation'),
790
])
791
def test_error_handling_strategies(self, s3_client, error_code, http_status, expected_action):
792
"""Test different error handling strategies."""
793
with Stubber(s3_client) as stubber:
794
stubber.add_client_error(
795
'get_object',
796
service_error_code=error_code,
797
http_status_code=http_status
798
)
799
800
with pytest.raises(ClientError) as exc_info:
801
s3_client.get_object(Bucket='test-bucket', Key='test-key')
802
803
error = exc_info.value
804
assert error.response['Error']['Code'] == error_code
805
assert error.response['ResponseMetadata']['HTTPStatusCode'] == http_status
806
807
# Test appropriate error handling strategy
808
# Implementation would call appropriate handler based on expected_action
809
```
810
811
### Validation and Cleanup
812
813
Ensure thorough test validation and cleanup.
814
815
```python
816
def test_complete_workflow_with_validation():
817
"""Test with comprehensive validation and cleanup."""
818
session = botocore.session.get_session()
819
s3_client = session.create_client('s3', region_name='us-east-1')
820
821
stubber = Stubber(s3_client)
822
823
try:
824
# Add multiple responses
825
stubber.add_response('create_bucket', {'Location': '/test-bucket'})
826
stubber.add_response('put_object', {'ETag': '"abc123"'})
827
stubber.add_response('get_object', {'Body': MockStreamingBody(b'test')})
828
829
stubber.activate()
830
831
# Execute operations
832
s3_client.create_bucket(Bucket='test-bucket')
833
s3_client.put_object(Bucket='test-bucket', Key='test.txt', Body=b'test')
834
result = s3_client.get_object(Bucket='test-bucket', Key='test.txt')
835
836
# Validate results
837
assert result['Body'].read() == b'test'
838
839
# Ensure all stubs were used
840
stubber.assert_no_pending_responses()
841
842
finally:
843
# Always clean up
844
stubber.deactivate()
845
```
846
847
The testing framework provides comprehensive support for testing AWS service interactions without making actual API calls, enabling fast, reliable unit tests that validate both successful operations and error conditions.