Use mocking for tests/test_pipeline_files.py::TestS3FilesStore::test_persist

scrapy/utils/test.py

@@ -13,15 +13,6 @@ from twisted.trial.unittest import SkipTest
 from scrapy.utils.boto import is_botocore_available
 
 
-def assert_aws_environ():
-    """Asserts the current environment is suitable for running AWS testsi.
-    Raises SkipTest with the reason if it's not.
-    """
-    skip_if_no_boto()
-    if 'AWS_ACCESS_KEY_ID' not in os.environ:
-        raise SkipTest("AWS keys not found")
-
-
 def assert_gcs_environ():
     if 'GCS_PROJECT_ID' not in os.environ:
         raise SkipTest("GCS_PROJECT_ID not found")

tests/test_pipeline_files.py

@@ -1,6 +1,7 @@
 import os
 import random
 import time
+from datetime import datetime
 from io import BytesIO
 from shutil import rmtree
 from tempfile import mkdtemp
@@ -23,11 +24,11 @@ from scrapy.pipelines.files import (
 )
 from scrapy.settings import Settings
 from scrapy.utils.test import (
-    assert_aws_environ,
     assert_gcs_environ,
     get_ftp_content_and_delete,
     get_gcs_content_and_delete,
     get_s3_content_and_delete,
+    skip_if_no_boto,
 )
 
 
@@ -414,32 +415,87 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
 
 
 class TestS3FilesStore(unittest.TestCase):
+
     @defer.inlineCallbacks
     def test_persist(self):
-        assert_aws_environ()
-        uri = os.environ.get('S3_TEST_FILE_URI')
-        if not uri:
-            raise unittest.SkipTest("No S3 URI available for testing")
-        data = b"TestS3FilesStore: \xe2\x98\x83"
-        buf = BytesIO(data)
+        skip_if_no_boto()
+
+        bucket = 'mybucket'
+        key = 'export.csv'
+        uri = f's3://{bucket}/{key}'
+        buffer = mock.MagicMock()
         meta = {'foo': 'bar'}
         path = ''
+        content_type = 'image/png'
+
         store = S3FilesStore(uri)
-        yield store.persist_file(
-            path, buf, info=None, meta=meta,
-            headers={'Content-Type': 'image/png'})
-        s = yield store.stat_file(path, info=None)
-        self.assertIn('last_modified', s)
-        self.assertIn('checksum', s)
-        self.assertEqual(s['checksum'], '3187896a9657a28163abb31667df64c8')
-        u = urlparse(uri)
-        content, key = get_s3_content_and_delete(
-            u.hostname, u.path[1:], with_key=True)
-        self.assertEqual(content, data)
-        self.assertEqual(key['Metadata'], {'foo': 'bar'})
-        self.assertEqual(
-            key['CacheControl'], S3FilesStore.HEADERS['Cache-Control'])
-        self.assertEqual(key['ContentType'], 'image/png')
+        from botocore.stub import Stubber
+        with Stubber(store.s3_client) as stub:
+            stub.add_response(
+                'put_object',
+                expected_params={
+                    'ACL': S3FilesStore.POLICY,
+                    'Body': buffer,
+                    'Bucket': bucket,
+                    'CacheControl': S3FilesStore.HEADERS['Cache-Control'],
+                    'ContentType': content_type,
+                    'Key': key,
+                    'Metadata': meta,
+                },
+                service_response={},
+            )
+
+            yield store.persist_file(
+                path,
+                buffer,
+                info=None,
+                meta=meta,
+                headers={'Content-Type': content_type},
+            )
+
+            stub.assert_no_pending_responses()
+            self.assertEqual(
+                buffer.method_calls,
+                [
+                    mock.call.seek(0),
+                    # The call to read does not happen with Stubber
+                ]
+            )
+
+    @defer.inlineCallbacks
+    def test_stat(self):
+        skip_if_no_boto()
+
+        bucket = 'mybucket'
+        key = 'export.csv'
+        uri = f's3://{bucket}/{key}'
+        checksum = '3187896a9657a28163abb31667df64c8'
+
+        store = S3FilesStore(uri)
+        from botocore.stub import Stubber
+        with Stubber(store.s3_client) as stub:
+            stub.add_response(
+                'head_object',
+                expected_params={
+                    'Bucket': bucket,
+                    'Key': key,
+                },
+                service_response={
+                    'ETag': f'"{checksum}"',
+                    'LastModified': datetime(2019, 12, 1),
+                },
+            )
+
+            file_stats = yield store.stat_file('', info=None)
+            self.assertEqual(
+                file_stats,
+                {
+                    'checksum': checksum,
+                    'last_modified': 1575154800,
+                },
+            )
+
+            stub.assert_no_pending_responses()
 
 
 class TestGCSFilesStore(unittest.TestCase):