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
|
import logging
from unittest import TestCase
from unittest import skipIf
import requests_mock
from parameterized import parameterized
from hvac.adapters import JSONAdapter
from hvac.api.secrets_engines.azure import Azure, DEFAULT_MOUNT_POINT
from tests import utils
@skipIf(
utils.vault_version_lt("0.11.0"),
"Azure secret engine not available before Vault version 0.11.0",
)
class TestAzure(TestCase):
@parameterized.expand(
[
("create role", None),
]
)
@requests_mock.Mocker()
def test_create_or_update_role(self, test_label, azure_roles, requests_mocker):
expected_status_code = 204
role_name = "hvac"
if azure_roles is None:
azure_roles = [
{
"role_name": "Contributor",
"scope": "/subscriptions/95e675fa-307a-455e-8cdf-0a66aeaa35ae",
},
]
mock_url = "http://localhost:8200/v1/{mount_point}/roles/{name}".format(
mount_point=DEFAULT_MOUNT_POINT,
name=role_name,
)
requests_mocker.register_uri(
method="POST",
url=mock_url,
status_code=expected_status_code,
# json=mock_response,
)
azure = Azure(adapter=JSONAdapter())
create_or_update_role_response = azure.create_or_update_role(
name=role_name, azure_roles=azure_roles, mount_point=DEFAULT_MOUNT_POINT
)
logging.debug(
"create_or_update_role_response: %s" % create_or_update_role_response
)
self.assertEqual(
first=expected_status_code,
second=create_or_update_role_response.status_code,
)
@parameterized.expand(
[
("some_test",),
]
)
@requests_mock.Mocker()
def test_list_roles(self, test_label, requests_mocker):
expected_status_code = 200
role_names = ["hvac"]
mock_response = {
"data": {
"roles": role_names,
},
}
mock_url = "http://localhost:8200/v1/{mount_point}/roles".format(
mount_point=DEFAULT_MOUNT_POINT,
)
requests_mocker.register_uri(
method="LIST",
url=mock_url,
status_code=expected_status_code,
json=mock_response,
)
azure = Azure(adapter=JSONAdapter())
list_roles_response = azure.list_roles(mount_point=DEFAULT_MOUNT_POINT)
logging.debug("list_roles_response: %s" % list_roles_response)
self.assertEqual(
first=mock_response["data"],
second=list_roles_response,
)
@parameterized.expand(
[
("some_test",),
]
)
@requests_mock.Mocker()
def test_generate_credentials(self, test_label, requests_mocker):
expected_status_code = 200
role_name = "hvac"
mock_response = {
"data": {
"client_id": "some_client_id",
"client_secret": "some_client_secret",
},
}
mock_url = "http://localhost:8200/v1/{mount_point}/creds/{name}".format(
mount_point=DEFAULT_MOUNT_POINT,
name=role_name,
)
requests_mocker.register_uri(
method="GET",
url=mock_url,
status_code=expected_status_code,
json=mock_response,
)
azure = Azure(adapter=JSONAdapter())
generate_credentials_response = azure.generate_credentials(
name=role_name, mount_point=DEFAULT_MOUNT_POINT
)
logging.debug(
"generate_credentials_response: %s" % generate_credentials_response
)
self.assertEqual(
first=mock_response["data"],
second=generate_credentials_response,
)
|
