# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import sys
import datetime

try:
    import simplejson as json
except ImportError:
    import json

from mock import Mock

from libcloud.utils.py3 import httplib
from libcloud.common.openstack import OpenStackBaseConnection
from libcloud.common.openstack_identity import AUTH_TOKEN_EXPIRES_GRACE_SECONDS
from libcloud.common.openstack_identity import get_class_for_auth_version
from libcloud.common.openstack_identity import OpenStackServiceCatalog
from libcloud.common.openstack_identity import OpenStackIdentity_2_0_Connection
from libcloud.common.openstack_identity import OpenStackIdentity_3_0_Connection
from libcloud.common.openstack_identity import OpenStackIdentityUser
from libcloud.compute.drivers.openstack import OpenStack_1_0_NodeDriver

from libcloud.test import unittest
from libcloud.test import MockHttp
from libcloud.test.secrets import OPENSTACK_PARAMS
from libcloud.test.file_fixtures import ComputeFileFixtures
from libcloud.test.compute.test_openstack import OpenStackMockHttp
from libcloud.test.compute.test_openstack import OpenStack_2_0_MockHttp


class OpenStackIdentityConnectionTestCase(unittest.TestCase):
    def setUp(self):
        OpenStackBaseConnection.auth_url = None
        OpenStackBaseConnection.conn_classes = (OpenStackMockHttp,
                                                OpenStackMockHttp)

    def test_auth_url_is_correctly_assembled(self):
        tuples = [
            ('1.0', OpenStackMockHttp),
            ('1.1', OpenStackMockHttp),
            ('2.0', OpenStack_2_0_MockHttp),
            ('2.0_apikey', OpenStack_2_0_MockHttp),
            ('2.0_password', OpenStack_2_0_MockHttp)
        ]

        APPEND = 0
        NOTAPPEND = 1

        auth_urls = [
            ('https://auth.api.example.com', APPEND, ''),
            ('https://auth.api.example.com/', NOTAPPEND, '/'),
            ('https://auth.api.example.com/foo/bar', NOTAPPEND, '/foo/bar'),
            ('https://auth.api.example.com/foo/bar/', NOTAPPEND, '/foo/bar/')
        ]

        actions = {
            '1.0': '/v1.0',
            '1.1': '/v1.1/auth',
            '2.0': '/v2.0/tokens',
            '2.0_apikey': '/v2.0/tokens',
            '2.0_password': '/v2.0/tokens'
        }

        user_id = OPENSTACK_PARAMS[0]
        key = OPENSTACK_PARAMS[1]

        for (auth_version, mock_http_class) in tuples:
            for (url, should_append_default_path, expected_path) in auth_urls:
                connection = \
                    self._get_mock_connection(mock_http_class=mock_http_class,
                                              auth_url=url)

                auth_url = connection.auth_url
                cls = get_class_for_auth_version(auth_version=auth_version)
                osa = cls(auth_url=auth_url,
                          user_id=user_id,
                          key=key,
                          parent_conn=connection)

                try:
                    osa = osa.authenticate()
                except:
                    pass

                if (should_append_default_path == APPEND):
                    expected_path = actions[auth_version]

                self.assertEqual(osa.action, expected_path)

    def test_basic_authentication(self):
        tuples = [
            ('1.0', OpenStackMockHttp),
            ('1.1', OpenStackMockHttp),
            ('2.0', OpenStack_2_0_MockHttp),
            ('2.0_apikey', OpenStack_2_0_MockHttp),
            ('2.0_password', OpenStack_2_0_MockHttp)
        ]

        user_id = OPENSTACK_PARAMS[0]
        key = OPENSTACK_PARAMS[1]

        for (auth_version, mock_http_class) in tuples:
            connection = \
                self._get_mock_connection(mock_http_class=mock_http_class)
            auth_url = connection.auth_url

            cls = get_class_for_auth_version(auth_version=auth_version)
            osa = cls(auth_url=auth_url, user_id=user_id, key=key,
                      parent_conn=connection)

            self.assertEqual(osa.urls, {})
            self.assertEqual(osa.auth_token, None)
            self.assertEqual(osa.auth_user_info, None)
            osa = osa.authenticate()

            self.assertTrue(len(osa.urls) >= 1)
            self.assertTrue(osa.auth_token is not None)

            if auth_version in ['1.1', '2.0', '2.0_apikey', '2.0_password']:
                self.assertTrue(osa.auth_token_expires is not None)

            if auth_version in ['2.0', '2.0_apikey', '2.0_password']:
                self.assertTrue(osa.auth_user_info is not None)

    def test_token_expiration_and_force_reauthentication(self):
        user_id = OPENSTACK_PARAMS[0]
        key = OPENSTACK_PARAMS[1]

        connection = self._get_mock_connection(OpenStack_2_0_MockHttp)
        auth_url = connection.auth_url

        yesterday = datetime.datetime.today() - datetime.timedelta(1)
        tomorrow = datetime.datetime.today() + datetime.timedelta(1)

        osa = OpenStackIdentity_2_0_Connection(auth_url=auth_url,
                                               user_id=user_id,
                                               key=key,
                                               parent_conn=connection)

        mocked_auth_method = Mock(wraps=osa._authenticate_2_0_with_body)
        osa._authenticate_2_0_with_body = mocked_auth_method

        # Force re-auth, expired token
        osa.auth_token = None
        osa.auth_token_expires = yesterday
        count = 5

        for i in range(0, count):
            osa.authenticate(force=True)

        self.assertEqual(mocked_auth_method.call_count, count)

        # No force reauth, expired token
        osa.auth_token = None
        osa.auth_token_expires = yesterday

        mocked_auth_method.call_count = 0
        self.assertEqual(mocked_auth_method.call_count, 0)

        for i in range(0, count):
            osa.authenticate(force=False)

        self.assertEqual(mocked_auth_method.call_count, 1)

        # No force reauth, valid / non-expired token
        osa.auth_token = None

        mocked_auth_method.call_count = 0
        self.assertEqual(mocked_auth_method.call_count, 0)

        for i in range(0, count):
            osa.authenticate(force=False)

            if i == 0:
                osa.auth_token_expires = tomorrow

        self.assertEqual(mocked_auth_method.call_count, 1)

        # No force reauth, valid / non-expired token which is about to expire in
        # less than AUTH_TOKEN_EXPIRES_GRACE_SECONDS
        soon = datetime.datetime.utcnow() + \
            datetime.timedelta(seconds=AUTH_TOKEN_EXPIRES_GRACE_SECONDS - 1)
        osa.auth_token = None

        mocked_auth_method.call_count = 0
        self.assertEqual(mocked_auth_method.call_count, 0)

        for i in range(0, count):
            if i == 0:
                osa.auth_token_expires = soon

            osa.authenticate(force=False)

        self.assertEqual(mocked_auth_method.call_count, 1)

    def _get_mock_connection(self, mock_http_class, auth_url=None):
        OpenStackBaseConnection.conn_classes = (mock_http_class,
                                                mock_http_class)

        if auth_url is None:
            auth_url = "https://auth.api.example.com"

        OpenStackBaseConnection.auth_url = auth_url
        connection = OpenStackBaseConnection(*OPENSTACK_PARAMS)

        connection._ex_force_base_url = "https://www.foo.com"
        connection.driver = OpenStack_1_0_NodeDriver(*OPENSTACK_PARAMS)

        return connection


