test: cover the container auth related code

Author: pyrooka

resources/cr-token.txt

@@ -0,0 +1 @@
+cr-token-1

resources/ibm-credentials-container.env

@@ -0,0 +1,9 @@
+SERVICE_1_AUTH_TYPE=container
+SERVICE_1_CR_TOKEN_FILENAME=crtoken.txt
+SERVICE_1_IAM_PROFILE_NAME=iam-user1
+SERVICE_1_IAM_PROFILE_ID=iam-id1
+SERVICE_1_AUTH_URL=https://iamhost/iam/api
+SERVICE_1_SCOPE=scope1
+SERVICE_1_CLIENT_ID=iam-client1
+SERVICE_1_CLIENT_SECRET=iam-secret1
+SERVICE_1_AUTH_DISABLE_SSL=true

test/test_container_authenticator.py

@@ -0,0 +1,67 @@
+# pylint: disable=missing-docstring
+import pytest
+
+from ibm_cloud_sdk_core.authenticators import ContainerAuthenticator
+
+
+def test_container_authenticator():
+    authenticator = ContainerAuthenticator(iam_profile_name='iam-user-123')
+    assert authenticator is not None
+    assert authenticator.token_manager.client_id is None
+    assert authenticator.token_manager.client_secret is None
+    assert authenticator.token_manager.disable_ssl_verification is False
+    assert authenticator.token_manager.headers is None
+    assert authenticator.token_manager.proxies is None
+    assert authenticator.token_manager.scope is None
+
+    authenticator.set_cr_token_filename('path/to/token')
+    assert authenticator.token_manager.cr_token_filename == 'path/to/token'
+
+    authenticator.set_iam_profile_name('iam-user-123')
+    assert authenticator.token_manager.iam_profile_name == 'iam-user-123'
+
+    authenticator.set_iam_profile_id('iam-id-123')
+    assert authenticator.token_manager.iam_profile_id == 'iam-id-123'
+
+    authenticator.set_client_id_and_secret('tom', 'jerry')
+    assert authenticator.token_manager.client_id == 'tom'
+    assert authenticator.token_manager.client_secret == 'jerry'
+
+    authenticator.set_scope('scope1 scope2 scope3')
+    assert authenticator.token_manager.scope == 'scope1 scope2 scope3'
+
+    with pytest.raises(TypeError) as err:
+        authenticator.set_headers('dummy')
+    assert str(err.value) == 'headers must be a dictionary'
+
+    authenticator.set_headers({'dummy': 'headers'})
+    assert authenticator.token_manager.headers == {'dummy': 'headers'}
+
+    with pytest.raises(TypeError) as err:
+        authenticator.set_proxies('dummy')
+    assert str(err.value) == 'proxies must be a dictionary'
+
+    authenticator.set_proxies({'dummy': 'proxies'})
+    assert authenticator.token_manager.proxies == {'dummy': 'proxies'}
+
+
+def test_container_authenticator_with_scope():
+    authenticator = ContainerAuthenticator(iam_profile_name='iam-user-123', scope='scope1 scope2')
+    assert authenticator is not None
+    assert authenticator.token_manager.scope == 'scope1 scope2'
+
+
+def test_authenticator_validate_failed():
+    with pytest.raises(ValueError) as err:
+        ContainerAuthenticator(None)
+    assert str(err.value) == 'At least one of iam_profile_name or iam_profile_id must be specified.'
+
+    with pytest.raises(ValueError) as err:
+        ContainerAuthenticator(iam_profile_name='iam-user-123', client_id='my_client_id')
+    assert str(
+        err.value) == 'Both client_id and client_secret should be initialized.'
+
+    with pytest.raises(ValueError) as err:
+        ContainerAuthenticator(iam_profile_name='iam-user-123', client_secret='my_client_secret')
+    assert str(
+        err.value) == 'Both client_id and client_secret should be initialized.'

test/test_container_token_manager.py

