# 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 mock
import uuid

import testtools

import shade
from shade import _utils
from shade import exc
from shade.tests import fakes
from shade.tests.unit import base


RANGE_DATA = [
    dict(id=1, key1=1, key2=5),
    dict(id=2, key1=1, key2=20),
    dict(id=3, key1=2, key2=10),
    dict(id=4, key1=2, key2=30),
    dict(id=5, key1=3, key2=40),
    dict(id=6, key1=3, key2=40),
]


class TestShade(base.RequestsMockTestCase):

    def setUp(self):
        # This set of tests are not testing neutron, they're testing
        # rebuilding servers, but we do several network calls in service
        # of a NORMAL rebuild to find the default_network. Putting
        # in all of the neutron mocks for that will make the tests harder
        # to read. SO - we're going mock neutron into the off position
        # and then turn it back on in the few tests that specifically do.
        # Maybe we should reorg these into two classes - one with neutron
        # mocked out - and one with it not mocked out
        super(TestShade, self).setUp()
        self.has_neutron = False

        def fake_has_service(*args, **kwargs):
            return self.has_neutron

        self.cloud.has_service = fake_has_service

    def test_openstack_cloud(self):
        self.assertIsInstance(self.cloud, shade.OpenStackCloud)

    @mock.patch.object(shade.OpenStackCloud, 'search_images')
    def test_get_images(self, mock_search):
        image1 = dict(id='123', name='mickey')
        mock_search.return_value = [image1]
        r = self.cloud.get_image('mickey')
        self.assertIsNotNone(r)
        self.assertDictEqual(image1, r)

    @mock.patch.object(shade.OpenStackCloud, 'search_images')
    def test_get_image_not_found(self, mock_search):
        mock_search.return_value = []
        r = self.cloud.get_image('doesNotExist')
        self.assertIsNone(r)

    def test_get_server(self):
        server1 = fakes.make_fake_server('123', 'mickey')
        server2 = fakes.make_fake_server('345', 'mouse')

        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail']),
                 json={'servers': [server1, server2]}),
        ])

        r = self.cloud.get_server('mickey')
        self.assertIsNotNone(r)
        self.assertEqual(server1['name'], r['name'])

        self.assert_calls()

    def test_get_server_not_found(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail']),
                 json={'servers': []}),
        ])

        r = self.cloud.get_server('doesNotExist')
        self.assertIsNone(r)

        self.assert_calls()

    def test_list_servers_exception(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail']),
                 status_code=400)
        ])

        self.assertRaises(exc.OpenStackCloudException,
                          self.cloud.list_servers)

        self.assert_calls()

    def test__neutron_exceptions_resource_not_found(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'networks.json']),
                 status_code=404)
        ])
        self.assertRaises(exc.OpenStackCloudResourceNotFound,
                          self.cloud.list_networks)
        self.assert_calls()

    def test__neutron_exceptions_url_not_found(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'networks.json']),
                 status_code=404)
        ])
        self.assertRaises(exc.OpenStackCloudURINotFound,
                          self.cloud.list_networks)
        self.assert_calls()

    def test_list_servers(self):
        server_id = str(uuid.uuid4())
        server_name = self.getUniqueString('name')
        fake_server = fakes.make_fake_server(server_id, server_name)
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail']),
                 json={'servers': [fake_server]}),
        ])

        r = self.cloud.list_servers()

        self.assertEqual(1, len(r))
        self.assertEqual(server_name, r[0]['name'])

        self.assert_calls()

    def test_list_server_private_ip(self):
        self.has_neutron = True
        fake_server = {
            "OS-EXT-STS:task_state": None,
            "addresses": {
                "private": [
                    {
                        "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07",
                        "version": 4,
                        "addr": "10.4.0.13",
                        "OS-EXT-IPS:type": "fixed"
                    }, {
                        "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07",
                        "version": 4,
                        "addr": "89.40.216.229",
                        "OS-EXT-IPS:type": "floating"
                    }]},
            "links": [
                {
                    "href": "http://example.com/images/95e4c4",
                    "rel": "self"
                }, {
                    "href": "http://example.com/images/95e4c4",
                    "rel": "bookmark"
                }
            ],
            "image": {
                "id": "95e4c449-8abf-486e-97d9-dc3f82417d2d",
                "links": [
                    {
                        "href": "http://example.com/images/95e4c4",
                        "rel": "bookmark"
                    }
                ]
            },
            "OS-EXT-STS:vm_state": "active",
            "OS-SRV-USG:launched_at": "2018-03-01T02:44:50.000000",
            "flavor": {
                "id": "3bd99062-2fe8-4eac-93f0-9200cc0f97ae",
                "links": [
                    {
                        "href": "http://example.com/flavors/95e4c4",
                        "rel": "bookmark"
                    }
                ]
            },
            "id": "97fe35e9-756a-41a2-960a-1d057d2c9ee4",
            "security_groups": [{"name": "default"}],
            "user_id": "c17534835f8f42bf98fc367e0bf35e09",
            "OS-DCF:diskConfig": "MANUAL",
            "accessIPv4": "",
            "accessIPv6": "",
            "progress": 0,
            "OS-EXT-STS:power_state": 1,
            "OS-EXT-AZ:availability_zone": "nova",
            "metadata": {},
            "status": "ACTIVE",
            "updated": "2018-03-01T02:44:51Z",
            "hostId": "",
            "OS-SRV-USG:terminated_at": None,
            "key_name": None,
            "name": "mttest",
            "created": "2018-03-01T02:44:46Z",
            "tenant_id": "65222a4d09ea4c68934fa1028c77f394",
            "os-extended-volumes:volumes_attached": [],
            "config_drive": ""
        }
        fake_networks = {"networks": [
            {
                "status": "ACTIVE",
                "router:external": True,
                "availability_zone_hints": [],
                "availability_zones": ["nova"],
                "description": None,
                "subnets": [
                    "df3e17fa-a4b2-47ae-9015-bc93eb076ba2",
                    "6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec",
                    "fc541f48-fc7f-48c0-a063-18de6ee7bdd7"
                ],
                "shared": False,
                "tenant_id": "a564613210ee43708b8a7fc6274ebd63",
                "tags": [],
                "ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d",
                "mtu": 1550,
                "is_default": False,
                "admin_state_up": True,
                "revision_number": 0,
                "ipv4_address_scope": None,
                "port_security_enabled": True,
                "project_id": "a564613210ee43708b8a7fc6274ebd63",
                "id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
                "name": "ext-net"
            }, {
                "status": "ACTIVE",
                "router:external": False,
                "availability_zone_hints": [],
                "availability_zones": ["nova"],
                "description": "",
                "subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"],
                "shared": False,
                "tenant_id": "65222a4d09ea4c68934fa1028c77f394",
                "created_at": "2016-10-22T13:46:26Z",
                "tags": [],
                "ipv6_address_scope": None,
                "updated_at": "2016-10-22T13:46:26Z",
                "admin_state_up": True,
                "mtu": 1500,
                "revision_number": 0,
                "ipv4_address_scope": None,
                "port_security_enabled": True,
                "project_id": "65222a4d09ea4c68934fa1028c77f394",
                "id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
                "name": "private"
            }]}
        fake_subnets = {
            "subnets": [
                {
                    "service_types": [],
                    "description": "",
                    "enable_dhcp": True,
                    "tags": [],
                    "network_id": "827c6bb6-492f-4168-9577-f3a131eb29e8",
                    "tenant_id": "65222a4d09ea4c68934fa1028c77f394",
                    "created_at": "2017-06-12T13:23:57Z",
                    "dns_nameservers": [],
                    "updated_at": "2017-06-12T13:23:57Z",
                    "gateway_ip": "10.24.4.1",
                    "ipv6_ra_mode": None,
                    "allocation_pools": [
                        {
                            "start": "10.24.4.2",
                            "end": "10.24.4.254"
                        }],
                    "host_routes": [],
                    "revision_number": 0,
                    "ip_version": 4,
                    "ipv6_address_mode": None,
                    "cidr": "10.24.4.0/24",
                    "project_id": "65222a4d09ea4c68934fa1028c77f394",
                    "id": "3f0642d9-4644-4dff-af25-bcf64f739698",
                    "subnetpool_id": None,
                    "name": "foo_subnet"
                }, {
                    "service_types": [],
                    "description": "",
                    "enable_dhcp": True,
                    "tags": [],
                    "network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
                    "tenant_id": "65222a4d09ea4c68934fa1028c77f394",
                    "created_at": "2016-10-22T13:46:26Z",
                    "dns_nameservers": ["89.36.90.101", "89.36.90.102"],
                    "updated_at": "2016-10-22T13:46:26Z",
                    "gateway_ip": "10.4.0.1",
                    "ipv6_ra_mode": None,
                    "allocation_pools": [
                        {
                            "start": "10.4.0.2",
                            "end": "10.4.0.200"
                        }],
                    "host_routes": [],
                    "revision_number": 0,
                    "ip_version": 4,
                    "ipv6_address_mode": None,
                    "cidr": "10.4.0.0/24",
                    "project_id": "65222a4d09ea4c68934fa1028c77f394",
                    "id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9",
                    "subnetpool_id": None,
                    "name": "private-subnet-ipv4"
                }]}
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail']),
                 json={'servers': [fake_server]}),
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'networks.json']),
                 json=fake_networks),
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'subnets.json']),
                 json=fake_subnets)
        ])

        r = self.cloud.get_server('97fe35e9-756a-41a2-960a-1d057d2c9ee4')

        self.assertEqual('10.4.0.13', r['private_v4'])

        self.assert_calls()

    def test_list_servers_all_projects(self):
        '''This test verifies that when list_servers is called with
        `all_projects=True` that it passes `all_tenants=True` to nova.'''
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail'],
                     qs_elements=['all_tenants=True']),
                 complete_qs=True,
                 json={'servers': []}),
        ])

        self.cloud.list_servers(all_projects=True)

        self.assert_calls()

    def test_list_servers_filters(self):
        '''This test verifies that when list_servers is called with
        `filters` dict that it passes it to nova.'''
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'compute', 'public', append=['servers', 'detail'],
                     qs_elements=[
                         'deleted=True',
                         'changes-since=2014-12-03T00:00:00Z'
                     ]),
                 complete_qs=True,
                 json={'servers': []}),
        ])

        self.cloud.list_servers(filters={
            'deleted': True,
            'changes-since': '2014-12-03T00:00:00Z'
        })

        self.assert_calls()

    def test_iterate_timeout_bad_wait(self):
        with testtools.ExpectedException(
                exc.OpenStackCloudException,
                "Wait value must be an int or float value."):
            for count in _utils._iterate_timeout(
                    1, "test_iterate_timeout_bad_wait", wait="timeishard"):
                pass

    @mock.patch('time.sleep')
    def test_iterate_timeout_str_wait(self, mock_sleep):
        iter = _utils._iterate_timeout(
            10, "test_iterate_timeout_str_wait", wait="1.6")
        next(iter)
        next(iter)
        mock_sleep.assert_called_with(1.6)

    @mock.patch('time.sleep')
    def test_iterate_timeout_int_wait(self, mock_sleep):
        iter = _utils._iterate_timeout(
            10, "test_iterate_timeout_int_wait", wait=1)
        next(iter)
        next(iter)
        mock_sleep.assert_called_with(1.0)

    @mock.patch('time.sleep')
    def test_iterate_timeout_timeout(self, mock_sleep):
        message = "timeout test"
        with testtools.ExpectedException(
                exc.OpenStackCloudTimeout,
                message):
            for count in _utils._iterate_timeout(0.1, message, wait=1):
                pass
        mock_sleep.assert_called_with(1.0)

    def test__nova_extensions(self):
        body = [
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "Multinic",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "NMN",
                "description": "Multiple network support."
            },
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "DiskConfig",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "OS-DCF",
                "description": "Disk Management Extension."
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri='{endpoint}/extensions'.format(
                     endpoint=fakes.COMPUTE_ENDPOINT),
                 json=dict(extensions=body))
        ])
        extensions = self.cloud._nova_extensions()
        self.assertEqual(set(['NMN', 'OS-DCF']), extensions)

        self.assert_calls()

    def test__nova_extensions_fails(self):
        self.register_uris([
            dict(method='GET',
                 uri='{endpoint}/extensions'.format(
                     endpoint=fakes.COMPUTE_ENDPOINT),
                 status_code=404),
        ])
        with testtools.ExpectedException(
            exc.OpenStackCloudURINotFound,
            "Error fetching extension list for nova"
        ):
            self.cloud._nova_extensions()

        self.assert_calls()

    def test__has_nova_extension(self):
        body = [
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "Multinic",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "NMN",
                "description": "Multiple network support."
            },
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "DiskConfig",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "OS-DCF",
                "description": "Disk Management Extension."
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri='{endpoint}/extensions'.format(
                     endpoint=fakes.COMPUTE_ENDPOINT),
                 json=dict(extensions=body))
        ])
        self.assertTrue(self.cloud._has_nova_extension('NMN'))

        self.assert_calls()

    def test__has_nova_extension_missing(self):
        body = [
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "Multinic",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "NMN",
                "description": "Multiple network support."
            },
            {
                "updated": "2014-12-03T00:00:00Z",
                "name": "DiskConfig",
                "links": [],
                "namespace": "http://openstack.org/compute/ext/fake_xml",
                "alias": "OS-DCF",
                "description": "Disk Management Extension."
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri='{endpoint}/extensions'.format(
                     endpoint=fakes.COMPUTE_ENDPOINT),
                 json=dict(extensions=body))
        ])
        self.assertFalse(self.cloud._has_nova_extension('invalid'))

        self.assert_calls()

    def test__neutron_extensions(self):
        body = [
            {
                "updated": "2014-06-1T10:00:00-00:00",
                "name": "Distributed Virtual Router",
                "links": [],
                "alias": "dvr",
                "description":
                    "Enables configuration of Distributed Virtual Routers."
            },
            {
                "updated": "2013-07-23T10:00:00-00:00",
                "name": "Allowed Address Pairs",
                "links": [],
                "alias": "allowed-address-pairs",
                "description": "Provides allowed address pairs"
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'extensions.json']),
                 json=dict(extensions=body))
        ])
        extensions = self.cloud._neutron_extensions()
        self.assertEqual(set(['dvr', 'allowed-address-pairs']), extensions)

        self.assert_calls()

    def test__neutron_extensions_fails(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'extensions.json']),
                 status_code=404)
        ])
        with testtools.ExpectedException(
            exc.OpenStackCloudURINotFound,
            "Error fetching extension list for neutron"
        ):
            self.cloud._neutron_extensions()

        self.assert_calls()

    def test__has_neutron_extension(self):
        body = [
            {
                "updated": "2014-06-1T10:00:00-00:00",
                "name": "Distributed Virtual Router",
                "links": [],
                "alias": "dvr",
                "description":
                    "Enables configuration of Distributed Virtual Routers."
            },
            {
                "updated": "2013-07-23T10:00:00-00:00",
                "name": "Allowed Address Pairs",
                "links": [],
                "alias": "allowed-address-pairs",
                "description": "Provides allowed address pairs"
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'extensions.json']),
                 json=dict(extensions=body))
        ])
        self.assertTrue(self.cloud._has_neutron_extension('dvr'))
        self.assert_calls()

    def test__has_neutron_extension_missing(self):
        body = [
            {
                "updated": "2014-06-1T10:00:00-00:00",
                "name": "Distributed Virtual Router",
                "links": [],
                "alias": "dvr",
                "description":
                    "Enables configuration of Distributed Virtual Routers."
            },
            {
                "updated": "2013-07-23T10:00:00-00:00",
                "name": "Allowed Address Pairs",
                "links": [],
                "alias": "allowed-address-pairs",
                "description": "Provides allowed address pairs"
            },
        ]
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'extensions.json']),
                 json=dict(extensions=body))
        ])
        self.assertFalse(self.cloud._has_neutron_extension('invalid'))
        self.assert_calls()

    def test_range_search(self):
        filters = {"key1": "min", "key2": "20"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(1, len(retval))
        self.assertEqual([RANGE_DATA[1]], retval)

    def test_range_search_2(self):
        filters = {"key1": "<=2", "key2": ">10"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(2, len(retval))
        self.assertEqual([RANGE_DATA[1], RANGE_DATA[3]], retval)

    def test_range_search_3(self):
        filters = {"key1": "2", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(0, len(retval))

    def test_range_search_4(self):
        filters = {"key1": "max", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(0, len(retval))

    def test_range_search_5(self):
        filters = {"key1": "min", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(1, len(retval))
        self.assertEqual([RANGE_DATA[0]], retval)