class OpenStackIdentity_2_0_ConnectionTests(unittest.TestCase):
    def setUp(self):
        mock_cls = OpenStackIdentity_2_0_MockHttp
        mock_cls.type = None
        OpenStackIdentity_2_0_Connection.conn_classes = (mock_cls, mock_cls)

        self.auth_instance = OpenStackIdentity_2_0_Connection(auth_url='http://none',
                                                              user_id='test',
                                                              key='test',
                                                              tenant_name='test')
        self.auth_instance.auth_token = 'mock'

    def test_list_projects(self):
        result = self.auth_instance.list_projects()
        self.assertEqual(len(result), 2)
        self.assertEqual(result[0].id, 'a')
        self.assertEqual(result[0].name, 'test')
        self.assertEqual(result[0].description, 'test project')
        self.assertTrue(result[0].enabled)


class OpenStackIdentity_3_0_ConnectionTests(unittest.TestCase):
    def setUp(self):
        mock_cls = OpenStackIdentity_3_0_MockHttp
        mock_cls.type = None
        OpenStackIdentity_3_0_Connection.conn_classes = (mock_cls, mock_cls)

        self.auth_instance = OpenStackIdentity_3_0_Connection(auth_url='http://none',
                                                              user_id='test',
                                                              key='test',
                                                              tenant_name='test')
        self.auth_instance.auth_token = 'mock'

    def test_token_scope_argument(self):
        # Invalid token_scope value
        expected_msg = 'Invalid value for "token_scope" argument: foo'
        self.assertRaisesRegexp(ValueError, expected_msg,
                                OpenStackIdentity_3_0_Connection,
                                auth_url='http://none',
                                user_id='test',
                                key='test',
                                token_scope='foo')

        # Missing tenant_name
        expected_msg = 'Must provide tenant_name and domain_name argument'
        self.assertRaisesRegexp(ValueError, expected_msg,
                                OpenStackIdentity_3_0_Connection,
                                auth_url='http://none',
                                user_id='test',
                                key='test',
                                token_scope='project')

        # Missing domain_name
        expected_msg = 'Must provide domain_name argument'
        self.assertRaisesRegexp(ValueError, expected_msg,
                                OpenStackIdentity_3_0_Connection,
                                auth_url='http://none',
                                user_id='test',
                                key='test',
                                token_scope='domain',
                                domain_name=None)

        # Scope to project all ok
        OpenStackIdentity_3_0_Connection(auth_url='http://none',
                                         user_id='test',
                                         key='test',
                                         token_scope='project',
                                         tenant_name='test',
                                         domain_name='Default')
        # Scope to domain
        OpenStackIdentity_3_0_Connection(auth_url='http://none',
                                         user_id='test',
                                         key='test',
                                         token_scope='domain',
                                         tenant_name=None,
                                         domain_name='Default')

    def test_list_supported_versions(self):
        OpenStackIdentity_3_0_MockHttp.type = 'v3'

        versions = self.auth_instance.list_supported_versions()
        self.assertEqual(len(versions), 2)
        self.assertEqual(versions[0].version, 'v2.0')
        self.assertEqual(versions[0].url,
                         'http://192.168.18.100:5000/v2.0/')
        self.assertEqual(versions[1].version, 'v3.0')
        self.assertEqual(versions[1].url,
                         'http://192.168.18.100:5000/v3/')

    def test_list_domains(self):
        domains = self.auth_instance.list_domains()
        self.assertEqual(len(domains), 1)
        self.assertEqual(domains[0].id, 'default')
        self.assertEqual(domains[0].name, 'Default')
        self.assertTrue(domains[0].enabled)

    def test_list_projects(self):
        projects = self.auth_instance.list_projects()
        self.assertEqual(len(projects), 4)
        self.assertEqual(projects[0].id, 'a')
        self.assertEqual(projects[0].domain_id, 'default')
        self.assertTrue(projects[0].enabled)
        self.assertEqual(projects[0].description, 'Test project')

    def test_list_users(self):
        users = self.auth_instance.list_users()
        self.assertEqual(len(users), 12)
        self.assertEqual(users[0].id, 'a')
        self.assertEqual(users[0].domain_id, 'default')
        self.assertEqual(users[0].enabled, True)
        self.assertEqual(users[0].email, 'openstack-test@localhost')

    def test_list_roles(self):
        roles = self.auth_instance.list_roles()
        self.assertEqual(len(roles), 2)
        self.assertEqual(roles[1].id, 'b')
        self.assertEqual(roles[1].name, 'admin')

    def test_list_user_projects(self):
        user = self.auth_instance.list_users()[0]
        projects = self.auth_instance.list_user_projects(user=user)
        self.assertEqual(len(projects), 0)

    def test_list_user_domain_roles(self):
        user = self.auth_instance.list_users()[0]
        domain = self.auth_instance.list_domains()[0]
        roles = self.auth_instance.list_user_domain_roles(domain=domain,
                                                          user=user)
        self.assertEqual(len(roles), 1)
        self.assertEqual(roles[0].name, 'admin')

    def test_get_domain(self):
        domain = self.auth_instance.get_domain(domain_id='default')
        self.assertEqual(domain.name, 'Default')

    def test_create_user(self):
        user = self.auth_instance.create_user(email='test2@localhost', password='test1',
                                              name='test2', domain_id='default')

        self.assertEqual(user.id, 'c')
        self.assertEqual(user.name, 'test2')

    def test_enable_user(self):
        user = self.auth_instance.list_users()[0]
        result = self.auth_instance.enable_user(user=user)
        self.assertTrue(isinstance(result, OpenStackIdentityUser))

    def test_disable_user(self):
        user = self.auth_instance.list_users()[0]
        result = self.auth_instance.disable_user(user=user)
        self.assertTrue(isinstance(result, OpenStackIdentityUser))

    def test_grant_domain_role_to_user(self):
        domain = self.auth_instance.list_domains()[0]
        role = self.auth_instance.list_roles()[0]
        user = self.auth_instance.list_users()[0]

        result = self.auth_instance.grant_domain_role_to_user(domain=domain,
                                                              role=role,
                                                              user=user)
        self.assertTrue(result)

    def test_revoke_domain_role_from_user(self):
        domain = self.auth_instance.list_domains()[0]
        role = self.auth_instance.list_roles()[0]
        user = self.auth_instance.list_users()[0]

        result = self.auth_instance.revoke_domain_role_from_user(domain=domain,
                                                                 role=role,
                                                                 user=user)
        self.assertTrue(result)

    def test_grant_project_role_to_user(self):
        project = self.auth_instance.list_projects()[0]
        role = self.auth_instance.list_roles()[0]
        user = self.auth_instance.list_users()[0]

        result = self.auth_instance.grant_project_role_to_user(project=project,
                                                               role=role,
                                                               user=user)
        self.assertTrue(result)

    def test_revoke_project_role_from_user(self):
        project = self.auth_instance.list_projects()[0]
        role = self.auth_instance.list_roles()[0]
        user = self.auth_instance.list_users()[0]

        result = self.auth_instance.revoke_project_role_from_user(project=project,
                                                                  role=role,
                                                                  user=user)
        self.assertTrue(result)


