Skip to content

Commit e600716

Browse files
pyrookapyrooka
committed
test: add unit tests for VPC stuffs
1 parent f90f4d8 commit e600716

File tree

4 files changed

+310
-0
lines changed

4 files changed

+310
-0
lines changed

‎resources/ibm-credentials-vpc.env

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
# VPC auth with default config
2+
SERVICE1_AUTH_TYPE=vpc
3+
4+
# VPC auth with profile CRN
5+
SERVICE2_AUTH_TYPE=vpc
6+
SERVICE2_IAM_PROFILE_CRN=crn:iam-profile1
7+
SERVICE2_AUTH_URL=http://vpc.imds.com/api
8+
9+
# VPC auth with profile ID
10+
SERVICE3_AUTH_TYPE=vpc
11+
SERVICE3_IAM_PROFILE_ID=iam-profile1-id

‎test/test_utils.py

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -376,6 +376,36 @@ def test_get_authenticator_from_credential_file():
376376
assert authenticator.token_manager.scope is None
377377
del os.environ['IBM_CREDENTIALS_FILE']
378378

379+
file_path = os.path.join(os.path.dirname(__file__),
380+
'../resources/ibm-credentials-vpc.env')
381+
os.environ['IBM_CREDENTIALS_FILE'] = file_path
382+
authenticator = get_authenticator_from_environment('service1')
383+
assert authenticator is not None
384+
assert authenticator.authentication_type() == Authenticator.AUTHTYPE_VPC
385+
assert authenticator.token_manager.iam_profile_crn is None
386+
assert authenticator.token_manager.iam_profile_id is None
387+
assert authenticator.token_manager.url == 'http://169.254.169.254'
388+
389+
file_path = os.path.join(os.path.dirname(__file__),
390+
'../resources/ibm-credentials-vpc.env')
391+
os.environ['IBM_CREDENTIALS_FILE'] = file_path
392+
authenticator = get_authenticator_from_environment('service2')
393+
assert authenticator is not None
394+
assert authenticator.authentication_type() == Authenticator.AUTHTYPE_VPC
395+
assert authenticator.token_manager.iam_profile_crn == 'crn:iam-profile1'
396+
assert authenticator.token_manager.iam_profile_id is None
397+
assert authenticator.token_manager.url == 'http://vpc.imds.com/api'
398+
399+
file_path = os.path.join(os.path.dirname(__file__),
400+
'../resources/ibm-credentials-vpc.env')
401+
os.environ['IBM_CREDENTIALS_FILE'] = file_path
402+
authenticator = get_authenticator_from_environment('service3')
403+
assert authenticator is not None
404+
assert authenticator.authentication_type() == Authenticator.AUTHTYPE_VPC
405+
assert authenticator.token_manager.iam_profile_crn is None
406+
assert authenticator.token_manager.iam_profile_id == 'iam-profile1-id'
407+
assert authenticator.token_manager.url == 'http://169.254.169.254'
408+
379409

