(
self,
mock_firestore_ttl,
mock_get_blob,
mock_bucket_prop,
base_path,
gcs_ttl,
)
| 120 | @patch.object(GCSBackend, '_get_blob') |
| 121 | @patch.object(GCSBackend, '_is_firestore_ttl_policy_enabled') |
| 122 | def test_set_key( |
| 123 | self, |
| 124 | mock_firestore_ttl, |
| 125 | mock_get_blob, |
| 126 | mock_bucket_prop, |
| 127 | base_path, |
| 128 | gcs_ttl, |
| 129 | ): |
| 130 | self.app.conf.gcs_base_path = base_path |
| 131 | self.app.conf.gcs_ttl = gcs_ttl |
| 132 | |
| 133 | mock_firestore_ttl.return_value = True |
| 134 | mock_blob = Mock() |
| 135 | mock_get_blob.return_value = mock_blob |
| 136 | mock_bucket_prop.lifecycle_rules = [{'action': {'type': 'Delete'}}] |
| 137 | backend = GCSBackend(app=self.app) |
| 138 | backend.set('testkey', 'testvalue') |
| 139 | mock_get_blob.assert_called_once_with('testkey') |
| 140 | mock_blob.upload_from_string.assert_called_once_with( |
| 141 | 'testvalue', retry=backend._retry_policy |
| 142 | ) |
| 143 | if gcs_ttl: |
| 144 | assert mock_blob.custom_time >= datetime.utcnow() |
| 145 | |
| 146 | @patch.object(GCSBackend, '_get_blob') |
| 147 | @patch.object(GCSBackend, '_is_firestore_ttl_policy_enabled') |
nothing calls this directly
no test coverage detected