1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
from copy import deepcopy
from typing import Any, Optional
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.common_models import BaseModel
from moto.core.utils import unix_time
from moto.moto_api._internal import mock_random
from moto.utilities.tagging_service import TaggingService
from moto.utilities.utils import get_partition
from .exceptions import AlreadyExistsException, ResourceNotFoundException
class Plan(BaseModel):
def __init__(
self,
backup_plan: dict[str, Any],
creator_request_id: str,
backend: "BackupBackend",
):
self.backup_plan_id = str(mock_random.uuid4())
self.backup_plan_arn = f"arn:{get_partition(backend.region_name)}:backup:{backend.region_name}:{backend.account_id}:backup-plan:{self.backup_plan_id}"
self.creation_date = unix_time()
ran_str = mock_random.get_random_string(length=48)
self.version_id = ran_str
self.creator_request_id = creator_request_id
self.backup_plan = backup_plan
adv_settings = backup_plan.get("AdvancedBackupSettings")
self.advanced_backup_settings = adv_settings or []
self.deletion_date: Optional[float] = None
# Deletion Date is updated when the backup_plan is deleted
self.last_execution_date = None # start_restore_job not yet supported
rules = backup_plan["Rules"]
for rule in rules:
rule["ScheduleExpression"] = rule.get(
"ScheduleExpression", "cron(0 5 ? * * *)"
) # Default CRON expression in UTC
rule["StartWindowMinutes"] = rule.get(
"StartWindowMinutes", 480
) # Default=480
rule["CompletionWindowMinutes"] = rule.get(
"CompletionWindowMinutes", 10080
) # Default=10080
rule["ScheduleExpressionTimezone"] = rule.get(
"ScheduleExpressionTimezone", "Etc/UTC"
) # set to Etc/UTc by default
rule["RuleId"] = str(mock_random.uuid4())
def to_dict(self) -> dict[str, Any]:
dct = {
"BackupPlanId": self.backup_plan_id,
"BackupPlanArn": self.backup_plan_arn,
"CreationDate": self.creation_date,
"VersionId": self.version_id,
"AdvancedBackupSettings": self.advanced_backup_settings,
}
return {k: v for k, v in dct.items() if v}
def to_get_dict(self) -> dict[str, Any]:
dct = self.to_dict()
dct_options = {
"BackupPlan": self.backup_plan,
"CreatorRequestId": self.creator_request_id,
"DeletionDate": self.deletion_date,
"LastExecutionDate": self.last_execution_date,
}
for key, value in dct_options.items():
if value is not None:
dct[key] = value
return dct
def to_list_dict(self) -> dict[str, Any]:
dct = self.to_get_dict()
dct.pop("BackupPlan")
dct["BackupPlanName"] = self.backup_plan.get("BackupPlanName")
return dct
class Vault(BaseModel):
def __init__(
self,
backup_vault_name: str,
encryption_key_arn: str,
creator_request_id: str,
backend: "BackupBackend",
):
self.backup_vault_name = backup_vault_name
self.backup_vault_arn = f"arn:{get_partition(backend.region_name)}:backup:{backend.region_name}:{backend.account_id}:backup-vault:{backup_vault_name}"
self.creation_date = unix_time()
self.encryption_key_arn = encryption_key_arn
self.creator_request_id = creator_request_id
self.num_of_recovery_points = 0 # start_backup_job not yet supported
self.locked = False # put_backup_vault_lock_configuration
self.min_retention_days = 0 # put_backup_vault_lock_configuration
self.max_retention_days = 0 # put_backup_vault_lock_configuration
self.lock_date = None # put_backup_vault_lock_configuration
def to_dict(self) -> dict[str, Any]:
dct = {
"BackupVaultName": self.backup_vault_name,
"BackupVaultArn": self.backup_vault_arn,
"CreationDate": self.creation_date,
}
return dct
def to_list_dict(self) -> dict[str, Any]:
dct = self.to_dict()
dct_options: dict[str, Any] = {}
dct_options = {
"EncryptionKeyArn": self.encryption_key_arn,
"CreatorRequestId": self.creator_request_id,
"NumberOfRecoveryPoints": self.num_of_recovery_points,
"Locked": self.locked,
"MinRetentionDays": self.min_retention_days,
"MaxRetentionDays": self.max_retention_days,
"LockDate": self.lock_date,
}
for key, value in dct_options.items():
if value is not None:
dct[key] = value
return dct
class BackupBackend(BaseBackend):
"""Implementation of Backup APIs."""
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.vaults: dict[str, Vault] = {}
self.plans: dict[str, Plan] = {}
self.tagger = TaggingService()
def create_backup_plan(
self,
backup_plan: dict[str, Any],
backup_plan_tags: dict[str, str],
creator_request_id: str,
) -> Plan:
if backup_plan["BackupPlanName"] in [
p.backup_plan["BackupPlanName"] for p in list(self.plans.values())
]:
raise AlreadyExistsException(
msg="Backup plan with the same plan document already exists"
)
plan = Plan(
backup_plan=backup_plan,
creator_request_id=creator_request_id,
backend=self,
)
if backup_plan_tags:
self.tag_resource(plan.backup_plan_arn, backup_plan_tags)
self.plans[plan.backup_plan_id] = plan
return plan
def get_backup_plan(self, backup_plan_id: str, version_id: Optional[Any]) -> Plan:
msg = "Failed reading Backup plan with provided version"
if backup_plan_id not in self.plans:
raise ResourceNotFoundException(msg=msg)
plan = self.plans[backup_plan_id]
if version_id:
if plan.version_id == version_id:
return plan
else:
raise ResourceNotFoundException(msg=msg)
return plan
def delete_backup_plan(self, backup_plan_id: str) -> tuple[str, str, float, str]:
if backup_plan_id not in self.plans:
raise ResourceNotFoundException(
msg="Failed reading Backup plan with provided version"
)
deletion_date = unix_time()
res = self.plans[backup_plan_id]
res.deletion_date = deletion_date
return res.backup_plan_id, res.backup_plan_arn, deletion_date, res.version_id
def list_backup_plans(self, include_deleted: Any) -> list[Plan]:
"""
Pagination is not yet implemented
"""
plans_list = deepcopy(self.plans)
for plan in list(plans_list.values()):
backup_plan_id = plan.backup_plan_id
if plan.deletion_date is not None:
plans_list.pop(backup_plan_id)
if include_deleted:
return list(self.plans.values())
return list(plans_list.values())
def create_backup_vault(
self,
backup_vault_name: str,
backup_vault_tags: dict[str, str],
encryption_key_arn: str,
creator_request_id: str,
) -> Vault:
if backup_vault_name in self.vaults:
raise AlreadyExistsException(
msg="Backup vault with the same name already exists"
)
vault = Vault(
backup_vault_name=backup_vault_name,
encryption_key_arn=encryption_key_arn,
creator_request_id=creator_request_id,
backend=self,
)
if backup_vault_tags:
self.tag_resource(vault.backup_vault_arn, backup_vault_tags)
self.vaults[backup_vault_name] = vault
return vault
def list_backup_vaults(self) -> list[Vault]:
"""
Pagination is not yet implemented
"""
return list(self.vaults.values())
def list_tags(self, resource_arn: str) -> dict[str, str]:
"""
Pagination is not yet implemented
"""
return self.tagger.get_tag_dict_for_resource(resource_arn)
def tag_resource(self, resource_arn: str, tags: dict[str, str]) -> None:
tags_input = TaggingService.convert_dict_to_tags_input(tags or {})
self.tagger.tag_resource(resource_arn, tags_input)
def untag_resource(self, resource_arn: str, tag_key_list: list[str]) -> None:
self.tagger.untag_resource_using_names(resource_arn, tag_key_list)
backup_backends = BackendDict(BackupBackend, "backup")
|
