path: root/tests/test_auth.py

import contextlib

from testtools import TestCase
from reddwarfclient import auth
from mock import Mock

from reddwarfclient import exceptions

"""
Unit tests for the classes and functions in auth.py.
"""


def check_url_none(test_case, auth_class):
    # url is None, it must throw exception
    authObj = auth_class(url=None, type=auth_class, client=None,
                         username=None, password=None, tenant=None)
    try:
        authObj.authenticate()
        test_case.fail("AuthUrlNotGiven exception expected")
    except exceptions.AuthUrlNotGiven:
        pass


class AuthenticatorTest(TestCase):

    def setUp(self):
        super(AuthenticatorTest, self).setUp()
        self.orig_load = auth.ServiceCatalog._load
        self.orig__init = auth.ServiceCatalog.__init__

    def tearDown(self):
        super(AuthenticatorTest, self).tearDown()
        auth.ServiceCatalog._load = self.orig_load
        auth.ServiceCatalog.__init__ = self.orig__init

    def test_get_authenticator_cls(self):
        class_list = (auth.KeyStoneV2Authenticator,
                      auth.RaxAuthenticator,
                      auth.Auth1_1,
                      auth.FakeAuth)

        for c in class_list:
            self.assertEqual(c, auth.get_authenticator_cls(c))

        class_names = {"keystone": auth.KeyStoneV2Authenticator,
                       "rax": auth.RaxAuthenticator,
                       "auth1.1": auth.Auth1_1,
                       "fake": auth.FakeAuth}

        for cn in class_names.keys():
            self.assertEqual(class_names[cn], auth.get_authenticator_cls(cn))

        cls_or_name = "_unknown_"
        self.assertRaises(ValueError, auth.get_authenticator_cls, cls_or_name)

    def test__authenticate(self):
        authObj = auth.Authenticator(Mock(), auth.KeyStoneV2Authenticator,
                                     Mock(), Mock(), Mock(), Mock())
        # test response code 200
        resp = Mock()
        resp.status = 200
        body = "test_body"

        auth.ServiceCatalog._load = Mock(return_value=1)
        authObj.client._time_request = Mock(return_value=(resp, body))

        sc = authObj._authenticate(Mock(), Mock())
        self.assertEqual(body, sc.catalog)

        # test AmbiguousEndpoints exception
        auth.ServiceCatalog.__init__ = \
            Mock(side_effect=exceptions.AmbiguousEndpoints)
        self.assertRaises(exceptions.AmbiguousEndpoints,
                          authObj._authenticate, Mock(), Mock())

        # test handling KeyError and raising AuthorizationFailure exception
        auth.ServiceCatalog.__init__ = Mock(side_effect=KeyError)
        self.assertRaises(exceptions.AuthorizationFailure,
                          authObj._authenticate, Mock(), Mock())

        # test EndpointNotFound exception
        mock = Mock(side_effect=exceptions.EndpointNotFound)
        auth.ServiceCatalog.__init__ = mock
        self.assertRaises(exceptions.EndpointNotFound,
                          authObj._authenticate, Mock(), Mock())
        mock.side_effect = None

        # test response code 305
        resp.__getitem__ = Mock(return_value='loc')
        resp.status = 305
        body = "test_body"
        authObj.client._time_request = Mock(return_value=(resp, body))

        l = authObj._authenticate(Mock(), Mock())
        self.assertEqual('loc', l)

        # test any response code other than 200 and 305
        resp.status = 404
        exceptions.from_response = Mock(side_effect=ValueError)
        self.assertRaises(ValueError, authObj._authenticate, Mock(), Mock())

    def test_authenticate(self):
        authObj = auth.Authenticator(Mock(), auth.KeyStoneV2Authenticator,
                                     Mock(), Mock(), Mock(), Mock())
        self.assertRaises(NotImplementedError, authObj.authenticate)


class KeyStoneV2AuthenticatorTest(TestCase):

    def test_authenticate(self):
        # url is None
        check_url_none(self, auth.KeyStoneV2Authenticator)

        # url is not None, so it must not throw exception
        url = "test_url"
        cls_type = auth.KeyStoneV2Authenticator
        authObj = auth.KeyStoneV2Authenticator(url=url, type=cls_type,
                                               client=None, username=None,
                                               password=None, tenant=None)

        def side_effect_func(url):
            return url

        mock = Mock()
        mock.side_effect = side_effect_func
        authObj._v2_auth = mock
        r = authObj.authenticate()
        self.assertEqual(url, r)

    def test__v2_auth(self):
        username = "reddwarf_user"
        password = "reddwarf_password"
        tenant = "tenant"
        cls_type = auth.KeyStoneV2Authenticator
        authObj = auth.KeyStoneV2Authenticator(url=None, type=cls_type,
                                               client=None,
                                               username=username,
                                               password=password,
                                               tenant=tenant)

        def side_effect_func(url, body):
            return body
        mock = Mock()
        mock.side_effect = side_effect_func
        authObj._authenticate = mock
        body = authObj._v2_auth(Mock())
        self.assertEqual(username,
                         body['auth']['passwordCredentials']['username'])
        self.assertEqual(password,
                         body['auth']['passwordCredentials']['password'])
        self.assertEqual(tenant, body['auth']['tenantName'])


