<reponame>jogeo/rally-openstack
# Copyright 2014: Mirantis Inc.
# All Rights Reserved.
#
#    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.

from unittest import mock

import ddt
from neutronclient.common import exceptions as neutron_exceptions

from rally.common import utils

from rally_openstack.common import consts
from rally_openstack.common.wrappers import network
from tests.unit import test


SVC = "rally_openstack.common.wrappers.network."


class Owner(utils.RandomNameGeneratorMixin):
    task = {"uuid": "task-uuid"}


@ddt.ddt
class NeutronWrapperTestCase(test.TestCase):
    def setUp(self):
        super(NeutronWrapperTestCase, self).setUp()
        self.owner = Owner()
        self.owner.generate_random_name = mock.Mock()
        self.wrapper = network.NeutronWrapper(mock.MagicMock(),
                                              self.owner,
                                              config={})
        self._nc = self.wrapper.neutron.client

    def test_SUBNET_IP_VERSION(self):
        self.assertEqual(4, network.NeutronWrapper.SUBNET_IP_VERSION)

    @mock.patch(
        "rally_openstack.common.services.network.net_utils.generate_cidr")
    def test__generate_cidr(self, mock_generate_cidr):
        cidrs = iter(range(5))

        def fake_gen_cidr(ip_version=None, start_cidr=None):
            return 4, 3 + next(cidrs)

        mock_generate_cidr.side_effect = fake_gen_cidr

        self.assertEqual(3, self.wrapper._generate_cidr())
        self.assertEqual(4, self.wrapper._generate_cidr())
        self.assertEqual(5, self.wrapper._generate_cidr())
        self.assertEqual(6, self.wrapper._generate_cidr())
        self.assertEqual(7, self.wrapper._generate_cidr())
        self.assertEqual([mock.call(start_cidr=self.wrapper.start_cidr)] * 5,
                         mock_generate_cidr.call_args_list)

    def test_external_networks(self):
        self._nc.list_networks.return_value = {"networks": "foo_networks"}
        self.assertEqual("foo_networks", self.wrapper.external_networks)
        self._nc.list_networks.assert_called_once_with(
            **{"router:external": True})

    def test_get_network(self):
        neutron_net = {"id": "foo_id",
                       "name": "foo_name",
                       "tenant_id": "foo_tenant",
                       "status": "foo_status",
                       "router:external": "foo_external",
                       "subnets": "foo_subnets"}
        expected_net = {"id": "foo_id",
                        "name": "foo_name",
                        "tenant_id": "foo_tenant",
                        "status": "foo_status",
                        "external": "foo_external",
                        "router_id": None,
                        "subnets": "foo_subnets"}
        self._nc.show_network.return_value = {"network": neutron_net}
        net = self.wrapper.get_network(net_id="foo_id")
        self.assertEqual(expected_net, net)
        self._nc.show_network.assert_called_once_with("foo_id")

        self._nc.show_network.side_effect = (
            neutron_exceptions.NeutronClientException)
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.get_network,
                          net_id="foo_id")

        self._nc.list_networks.return_value = {"networks": [neutron_net]}
        net = self.wrapper.get_network(name="foo_name")
        self.assertEqual(expected_net, net)
        self._nc.list_networks.assert_called_once_with(name="foo_name")

        self._nc.list_networks.return_value = {"networks": []}
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.get_network,
                          name="foo_name")

    def test_create_v1_pool(self):
        subnet = "subnet_id"
        tenant = "foo_tenant"
        expected_pool = {"pool": {
            "id": "pool_id",
            "name": self.owner.generate_random_name.return_value,
            "subnet_id": subnet,
            "tenant_id": tenant}}
        self.wrapper.client.create_pool.return_value = expected_pool
        resultant_pool = self.wrapper.create_v1_pool(tenant, subnet)
        self.wrapper.client.create_pool.assert_called_once_with({
            "pool": {"lb_method": "ROUND_ROBIN",
                     "subnet_id": subnet,
                     "tenant_id": tenant,
                     "protocol": "HTTP",
                     "name": self.owner.generate_random_name.return_value}})
        self.assertEqual(expected_pool, resultant_pool)

    def test_create_network(self):
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network("foo_tenant")
        self._nc.create_network.assert_called_once_with({
            "network": {"tenant_id": "foo_tenant",
                        "name": self.owner.generate_random_name.return_value}})
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": None,
                          "subnets": []}, net)

    def test_create_network_with_subnets(self):
        subnets_num = 4
        subnets_ids = iter(range(subnets_num))
        self._nc.create_subnet.side_effect = lambda i: {
            "subnet": {"id": "subnet-%d" % next(subnets_ids)}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}

        net = self.wrapper.create_network("foo_tenant",
                                          subnets_num=subnets_num)

        self._nc.create_network.assert_called_once_with({
            "network": {"tenant_id": "foo_tenant",
                        "name": self.owner.generate_random_name.return_value}})
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "router_id": None,
                          "tenant_id": "foo_tenant",
                          "subnets": ["subnet-%d" % i
                                      for i in range(subnets_num)]}, net)
        self.assertEqual(
            [mock.call({"subnet":
                        {"name": self.owner.generate_random_name.return_value,
                         "network_id": "foo_id",
                         "tenant_id": "foo_tenant",
                         "ip_version": self.wrapper.SUBNET_IP_VERSION,
                         "dns_nameservers": ["8.8.8.8", "8.8.4.4"],
                         "cidr": mock.ANY}})
             for i in range(subnets_num)],
            self.wrapper.client.create_subnet.call_args_list
        )

    def test_create_network_with_router(self):
        self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network("foo_tenant", add_router=True)
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": "foo_router",
                          "subnets": []}, net)
        self._nc.create_router.assert_called_once_with({
            "router": {
                "name": self.owner.generate_random_name(),
                "tenant_id": "foo_tenant"
            }
        })

    def test_create_network_with_router_and_subnets(self):
        subnets_num = 4
        self.wrapper._generate_cidr = mock.Mock(return_value="foo_cidr")
        self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
        self._nc.create_subnet.return_value = {"subnet": {"id": "foo_subnet"}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network(
            "foo_tenant", add_router=True, subnets_num=subnets_num,
            dns_nameservers=["foo_nameservers"])
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": "foo_router",
                          "subnets": ["foo_subnet"] * subnets_num}, net)
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "tenant_id": "foo_tenant"}})
        self.assertEqual(
            [
                mock.call(
                    {"subnet": {
                        "name": self.owner.generate_random_name.return_value,
                        "network_id": "foo_id",
                        "tenant_id": "foo_tenant",
                        "ip_version": self.wrapper.SUBNET_IP_VERSION,
                        "dns_nameservers": ["foo_nameservers"],
                        "cidr": mock.ANY
                    }}
                )
            ] * subnets_num,
            self._nc.create_subnet.call_args_list,
        )
        self.assertEqual(self._nc.add_interface_router.call_args_list,
                         [mock.call("foo_router", {"subnet_id": "foo_subnet"})
                          for i in range(subnets_num)])

    def test_delete_v1_pool(self):
        pool = {"pool": {"id": "pool-id"}}
        self.wrapper.delete_v1_pool(pool["pool"]["id"])
        self.wrapper.client.delete_pool.assert_called_once_with("pool-id")

    def test_delete_network(self):
        self._nc.list_ports.return_value = {"ports": []}
        self._nc.list_subnets.return_value = {"subnets": []}
        self._nc.delete_network.return_value = "foo_deleted"
        self.wrapper.delete_network(
            {"id": "foo_id", "router_id": None, "subnets": [], "name": "x",
             "status": "y", "external": False})
        self.assertFalse(self._nc.remove_gateway_router.called)
        self.assertFalse(self._nc.remove_interface_router.called)
        self.assertFalse(self._nc.client.delete_router.called)
        self.assertFalse(self._nc.client.delete_subnet.called)
        self._nc.delete_network.assert_called_once_with("foo_id")

    def test_delete_network_with_router_and_ports_and_subnets(self):

        subnets = ["foo_subnet", "bar_subnet"]
        ports = [{"id": "foo_port", "device_owner": "network:router_interface",
                  "device_id": "rounttter"},
                 {"id": "bar_port", "device_owner": "network:dhcp"}]
        self._nc.list_ports.return_value = ({"ports": ports})
        self._nc.list_subnets.return_value = (
            {"subnets": [{"id": id_} for id_ in subnets]})

        self.wrapper.delete_network(
            {"id": "foo_id", "router_id": "foo_router", "subnets": subnets,
             "lb_pools": [], "name": "foo", "status": "x", "external": False})

        self.assertEqual(self._nc.remove_gateway_router.mock_calls,
                         [mock.call("foo_router")])
        self._nc.delete_port.assert_called_once_with(ports[1]["id"])
        self._nc.remove_interface_router.assert_called_once_with(
            ports[0]["device_id"], {"port_id": ports[0]["id"]})
        self.assertEqual(
            [mock.call(subnet_id) for subnet_id in subnets],
            self._nc.delete_subnet.call_args_list
        )
        self._nc.delete_network.assert_called_once_with("foo_id")

    @ddt.data({"exception_type": neutron_exceptions.NotFound,
               "should_raise": False},
              {"exception_type": neutron_exceptions.BadRequest,
               "should_raise": False},
              {"exception_type": KeyError,
               "should_raise": True})
    @ddt.unpack
    def test_delete_network_with_router_throw_exception(
            self, exception_type, should_raise):
        # Ensure cleanup context still move forward even
        # remove_interface_router throw NotFound/BadRequest exception

        self._nc.remove_interface_router.side_effect = exception_type
        subnets = ["foo_subnet", "bar_subnet"]
        ports = [{"id": "foo_port", "device_owner": "network:router_interface",
                  "device_id": "rounttter"},
                 {"id": "bar_port", "device_owner": "network:dhcp"}]
        self._nc.list_ports.return_value = {"ports": ports}
        self._nc.list_subnets.return_value = {"subnets": [
            {"id": id_} for id_ in subnets]}

        if should_raise:
            self.assertRaises(
                exception_type, self.wrapper.delete_network,
                {"id": "foo_id", "name": "foo", "router_id": "foo_router",
                 "subnets": subnets, "lb_pools": [], "status": "xxx",
                 "external": False})
            self.assertFalse(self._nc.delete_subnet.called)
            self.assertFalse(self._nc.delete_network.called)
        else:
            self.wrapper.delete_network(
                {"id": "foo_id", "name": "foo", "status": "xxx",
                 "router_id": "foo_router", "subnets": subnets,
                 "lb_pools": [], "external": False})

            self._nc.delete_port.assert_called_once_with(ports[1]["id"])
            self._nc.remove_interface_router.assert_called_once_with(
                ports[0]["device_id"], {"port_id": ports[0]["id"]})
            self.assertEqual(
                [mock.call(subnet_id) for subnet_id in subnets],
                self._nc.delete_subnet.call_args_list
            )
            self._nc.delete_network.assert_called_once_with("foo_id")

            self._nc.remove_gateway_router.assert_called_once_with(
                "foo_router")

    def test_list_networks(self):
        self._nc.list_networks.return_value = {"networks": "foo_nets"}
        self.assertEqual("foo_nets", self.wrapper.list_networks())
        self._nc.list_networks.assert_called_once_with()

    def test_create_floating_ip(self):
        self._nc.create_port.return_value = {"port": {"id": "port_id"}}
        self._nc.create_floatingip.return_value = {
            "floatingip": {"id": "fip_id", "floating_ip_address": "fip_ip"}}

        self.assertRaises(ValueError, self.wrapper.create_floating_ip)

        self._nc.list_networks.return_value = {"networks": []}
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.create_floating_ip,
                          tenant_id="foo_tenant")

        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
        fip = self.wrapper.create_floating_ip(
            tenant_id="foo_tenant", port_id="port_id")
        self.assertEqual({"id": "fip_id", "ip": "fip_ip"}, fip)

        self._nc.list_networks.return_value = {"networks": [
            {"id": "ext_net_id", "name": "ext_net", "router:external": True}]}
        self.wrapper.create_floating_ip(
            tenant_id="foo_tenant", ext_network="ext_net", port_id="port_id")

        self.assertRaises(
            network.NetworkWrapperException,
            self.wrapper.create_floating_ip, tenant_id="foo_tenant",
            ext_network="ext_net_2")

    def test_delete_floating_ip(self):
        self.wrapper.delete_floating_ip("fip_id")
        self.wrapper.delete_floating_ip("fip_id", ignored_kwarg="bar")
        self.assertEqual([mock.call("fip_id")] * 2,
                         self._nc.delete_floatingip.call_args_list)

    def test_create_router(self):
        self._nc.create_router.return_value = {"router": "foo_router"}
        self._nc.list_extensions.return_value = {
            "extensions": [{"alias": "ext-gw-mode"}]}
        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}

        router = self.wrapper.create_router()
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value}})
        self.assertEqual("foo_router", router)

        self.wrapper.create_router(external=True, flavor_id="bar")
        self._nc.create_router.assert_called_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "external_gateway_info": {
                            "network_id": "ext_id",
                            "enable_snat": True},
                        "flavor_id": "bar"}})

    def test_create_router_without_ext_gw_mode_extension(self):
        self._nc.create_router.return_value = {"router": "foo_router"}
        self._nc.list_extensions.return_value = {"extensions": []}
        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}

        router = self.wrapper.create_router()
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value}})
        self.assertEqual(router, "foo_router")

        self.wrapper.create_router(external=True, flavor_id="bar")
        self._nc.create_router.assert_called_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "external_gateway_info": {"network_id": "ext_id"},
                        "flavor_id": "bar"}})

    def test_create_port(self):
        self._nc.create_port.return_value = {"port": "foo_port"}

        port = self.wrapper.create_port("foo_net")
        self._nc.create_port.assert_called_once_with(
            {"port": {"network_id": "foo_net",
                      "name": self.owner.generate_random_name.return_value}})
        self.assertEqual("foo_port", port)

        port = self.wrapper.create_port("foo_net", foo="bar")
        self.wrapper.client.create_port.assert_called_with(
            {"port": {"network_id": "foo_net",
                      "name": self.owner.generate_random_name.return_value,
                      "foo": "bar"}})

    def test_supports_extension(self):
        self._nc.list_extensions.return_value = (
            {"extensions": [{"alias": "extension"}]})
        self.assertTrue(self.wrapper.supports_extension("extension")[0])

        self.wrapper.neutron._cached_supported_extensions = None
        self._nc.list_extensions.return_value = (
            {"extensions": [{"alias": "extension"}]})
        self.assertFalse(self.wrapper.supports_extension("dummy-group")[0])

        self.wrapper.neutron._cached_supported_extensions = None
        self._nc.list_extensions.return_value = {"extensions": []}
        self.assertFalse(self.wrapper.supports_extension("extension")[0])