@@ -0,0 +1,259 @@
+# pylint: disable=missing-docstring
+import json
+import time
+from urllib.parse import parse_qs
+
+import responses
+import pytest
+
+from ibm_cloud_sdk_core import ApiException, ContainerTokenManager
+from ibm_cloud_sdk_core.authenticators import ContainerAuthenticator
+
+# pylint: disable=line-too-long
+TEST_ACCESS_TOKEN_1 = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VybmFtZSI6ImhlbGxvIiwicm9sZSI6InVzZXIiLCJwZXJtaXNzaW9ucyI6WyJhZG1pbmlzdHJhdG9yIiwiZGVwbG95bWVudF9hZG1pbiJdLCJzdWIiOiJoZWxsbyIsImlzcyI6IkpvaG4iLCJhdWQiOiJEU1giLCJ1aWQiOiI5OTkiLCJpYXQiOjE1NjAyNzcwNTEsImV4cCI6MTU2MDI4MTgxOSwianRpIjoiMDRkMjBiMjUtZWUyZC00MDBmLTg2MjMtOGNkODA3MGI1NDY4In0.cIodB4I6CCcX8vfIImz7Cytux3GpWyObt9Gkur5g1QI'
+TEST_ACCESS_TOKEN_2 = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiIsImtpZCI6IjIzMDQ5ODE1MWMyMTRiNzg4ZGQ5N2YyMmI4NTQxMGE1In0.eyJ1c2VybmFtZSI6ImR1bW15Iiwicm9sZSI6IkFkbWluIiwicGVybWlzc2lvbnMiOlsiYWRtaW5pc3RyYXRvciIsIm1hbmFnZV9jYXRhbG9nIl0sInN1YiI6ImFkbWluIiwiaXNzIjoic3NzIiwiYXVkIjoic3NzIiwidWlkIjoic3NzIiwiaWF0IjozNjAwLCJleHAiOjE2MjgwMDcwODF9.zvUDpgqWIWs7S1CuKv40ERw1IZ5FqSFqQXsrwZJyfRM'
+TEST_REFRESH_TOKEN = 'Xj7Gle500MachEOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VybmFtZSI6ImhlbGxvIiwicm9sZSI6InVzZXIiLCJwZXJtaXNzaW9ucyI6WyJhZG1pbmlzdHJhdG9yIiwiZGVwbG95bWVudF9hZG1pbiJdLCJzdWIiOiJoZWxsbyIsImlzcyI6IkpvaG4iLCJhdWQiOiJEU1giLCJ1aWQiOiI5OTkiLCJpYXQiOjE1NjAyNzcwNTEsImV4cCI6MTU2MDI4MTgxOSwianRpIjoiMDRkMjBiMjUtZWUyZC00MDBmLTg2MjMtOGNkODA3MGI1NDY4In0.cIodB4I6CCcX8vfIImz7Cytux3GpWyObt9Gkur5g1QI'
+MOCK_CR_TOKEN_FILE = './resources/cr-token.txt'
+MOCK_IAM_PROFILE_NAME = 'iam-user-123'
+MOCK_CLIENT_ID = 'client-id-1'
+MOCK_CLIENT_SECRET = 'client-secret-1'
+
+
+def _get_current_time() -> int:
+    return int(time.time())
+
+
+def mock_iam_response(func):
+    """This is decorator function which extends `responses.activate`.
+    This sets up all the mock response stuffs.
+    """
+    def callback(request):
+        assert request.headers['Accept'] == 'application/json'
+        assert request.headers['Content-Type'] == 'application/x-www-form-urlencoded'
+
+        payload = parse_qs(request.body)
+
+        assert payload['cr_token'][0] == 'cr-token-1'
+        assert payload['grant_type'][0] == 'urn:ibm:params:oauth:grant-type:cr-token'
+        assert payload['profile_name'][0] or payload['iam_profile_id']
+
+        status_code = 200
+
+        scope = payload.get('scope')[0] if payload.get('scope') else None
+        if scope == 'send-second-token':
+            access_token = TEST_ACCESS_TOKEN_2
+        elif scope == 'status-bad-request':
+            access_token = None
+            status_code = 400
+        elif scope == 'check-basic-auth':
+            assert request.headers['Authorization'] == 'Basic Y2xpZW50LWlkLTE6Y2xpZW50LXNlY3JldC0x'
+            access_token = TEST_ACCESS_TOKEN_1
+        else:
+            access_token = TEST_ACCESS_TOKEN_1
+
+        response = json.dumps({
+            'access_token': access_token,
+            'token_type': 'Bearer',
+            'expires_in': 3600,
+            'expiration': _get_current_time()+3600,
+            'refresh_token': TEST_REFRESH_TOKEN,
+        })
+
+        return (status_code, {}, response)
+
+    @responses.activate
+    def wrapper():
+        response = responses.CallbackResponse(
+            method=responses.POST,
+            url='https://iam.cloud.ibm.com/identity/token',
+            callback=callback,
+        )
+
+        responses.add(response)
+
+        func()
+
+    return wrapper
+
+
+@mock_iam_response
+def test_request_token_auth_default():
+    iam_url = "https://iam.cloud.ibm.com/identity/token"
+
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+    )
+    token_manager.request_token()
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.url == iam_url
+    assert responses.calls[0].request.headers.get('Authorization') is None
+    assert json.loads(responses.calls[0].response.text)['access_token'] == TEST_ACCESS_TOKEN_1
+
+
+@mock_iam_response
+def test_request_token_auth_in_ctor():
+    default_auth_header = 'Basic Yng6Yng='
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+        client_id='foo',
+        client_secret='bar')
+
+    token_manager.request_token()
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.headers['Authorization'] != default_auth_header
+    assert json.loads(responses.calls[0].response.text)['access_token'] == TEST_ACCESS_TOKEN_1
+    assert 'scope' not in responses.calls[0].response.request.body
+
+
+@mock_iam_response
+def test_request_token_auth_in_ctor_with_scope():
+    default_auth_header = 'Basic Yng6Yng='
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+        client_id='foo',
+        client_secret='bar',
+        scope='john snow')
+
+    token_manager.request_token()
+
+    assert len(responses.calls) == 1
+    assert responses.calls[0].request.headers['Authorization'] != default_auth_header
+    assert json.loads(responses.calls[0].response.text)['access_token'] == TEST_ACCESS_TOKEN_1
+    assert 'scope=john+snow' in responses.calls[0].response.request.body
+
+
+def test_retrieve_cr_token_success():
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+    )
+
+    cr_token = token_manager.retrieve_cr_token()
+
+    assert cr_token == 'cr-token-1'
+
+
+def test_retrieve_cr_token_success_fail():
+    token_manager = ContainerTokenManager(
+        cr_token_filename='bogus-cr-token-file',
+    )
+
+    with pytest.raises(Exception) as err:
+        token_manager.retrieve_cr_token()
+
+    assert str(err.value) == 'Unable to retrieve the CR token value from file bogus-cr-token-file: [Errno 2] No such file or directory: \'bogus-cr-token-file\''
+
+
+@mock_iam_response
+def test_get_token_success():
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+    )
+
+    cr_token = token_manager.access_token
+    assert cr_token is None
+
+    cr_token = token_manager.get_token()
+    assert cr_token == TEST_ACCESS_TOKEN_1
+    assert token_manager.access_token == TEST_ACCESS_TOKEN_1
+
+    # Verify the token manager return the cached value.
+    # Before we call the `get_token` again, set the expiration and time.
+    # This is necessary because we are using a fix JWT response.
+    token_manager.expire_time = _get_current_time() + 3600
+    token_manager.refresh_time = _get_current_time() + 3600
+    token_manager.set_scope('send-second-token')
+    cr_token = token_manager.get_token()
+    assert cr_token == TEST_ACCESS_TOKEN_1
+    assert token_manager.access_token == TEST_ACCESS_TOKEN_1
+
+    # Force expiration to get the second token.
+    token_manager.expire_time = _get_current_time() - 1
+    cr_token = token_manager.get_token()
+    assert cr_token == TEST_ACCESS_TOKEN_2
+    assert token_manager.access_token == TEST_ACCESS_TOKEN_2
+
+
+@mock_iam_response
+def test_request_token_success():
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+    )
+
+    token_response = token_manager.request_token()
+    assert token_response['access_token'] == TEST_ACCESS_TOKEN_1
+
+
+@mock_iam_response
+def test_authenticate_success():
+    authenticator = ContainerAuthenticator(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name='iam-user-123')
+
+    request = {'headers': {}}
+
+    authenticator.authenticate(request)
+    assert request['headers']['Authorization'] == 'Bearer ' + TEST_ACCESS_TOKEN_1
+
+    # Verify the token manager return the cached value.
+    # Before we call the `get_token` again, set the expiration and time.
+    # This is necessary because we are using a fix JWT response.
+    authenticator.token_manager.expire_time = _get_current_time() + 3600
+    authenticator.token_manager.refresh_time = _get_current_time() + 3600
+    authenticator.token_manager.set_scope('send-second-token')
+    authenticator.authenticate(request)
+    assert request['headers']['Authorization'] == 'Bearer ' + TEST_ACCESS_TOKEN_1
+
+    # Force expiration to get the second token.
+    authenticator.token_manager.expire_time = _get_current_time() - 1
+    authenticator.authenticate(request)
+    assert request['headers']['Authorization'] == 'Bearer ' + TEST_ACCESS_TOKEN_2
+
+
+@mock_iam_response
+def test_authenticate_fail_no_cr_token():
+    authenticator = ContainerAuthenticator(
+        cr_token_filename='bogus-cr-token-file',
+        iam_profile_name='iam-user-123',
+        url='https://bogus.iam.endpoint')
+
+    request = {'headers': {}}
+
+    with pytest.raises(Exception) as err:
+        authenticator.authenticate(request)
+
+    assert str(err.value) == 'Unable to retrieve the CR token value from file bogus-cr-token-file: [Errno 2] No such file or directory: \'bogus-cr-token-file\''
+
+
+@mock_iam_response
+def test_authenticate_fail_iam():
+    authenticator = ContainerAuthenticator(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name='iam-user-123',
+        scope='status-bad-request')
+
+    request = {'headers': {}}
+
+    with pytest.raises(ApiException) as err:
+        authenticator.authenticate(request)
+
+    assert str(err.value) == 'Error: Bad Request, Code: 400'
+
+
+@mock_iam_response
+def test_client_id_and_secret():
+    token_manager = ContainerTokenManager(
+        cr_token_filename=MOCK_CR_TOKEN_FILE,
+        iam_profile_name=MOCK_IAM_PROFILE_NAME,
+    )
+
+    token_manager.set_client_id_and_secret(MOCK_CLIENT_ID, MOCK_CLIENT_SECRET)
+    token_manager.set_scope('check-basic-auth')
+    access_token = token_manager.get_token()
+    assert access_token == TEST_ACCESS_TOKEN_1

test/test_utils.py

@@ -291,6 +291,15 @@ def test_get_authenticator_from_credential_file():
     assert authenticator.username == 'my_username'
     del os.environ['IBM_CREDENTIALS_FILE']
 
+    file_path = os.path.join(os.path.dirname(__file__),
+                             '../resources/ibm-credentials-container.env')
+    os.environ['IBM_CREDENTIALS_FILE'] = file_path
+    authenticator = get_authenticator_from_environment('service 1')
+    assert authenticator is not None
+    assert authenticator.token_manager.cr_token_filename == 'crtoken.txt'
+    assert authenticator.token_manager.iam_profile_name == 'iam-user1'
+    del os.environ['IBM_CREDENTIALS_FILE']
+
     file_path = os.path.join(os.path.dirname(__file__),
                              '../resources/ibm-credentials-cp4d.env')
     os.environ['IBM_CREDENTIALS_FILE'] = file_path