class Auth1_1Test(TestCase):

    def test_authenticate(self):
        # handle when url is None
        check_url_none(self, auth.Auth1_1)

        # url is not none
        username = "reddwarf_user"
        password = "reddwarf_password"
        url = "test_url"
        authObj = auth.Auth1_1(url=url,
                               type=auth.Auth1_1,
                               client=None, username=username,
                               password=password, tenant=None)

        def side_effect_func(auth_url, body, root_key):
            return auth_url, body, root_key

        mock = Mock()
        mock.side_effect = side_effect_func
        authObj._authenticate = mock
        auth_url, body, root_key = authObj.authenticate()

        self.assertEqual(username, body['credentials']['username'])
        self.assertEqual(password, body['credentials']['key'])
        self.assertEqual(auth_url, url)
        self.assertEqual('auth', root_key)


class RaxAuthenticatorTest(TestCase):

    def test_authenticate(self):
        # url is None
        check_url_none(self, auth.RaxAuthenticator)

        # url is not None, so it must not throw exception
        url = "test_url"
        authObj = auth.RaxAuthenticator(url=url,
                                        type=auth.RaxAuthenticator,
                                        client=None, username=None,
                                        password=None, tenant=None)

        def side_effect_func(url):
            return url

        mock = Mock()
        mock.side_effect = side_effect_func
        authObj._rax_auth = mock
        r = authObj.authenticate()
        self.assertEqual(url, r)

    def test__rax_auth(self):
        username = "reddwarf_user"
        password = "reddwarf_password"
        tenant = "tenant"
        authObj = auth.RaxAuthenticator(url=None,
                                        type=auth.RaxAuthenticator,
                                        client=None, username=username,
                                        password=password, tenant=tenant)

        def side_effect_func(url, body):
            return body

        mock = Mock()
        mock.side_effect = side_effect_func
        authObj._authenticate = mock
        body = authObj._rax_auth(Mock())

        v = body['auth']['RAX-KSKEY:apiKeyCredentials']['username']
        self.assertEqual(username, v)

        v = body['auth']['RAX-KSKEY:apiKeyCredentials']['apiKey']
        self.assertEqual(password, v)

        v = body['auth']['RAX-KSKEY:apiKeyCredentials']['tenantName']
        self.assertEqual(tenant, v)


class FakeAuthTest(TestCase):

    def test_authenticate(self):
        tenant = "tenant"
        authObj = auth.FakeAuth(url=None,
                                type=auth.FakeAuth,
                                client=None, username=None,
                                password=None, tenant=tenant)

        fc = authObj.authenticate()
        public_url = "%s/%s" % ('http://localhost:8779/v1.0', tenant)
        self.assertEqual(public_url, fc.get_public_url())
        self.assertEqual(tenant, fc.get_token())