class FunctionsTestCase(test.TestCase):

    def test_wrap(self):
        mock_clients = mock.Mock()
        config = {"fakearg": "fake"}
        owner = Owner()

        mock_clients.services.return_value = {"foo": consts.Service.NEUTRON}
        wrapper = network.wrap(mock_clients, owner, config)
        self.assertIsInstance(wrapper, network.NeutronWrapper)
        self.assertEqual(wrapper.owner, owner)
        self.assertEqual(wrapper.config, config)

# Copyright 2014: Mirantis Inc.
# All Rights Reserved.
#
#    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.

from unittest import mock

import ddt
from neutronclient.common import exceptions as neutron_exceptions

from rally.common import utils

from rally_openstack.common import consts
from rally_openstack.common.wrappers import network
from tests.unit import test


SVC = "rally_openstack.common.wrappers.network."


class Owner(utils.RandomNameGeneratorMixin):
    task = {"uuid": "task-uuid"}


@ddt.ddt
class NeutronWrapperTestCase(test.TestCase):
    def setUp(self):
        super(NeutronWrapperTestCase, self).setUp()
        self.owner = Owner()
        self.owner.generate_random_name = mock.Mock()
        self.wrapper = network.NeutronWrapper(mock.MagicMock(),
                                              self.owner,
                                              config={})
        self._nc = self.wrapper.neutron.client

    def test_SUBNET_IP_VERSION(self):
        self.assertEqual(4, network.NeutronWrapper.SUBNET_IP_VERSION)

    @mock.patch(
        "rally_openstack.common.services.network.net_utils.generate_cidr")
    def test__generate_cidr(self, mock_generate_cidr):
        cidrs = iter(range(5))

        def fake_gen_cidr(ip_version=None, start_cidr=None):
            return 4, 3 + next(cidrs)

        mock_generate_cidr.side_effect = fake_gen_cidr

        self.assertEqual(3, self.wrapper._generate_cidr())
        self.assertEqual(4, self.wrapper._generate_cidr())
        self.assertEqual(5, self.wrapper._generate_cidr())
        self.assertEqual(6, self.wrapper._generate_cidr())
        self.assertEqual(7, self.wrapper._generate_cidr())
        self.assertEqual([mock.call(start_cidr=self.wrapper.start_cidr)] * 5,
                         mock_generate_cidr.call_args_list)

    def test_external_networks(self):
        self._nc.list_networks.return_value = {"networks": "foo_networks"}
        self.assertEqual("foo_networks", self.wrapper.external_networks)
        self._nc.list_networks.assert_called_once_with(
            **{"router:external": True})

    def test_get_network(self):
        neutron_net = {"id": "foo_id",
                       "name": "foo_name",
                       "tenant_id": "foo_tenant",
                       "status": "foo_status",
                       "router:external": "foo_external",
                       "subnets": "foo_subnets"}
        expected_net = {"id": "foo_id",
                        "name": "foo_name",
                        "tenant_id": "foo_tenant",
                        "status": "foo_status",
                        "external": "foo_external",
                        "router_id": None,
                        "subnets": "foo_subnets"}
        self._nc.show_network.return_value = {"network": neutron_net}
        net = self.wrapper.get_network(net_id="foo_id")
        self.assertEqual(expected_net, net)
        self._nc.show_network.assert_called_once_with("foo_id")

        self._nc.show_network.side_effect = (
            neutron_exceptions.NeutronClientException)
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.get_network,
                          net_id="foo_id")

        self._nc.list_networks.return_value = {"networks": [neutron_net]}
        net = self.wrapper.get_network(name="foo_name")
        self.assertEqual(expected_net, net)
        self._nc.list_networks.assert_called_once_with(name="foo_name")

        self._nc.list_networks.return_value = {"networks": []}
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.get_network,
                          name="foo_name")

    def test_create_v1_pool(self):
        subnet = "subnet_id"
        tenant = "foo_tenant"
        expected_pool = {"pool": {
            "id": "pool_id",
            "name": self.owner.generate_random_name.return_value,
            "subnet_id": subnet,
            "tenant_id": tenant}}
        self.wrapper.client.create_pool.return_value = expected_pool
        resultant_pool = self.wrapper.create_v1_pool(tenant, subnet)
        self.wrapper.client.create_pool.assert_called_once_with({
            "pool": {"lb_method": "ROUND_ROBIN",
                     "subnet_id": subnet,
                     "tenant_id": tenant,
                     "protocol": "HTTP",
                     "name": self.owner.generate_random_name.return_value}})
        self.assertEqual(expected_pool, resultant_pool)

    def test_create_network(self):
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network("foo_tenant")
        self._nc.create_network.assert_called_once_with({
            "network": {"tenant_id": "foo_tenant",
                        "name": self.owner.generate_random_name.return_value}})
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": None,
                          "subnets": []}, net)

    def test_create_network_with_subnets(self):
        subnets_num = 4
        subnets_ids = iter(range(subnets_num))
        self._nc.create_subnet.side_effect = lambda i: {
            "subnet": {"id": "subnet-%d" % next(subnets_ids)}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}

        net = self.wrapper.create_network("foo_tenant",
                                          subnets_num=subnets_num)

        self._nc.create_network.assert_called_once_with({
            "network": {"tenant_id": "foo_tenant",
                        "name": self.owner.generate_random_name.return_value}})
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "router_id": None,
                          "tenant_id": "foo_tenant",
                          "subnets": ["subnet-%d" % i
                                      for i in range(subnets_num)]}, net)
        self.assertEqual(
            [mock.call({"subnet":
                        {"name": self.owner.generate_random_name.return_value,
                         "network_id": "foo_id",
                         "tenant_id": "foo_tenant",
                         "ip_version": self.wrapper.SUBNET_IP_VERSION,
                         "dns_nameservers": ["8.8.8.8", "8.8.4.4"],
                         "cidr": mock.ANY}})
             for i in range(subnets_num)],
            self.wrapper.client.create_subnet.call_args_list
        )

    def test_create_network_with_router(self):
        self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network("foo_tenant", add_router=True)
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": "foo_router",
                          "subnets": []}, net)
        self._nc.create_router.assert_called_once_with({
            "router": {
                "name": self.owner.generate_random_name(),
                "tenant_id": "foo_tenant"
            }
        })

    def test_create_network_with_router_and_subnets(self):
        subnets_num = 4
        self.wrapper._generate_cidr = mock.Mock(return_value="foo_cidr")
        self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
        self._nc.create_subnet.return_value = {"subnet": {"id": "foo_subnet"}}
        self._nc.create_network.return_value = {
            "network": {"id": "foo_id",
                        "name": self.owner.generate_random_name.return_value,
                        "status": "foo_status"}}
        net = self.wrapper.create_network(
            "foo_tenant", add_router=True, subnets_num=subnets_num,
            dns_nameservers=["foo_nameservers"])
        self.assertEqual({"id": "foo_id",
                          "name": self.owner.generate_random_name.return_value,
                          "status": "foo_status",
                          "external": False,
                          "tenant_id": "foo_tenant",
                          "router_id": "foo_router",
                          "subnets": ["foo_subnet"] * subnets_num}, net)
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "tenant_id": "foo_tenant"}})
        self.assertEqual(
            [
                mock.call(
                    {"subnet": {
                        "name": self.owner.generate_random_name.return_value,
                        "network_id": "foo_id",
                        "tenant_id": "foo_tenant",
                        "ip_version": self.wrapper.SUBNET_IP_VERSION,
                        "dns_nameservers": ["foo_nameservers"],
                        "cidr": mock.ANY
                    }}
                )
            ] * subnets_num,
            self._nc.create_subnet.call_args_list,
        )
        self.assertEqual(self._nc.add_interface_router.call_args_list,
                         [mock.call("foo_router", {"subnet_id": "foo_subnet"})
                          for i in range(subnets_num)])

    def test_delete_v1_pool(self):
        pool = {"pool": {"id": "pool-id"}}
        self.wrapper.delete_v1_pool(pool["pool"]["id"])
        self.wrapper.client.delete_pool.assert_called_once_with("pool-id")

    def test_delete_network(self):
        self._nc.list_ports.return_value = {"ports": []}
        self._nc.list_subnets.return_value = {"subnets": []}
        self._nc.delete_network.return_value = "foo_deleted"
        self.wrapper.delete_network(
            {"id": "foo_id", "router_id": None, "subnets": [], "name": "x",
             "status": "y", "external": False})
        self.assertFalse(self._nc.remove_gateway_router.called)
        self.assertFalse(self._nc.remove_interface_router.called)
        self.assertFalse(self._nc.client.delete_router.called)
        self.assertFalse(self._nc.client.delete_subnet.called)
        self._nc.delete_network.assert_called_once_with("foo_id")

    def test_delete_network_with_router_and_ports_and_subnets(self):

        subnets = ["foo_subnet", "bar_subnet"]
        ports = [{"id": "foo_port", "device_owner": "network:router_interface",
                  "device_id": "rounttter"},
                 {"id": "bar_port", "device_owner": "network:dhcp"}]
        self._nc.list_ports.return_value = ({"ports": ports})
        self._nc.list_subnets.return_value = (
            {"subnets": [{"id": id_} for id_ in subnets]})

        self.wrapper.delete_network(
            {"id": "foo_id", "router_id": "foo_router", "subnets": subnets,
             "lb_pools": [], "name": "foo", "status": "x", "external": False})

        self.assertEqual(self._nc.remove_gateway_router.mock_calls,
                         [mock.call("foo_router")])
        self._nc.delete_port.assert_called_once_with(ports[1]["id"])
        self._nc.remove_interface_router.assert_called_once_with(
            ports[0]["device_id"], {"port_id": ports[0]["id"]})
        self.assertEqual(
            [mock.call(subnet_id) for subnet_id in subnets],
            self._nc.delete_subnet.call_args_list
        )
        self._nc.delete_network.assert_called_once_with("foo_id")

    @ddt.data({"exception_type": neutron_exceptions.NotFound,
               "should_raise": False},
              {"exception_type": neutron_exceptions.BadRequest,
               "should_raise": False},
              {"exception_type": KeyError,
               "should_raise": True})
    @ddt.unpack
    def test_delete_network_with_router_throw_exception(
            self, exception_type, should_raise):
        # Ensure cleanup context still move forward even
        # remove_interface_router throw NotFound/BadRequest exception

        self._nc.remove_interface_router.side_effect = exception_type
        subnets = ["foo_subnet", "bar_subnet"]
        ports = [{"id": "foo_port", "device_owner": "network:router_interface",
                  "device_id": "rounttter"},
                 {"id": "bar_port", "device_owner": "network:dhcp"}]
        self._nc.list_ports.return_value = {"ports": ports}
        self._nc.list_subnets.return_value = {"subnets": [
            {"id": id_} for id_ in subnets]}

        if should_raise:
            self.assertRaises(
                exception_type, self.wrapper.delete_network,
                {"id": "foo_id", "name": "foo", "router_id": "foo_router",
                 "subnets": subnets, "lb_pools": [], "status": "xxx",
                 "external": False})
            self.assertFalse(self._nc.delete_subnet.called)
            self.assertFalse(self._nc.delete_network.called)
        else:
            self.wrapper.delete_network(
                {"id": "foo_id", "name": "foo", "status": "xxx",
                 "router_id": "foo_router", "subnets": subnets,
                 "lb_pools": [], "external": False})

            self._nc.delete_port.assert_called_once_with(ports[1]["id"])
            self._nc.remove_interface_router.assert_called_once_with(
                ports[0]["device_id"], {"port_id": ports[0]["id"]})
            self.assertEqual(
                [mock.call(subnet_id) for subnet_id in subnets],
                self._nc.delete_subnet.call_args_list
            )
            self._nc.delete_network.assert_called_once_with("foo_id")

            self._nc.remove_gateway_router.assert_called_once_with(
                "foo_router")

    def test_list_networks(self):
        self._nc.list_networks.return_value = {"networks": "foo_nets"}
        self.assertEqual("foo_nets", self.wrapper.list_networks())
        self._nc.list_networks.assert_called_once_with()

    def test_create_floating_ip(self):
        self._nc.create_port.return_value = {"port": {"id": "port_id"}}
        self._nc.create_floatingip.return_value = {
            "floatingip": {"id": "fip_id", "floating_ip_address": "fip_ip"}}

        self.assertRaises(ValueError, self.wrapper.create_floating_ip)

        self._nc.list_networks.return_value = {"networks": []}
        self.assertRaises(network.NetworkWrapperException,
                          self.wrapper.create_floating_ip,
                          tenant_id="foo_tenant")

        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
        fip = self.wrapper.create_floating_ip(
            tenant_id="foo_tenant", port_id="port_id")
        self.assertEqual({"id": "fip_id", "ip": "fip_ip"}, fip)

        self._nc.list_networks.return_value = {"networks": [
            {"id": "ext_net_id", "name": "ext_net", "router:external": True}]}
        self.wrapper.create_floating_ip(
            tenant_id="foo_tenant", ext_network="ext_net", port_id="port_id")

        self.assertRaises(
            network.NetworkWrapperException,
            self.wrapper.create_floating_ip, tenant_id="foo_tenant",
            ext_network="ext_net_2")

    def test_delete_floating_ip(self):
        self.wrapper.delete_floating_ip("fip_id")
        self.wrapper.delete_floating_ip("fip_id", ignored_kwarg="bar")
        self.assertEqual([mock.call("fip_id")] * 2,
                         self._nc.delete_floatingip.call_args_list)

    def test_create_router(self):
        self._nc.create_router.return_value = {"router": "foo_router"}
        self._nc.list_extensions.return_value = {
            "extensions": [{"alias": "ext-gw-mode"}]}
        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}

        router = self.wrapper.create_router()
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value}})
        self.assertEqual("foo_router", router)

        self.wrapper.create_router(external=True, flavor_id="bar")
        self._nc.create_router.assert_called_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "external_gateway_info": {
                            "network_id": "ext_id",
                            "enable_snat": True},
                        "flavor_id": "bar"}})

    def test_create_router_without_ext_gw_mode_extension(self):
        self._nc.create_router.return_value = {"router": "foo_router"}
        self._nc.list_extensions.return_value = {"extensions": []}
        self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}

        router = self.wrapper.create_router()
        self._nc.create_router.assert_called_once_with(
            {"router": {"name": self.owner.generate_random_name.return_value}})
        self.assertEqual(router, "foo_router")

        self.wrapper.create_router(external=True, flavor_id="bar")
        self._nc.create_router.assert_called_with(
            {"router": {"name": self.owner.generate_random_name.return_value,
                        "external_gateway_info": {"network_id": "ext_id"},
                        "flavor_id": "bar"}})

    def test_create_port(self):
        self._nc.create_port.return_value = {"port": "foo_port"}

        port = self.wrapper.create_port("foo_net")
        self._nc.create_port.assert_called_once_with(
            {"port": {"network_id": "foo_net",
                      "name": self.owner.generate_random_name.return_value}})
        self.assertEqual("foo_port", port)

        port = self.wrapper.create_port("foo_net", foo="bar")
        self.wrapper.client.create_port.assert_called_with(
            {"port": {"network_id": "foo_net",
                      "name": self.owner.generate_random_name.return_value,
                      "foo": "bar"}})

    def test_supports_extension(self):
        self._nc.list_extensions.return_value = (
            {"extensions": [{"alias": "extension"}]})
        self.assertTrue(self.wrapper.supports_extension("extension")[0])

        self.wrapper.neutron._cached_supported_extensions = None
        self._nc.list_extensions.return_value = (
            {"extensions": [{"alias": "extension"}]})
        self.assertFalse(self.wrapper.supports_extension("dummy-group")[0])

        self.wrapper.neutron._cached_supported_extensions = None
        self._nc.list_extensions.return_value = {"extensions": []}
        self.assertFalse(self.wrapper.supports_extension("extension")[0])


class FunctionsTestCase(test.TestCase):

    def test_wrap(self):
        mock_clients = mock.Mock()
        config = {"fakearg": "fake"}
        owner = Owner()

        mock_clients.services.return_value = {"foo": consts.Service.NEUTRON}
        wrapper = network.wrap(mock_clients, owner, config)
        self.assertIsInstance(wrapper, network.NeutronWrapper)
        self.assertEqual(wrapper.owner, owner)
        self.assertEqual(wrapper.config, config)
# Copyright 2014: Mirantis Inc. # All Rights Reserved. # #    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. # Ensure cleanup context still move forward even # remove_interface_router throw NotFound/BadRequest exception