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
|
from collections import namedtuple
import io
import os
import requests
import time
from azure.mgmt.keyvault import KeyVaultManagementClient
from azure.mgmt.keyvault.models import (
SecretPermissions,
KeyPermissions,
CertificatePermissions,
StoragePermissions,
Permissions,
Sku,
SkuName,
AccessPolicyEntry,
VaultProperties,
VaultCreateOrUpdateParameters,
)
from devtools_testutils import AzureMgmtPreparer, AzureTestError, ResourceGroupPreparer
from devtools_testutils.resource_testcase import RESOURCE_GROUP_PARAM
VAULT_PARAM = "vault"
FakeVault = namedtuple("FakeVault", ["name", "location", "properties"])
DEFAULT_PERMISSIONS = Permissions(
keys=[perm.value for perm in KeyPermissions],
secrets=[perm.value for perm in SecretPermissions],
certificates=[perm.value for perm in CertificatePermissions],
storage=[perm.value for perm in StoragePermissions],
)
DEFAULT_SKU = SkuName.premium.value
CLIENT_OID = "00000000-0000-0000-0000-000000000000"
class KeyVaultPreparer(AzureMgmtPreparer):
def __init__(
self,
sku=None,
permissions=None,
enabled_for_deployment=True,
enabled_for_disk_encryption=True,
enabled_for_template_deployment=True,
enable_soft_delete=None,
name_prefix="vault",
location="westus",
parameter_name=VAULT_PARAM,
resource_group_parameter_name=RESOURCE_GROUP_PARAM,
disable_recording=True,
playback_fake_resource=None,
client_kwargs=None,
):
super(KeyVaultPreparer, self).__init__(
name_prefix,
24,
disable_recording=disable_recording,
playback_fake_resource=playback_fake_resource,
client_kwargs=client_kwargs,
)
self.sku = sku or DEFAULT_SKU
self.permissions = permissions or DEFAULT_PERMISSIONS
self.enabled_for_deployment = enabled_for_deployment
self.enabled_for_disk_encryption = enabled_for_disk_encryption
self.enabled_for_template_deployment = enabled_for_template_deployment
self.enable_soft_delete = enable_soft_delete
self.location = location
self.resource_group_parameter_name = resource_group_parameter_name
self.parameter_name = parameter_name
self.creds_parameter = "credentials"
self.parameter_name_for_location = "location"
self.client_oid = None
def _get_resource_group(self, **kwargs):
try:
return kwargs[self.resource_group_parameter_name]
except KeyError:
template = (
"To create a key vault a resource group is required. Please add "
"decorator @{} in front of this storage account preparer."
)
raise AzureTestError(template.format(ResourceGroupPreparer.__name__))
def create_resource(self, name, **kwargs):
self.client_oid = self.test_class_instance.set_value_to_scrub("CLIENT_OID", CLIENT_OID)
group = self._get_resource_group(**kwargs).name
access_policies = [
AccessPolicyEntry(
tenant_id=self.test_class_instance.settings.TENANT_ID,
object_id=self.client_oid,
permissions=self.permissions,
)
]
properties = VaultProperties(
tenant_id=self.test_class_instance.settings.TENANT_ID,
sku=Sku(name=self.sku),
access_policies=access_policies,
vault_uri=None,
enabled_for_deployment=self.enabled_for_deployment,
enabled_for_disk_encryption=self.enabled_for_disk_encryption,
enabled_for_template_deployment=self.enabled_for_template_deployment,
enable_soft_delete=self.enable_soft_delete,
enable_purge_protection=None,
)
if self.is_live:
self.client = self.create_mgmt_client(KeyVaultManagementClient)
parameters = VaultCreateOrUpdateParameters(location=self.location, properties=properties)
self.resource = self.client.vaults.create_or_update(group, name, parameters).result()
else:
properties.vault_uri = "https://{}.vault.azure.net/".format(name)
self.resource = FakeVault(name=name, location=self.location, properties=properties)
return {self.parameter_name: self.resource}
def remove_resource(self, name, **kwargs):
if self.is_live:
group = self._get_resource_group(**kwargs).name
self.client.vaults.delete(group, name)
if self.enable_soft_delete:
self.client.vaults.purge_deleted(name, self.location).wait()
|
