Ensure presignings do not get processed by stubber

botocore/signers.py

@@ -557,7 +557,9 @@ def generate_presigned_url(self, ClientMethod, Params=None, ExpiresIn=3600,
         params = {}
     expires_in = ExpiresIn
     http_method = HttpMethod
-    context = {}
+    context = {
+        'is_presign_request': True
+    }
 
     request_signer = self._request_signer
     serializer = self._serializer

botocore/stub.py

@@ -331,12 +331,14 @@ def _assert_expected_call_order(self, model, params):
                 operation_name=model.name,
                 reason='Operation mismatch: found response for %s.' % name)
 
-    def _get_response_handler(self, model, params, **kwargs):
+    def _get_response_handler(self, model, params, context, **kwargs):
         self._assert_expected_call_order(model, params)
         # Pop off the entire response once everything has been validated
         return self._queue.popleft()['response']
 
-    def _assert_expected_params(self, model, params, **kwargs):
+    def _assert_expected_params(self, model, params, context, **kwargs):
+        if self._should_not_stub(context):
+            return
         self._assert_expected_call_order(model, params)
         expected_params = self._queue[0]['expected_params']
         if expected_params is None:
@@ -357,6 +359,13 @@ def _assert_expected_params(self, model, params, **kwargs):
                 reason='Expected parameters:\n%s,\nbut received:\n%s' % (
                     pformat(expected_params), pformat(params)))
 
+    def _should_not_stub(self, context):
+        # Do not include presign requests when processing stubbed client calls
+        # as a presign request will never have an HTTP request sent over the
+        # wire for it and therefore not receive a response back.
+        if context and context.get('is_presign_request'):
+            return True
+
     def _validate_response(self, operation_name, service_response):
         service_model = self.client.meta.service_model
         operation_model = service_model.operation_model(operation_name)

tests/functional/test_sts.py

@@ -23,6 +23,9 @@ class TestSTSPresignedUrl(BaseSessionTest):
     def setUp(self):
         super(TestSTSPresignedUrl, self).setUp()
         self.client = self.session.create_client('sts', 'us-west-2')
+        # Makes sure that no requests will go through
+        self.stubber = Stubber(self.client)
+        self.stubber.activate()
 
     def test_presigned_url_contains_no_content_type(self):
         timestamp = datetime(2017, 3, 22, 0, 0)

tests/functional/test_stub.py

@@ -28,9 +28,12 @@
 class TestStubber(unittest.TestCase):
     def setUp(self):
         session = botocore.session.get_session()
-        config = botocore.config.Config(signature_version=botocore.UNSIGNED)
-        self.client = session.create_client('s3', config=config)
-
+        config = botocore.config.Config(
+            signature_version=botocore.UNSIGNED,
+            s3={'addressing_style': 'path'}
+        )
+        self.client = session.create_client(
+            's3', region_name='us-east-1', config=config)
         self.stubber = Stubber(self.client)
 
     def test_stubber_returns_response(self):
@@ -270,3 +273,43 @@ def test_many_expected_params(self):
         except StubAssertionError:
             self.fail(
                 "Stubber inappropriately raised error for same parameters.")
+
+    def test_no_stub_for_presign_url(self):
+        try:
+            with self.stubber:
+                url = self.client.generate_presigned_url(
+                    ClientMethod='get_object',
+                    Params={
+                        'Bucket': 'mybucket',
+                        'Key': 'mykey'
+                    }
+                )
+                self.assertEqual(
+                    url, 'https://s3.amazonaws.com/mybucket/mykey')
+        except StubResponseError:
+            self.fail(
+                'Stubbed responses should not be required for generating '
+                'presigned requests'
+            )
+
+    def test_can_stub_with_presign_url_mixed_in(self):
+        desired_response = {}
+        expected_params = {
+            'Bucket': 'mybucket',
+            'Prefix': 'myprefix',
+        }
+        self.stubber.add_response(
+            'list_objects', desired_response, expected_params)
+        with self.stubber:
+            url = self.client.generate_presigned_url(
+                ClientMethod='get_object',
+                Params={
+                    'Bucket': 'myotherbucket',
+                    'Key': 'myotherkey'
+                }
+            )
+            self.assertEqual(
+                    url, 'https://s3.amazonaws.com/myotherbucket/myotherkey')
+            actual_response = self.client.list_objects(**expected_params)
+            self.assertEqual(desired_response, actual_response)
+        self.stubber.assert_no_pending_responses()

tests/unit/test_signers.py

@@ -815,6 +815,22 @@ def test_generate_presigned_url_emits_param_events(self):
             ]
         )
 
+    def test_generate_presign_url_emits_is_presign_in_context(self):
+        emitter = mock.Mock(HierarchicalEmitter)
+        emitter.emit.return_value = []
+        self.client.meta.events = emitter
+        self.client.generate_presigned_url(
+            'get_object', Params={'Bucket': self.bucket, 'Key': self.key})
+        kwargs_emitted = [
+            emit_call[1] for emit_call in emitter.emit.call_args_list
+        ]
+        for kwargs in kwargs_emitted:
+            self.assertTrue(
+                kwargs.get('context', {}).get('is_presign_request'),
+                'The context did not have is_presign_request set to True for '
+                'the following kwargs emitted: %s' % kwargs
+            )
+
 
 class TestGeneratePresignedPost(unittest.TestCase):
     def setUp(self):