# Licensed 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 uuid

import http.client
from testtools import matchers

from keystone.common import provider_api
import keystone.conf
from keystone.credential.providers import fernet as credential_fernet
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import ksfixtures
from keystone.tests.unit import test_v3
from keystone.tests.unit import utils as test_utils


CONF = keystone.conf.CONF
PROVIDERS = provider_api.ProviderAPIs


class ResourceTestCase(test_v3.RestfulTestCase,
                       test_v3.AssignmentTestMixin):
    """Test domains and projects."""

    def setUp(self):
        super(ResourceTestCase, self).setUp()
        self.useFixture(
            ksfixtures.KeyRepository(
                self.config_fixture,
                'credential',
                credential_fernet.MAX_ACTIVE_KEYS
            )
        )

    # Domain CRUD tests

    def test_create_domain(self):
        """Call ``POST /domains``."""
        ref = unit.new_domain_ref()
        r = self.post(
            '/domains',
            body={'domain': ref})
        return self.assertValidDomainResponse(r, ref)

    def test_create_domain_case_sensitivity(self):
        """Call `POST /domains`` twice with upper() and lower() cased name."""
        ref = unit.new_domain_ref()

        # ensure the name is lowercase
        ref['name'] = ref['name'].lower()
        r = self.post(
            '/domains',
            body={'domain': ref})
        self.assertValidDomainResponse(r, ref)

        # ensure the name is uppercase
        ref['name'] = ref['name'].upper()
        r = self.post(
            '/domains',
            body={'domain': ref})
        self.assertValidDomainResponse(r, ref)

    def test_create_domain_bad_request(self):
        """Call ``POST /domains``."""
        self.post('/domains', body={'domain': {}},
                  expected_status=http.client.BAD_REQUEST)

    def test_create_domain_unsafe(self):
        """Call ``POST /domains with unsafe names``."""
        unsafe_name = 'i am not / safe'

        self.config_fixture.config(group='resource',
                                   domain_name_url_safe='off')
        ref = unit.new_domain_ref(name=unsafe_name)
        self.post(
            '/domains',
            body={'domain': ref})

        for config_setting in ['new', 'strict']:
            self.config_fixture.config(group='resource',
                                       domain_name_url_safe=config_setting)
            ref = unit.new_domain_ref(name=unsafe_name)
            self.post(
                '/domains',
                body={'domain': ref},
                expected_status=http.client.BAD_REQUEST)

    def test_create_domain_unsafe_default(self):
        """Check default for unsafe names for ``POST /domains``."""
        unsafe_name = 'i am not / safe'

        # By default, we should be able to create unsafe names
        ref = unit.new_domain_ref(name=unsafe_name)
        self.post(
            '/domains',
            body={'domain': ref})

    def test_create_domain_creates_is_domain_project(self):
        """Check a project that acts as a domain is created.

        Call ``POST /domains``.
        """
        # Create a new domain
        domain_ref = unit.new_domain_ref()
        r = self.post('/domains', body={'domain': domain_ref})
        self.assertValidDomainResponse(r, domain_ref)

        # Retrieve its correspondent project
        r = self.get('/projects/%(project_id)s' % {
            'project_id': r.result['domain']['id']})
        self.assertValidProjectResponse(r)

        # The created project has is_domain flag as True
        self.assertTrue(r.result['project']['is_domain'])

        # And its parent_id and domain_id attributes are equal
        self.assertIsNone(r.result['project']['parent_id'])
        self.assertIsNone(r.result['project']['domain_id'])

    def test_create_is_domain_project_creates_domain(self):
        """Call ``POST /projects`` is_domain and check a domain is created."""
        # Create a new project that acts as a domain
        project_ref = unit.new_project_ref(domain_id=None, is_domain=True)
        r = self.post('/projects', body={'project': project_ref})
        self.assertValidProjectResponse(r)

        # Retrieve its correspondent domain
        r = self.get('/domains/%(domain_id)s' % {
            'domain_id': r.result['project']['id']})
        self.assertValidDomainResponse(r)
        self.assertIsNotNone(r.result['domain'])

    def test_create_domain_valid_explicit_id(self):
        """Call ``POST /domains`` with a valid `explicit_domain_id` set."""
        ref = unit.new_domain_ref()
        explicit_domain_id = '9aea63518f0040c6b4518d8d2242911c'

        ref['explicit_domain_id'] = explicit_domain_id
        r = self.post(
            '/domains',
            body={'domain': ref})
        self.assertValidDomainResponse(r, ref)

        r = self.get('/domains/%(domain_id)s' % {
            'domain_id': explicit_domain_id})
        self.assertValidDomainResponse(r)
        self.assertIsNotNone(r.result['domain'])

    def test_create_second_domain_valid_explicit_id_fails(self):
        """Call ``POST /domains`` with a valid `explicit_domain_id` set."""
        ref = unit.new_domain_ref()
        explicit_domain_id = '9aea63518f0040c6b4518d8d2242911c'

        ref['explicit_domain_id'] = explicit_domain_id
        r = self.post(
            '/domains',
            body={'domain': ref})
        self.assertValidDomainResponse(r, ref)

        # second one should fail
        r = self.post(
            '/domains',
            body={'domain': ref},
            expected_status=http.client.CONFLICT)

    def test_create_domain_invalid_explicit_ids(self):
        """Call ``POST /domains`` with various invalid explicit_domain_ids."""
        ref = unit.new_domain_ref()

        bad_ids = ['bad!',
                   '',
                   '9aea63518f0040c',
                   '1234567890123456789012345678901234567890',
                   '9aea63518f0040c6b4518d8d2242911c9aea63518f0040c6b45']

        for explicit_domain_id in bad_ids:
            ref['explicit_domain_id'] = explicit_domain_id
            self.post('/domains', body={'domain': {}},
                      expected_status=http.client.BAD_REQUEST)

    def test_list_head_domains(self):
        """Call ``GET & HEAD /domains``."""
        resource_url = '/domains'
        r = self.get(resource_url)
        self.assertValidDomainListResponse(r, ref=self.domain,
                                           resource_url=resource_url)
        self.head(resource_url, expected_status=http.client.OK)

    def test_list_limit_for_domains(self):
        for x in range(6):
            domain = {'domain': unit.new_domain_ref()}
            self.post('/domains', body=domain)

        for expected_length in range(1, 6):
            self.config_fixture.config(
                group='resource', list_limit=expected_length
            )
            response = self.get('/domains')
            domain_list = response.json_body['domains']
            self.assertEqual(expected_length, len(domain_list))

    def test_get_head_domain(self):
        """Call ``GET /domains/{domain_id}``."""
        resource_url = '/domains/%(domain_id)s' % {
            'domain_id': self.domain_id}
        r = self.get(resource_url)
        self.assertValidDomainResponse(r, self.domain)
        self.head(resource_url, expected_status=http.client.OK)

    def test_update_domain(self):
        """Call ``PATCH /domains/{domain_id}``."""
        ref = unit.new_domain_ref()
        del ref['id']
        r = self.patch('/domains/%(domain_id)s' % {
            'domain_id': self.domain_id},
            body={'domain': ref})
        self.assertValidDomainResponse(r, ref)

    def test_update_domain_unsafe(self):
        """Call ``POST /domains/{domain_id} with unsafe names``."""
        unsafe_name = 'i am not / safe'

        self.config_fixture.config(group='resource',
                                   domain_name_url_safe='off')
        ref = unit.new_domain_ref(name=unsafe_name)
        del ref['id']
        self.patch('/domains/%(domain_id)s' % {
            'domain_id': self.domain_id},
            body={'domain': ref})

        unsafe_name = 'i am still not / safe'
        for config_setting in ['new', 'strict']:
            self.config_fixture.config(group='resource',
                                       domain_name_url_safe=config_setting)
            ref = unit.new_domain_ref(name=unsafe_name)
            del ref['id']
            self.patch('/domains/%(domain_id)s' % {
                'domain_id': self.domain_id},
                body={'domain': ref},
                expected_status=http.client.BAD_REQUEST)

    def test_update_domain_unsafe_default(self):
        """Check default for unsafe names for ``POST /domains``."""
        unsafe_name = 'i am not / safe'

        # By default, we should be able to create unsafe names
        ref = unit.new_domain_ref(name=unsafe_name)
        del ref['id']
        self.patch('/domains/%(domain_id)s' % {
            'domain_id': self.domain_id},
            body={'domain': ref})

    def test_update_domain_updates_is_domain_project(self):
        """Check the project that acts as a domain is updated.

        Call ``PATCH /domains``.
        """
        # Create a new domain
        domain_ref = unit.new_domain_ref()
        r = self.post('/domains', body={'domain': domain_ref})
        self.assertValidDomainResponse(r, domain_ref)

        # Disable it
        self.patch('/domains/%s' % r.result['domain']['id'],
                   body={'domain': {'enabled': False}})

        # Retrieve its correspondent project
        r = self.get('/projects/%(project_id)s' % {
            'project_id': r.result['domain']['id']})
        self.assertValidProjectResponse(r)

        # The created project is disabled as well
        self.assertFalse(r.result['project']['enabled'])

    def test_disable_domain(self):
        """Call ``PATCH /domains/{domain_id}`` (set enabled=False)."""
        # Create a 2nd set of entities in a 2nd domain
        domain2 = unit.new_domain_ref()
        PROVIDERS.resource_api.create_domain(domain2['id'], domain2)

        project2 = unit.new_project_ref(domain_id=domain2['id'])
        PROVIDERS.resource_api.create_project(project2['id'], project2)

        user2 = unit.create_user(PROVIDERS.identity_api,
                                 domain_id=domain2['id'],
                                 project_id=project2['id'])

        role_member = unit.new_role_ref()
        PROVIDERS.role_api.create_role(role_member['id'], role_member)

        PROVIDERS.assignment_api.add_role_to_user_and_project(
            user2['id'], project2['id'], role_member['id']
        )

        # First check a user in that domain can authenticate..
        auth_data = self.build_authentication_request(
            user_id=user2['id'],
            password=user2['password'],
            project_id=project2['id'])
        self.v3_create_token(auth_data)

        # Now disable the domain
        domain2['enabled'] = False
        r = self.patch('/domains/%(domain_id)s' % {
            'domain_id': domain2['id']},
            body={'domain': {'enabled': False}})
        self.assertValidDomainResponse(r, domain2)

        # Try looking up in v3 by name and id
        auth_data = self.build_authentication_request(
            user_id=user2['id'],
            password=user2['password'],
            project_id=project2['id'])
        self.v3_create_token(auth_data,
                             expected_status=http.client.UNAUTHORIZED)

        auth_data = self.build_authentication_request(
            username=user2['name'],
            user_domain_id=domain2['id'],
            password=user2['password'],
            project_id=project2['id'])
        self.v3_create_token(auth_data,
                             expected_status=http.client.UNAUTHORIZED)

    def test_delete_enabled_domain_fails(self):
        """Call ``DELETE /domains/{domain_id}`` (when domain enabled)."""
        # Try deleting an enabled domain, which should fail
        self.delete('/domains/%(domain_id)s' % {
            'domain_id': self.domain['id']},
            expected_status=exception.ForbiddenAction.code)

    def test_delete_domain(self):
        """Call ``DELETE /domains/{domain_id}``.

        The sample data set up already has a user and project that is part of
        self.domain. Additionally we will create a group and a credential
        within it. Since we will authenticate in this domain,
        we create another set of entities in a second domain.  Deleting this
        second domain should delete all these new entities. In addition,
        all the entities in the regular self.domain should be unaffected
        by the delete.

        Test Plan:

        - Create domain2 and a 2nd set of entities
        - Disable domain2
        - Delete domain2
        - Check entities in domain2 have been deleted
        - Check entities in self.domain are unaffected

        """
        # Create a group and a credential in the main domain
        group = unit.new_group_ref(domain_id=self.domain_id)
        group = PROVIDERS.identity_api.create_group(group)

        credential = unit.new_credential_ref(user_id=self.user['id'],
                                             project_id=self.project_id)
        PROVIDERS.credential_api.create_credential(
            credential['id'], credential
        )

        # Create a 2nd set of entities in a 2nd domain
        domain2 = unit.new_domain_ref()
        PROVIDERS.resource_api.create_domain(domain2['id'], domain2)

        project2 = unit.new_project_ref(domain_id=domain2['id'])
        project2 = PROVIDERS.resource_api.create_project(
            project2['id'], project2
        )

        user2 = unit.new_user_ref(domain_id=domain2['id'],
                                  project_id=project2['id'])
        user2 = PROVIDERS.identity_api.create_user(user2)

        group2 = unit.new_group_ref(domain_id=domain2['id'])
        group2 = PROVIDERS.identity_api.create_group(group2)

        credential2 = unit.new_credential_ref(user_id=user2['id'],
                                              project_id=project2['id'])
        PROVIDERS.credential_api.create_credential(
            credential2['id'], credential2
        )

        # Now disable the new domain and delete it
        domain2['enabled'] = False
        r = self.patch('/domains/%(domain_id)s' % {
            'domain_id': domain2['id']},
            body={'domain': {'enabled': False}})
        self.assertValidDomainResponse(r, domain2)
        self.delete('/domains/%(domain_id)s' % {'domain_id': domain2['id']})

        # Check all the domain2 relevant entities are gone
        self.assertRaises(exception.DomainNotFound,
                          PROVIDERS.resource_api.get_domain,
                          domain2['id'])
        self.assertRaises(exception.ProjectNotFound,
                          PROVIDERS.resource_api.get_project,
                          project2['id'])
        self.assertRaises(exception.GroupNotFound,
                          PROVIDERS.identity_api.get_group,
                          group2['id'])
        self.assertRaises(exception.UserNotFound,
                          PROVIDERS.identity_api.get_user,
                          user2['id'])
        self.assertRaises(exception.CredentialNotFound,
                          PROVIDERS.credential_api.get_credential,
                          credential2['id'])

        # ...and that all self.domain entities are still here
        r = PROVIDERS.resource_api.get_domain(self.domain['id'])
        self.assertDictEqual(self.domain, r)
        r = PROVIDERS.resource_api.get_project(self.project['id'])
        self.assertDictEqual(self.project, r)
        r = PROVIDERS.identity_api.get_group(group['id'])
        self.assertDictEqual(group, r)
        r = PROVIDERS.identity_api.get_user(self.user['id'])
        self.user.pop('password')
        self.assertDictEqual(self.user, r)
        r = PROVIDERS.credential_api.get_credential(credential['id'])
        self.assertDictEqual(credential, r)

    def test_delete_domain_with_idp(self):
        # Create a new domain
        domain_ref = unit.new_domain_ref()
        r = self.post('/domains', body={'domain': domain_ref})
        self.assertValidDomainResponse(r, domain_ref)
        domain_id = r.result['domain']['id']
        # Create a Idp in the domain
        self.put('/OS-FEDERATION/identity_providers/test_idp',
                 body={"identity_provider": {
                     "domain_id": domain_id}},
                 expected_status=http.client.CREATED)
        # Disable and delete the domain with no error.
        self.patch('/domains/%(domain_id)s' % {
            'domain_id': domain_id},
            body={'domain': {'enabled': False}})
        self.delete('/domains/%s' % domain_id)
        # The Idp is deleted as well
        self.get('/OS-FEDERATION/identity_providers/test_idp',
                 expected_status=http.client.NOT_FOUND)

    def test_delete_domain_deletes_is_domain_project(self):
        """Check the project that acts as a domain is deleted.

        Call ``DELETE /domains``.
        """
        # Create a new domain
        domain_ref = unit.new_domain_ref()
        r = self.post('/domains', body={'domain': domain_ref})
        self.assertValidDomainResponse(r, domain_ref)

        # Retrieve its correspondent project
        self.get('/projects/%(project_id)s' % {
            'project_id': r.result['domain']['id']})

        # Delete the domain
        self.patch('/domains/%s' % r.result['domain']['id'],
                   body={'domain': {'enabled': False}})
        self.delete('/domains/%s' % r.result['domain']['id'])

        # The created project is deleted as well
        self.get('/projects/%(project_id)s' % {
            'project_id': r.result['domain']['id']}, expected_status=404)