diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py index 27fcc3996c02..c84d45e358ea 100644 --- 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 index 28d1c4160b1e..705e5d9f28cf 100644 --- 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