|
12 | 12 | # See the License for the specific language governing permissions and
|
13 | 13 |
|
14 | 14 | import base64
|
| 15 | +from datetime import timedelta |
15 | 16 | import os
|
16 | 17 | import time
|
17 | 18 | from typing import Iterator, Optional, Tuple, Union
|
18 | 19 | import uuid
|
19 | 20 |
|
20 | 21 | from google.api_core import exceptions, retry
|
21 | 22 | from google.cloud import secretmanager
|
| 23 | +from google.protobuf.duration_pb2 import Duration |
| 24 | + |
22 | 25 | import pytest
|
23 | 26 |
|
24 | 27 | from access_secret_version import access_secret_version
|
25 | 28 | from add_secret_version import add_secret_version
|
26 | 29 | from consume_event_notification import consume_event_notification
|
27 | 30 | from create_secret import create_secret
|
28 | 31 | from create_secret_with_annotations import create_secret_with_annotations
|
| 32 | +from create_secret_with_delayed_destroy import create_secret_with_delayed_destroy |
29 | 33 | from create_secret_with_labels import create_secret_with_labels
|
30 | 34 | from create_secret_with_user_managed_replication import create_ummr_secret
|
31 | 35 | from create_update_secret_label import create_update_secret_label
|
|
36 | 40 | from destroy_secret_version_with_etag import destroy_secret_version_with_etag
|
37 | 41 | from disable_secret_version import disable_secret_version
|
38 | 42 | from disable_secret_version_with_etag import disable_secret_version_with_etag
|
| 43 | +from disable_secret_with_delayed_destroy import disable_secret_with_delayed_destroy |
39 | 44 | from edit_secret_annotations import edit_secret_annotations
|
40 | 45 | from enable_secret_version import enable_secret_version
|
41 | 46 | from enable_secret_version_with_etag import enable_secret_version_with_etag
|
|
50 | 55 | from quickstart import quickstart
|
51 | 56 | from update_secret import update_secret
|
52 | 57 | from update_secret_with_alias import update_secret_with_alias
|
| 58 | +from update_secret_with_delayed_destroy import update_secret_with_delayed_destroy |
53 | 59 | from update_secret_with_etag import update_secret_with_etag
|
54 | 60 | from view_secret_annotations import view_secret_annotations
|
55 | 61 | from view_secret_labels import view_secret_labels
|
@@ -95,6 +101,11 @@ def annotation_value() -> str:
|
95 | 101 | return "annotationvalue"
|
96 | 102 |
|
97 | 103 |
|
| 104 | +@pytest.fixture() |
| 105 | +def version_destroy_ttl() -> str: |
| 106 | + return 604800 # 7 days in seconds |
| 107 | + |
| 108 | + |
98 | 109 | @retry.Retry()
|
99 | 110 | def retry_client_create_secret(
|
100 | 111 | client: secretmanager.SecretManagerServiceClient,
|
@@ -180,6 +191,33 @@ def secret(
|
180 | 191 | yield project_id, secret_id, secret.etag
|
181 | 192 |
|
182 | 193 |
|
| 194 | +@pytest.fixture() |
| 195 | +def secret_with_delayed_destroy( |
| 196 | + client: secretmanager.SecretManagerServiceClient, |
| 197 | + project_id: str, |
| 198 | + secret_id: str, |
| 199 | + version_destroy_ttl: int, |
| 200 | + ttl: Optional[str], |
| 201 | +) -> Iterator[Tuple[str, str]]: |
| 202 | + print("creating secret with given secret id.") |
| 203 | + |
| 204 | + parent = f"projects/{project_id}" |
| 205 | + time.sleep(5) |
| 206 | + retry_client_create_secret( |
| 207 | + client, |
| 208 | + request={ |
| 209 | + "parent": parent, |
| 210 | + "secret_id": secret_id, |
| 211 | + "secret": { |
| 212 | + "replication": {"automatic": {}}, |
| 213 | + "version_destroy_ttl": Duration(seconds=version_destroy_ttl), |
| 214 | + }, |
| 215 | + }, |
| 216 | + ) |
| 217 | + |
| 218 | + yield project_id, secret_id |
| 219 | + |
| 220 | + |
183 | 221 | @pytest.fixture()
|
184 | 222 | def secret_version(
|
185 | 223 | client: secretmanager.SecretManagerServiceClient, secret: Tuple[str, str, str]
|
@@ -288,6 +326,15 @@ def test_create_secret_with_annotations(
|
288 | 326 | assert secret_id in secret.name
|
289 | 327 |
|
290 | 328 |
|
| 329 | +def test_create_secret_with_delayed_destroy( |
| 330 | + client: secretmanager.SecretManagerServiceClient, |
| 331 | + project_id: str, secret_id: str, version_destroy_ttl: int |
| 332 | +) -> None: |
| 333 | + secret = create_secret_with_delayed_destroy(project_id, secret_id, version_destroy_ttl) |
| 334 | + assert secret_id in secret.name |
| 335 | + assert timedelta(seconds=version_destroy_ttl) == secret.version_destroy_ttl |
| 336 | + |
| 337 | + |
291 | 338 | def test_delete_secret(
|
292 | 339 | client: secretmanager.SecretManagerServiceClient, secret: Tuple[str, str, str]
|
293 | 340 | ) -> None:
|
@@ -341,6 +388,15 @@ def test_destroy_secret_version_with_etag(
|
341 | 388 | assert version.destroy_time
|
342 | 389 |
|
343 | 390 |
|
| 391 | +def test_disable_secret_with_delayed_destroy( |
| 392 | + client: secretmanager.SecretManagerServiceClient, |
| 393 | + secret_with_delayed_destroy: Tuple[str, str], |
| 394 | +) -> None: |
| 395 | + project_id, secret_id = secret_with_delayed_destroy |
| 396 | + updated_secret = disable_secret_with_delayed_destroy(project_id, secret_id) |
| 397 | + assert updated_secret.version_destroy_ttl == timedelta(0) |
| 398 | + |
| 399 | + |
344 | 400 | def test_enable_disable_secret_version(
|
345 | 401 | client: secretmanager.SecretManagerServiceClient,
|
346 | 402 | secret_version: Tuple[str, str, str, str],
|
@@ -532,3 +588,10 @@ def test_update_secret_with_alias(secret_version: Tuple[str, str, str, str]) ->
|
532 | 588 | project_id, secret_id, version_id, _ = secret_version
|
533 | 589 | secret = update_secret_with_alias(project_id, secret_id)
|
534 | 590 | assert secret.version_aliases["test"] == 1
|
| 591 | + |
| 592 | + |
| 593 | +def test_update_secret_with_delayed_destroy(secret_with_delayed_destroy: Tuple[str, str], version_destroy_ttl: str) -> None: |
| 594 | + project_id, secret_id = secret_with_delayed_destroy |
| 595 | + updated_version_destroy_ttl_value = 118400 |
| 596 | + updated_secret = update_secret_with_delayed_destroy(project_id, secret_id, updated_version_destroy_ttl_value) |
| 597 | + assert updated_secret.version_destroy_ttl == timedelta(seconds=updated_version_destroy_ttl_value) |
0 commit comments