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
|
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.auth_methods import Azure
from tests import utils
@skipIf(
utils.vault_version_lt("0.10.0"),
"Azure auth method not available before Vault version 0.10.0",
)
class TestAzure(TestCase):
TEST_MOUNT_POINT = "azure-test"
@parameterized.expand(
[
(
"success",
dict(),
None,
),
(
"with subscription_id",
dict(subscription_id="my_subscription_id"),
None,
),
(
"with resource_group_name",
dict(resource_group_name="my_resource_group_name"),
None,
),
(
"with vm_name",
dict(vm_name="my_vm_name"),
None,
),
(
"with vmss_name",
dict(vmss_name="my_vmss_name"),
None,
),
(
"with vm_name and vmss_name",
dict(vm_name="my_vm_name", vmss_name="my_vmss_name"),
None,
),
]
)
@requests_mock.Mocker()
def test_login(self, label, test_params, raises, requests_mocker):
role_name = "hvac"
test_policies = [
"default",
"dev",
"prod",
]
expected_status_code = 200
mock_url = "http://localhost:8200/v1/auth/{mount_point}/login".format(
mount_point=self.TEST_MOUNT_POINT,
)
mock_response = {
"auth": {
"client_token": "f33f8c72-924e-11f8-cb43-ac59d697597c",
"accessor": "0e9e354a-520f-df04-6867-ee81cae3d42d",
"policies": test_policies,
"lease_duration": 2764800,
"renewable": True,
},
}
requests_mocker.register_uri(
method="POST",
url=mock_url,
status_code=expected_status_code,
json=mock_response,
)
azure = Azure(adapter=JSONAdapter())
if raises is not None:
with self.assertRaises(raises):
azure.login(
role=role_name,
jwt="my-jwt",
mount_point=self.TEST_MOUNT_POINT,
**test_params
)
else:
login_response = azure.login(
role=role_name,
jwt="my-jwt",
mount_point=self.TEST_MOUNT_POINT,
**test_params
)
logging.debug("login_response: %s" % login_response)
self.assertEqual(
first=login_response["auth"]["policies"],
second=test_policies,
)
|
