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
|
import logging
from unittest import TestCase
from unittest import skipIf
from parameterized import parameterized
from hvac import exceptions
from tests import utils
from tests.utils.hvac_integration_test_case import HvacIntegrationTestCase
@skipIf(
utils.vault_version_lt("0.11.0"),
"Azure secret engine not available before Vault version 0.11.0",
)
class TestAzure(HvacIntegrationTestCase, TestCase):
TENANT_ID = "00000000-0000-0000-0000-000000000000"
SUBSCRIPTION_ID = "00000000-0000-0000-0000-000000000000"
DEFAULT_MOUNT_POINT = "azure-integration-test"
def setUp(self):
super().setUp()
self.client.sys.enable_secrets_engine(
backend_type="azure",
path=self.DEFAULT_MOUNT_POINT,
)
def tearDown(self):
self.client.sys.disable_secrets_engine(path=self.DEFAULT_MOUNT_POINT)
super().tearDown()
@parameterized.expand(
[
("no parameters",),
("valid environment argument", "AzureUSGovernmentCloud"),
(
"invalid environment argument",
"AzureCityKity",
exceptions.ParamValidationError,
"invalid environment argument provided",
),
]
)
def test_configure_and_read_configuration(
self, test_label, environment=None, raises=False, exception_message=""
):
configure_arguments = {
"subscription_id": self.SUBSCRIPTION_ID,
"tenant_id": self.TENANT_ID,
"mount_point": self.DEFAULT_MOUNT_POINT,
}
if environment is not None:
configure_arguments["environment"] = environment
if raises:
with self.assertRaises(raises) as cm:
self.client.secrets.azure.configure(**configure_arguments)
self.assertIn(
member=exception_message,
container=str(cm.exception),
)
else:
configure_response = self.client.secrets.azure.configure(
**configure_arguments
)
logging.debug("configure_response: %s" % configure_response)
read_configuration_response = self.client.secrets.azure.read_config(
mount_point=self.DEFAULT_MOUNT_POINT,
)
logging.debug(
"read_configuration_response: %s" % read_configuration_response
)
# raise Exception()
self.assertEqual(
first=self.SUBSCRIPTION_ID,
second=read_configuration_response["subscription_id"],
)
self.assertEqual(
first=self.TENANT_ID,
second=read_configuration_response["tenant_id"],
)
if environment is not None:
self.assertEqual(
first=environment,
second=read_configuration_response["environment"],
)
@parameterized.expand(
[
("create and then delete config",),
]
)
def test_delete_config(self, test_label):
configure_response = self.client.secrets.azure.configure(
subscription_id=self.SUBSCRIPTION_ID,
tenant_id=self.TENANT_ID,
mount_point=self.DEFAULT_MOUNT_POINT,
)
logging.debug("configure_response: %s" % configure_response)
self.client.secrets.azure.delete_config(
mount_point=self.DEFAULT_MOUNT_POINT,
)
read_configuration_response = self.client.secrets.azure.read_config(
mount_point=self.DEFAULT_MOUNT_POINT,
)
logging.debug("read_configuration_response: %s" % read_configuration_response)
read_expected_response = {
"client_id": "",
"environment": "",
"subscription_id": "",
"tenant_id": "",
}
if utils.vault_version_ge("1.9.0"):
read_expected_response["root_password_ttl"] = 0
if utils.vault_version_lt("1.12.0"):
read_expected_response["use_microsoft_graph_api"] = False
self.assertEqual(
first=read_expected_response,
second=read_configuration_response,
)
|