380410
def test_get_authenticator_from_credential_file_scope():
381411
file_path = os.path.join(os.path.dirname(__file__),

‎test/test_vpc_instance_authenticator.py

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
# pylint: disable=missing-docstring
2+
import pytest
3+
4+
from ibm_cloud_sdk_core.authenticators import VPCInstanceAuthenticator, Authenticator
5+
6+
7+
TEST_IAM_PROFILE_CRN = 'crn:iam-profile:123'
8+
TEST_IAM_PROFILE_ID = 'iam-id-123'
9+
10+
11+
def test_constructor():
12+
authenticator =VPCInstanceAuthenticator(iam_profile_id=TEST_IAM_PROFILE_ID, url='someurl.com')
13+
assert authenticator is not None
14+
assert authenticator.authentication_type() == Authenticator.AUTHTYPE_VPC
15+
assert authenticator.token_manager.iam_profile_crn is None
16+
assert authenticator.token_manager.iam_profile_id == TEST_IAM_PROFILE_ID
17+
assert authenticator.token_manager.url == 'someurl.com'
18+
19+
20+
def test_setters():
21+
authenticator =VPCInstanceAuthenticator(iam_profile_id=TEST_IAM_PROFILE_ID, url='someurl.com')
22+
assert authenticator is not None
23+
assert authenticator.authentication_type() == Authenticator.AUTHTYPE_VPC
24+
assert authenticator.token_manager.iam_profile_crn is None
25+
assert authenticator.token_manager.iam_profile_id == TEST_IAM_PROFILE_ID
26+
assert authenticator.token_manager.url == 'someurl.com'
27+
28+
# Set the IAM profile CRN to trigger a validation which will fail,
29+
# because at most one of iam_profile_crn or iam_profile_id may be specified.
30+
with pytest.raises(ValueError) as err:
31+
authenticator.set_iam_profile_crn(TEST_IAM_PROFILE_CRN)
32+
assert str(
33+
err.value) == 'At most one of "iam_profile_id" or "iam_profile_crn" may be specified.'
34+
35+
authenticator.set_iam_profile_id(None)
36+
assert authenticator.token_manager.iam_profile_id is None
37+
38+
authenticator.set_iam_profile_crn(TEST_IAM_PROFILE_CRN)
39+
assert authenticator.token_manager.iam_profile_crn == TEST_IAM_PROFILE_CRN
40+
41+
42+
def test_constructor_validate_failed():
43+
with pytest.raises(ValueError) as err:
44+
VPCInstanceAuthenticator(
45+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
46+
iam_profile_id=TEST_IAM_PROFILE_ID,
47+
)
48+
assert str(
49+
err.value) == 'At most one of "iam_profile_id" or "iam_profile_crn" may be specified.'
50+
51+
52+
def test_authenticate():
53+
def mock_get_token():
54+
return 'mock_token'
55+
56+
authenticator = VPCInstanceAuthenticator(iam_profile_crn=TEST_IAM_PROFILE_CRN)
57+
authenticator.token_manager.get_token = mock_get_token
58+
59+
# Simulate an SDK API request that needs to be authenticated.
60+
request = {'headers': {}}
61+
62+
# Trigger the "get token" processing to obtain the access token and add to the "SDK request".
63+
authenticator.authenticate(request)
64+
65+
# Verify that the "authenticate()" method added the Authorization header
66+
assert request['headers']['Authorization'] == 'Bearer mock_token'

‎test/test_vpc_instance_token_manager.py

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
# coding: utf-8
2+
3+
# Copyright 2021 IBM All Rights Reserved.
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
# pylint: disable=missing-docstring
18+
import json
19+
import logging
20+
21+
import pytest
22+
import responses
23+
24+
from ibm_cloud_sdk_core import ApiException, VPCInstanceTokenManager
25+
26+
27+
TEST_TOKEN = 'abc123'
28+
TEST_IAM_TOKEN = 'iam-abc123'
29+
TEST_IAM_PROFILE_CRN = 'crn:iam-profile:123'
30+
TEST_IAM_PROFILE_ID = 'iam-id-123'
31+
32+
33+
def test_constructor():
34+
token_manager = VPCInstanceTokenManager(
35+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
36+
)
37+
38+
assert token_manager.iam_profile_crn is TEST_IAM_PROFILE_CRN
39+
assert token_manager.iam_profile_id is None
40+
assert token_manager.access_token is None
41+
42+
43+
def test_setters():
44+
token_manager = VPCInstanceTokenManager(
45+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
46+
)
47+
48+
assert token_manager.iam_profile_crn is TEST_IAM_PROFILE_CRN
49+
assert token_manager.iam_profile_id is None
50+
assert token_manager.access_token is None
51+
52+
token_manager.set_iam_profile_crn(None)
53+
assert token_manager.iam_profile_crn is None
54+
55+
token_manager.set_iam_profile_id(TEST_IAM_PROFILE_ID)
56+
assert token_manager.iam_profile_id == TEST_IAM_PROFILE_ID
57+
58+
59+
@responses.activate
60+
def test_retrieve_instance_identity_token(caplog):
61+
caplog.set_level(logging.DEBUG)
62+
63+
token_manager = VPCInstanceTokenManager(
64+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
65+
url='http://someurl.com',
66+
)
67+
68+
response = {
69+
'access_token': TEST_TOKEN,
70+
}
71+
72+
responses.add(responses.PUT, 'http://someurl.com/instance_identity/v1/token',
73+
body=json.dumps(response), status=200)
74+
75+
ii_token = token_manager.retrieve_instance_identity_token()
76+
assert len(responses.calls) == 1
77+
assert responses.calls[0].request.headers['version'] == '2021-09-20'
78+
assert responses.calls[0].request.headers['Content-Type'] == 'application/json'
79+
assert responses.calls[0].request.headers['Accept'] == 'application/json'
80+
assert responses.calls[0].request.headers['Metadata-Flavor'] == 'ibm'
81+
assert responses.calls[0].request.body == '{"expires_in": 300}'
82+
assert ii_token == TEST_TOKEN
83+
# Check the logs.
84+
#pylint: disable=line-too-long
85+
assert caplog.record_tuples[0][2] == 'Invoking VPC \'create_access_token\' operation: http://someurl.com/instance_identity/v1/token'
86+
assert caplog.record_tuples[1][2] == 'Returned from VPC \'create_access_token\' operation."'
87+
88+
@responses.activate
89+
def test_retrieve_instance_identity_token_failed(caplog):
90+
caplog.set_level(logging.DEBUG)
91+
92+
token_manager = VPCInstanceTokenManager(
93+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
94+
url='http://someurl.com',
95+
)
96+
97+
response = {
98+
'errors': ['Ooops'],
99+
}
100+
101+
responses.add(responses.PUT, 'http://someurl.com/instance_identity/v1/token',
102+
body=json.dumps(response), status=400)
103+
104+
with pytest.raises(ApiException):
105+
token_manager.retrieve_instance_identity_token()
106+
107+
assert len(responses.calls) == 1
108+
# Check the logs.
109+
#pylint: disable=line-too-long
110+
assert caplog.record_tuples[0][2] == 'Invoking VPC \'create_access_token\' operation: http://someurl.com/instance_identity/v1/token'
111+
112+
@responses.activate
113+
def test_request_token_with_crn(caplog):
114+
caplog.set_level(logging.DEBUG)
115+
116+
token_manager = VPCInstanceTokenManager(
117+
iam_profile_crn=TEST_IAM_PROFILE_CRN,
118+
)
119+
120+
# Mock the retrieve instance identity token method.
121+
def mock_retrieve_instance_identity_token():
122+
return TEST_TOKEN
123+
token_manager.retrieve_instance_identity_token = mock_retrieve_instance_identity_token
124+
125+
response = {
126+
'access_token': TEST_IAM_TOKEN,
127+
}
128+
129+
responses.add(responses.POST, 'http://169.254.169.254/instance_identity/v1/iam_token',
130+
body=json.dumps(response), status=200)
131+
132+
response = token_manager.request_token()
133+
assert len(responses.calls) == 1
134+
assert responses.calls[0].request.headers['Content-Type'] == 'application/json'
135+
assert responses.calls[0].request.headers['Accept'] == 'application/json'
136+
assert responses.calls[0].request.headers['Authorization'] == 'Bearer ' + TEST_TOKEN
137+
assert responses.calls[0].request.body == '{"trusted_profile": "{\\"crn\\": \\"crn:iam-profile:123\\"}"}'
138+
assert responses.calls[0].request.params['version'] == '2021-09-20'
139+
# Check the logs.
140+
#pylint: disable=line-too-long
141+
assert caplog.record_tuples[0][2] == 'Invoking VPC \'create_iam_token\' operation: http://169.254.169.254/instance_identity/v1/iam_token'
142+
assert caplog.record_tuples[1][2] == 'Returned from VPC \'create_iam_token\' operation."'
143+
144+
145+
@responses.activate
146+
def test_request_token_with_id(caplog):
147+
caplog.set_level(logging.DEBUG)
148+
149+
token_manager = VPCInstanceTokenManager(
150+
iam_profile_id=TEST_IAM_PROFILE_ID,
151+
)
152+
153+
# Mock the retrieve instance identity token method.
154+
def mock_retrieve_instance_identity_token():
155+
return TEST_TOKEN
156+
token_manager.retrieve_instance_identity_token = mock_retrieve_instance_identity_token
157+
158+
response = {
159+
'access_token': TEST_IAM_TOKEN,
160+
}
161+
162+
responses.add(responses.POST, 'http://169.254.169.254/instance_identity/v1/iam_token',
163+
body=json.dumps(response), status=200)
164+
165+
response = token_manager.request_token()
166+
assert len(responses.calls) == 1
167+
assert responses.calls[0].request.headers['Content-Type'] == 'application/json'
168+
assert responses.calls[0].request.headers['Accept'] == 'application/json'
169+
assert responses.calls[0].request.headers['Authorization'] == 'Bearer ' + TEST_TOKEN
170+
assert responses.calls[0].request.body == '{"trusted_profile": "{\\"id\\": \\"iam-id-123\\"}"}'
171+
assert responses.calls[0].request.params['version'] == '2021-09-20'
172+
# Check the logs.
173+
#pylint: disable=line-too-long
174+
assert caplog.record_tuples[0][2] == 'Invoking VPC \'create_iam_token\' operation: http://169.254.169.254/instance_identity/v1/iam_token'
175+
assert caplog.record_tuples[1][2] == 'Returned from VPC \'create_iam_token\' operation."'
176+
177+
178+
@responses.activate
179+
def test_request_token_with_failed(caplog):
180+
caplog.set_level(logging.DEBUG)
181+
182+
token_manager = VPCInstanceTokenManager(
183+
iam_profile_id=TEST_IAM_PROFILE_ID,
184+
)
185+
186+
# Mock the retrieve instance identity token method.
187+
def mock_retrieve_instance_identity_token():
188+
return TEST_TOKEN
189+
token_manager.retrieve_instance_identity_token = mock_retrieve_instance_identity_token
190+
191+
response = {
192+
'errors': ['Ooops'],
193+
}
194+
195+
responses.add(responses.POST, 'http://169.254.169.254/instance_identity/v1/iam_token',
196+
body=json.dumps(response), status=400)
197+
198+
with pytest.raises(ApiException):
199+
token_manager.request_token()
200+
assert len(responses.calls) == 1
201+
# Check the logs.
202+
#pylint: disable=line-too-long
203+
assert caplog.record_tuples[0][2] == 'Invoking VPC \'create_iam_token\' operation: http://169.254.169.254/instance_identity/v1/iam_token'

0 commit comments

Comments
 (0)