Add blob properties to support retention policy feature. (#446)

tseaver · tseaver · commit cee0807da184 · 2018-08-28T14:27:24.000-04:00
Toward #445.
diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py
@@ -1590,6 +1590,16 @@ def etag(self):
         """
         return self._properties.get('etag')
 
+    event_based_hold = _scalar_property('eventBasedHold')
+    """Is an event-based hold active on the object?
+
+    See `API reference docs`_.
+
+    If the property is not set locally, returns :data:`None`.
+
+    :rtype: bool or ``NoneType``
+    """
+
     @property
     def generation(self):
         """Retrieve the generation for the object.
@@ -1696,6 +1706,20 @@ def owner(self):
         """
         return copy.deepcopy(self._properties.get('owner'))
 
+    @property
+    def retention_expiration_time(self):
+        """Retrieve timestamp at which the object's retention period expires.
+
+        See https://cloud.google.com/storage/docs/json_api/v1/objects
+
+        :rtype: :class:`datetime.datetime` or ``NoneType``
+        :returns: Datetime object parsed from RFC3339 valid timestamp, or
+                  ``None`` if the property is not set locally.
+        """
+        value = self._properties.get('retentionExpirationTime')
+        if value is not None:
+            return _rfc3339_to_datetime(value)
+
     @property
     def self_link(self):
         """Retrieve the URI for the object.
@@ -1749,6 +1773,16 @@ def kms_key_name(self):
               "DURABLE_REDUCED_AVAILABILITY", else ``None``.
     """
 
+    temporary_hold = _scalar_property('temporaryHold')
+    """Is a temporary hold active on the object?
+
+    See `API reference docs`_.
+
+    If the property is not set locally, returns :data:`None`.
+
+    :rtype: bool or ``NoneType``
+    """
+
     @property
     def time_deleted(self):
         """Retrieve the timestamp at which the object was deleted.
diff --git a/storage/tests/unit/test_blob.py b/storage/tests/unit/test_blob.py
@@ -2667,6 +2667,35 @@ def test_etag(self):
         blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
         self.assertEqual(blob.etag, ETAG)
 
+    def test_event_based_hold_getter_missing(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertIsNone(blob.event_based_hold)
+
+    def test_event_based_hold_getter_false(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {'eventBasedHold': False}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertFalse(blob.event_based_hold)
+
+    def test_event_based_hold_getter_true(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {'eventBasedHold': True}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertTrue(blob.event_based_hold)
+
+    def test_event_based_hold_setter(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+        self.assertIsNone(blob.event_based_hold)
+        blob.event_based_hold = True
+        self.assertEqual(blob.event_based_hold, True)
+
     def test_generation(self):
         BUCKET = object()
         GENERATION = 42
@@ -2766,6 +2795,23 @@ def test_owner(self):
         self.assertEqual(owner['entity'], 'project-owner-12345')
         self.assertEqual(owner['entityId'], '23456')
 
+    def test_retention_expiration_time(self):
+        from google.cloud._helpers import _RFC3339_MICROS
+        from google.cloud._helpers import UTC
+
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        TIMESTAMP = datetime.datetime(2014, 11, 5, 20, 34, 37, tzinfo=UTC)
+        TIME_CREATED = TIMESTAMP.strftime(_RFC3339_MICROS)
+        properties = {'retentionExpirationTime': TIME_CREATED}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertEqual(blob.retention_expiration_time, TIMESTAMP)
+
+    def test_retention_expiration_time_unset(self):
+        BUCKET = object()
+        blob = self._make_one('blob-name', bucket=BUCKET)
+        self.assertIsNone(blob.retention_expiration_time)
+
     def test_self_link(self):
         BLOB_NAME = 'blob-name'
         bucket = _Bucket()
@@ -2811,6 +2857,35 @@ def test_storage_class_setter(self):
         self.assertEqual(blob.storage_class, storage_class)
         self.assertEqual(blob._properties, {'storageClass': storage_class})
 
+    def test_temporary_hold_getter_missing(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertIsNone(blob.temporary_hold)
+
+    def test_temporary_hold_getter_false(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {'temporaryHold': False}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertFalse(blob.temporary_hold)
+
+    def test_temporary_hold_getter_true(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        properties = {'temporaryHold': True}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
+        self.assertTrue(blob.temporary_hold)
+
+    def test_temporary_hold_setter(self):
+        BLOB_NAME = 'blob-name'
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+        self.assertIsNone(blob.temporary_hold)
+        blob.temporary_hold = True
+        self.assertEqual(blob.temporary_hold, True)
+
     def test_time_deleted(self):
         from google.cloud._helpers import _RFC3339_MICROS
         from google.cloud._helpers import UTC
