Skip to content
This repository was archived by the owner on Jun 8, 2026. It is now read-only.
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 41 additions & 36 deletions tests/system/test_backup_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,9 +79,9 @@ def diff_config_instance(
_helpers.scrub_instance_ignore_not_found(diff_config_instance)


@pytest.fixture(scope="session")
def database_version_time():
return datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc)
@pytest.fixture(scope="function")
def database_version_time(shared_database): # Ensure database exists.
return datetime.datetime.now(datetime.timezone.utc)


@pytest.fixture(scope="session")
Expand Down Expand Up @@ -124,8 +124,9 @@ def test_backup_workflow(
)

backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)
encryption_enum = CreateBackupEncryptionConfig.EncryptionType
encryption_config = CreateBackupEncryptionConfig(
encryption_type=encryption_enum.GOOGLE_DEFAULT_ENCRYPTION,
Expand Down Expand Up @@ -162,8 +163,9 @@ def test_backup_workflow(
)

# Update with valid argument.
valid_expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=7)
valid_expire_time = valid_expire_time.replace(tzinfo=datetime.timezone.utc)
valid_expire_time = datetime.datetime.now(
datetime.timezone.utc
) + datetime.timedelta(days=7)
backup.update_expire_time(valid_expire_time)
assert valid_expire_time == backup.expire_time

Expand Down Expand Up @@ -193,15 +195,12 @@ def test_backup_workflow(


def test_backup_create_w_version_time_dflt_to_create_time(
shared_instance,
shared_database,
database_version_time,
backups_to_delete,
databases_to_delete,
shared_instance, shared_database, backups_to_delete, databases_to_delete,
):
backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)

# Create backup.
backup = shared_instance.backup(
Expand All @@ -228,7 +227,7 @@ def test_backup_create_w_version_time_dflt_to_create_time(

def test_backup_create_w_invalid_expire_time(shared_instance, shared_database):
backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc)

backup = shared_instance.backup(
backup_id, database=shared_database, expire_time=expire_time
Expand All @@ -243,10 +242,12 @@ def test_backup_create_w_invalid_version_time_past(
shared_instance, shared_database,
):
backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
version_time = datetime.datetime.utcnow() - datetime.timedelta(days=10)
version_time = version_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)
version_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
days=10
)

backup = shared_instance.backup(
backup_id,
Expand All @@ -264,10 +265,12 @@ def test_backup_create_w_invalid_version_time_future(
shared_instance, shared_database,
):
backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
version_time = datetime.datetime.utcnow() + datetime.timedelta(days=2)
version_time = version_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)
version_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=2
)

backup = shared_instance.backup(
backup_id,
Expand All @@ -289,8 +292,9 @@ def test_database_restore_to_diff_instance(
databases_to_delete,
):
backup_id = _helpers.unique_id("backup_id", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)

# Create backup.
backup = shared_instance.backup(
Expand Down Expand Up @@ -322,8 +326,9 @@ def test_multi_create_cancel_update_error_restore_errors(
):
backup_id_1 = _helpers.unique_id("backup_id1", separator="_")
backup_id_2 = _helpers.unique_id("backup_id2", separator="_")
expire_time = datetime.datetime.utcnow() + datetime.timedelta(days=3)
expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=3
)

backup1 = shared_instance.backup(
backup_id_1, database=shared_database, expire_time=expire_time
Expand Down Expand Up @@ -354,8 +359,8 @@ def test_multi_create_cancel_update_error_restore_errors(

# Update expire time to invalid value.
max_expire_days = 366 # documented maximum
invalid_expire_time = datetime.datetime.now().replace(
tzinfo=datetime.timezone.utc
invalid_expire_time = datetime.datetime.now(
datetime.timezone.utc
) + datetime.timedelta(days=max_expire_days + 1)
with pytest.raises(exceptions.InvalidArgument):
backup1.update_expire_time(invalid_expire_time)
Expand Down Expand Up @@ -385,8 +390,9 @@ def test_instance_list_backups(
backup_id_1 = _helpers.unique_id("backup_id1", separator="_")
backup_id_2 = _helpers.unique_id("backup_id2", separator="_")

expire_time_1 = datetime.datetime.utcnow() + datetime.timedelta(days=21)
expire_time_1 = expire_time_1.replace(tzinfo=datetime.timezone.utc)
expire_time_1 = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=21
)
expire_time_1_stamp = expire_time_1.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

backup1 = shared_instance.backup(
Expand All @@ -396,8 +402,9 @@ def test_instance_list_backups(
version_time=database_version_time,
)

expire_time_2 = datetime.datetime.utcnow() + datetime.timedelta(days=1)
expire_time_2 = expire_time_2.replace(tzinfo=datetime.timezone.utc)
expire_time_2 = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
days=1
)
backup2 = shared_instance.backup(
backup_id_2, database=second_database, expire_time=expire_time_2
)
Expand All @@ -408,9 +415,7 @@ def test_instance_list_backups(
op1.result() # blocks indefinitely
backup1.reload()

create_time_compare = datetime.datetime.utcnow().replace(
tzinfo=datetime.timezone.utc
)
create_time_compare = datetime.datetime.now(datetime.timezone.utc)
create_time_stamp = create_time_compare.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

backup2.create()
Expand Down