class OpenStackServiceCatalogTestCase(unittest.TestCase):
    fixtures = ComputeFileFixtures('openstack')

    def test_parsing_auth_v1_1(self):
        data = self.fixtures.load('_v1_1__auth.json')
        data = json.loads(data)
        service_catalog = data['auth']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='1.0')
        entries = catalog.get_entries()
        self.assertEqual(len(entries), 3)

        entry = [e for e in entries if e.service_type == 'cloudFilesCDN'][0]
        self.assertEqual(entry.service_type, 'cloudFilesCDN')
        self.assertEqual(entry.service_name, None)
        self.assertEqual(len(entry.endpoints), 2)
        self.assertEqual(entry.endpoints[0].region, 'ORD')
        self.assertEqual(entry.endpoints[0].url,
                         'https://cdn2.clouddrive.com/v1/MossoCloudFS')
        self.assertEqual(entry.endpoints[0].endpoint_type, 'external')
        self.assertEqual(entry.endpoints[1].region, 'LON')
        self.assertEqual(entry.endpoints[1].endpoint_type, 'external')

    def test_parsing_auth_v2(self):
        data = self.fixtures.load('_v2_0__auth.json')
        data = json.loads(data)
        service_catalog = data['access']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='2.0')
        entries = catalog.get_entries()
        self.assertEqual(len(entries), 6)

        entry = [e for e in entries if e.service_name == 'cloudServers'][0]
        self.assertEqual(entry.service_type, 'compute')
        self.assertEqual(entry.service_name, 'cloudServers')
        self.assertEqual(len(entry.endpoints), 1)
        self.assertEqual(entry.endpoints[0].region, None)
        self.assertEqual(entry.endpoints[0].url,
                         'https://servers.api.rackspacecloud.com/v1.0/1337')
        self.assertEqual(entry.endpoints[0].endpoint_type, 'external')

    def test_parsing_auth_v3(self):
        data = self.fixtures.load('_v3__auth.json')
        data = json.loads(data)
        service_catalog = data['token']['catalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='3.x')
        entries = catalog.get_entries()
        self.assertEqual(len(entries), 6)
        entry = [e for e in entries if e.service_type == 'volume'][0]
        self.assertEqual(entry.service_type, 'volume')
        self.assertEqual(entry.service_name, None)
        self.assertEqual(len(entry.endpoints), 3)
        self.assertEqual(entry.endpoints[0].region, 'regionOne')
        self.assertEqual(entry.endpoints[0].endpoint_type, 'external')
        self.assertEqual(entry.endpoints[1].region, 'regionOne')
        self.assertEqual(entry.endpoints[1].endpoint_type, 'admin')
        self.assertEqual(entry.endpoints[2].region, 'regionOne')
        self.assertEqual(entry.endpoints[2].endpoint_type, 'internal')

    def test_get_public_urls(self):
        data = self.fixtures.load('_v2_0__auth.json')
        data = json.loads(data)
        service_catalog = data['access']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='2.0')

        public_urls = catalog.get_public_urls(service_type='object-store')
        expected_urls = ['https://storage101.lon1.clouddrive.com/v1/MossoCloudFS_11111-111111111-1111111111-1111111',
                         'https://storage101.ord1.clouddrive.com/v1/MossoCloudFS_11111-111111111-1111111111-1111111']
        self.assertEqual(public_urls, expected_urls)

    def test_get_regions(self):
        data = self.fixtures.load('_v2_0__auth.json')
        data = json.loads(data)
        service_catalog = data['access']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='2.0')

        regions = catalog.get_regions(service_type='object-store')
        self.assertEqual(regions, ['LON', 'ORD'])

        regions = catalog.get_regions(service_type='invalid')
        self.assertEqual(regions, [])

    def test_get_service_types(self):
        data = self.fixtures.load('_v2_0__auth.json')
        data = json.loads(data)
        service_catalog = data['access']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='2.0')
        service_types = catalog.get_service_types()
        self.assertEqual(service_types, ['compute', 'object-store',
                                         'rax:object-cdn'])

        service_types = catalog.get_service_types(region='ORD')
        self.assertEqual(service_types, ['rax:object-cdn'])

    def test_get_service_names(self):
        data = self.fixtures.load('_v2_0__auth.json')
        data = json.loads(data)
        service_catalog = data['access']['serviceCatalog']

        catalog = OpenStackServiceCatalog(service_catalog=service_catalog,
                                          auth_version='2.0')

        service_names = catalog.get_service_names()
        self.assertEqual(service_names, ['cloudFiles', 'cloudFilesCDN',
                                         'cloudServers',
                                         'cloudServersOpenStack',
                                         'cloudServersPreprod',
                                         'nova'])

        service_names = catalog.get_service_names(service_type='compute')
        self.assertEqual(service_names, ['cloudServers',
                                         'cloudServersOpenStack',
                                         'cloudServersPreprod',
                                         'nova'])


