Add system tests for blob generations.

storage/tests/system.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import io
 import os
 import tempfile
 import re
@@ -78,6 +79,7 @@ def setUpModule():
     # In the **very** rare case the bucket name is reserved, this
     # fails with a ConnectionError.
     Config.TEST_BUCKET = Config.CLIENT.bucket(bucket_name)
+    Config.TEST_BUCKET.versioning_enabled = True
     retry_429(Config.TEST_BUCKET.create)()
 
 
@@ -414,29 +416,50 @@ def test_crud_blob_w_user_project(self):
 
         # Exercise 'objects.insert' w/ userProject.
         blob.upload_from_filename(file_data["path"])
+        gen0 = blob.generation
+
+        # Upload a second generation of the blob
+        blob.upload_from_file(io.StringIO(six.text_type('gen1')))
+        gen1 = blob.generation
+
+        blob0 = with_user_project.blob("SmallFile", generation=gen0)
+        blob1 = with_user_project.blob("SmallFile", generation=gen1)
+
+        # Exercise 'objects.get' w/ generation
+        self.assertEqual(with_user_project.get_blob(blob.name).generation, gen1)
+        self.assertEqual(with_user_project.get_blob(blob.name, generation=gen0).generation, gen0)
 
         try:
             # Exercise 'objects.get' (metadata) w/ userProject.
             self.assertTrue(blob.exists())
             blob.reload()
 
             # Exercise 'objects.get' (media) w/ userProject.
-            downloaded = blob.download_as_string()
-            self.assertEqual(downloaded, file_contents)
+            self.assertEqual(blob0.download_as_string(), file_contents)
+            self.assertEqual(blob1.download_as_string(), "gen1")
 
             # Exercise 'objects.patch' w/ userProject.
-            blob.content_language = "en"
-            blob.patch()
-            self.assertEqual(blob.content_language, "en")
+            blob0.content_language = "en"
+            blob0.patch()
+            self.assertEqual(blob0.content_language, "en")
+            self.assertIsNone(blob1.content_language)
 
             # Exercise 'objects.update' w/ userProject.
             metadata = {"foo": "Foo", "bar": "Bar"}
-            blob.metadata = metadata
-            blob.update()
-            self.assertEqual(blob.metadata, metadata)
+            blob0.metadata = metadata
+            blob0.update()
+            self.assertEqual(blob0.metadata, metadata)
+            self.assertIsNone(blob1.metadata)
         finally:
             # Exercise 'objects.delete' (metadata) w/ userProject.
-            blob.delete()
+            blobs = with_user_project.list_blobs(prefix=blob.name, versions=True)
+            self.assertEqual([each.generation for each in blobs], [gen0, gen1])
+
+            blob0.delete()
+            blobs = with_user_project.list_blobs(prefix=blob.name, versions=True)
+            self.assertEqual([each.generation for each in blobs], [gen1])
+
+            blob1.delete()
 
     @unittest.skipUnless(USER_PROJECT, "USER_PROJECT not set in environment.")
     def test_blob_acl_w_user_project(self):

storage/tests/unit/test_bucket.py

@@ -256,6 +256,21 @@ def test_blob_w_encryption_key(self):
         self.assertEqual(blob._encryption_key, KEY)
         self.assertIsNone(blob.kms_key_name)
 
+    def test_blob_w_generation(self):
+        from google.cloud.storage.blob import Blob
+
+        BUCKET_NAME = "BUCKET_NAME"
+        BLOB_NAME = "BLOB_NAME"
+        GENERATION = 123
+
+        bucket = self._make_one(name=BUCKET_NAME)
+        blob = bucket.blob(BLOB_NAME, generation=GENERATION)
+        self.assertIsInstance(blob, Blob)
+        self.assertIs(blob.bucket, bucket)
+        self.assertIs(blob.client, bucket.client)
+        self.assertEqual(blob.name, BLOB_NAME)
+        self.assertEqual(blob.generation, GENERATION)
+
     def test_blob_w_kms_key_name(self):
         from google.cloud.storage.blob import Blob