Add tests for GCS Storage

scrapy/utils/test.py

@@ -20,6 +20,12 @@ def assert_aws_environ():
     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")
+
+
 def skip_if_no_boto():
     try:
         is_botocore()
@@ -45,6 +51,15 @@ def get_s3_content_and_delete(bucket, path, with_key=False):
         bucket.delete_key(path)
     return (content, key) if with_key else content
 
+def get_gcs_content_and_delete(bucket, path):
+    from google.cloud import storage
+    client = storage.Client(project=os.environ.get('GCS_PROJECT_ID'))
+    bucket = client.get_bucket(bucket)
+    blob = bucket.get_blob(path)
+    content = blob.download_as_string()
+    bucket.delete_blob(path)
+    return content, blob
+
 def get_crawler(spidercls=None, settings_dict=None):
     """Return an unconfigured Crawler object. If settings_dict is given, it
     will be used to populate the crawler settings with a project level

tests/test_pipeline_files.py

@@ -11,12 +11,13 @@ from six import BytesIO
 from twisted.trial import unittest
 from twisted.internet import defer
 
-from scrapy.pipelines.files import FilesPipeline, FSFilesStore, S3FilesStore
+from scrapy.pipelines.files import FilesPipeline, FSFilesStore, S3FilesStore, GCSFilesStore
 from scrapy.item import Item, Field
 from scrapy.http import Request, Response
 from scrapy.settings import Settings
 from scrapy.utils.python import to_bytes
 from scrapy.utils.test import assert_aws_environ, get_s3_content_and_delete
+from scrapy.utils.test import assert_gcs_environ, get_gcs_content_and_delete
 from scrapy.utils.boto import is_botocore
 
 from tests import mock
@@ -375,6 +376,31 @@ class TestS3FilesStore(unittest.TestCase):
             self.assertEqual(key.content_type, 'image/png')
 
 
+class TestGCSFilesStore(unittest.TestCase):
+    @defer.inlineCallbacks
+    def test_persist(self):
+        assert_gcs_environ()
+        uri = os.environ.get('GCS_TEST_FILE_URI')
+        if not uri:
+            raise unittest.SkipTest("No GCS URI available for testing")
+        data = b"TestGCSFilesStore: \xe2\x98\x83"
+        buf = BytesIO(data)
+        meta = {'foo': 'bar'}
+        path = 'full/filename'
+        store = GCSFilesStore(uri)
+        yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
+        s = yield store.stat_file(path, info=None)
+        self.assertIn('last_modified', s)
+        self.assertIn('checksum', s)
+        self.assertEqual(s['checksum'], 'zc2oVgXkbQr2EQdSdw3OPA==')
+        u = urlparse(uri)
+        content, blob = get_gcs_content_and_delete(u.hostname, u.path[1:]+path)
+        self.assertEqual(content, data)
+        self.assertEqual(blob.metadata, {'foo': 'bar'})
+        self.assertEqual(blob.cache_control, GCSFilesStore.CACHE_CONTROL)
+        self.assertEqual(blob.content_type, 'application/octet-stream')
+
+
 class ItemWithFiles(Item):
     file_urls = Field()
     files = Field()

tox.ini

@@ -11,6 +11,7 @@ deps =
     -rrequirements.txt
     # Extras
     botocore
+    google-cloud-storage
     Pillow != 3.0.0
     leveldb
     -rtests/requirements.txt
@@ -18,6 +19,8 @@ passenv =
     S3_TEST_FILE_URI
     AWS_ACCESS_KEY_ID
     AWS_SECRET_ACCESS_KEY
+    GCS_TEST_FILE_URI
+    GCS_PROJECT_ID
 commands =
     py.test --cov=scrapy --cov-report= {posargs:scrapy tests}