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
|
#!/usr/bin/env python
import logging
from unittest import TestCase
import requests_mock
from parameterized import parameterized, param
from hvac.adapters import JSONAdapter
from hvac.api.secrets_engines import Aws
from hvac.api.secrets_engines.aws import DEFAULT_MOUNT_POINT
from hvac.exceptions import ParamValidationError
class TestAws(TestCase):
@parameterized.expand(
[
param(
"success",
),
]
)
def test_rotate_root_iam_credentials(
self, test_label, mount_point=DEFAULT_MOUNT_POINT
):
expected_status_code = 200
mock_response = {"data": {"access_key": "AKIA..."}}
aws = Aws(adapter=JSONAdapter())
mock_url = "http://localhost:8200/v1/{mount_point}/config/rotate-root".format(
mount_point=mount_point,
)
logging.debug("Mocking URL: %s" % mock_url)
with requests_mock.mock() as requests_mocker:
requests_mocker.register_uri(
method="POST",
url=mock_url,
status_code=expected_status_code,
json=mock_response,
)
rotate_root_response = aws.rotate_root_iam_credentials(
mount_point=mount_point,
)
logging.debug("rotate_root_response: %s" % rotate_root_response)
self.assertEqual(
first=mock_response,
second=rotate_root_response,
)
@parameterized.expand(
[
param(
"success",
),
param(
"invalid endpoint",
endpoint="cats",
raises=ParamValidationError,
exception_msg="cats",
),
]
)
def test_generate_credentials(
self,
test_label,
role_name="hvac-test-role",
mount_point=DEFAULT_MOUNT_POINT,
endpoint="creds",
raises=None,
exception_msg="",
):
expected_status_code = 200
mock_response = {
"data": {
"access_key": "AKIA...",
"secret_key": "xlCs...",
"security_token": None,
}
}
mock_url = "http://localhost:8200/v1/{mount_point}/creds/{role_name}".format(
mount_point=mount_point,
role_name=role_name,
)
logging.debug("Mocking URL: %s" % mock_url)
aws = Aws(adapter=JSONAdapter())
with requests_mock.mock() as requests_mocker:
requests_mocker.register_uri(
method="GET",
url=mock_url,
status_code=expected_status_code,
json=mock_response,
)
if raises:
with self.assertRaises(raises) as cm:
aws.generate_credentials(
name=role_name,
endpoint=endpoint,
mount_point=mount_point,
)
self.assertIn(
member=exception_msg,
container=str(cm.exception),
)
else:
gen_creds_response = aws.generate_credentials(
name=role_name,
endpoint=endpoint,
mount_point=mount_point,
)
logging.debug("gen_creds_response: %s" % gen_creds_response)
self.assertEqual(
first=mock_response,
second=gen_creds_response,
)
|