class OpenStackIdentity_2_0_MockHttp(MockHttp):
    fixtures = ComputeFileFixtures('openstack_identity/v2')
    json_content_headers = {'content-type': 'application/json; charset=UTF-8'}

    def _v2_0_tenants(self, method, url, body, headers):
        if method == 'GET':
            body = self.fixtures.load('v2_0_tenants.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()


class OpenStackIdentity_3_0_MockHttp(MockHttp):
    fixtures = ComputeFileFixtures('openstack_identity/v3')
    json_content_headers = {'content-type': 'application/json; charset=UTF-8'}

    def _v3(self, method, url, body, headers):
        if method == 'GET':
            body = self.fixtures.load('v3_versions.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_domains(self, method, url, body, headers):
        if method == 'GET':
            body = self.fixtures.load('v3_domains.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_projects(self, method, url, body, headers):
        if method == 'GET':
            body = self.fixtures.load('v3_projects.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_users(self, method, url, body, headers):
        if method == 'GET':
            # list users
            body = self.fixtures.load('v3_users.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        elif method == 'POST':
            # create user
            body = self.fixtures.load('v3_create_user.json')
            return (httplib.CREATED, body, self.json_content_headers,
                    httplib.responses[httplib.CREATED])
        raise NotImplementedError()

    def _v3_users_a(self, method, url, body, headers):
        if method == 'PATCH':
            # enable / disable user
            body = self.fixtures.load('v3_users_a.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_roles(self, method, url, body, headers):
        if method == 'GET':
            body = self.fixtures.load('v3_roles.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_domains_default_users_a_roles_a(self, method, url, body, headers):
        if method == 'PUT':
            # grant domain role
            body = ''
            return (httplib.NO_CONTENT, body, self.json_content_headers,
                    httplib.responses[httplib.NO_CONTENT])
        elif method == 'DELETE':
            # revoke domain role
            body = ''
            return (httplib.NO_CONTENT, body, self.json_content_headers,
                    httplib.responses[httplib.NO_CONTENT])
        raise NotImplementedError()

    def _v3_projects_a_users_a_roles_a(self, method, url, body, headers):
        if method == 'PUT':
            # grant project role
            body = ''
            return (httplib.NO_CONTENT, body, self.json_content_headers,
                    httplib.responses[httplib.NO_CONTENT])
        elif method == 'DELETE':
            # revoke project role
            body = ''
            return (httplib.NO_CONTENT, body, self.json_content_headers,
                    httplib.responses[httplib.NO_CONTENT])
        raise NotImplementedError()

    def _v3_domains_default(self, method, url, body, headers):
        if method == 'GET':
            # get domain
            body = self.fixtures.load('v3_domains_default.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_users_a_projects(self, method, url, body, headers):
        if method == 'GET':
            # get user projects
            body = self.fixtures.load('v3_users_a_projects.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()

    def _v3_domains_default_users_a_roles(self, method, url, body, headers):
        if method == 'GET':
            # get user domain roles
            body = self.fixtures.load('v3_domains_default_users_a_roles.json')
            return (httplib.OK, body, self.json_content_headers, httplib.responses[httplib.OK])
        raise NotImplementedError()


if __name__ == '__main__':
    sys.exit(unittest.main())