class ServiceCatalogTest(TestCase):

    def setUp(self):
        super(ServiceCatalogTest, self).setUp()
        self.orig_url_for = auth.ServiceCatalog._url_for
        self.orig__init__ = auth.ServiceCatalog.__init__
        auth.ServiceCatalog.__init__ = Mock(return_value=None)
        self.test_url = "http://localhost:1234/test"

    def tearDown(self):
        super(ServiceCatalogTest, self).tearDown()
        auth.ServiceCatalog._url_for = self.orig_url_for
        auth.ServiceCatalog.__init__ = self.orig__init__

    def test__load(self):
        url = "random_url"
        auth.ServiceCatalog._url_for = Mock(return_value=url)

        # when service_url is None
        scObj = auth.ServiceCatalog()
        scObj.region = None
        scObj.service_url = None
        scObj._load()
        self.assertEqual(url, scObj.public_url)
        self.assertEqual(url, scObj.management_url)

        # service url is not None
        service_url = "service_url"
        scObj = auth.ServiceCatalog()
        scObj.region = None
        scObj.service_url = service_url
        scObj._load()
        self.assertEqual(service_url, scObj.public_url)
        self.assertEqual(service_url, scObj.management_url)

    def test_get_token(self):
        test_id = "test_id"
        scObj = auth.ServiceCatalog()
        scObj.root_key = "root_key"
        scObj.catalog = dict()
        scObj.catalog[scObj.root_key] = dict()
        scObj.catalog[scObj.root_key]['token'] = dict()
        scObj.catalog[scObj.root_key]['token']['id'] = test_id
        self.assertEqual(test_id, scObj.get_token())

    def test_get_management_url(self):
        test_mng_url = "test_management_url"
        scObj = auth.ServiceCatalog()
        scObj.management_url = test_mng_url
        self.assertEqual(test_mng_url, scObj.get_management_url())

    def test_get_public_url(self):
        test_public_url = "test_public_url"
        scObj = auth.ServiceCatalog()
        scObj.public_url = test_public_url
        self.assertEqual(test_public_url, scObj.get_public_url())

    def test__url_for(self):
        scObj = auth.ServiceCatalog()

        # case for no endpoint found
        self.case_no_endpoint_match(scObj)

        # case for empty service catalog
        self.case_endpoing_with_empty_catalog(scObj)

        # more than one matching endpoints
        self.case_ambiguous_endpoint(scObj)

        # happy case
        self.case_unique_endpoint(scObj)

        # testing if-statements in for-loop to iterate services in catalog
        self.case_iterating_services_in_catalog(scObj)

    def case_no_endpoint_match(self, scObj):
        # empty endpoint list
        scObj.catalog = dict()
        scObj.catalog['endpoints'] = list()
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)

        def side_effect_func_ep(attr):
            return "test_attr_value"

        # simulating dict
        endpoint = Mock()
        mock = Mock()
        mock.side_effect = side_effect_func_ep
        endpoint.__getitem__ = mock
        scObj.catalog['endpoints'].append(endpoint)

        # not-empty list but not matching endpoint
        filter_value = "not_matching_value"
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
                          attr="test_attr", filter_value=filter_value)

        filter_value = "test_attr_value"  # so that we have an endpoint match
        scObj.root_key = "access"
        scObj.catalog[scObj.root_key] = dict()
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
                          attr="test_attr", filter_value=filter_value)

    def case_endpoing_with_empty_catalog(self, scObj):
        # first, test with empty catalog, this should pass since
        # there is already enpoint added
        scObj.catalog[scObj.root_key]['serviceCatalog'] = list()

        endpoint = scObj.catalog['endpoints'][0]
        endpoint.get = Mock(return_value=self.test_url)
        r_url = scObj._url_for(attr="test_attr",
                               filter_value="test_attr_value")
        self.assertEqual(self.test_url, r_url)

    def case_ambiguous_endpoint(self, scObj):
        scObj.service_type = "reddwarf"
        scObj.service_name = "test_service_name"

        def side_effect_func_service(key):
            if key == "type":
                return "reddwarf"
            elif key == "name":
                return "test_service_name"
            return None

        mock1 = Mock()
        mock1.side_effect = side_effect_func_service
        service1 = Mock()
        service1.get = mock1

        endpoint2 = {"test_attr": "test_attr_value"}
        service1.__getitem__ = Mock(return_value=[endpoint2])
        scObj.catalog[scObj.root_key]['serviceCatalog'] = [service1]
        self.assertRaises(exceptions.AmbiguousEndpoints, scObj._url_for,
                          attr="test_attr", filter_value="test_attr_value")

    def case_unique_endpoint(self, scObj):
        # changing the endpoint2 attribute to pass the filter
        service1 = scObj.catalog[scObj.root_key]['serviceCatalog'][0]
        endpoint2 = service1[0][0]
        endpoint2["test_attr"] = "new value not matching filter"
        r_url = scObj._url_for(attr="test_attr",
                               filter_value="test_attr_value")
        self.assertEqual(self.test_url, r_url)

    def case_iterating_services_in_catalog(self, scObj):
        service1 = scObj.catalog[scObj.root_key]['serviceCatalog'][0]

        scObj.catalog = dict()
        scObj.root_key = "access"
        scObj.catalog[scObj.root_key] = dict()
        scObj.service_type = "no_match"

        scObj.catalog[scObj.root_key]['serviceCatalog'] = [service1]
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)

        scObj.service_type = "reddwarf"
        scObj.service_name = "no_match"
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for)

        # no endpoints and no 'serviceCatalog' in catalog => raise exception
        scObj = auth.ServiceCatalog()
        scObj.catalog = dict()
        scObj.root_key = "access"
        scObj.catalog[scObj.root_key] = dict()
        scObj.catalog[scObj.root_key]['serviceCatalog'] = []
        self.assertRaises(exceptions.EndpointNotFound, scObj._url_for,
                          attr="test_attr", filter_value="test_attr_value")