# Copyright 2013 OpenStack Foundation
#
#    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 copy
import uuid

from six.moves import http_client
from testtools import matchers

from keystone import catalog
from keystone.tests import unit
from keystone.tests.unit.ksfixtures import database
from keystone.tests.unit import test_v3


class CatalogTestCase(test_v3.RestfulTestCase):
    """Test service & endpoint CRUD."""

    # region crud tests

    def test_create_region_with_id(self):
        """Call ``PUT /regions/{region_id}`` w/o an ID in the request body."""
        ref = self.new_region_ref()
        region_id = ref.pop('id')
        r = self.put(
            '/regions/%s' % region_id,
            body={'region': ref},
            expected_status=http_client.CREATED)
        self.assertValidRegionResponse(r, ref)
        # Double-check that the region ID was kept as-is and not
        # populated with a UUID, as is the case with POST /v3/regions
        self.assertEqual(region_id, r.json['region']['id'])

    def test_create_region_with_matching_ids(self):
        """Call ``PUT /regions/{region_id}`` with an ID in the request body."""
        ref = self.new_region_ref()
        region_id = ref['id']
        r = self.put(
            '/regions/%s' % region_id,
            body={'region': ref},
            expected_status=http_client.CREATED)
        self.assertValidRegionResponse(r, ref)
        # Double-check that the region ID was kept as-is and not
        # populated with a UUID, as is the case with POST /v3/regions
        self.assertEqual(region_id, r.json['region']['id'])

    def test_create_region_with_duplicate_id(self):
        """Call ``PUT /regions/{region_id}``."""
        ref = dict(description="my region")
        self.put(
            '/regions/myregion',
            body={'region': ref}, expected_status=http_client.CREATED)
        # Create region again with duplicate id
        self.put(
            '/regions/myregion',
            body={'region': ref}, expected_status=409)

    def test_create_region(self):
        """Call ``POST /regions`` with an ID in the request body."""
        # the ref will have an ID defined on it
        ref = self.new_region_ref()
        r = self.post(
            '/regions',
            body={'region': ref})
        self.assertValidRegionResponse(r, ref)

        # we should be able to get the region, having defined the ID ourselves
        r = self.get(
            '/regions/%(region_id)s' % {
                'region_id': ref['id']})
        self.assertValidRegionResponse(r, ref)

    def test_create_region_with_empty_id(self):
        """Call ``POST /regions`` with an empty ID in the request body."""
        ref = self.new_region_ref()
        ref['id'] = ''

        r = self.post('/regions', body={'region': ref})
        self.assertValidRegionResponse(r, ref)
        self.assertNotEmpty(r.result['region'].get('id'))

    def test_create_region_without_id(self):
        """Call ``POST /regions`` without an ID in the request body."""
        ref = self.new_region_ref()

        # instead of defining the ID ourselves...
        del ref['id']

        # let the service define the ID
        r = self.post('/regions', body={'region': ref})
        self.assertValidRegionResponse(r, ref)

    def test_create_region_without_description(self):
        """Call ``POST /regions`` without description in the request body."""
        ref = self.new_region_ref()

        del ref['description']

        r = self.post('/regions', body={'region': ref})
        # Create the description in the reference to compare to since the
        # response should now have a description, even though we didn't send
        # it with the original reference.
        ref['description'] = ''
        self.assertValidRegionResponse(r, ref)

    def test_create_regions_with_same_description_string(self):
        """Call ``POST /regions`` with same description in the request bodies.
        """
        # NOTE(lbragstad): Make sure we can create two regions that have the
        # same description.
        ref1 = self.new_region_ref()
        ref2 = self.new_region_ref()

        region_desc = 'Some Region Description'

        ref1['description'] = region_desc
        ref2['description'] = region_desc

        resp1 = self.post('/regions', body={'region': ref1})
        self.assertValidRegionResponse(resp1, ref1)

        resp2 = self.post('/regions', body={'region': ref2})
        self.assertValidRegionResponse(resp2, ref2)

    def test_create_regions_without_descriptions(self):
        """Call ``POST /regions`` with no description in the request bodies.
        """
        # NOTE(lbragstad): Make sure we can create two regions that have
        # no description in the request body. The description should be
        # populated by Catalog Manager.
        ref1 = self.new_region_ref()
        ref2 = self.new_region_ref()

        del ref1['description']
        ref2['description'] = None

        resp1 = self.post('/regions', body={'region': ref1})

        resp2 = self.post('/regions', body={'region': ref2})
        # Create the descriptions in the references to compare to since the
        # responses should now have descriptions, even though we didn't send
        # a description with the original references.
        ref1['description'] = ''
        ref2['description'] = ''
        self.assertValidRegionResponse(resp1, ref1)
        self.assertValidRegionResponse(resp2, ref2)

    def test_create_region_with_conflicting_ids(self):
        """Call ``PUT /regions/{region_id}`` with conflicting region IDs."""
        # the region ref is created with an ID
        ref = self.new_region_ref()

        # but instead of using that ID, make up a new, conflicting one
        self.put(
            '/regions/%s' % uuid.uuid4().hex,
            body={'region': ref},
            expected_status=http_client.BAD_REQUEST)

    def test_list_regions(self):
        """Call ``GET /regions``."""
        r = self.get('/regions')
        self.assertValidRegionListResponse(r, ref=self.region)

    def _create_region_with_parent_id(self, parent_id=None):
        ref = self.new_region_ref()
        ref['parent_region_id'] = parent_id
        return self.post(
            '/regions',
            body={'region': ref})

    def test_list_regions_filtered_by_parent_region_id(self):
        """Call ``GET /regions?parent_region_id={parent_region_id}``."""
        new_region = self._create_region_with_parent_id()
        parent_id = new_region.result['region']['id']

        new_region = self._create_region_with_parent_id(parent_id)
        new_region = self._create_region_with_parent_id(parent_id)

        r = self.get('/regions?parent_region_id=%s' % parent_id)

        for region in r.result['regions']:
            self.assertEqual(parent_id, region['parent_region_id'])

    def test_get_region(self):
        """Call ``GET /regions/{region_id}``."""
        r = self.get('/regions/%(region_id)s' % {
            'region_id': self.region_id})
        self.assertValidRegionResponse(r, self.region)

    def test_update_region(self):
        """Call ``PATCH /regions/{region_id}``."""
        region = self.new_region_ref()
        del region['id']
        r = self.patch('/regions/%(region_id)s' % {
            'region_id': self.region_id},
            body={'region': region})
        self.assertValidRegionResponse(r, region)

    def test_update_region_without_description_keeps_original(self):
        """Call ``PATCH /regions/{region_id}``."""
        region_ref = self.new_region_ref()

        resp = self.post('/regions', body={'region': region_ref})

        region_updates = {
            # update with something that's not the description
            'parent_region_id': self.region_id,
        }
        resp = self.patch('/regions/%s' % region_ref['id'],
                          body={'region': region_updates})

        # NOTE(dstanek): Keystone should keep the original description.
        self.assertEqual(region_ref['description'],
                         resp.result['region']['description'])

    def test_update_region_with_null_description(self):
        """Call ``PATCH /regions/{region_id}``."""
        region = self.new_region_ref()
        del region['id']
        region['description'] = None
        r = self.patch('/regions/%(region_id)s' % {
            'region_id': self.region_id},
            body={'region': region})

        # NOTE(dstanek): Keystone should turn the provided None value into
        # an empty string before storing in the backend.
        region['description'] = ''
        self.assertValidRegionResponse(r, region)

    def test_delete_region(self):
        """Call ``DELETE /regions/{region_id}``."""

        ref = self.new_region_ref()
        r = self.post(
            '/regions',
            body={'region': ref})
        self.assertValidRegionResponse(r, ref)

        self.delete('/regions/%(region_id)s' % {
            'region_id': ref['id']})

    # service crud tests

    def test_create_service(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        r = self.post(
            '/services',
            body={'service': ref})
        self.assertValidServiceResponse(r, ref)

    def test_create_service_no_name(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        del ref['name']
        r = self.post(
            '/services',
            body={'service': ref})
        ref['name'] = ''
        self.assertValidServiceResponse(r, ref)

    def test_create_service_no_enabled(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        del ref['enabled']
        r = self.post(
            '/services',
            body={'service': ref})
        ref['enabled'] = True
        self.assertValidServiceResponse(r, ref)
        self.assertIs(True, r.result['service']['enabled'])

    def test_create_service_enabled_false(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        ref['enabled'] = False
        r = self.post(
            '/services',
            body={'service': ref})
        self.assertValidServiceResponse(r, ref)
        self.assertIs(False, r.result['service']['enabled'])

    def test_create_service_enabled_true(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        ref['enabled'] = True
        r = self.post(
            '/services',
            body={'service': ref})
        self.assertValidServiceResponse(r, ref)
        self.assertIs(True, r.result['service']['enabled'])

    def test_create_service_enabled_str_true(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        ref['enabled'] = 'True'
        self.post('/services', body={'service': ref},
                  expected_status=http_client.BAD_REQUEST)

    def test_create_service_enabled_str_false(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        ref['enabled'] = 'False'
        self.post('/services', body={'service': ref},
                  expected_status=http_client.BAD_REQUEST)

    def test_create_service_enabled_str_random(self):
        """Call ``POST /services``."""
        ref = self.new_service_ref()
        ref['enabled'] = 'puppies'
        self.post('/services', body={'service': ref},
                  expected_status=http_client.BAD_REQUEST)

    def test_list_services(self):
        """Call ``GET /services``."""
        r = self.get('/services')
        self.assertValidServiceListResponse(r, ref=self.service)

    def _create_random_service(self):
        ref = self.new_service_ref()
        ref['enabled'] = True
        response = self.post(
            '/services',
            body={'service': ref})
        return response.json['service']

    def test_filter_list_services_by_type(self):
        """Call ``GET /services?type=<some type>``."""
        target_ref = self._create_random_service()

        # create unrelated services
        self._create_random_service()
        self._create_random_service()

        response = self.get('/services?type=' + target_ref['type'])
        self.assertValidServiceListResponse(response, ref=target_ref)

        filtered_service_list = response.json['services']
        self.assertEqual(1, len(filtered_service_list))

        filtered_service = filtered_service_list[0]
        self.assertEqual(target_ref['type'], filtered_service['type'])

    def test_filter_list_services_by_name(self):
        """Call ``GET /services?name=<some name>``."""
        target_ref = self._create_random_service()

        # create unrelated services
        self._create_random_service()
        self._create_random_service()

        response = self.get('/services?name=' + target_ref['name'])
        self.assertValidServiceListResponse(response, ref=target_ref)

        filtered_service_list = response.json['services']
        self.assertEqual(1, len(filtered_service_list))

        filtered_service = filtered_service_list[0]
        self.assertEqual(target_ref['name'], filtered_service['name'])

    def test_get_service(self):
        """Call ``GET /services/{service_id}``."""
        r = self.get('/services/%(service_id)s' % {
            'service_id': self.service_id})
        self.assertValidServiceResponse(r, self.service)

    def test_update_service(self):
        """Call ``PATCH /services/{service_id}``."""
        service = self.new_service_ref()
        del service['id']
        r = self.patch('/services/%(service_id)s' % {
            'service_id': self.service_id},
            body={'service': service})
        self.assertValidServiceResponse(r, service)

    def test_delete_service(self):
        """Call ``DELETE /services/{service_id}``."""
        self.delete('/services/%(service_id)s' % {
            'service_id': self.service_id})

    # endpoint crud tests

    def test_list_endpoints(self):
        """Call ``GET /endpoints``."""
        r = self.get('/endpoints')
        self.assertValidEndpointListResponse(r, ref=self.endpoint)

    def _create_random_endpoint(self, interface='public',
                                parent_region_id=None):
        region = self._create_region_with_parent_id(
            parent_id=parent_region_id)
        service = self._create_random_service()
        ref = self.new_endpoint_ref(
            service_id=service['id'],
            interface=interface,
            region_id=region.result['region']['id'])

        response = self.post(
            '/endpoints',
            body={'endpoint': ref})
        return response.json['endpoint']

    def test_list_endpoints_filtered_by_interface(self):
        """Call ``GET /endpoints?interface={interface}``."""
        ref = self._create_random_endpoint(interface='internal')

        response = self.get('/endpoints?interface=%s' % ref['interface'])
        self.assertValidEndpointListResponse(response, ref=ref)

        for endpoint in response.json['endpoints']:
            self.assertEqual(ref['interface'], endpoint['interface'])

    def test_list_endpoints_filtered_by_service_id(self):
        """Call ``GET /endpoints?service_id={service_id}``."""
        ref = self._create_random_endpoint()

        response = self.get('/endpoints?service_id=%s' % ref['service_id'])
        self.assertValidEndpointListResponse(response, ref=ref)

        for endpoint in response.json['endpoints']:
            self.assertEqual(ref['service_id'], endpoint['service_id'])

    def test_list_endpoints_filtered_by_region_id(self):
        """Call ``GET /endpoints?region_id={region_id}``."""
        ref = self._create_random_endpoint()

        response = self.get('/endpoints?region_id=%s' % ref['region_id'])
        self.assertValidEndpointListResponse(response, ref=ref)

        for endpoint in response.json['endpoints']:
            self.assertEqual(ref['region_id'], endpoint['region_id'])

    def test_list_endpoints_filtered_by_parent_region_id(self):
        """Call ``GET /endpoints?region_id={region_id}``.

        Ensure passing the parent_region_id as filter returns an
        empty list.

        """
        parent_region = self._create_region_with_parent_id()
        parent_region_id = parent_region.result['region']['id']
        self._create_random_endpoint(parent_region_id=parent_region_id)

        response = self.get('/endpoints?region_id=%s' % parent_region_id)
        self.assertEqual(0, len(response.json['endpoints']))

    def test_list_endpoints_with_multiple_filters(self):
        """Call ``GET /endpoints?interface={interface}...``.

        Ensure passing different combinations of interface, region_id and
        service_id as filters will return the correct result.

        """
        # interface and region_id specified
        ref = self._create_random_endpoint(interface='internal')
        response = self.get('/endpoints?interface=%s&region_id=%s' %
                            (ref['interface'], ref['region_id']))
        self.assertValidEndpointListResponse(response, ref=ref)

        for endpoint in response.json['endpoints']: