# Copyright 2013 Openstack Foundation
# 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.

import contextlib
import copy
import time
import uuid

import mock
from neutron.api.v2 import attributes as neutron_attrs
from neutron.common import exceptions
from neutron.openstack.common.notifier import api as notifier_api
from oslo.config import cfg

from quark.db import models
from quark.tests import test_quark_plugin


class TestQuarkGetSubnetCount(test_quark_plugin.TestQuarkPlugin):
    def test_get_subnet_count(self):
        """This isn't really testable."""
        with mock.patch("quark.db.api.subnet_count_all"):
            self.plugin.get_subnets_count(self.context, {})


class TestQuarkGetSubnets(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnets=None, routes=None):
        if routes is None:
            routes = []
        route_models = []
        for route in routes:
            r = models.Route()
            r.update(route)
            route_models.append(r)

        if isinstance(subnets, list):
            subnet_models = []
            for subnet in subnets:
                s_dict = subnet.copy()
                s_dict["routes"] = route_models
                s = models.Subnet(network=models.Network())
                s.update(s_dict)
                subnet_models.append(s)
        elif subnets:
            mod = models.Subnet(network=models.Network())
            mod.update(subnets)
            mod["routes"] = route_models
            subnet_models = mod
        else:
            subnet_models = None

        with mock.patch("quark.db.api.subnet_find") as subnet_find:
            subnet_find.return_value = subnet_models
            yield

    def test_subnets_list(self):
        subnet_id = str(uuid.uuid4())
        route = dict(id=1, cidr="0.0.0.0/0", gateway="192.168.0.1")

        subnet = dict(id=subnet_id, network_id=1, name=subnet_id,
                      tenant_id=self.context.tenant_id, ip_version=4,
                      cidr="192.168.0.0/24", gateway_ip="192.168.0.1",
                      dns_nameservers=[],
                      enable_dhcp=None)
        expected_route = dict(destination=route["cidr"],
                              nexthop=route["gateway"])

        with self._stubs(subnets=[subnet], routes=[route]):
            res = self.plugin.get_subnets(self.context, {}, {})
            # Compare routes separately
            routes = res[0].pop("host_routes")
            for key in subnet.keys():
                self.assertEqual(res[0][key], subnet[key])
            for key in expected_route.keys():
                self.assertEqual(routes[0][key], expected_route[key])

    def test_subnet_show_fail(self):
        with self._stubs():
            with self.assertRaises(exceptions.SubnetNotFound):
                self.plugin.get_subnet(self.context, 1)

    def test_subnet_show(self):
        subnet_id = str(uuid.uuid4())
        route = dict(id=1, cidr="0.0.0.0/0", gateway="192.168.0.1",
                     subnet_id=subnet_id)

        expected_route = dict(destination=route["cidr"],
                              nexthop=route["gateway"])

        subnet = dict(id=subnet_id, network_id=1, name=subnet_id,
                      tenant_id=self.context.tenant_id, ip_version=4,
                      cidr="192.168.0.0/24", gateway_ip="192.168.0.1",
                      dns_nameservers=[],
                      enable_dhcp=None)

        with self._stubs(subnets=subnet, routes=[route]):
            res = self.plugin.get_subnet(self.context, subnet_id)

            # Compare routes separately
            routes = res.pop("host_routes")
            for key in subnet.keys():
                self.assertEqual(res[key], subnet[key])
            for key in expected_route.keys():
                self.assertEqual(routes[0][key], expected_route[key])


class TestQuarkCreateSubnetOverlapping(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnets=None):
        if subnets is None:
            subnets = []
        subnet_models = []
        for subnet in subnets:
            s = models.Subnet()
            s.update(subnet)
            subnet_models.append(s)
        network = models.Network()
        network.update(dict(id=1, subnets=subnet_models))
        with contextlib.nested(
            mock.patch("quark.db.api.network_find"),
            mock.patch("quark.db.api.subnet_find"),
            mock.patch("quark.db.api.subnet_create")
        ) as (net_find, subnet_find, subnet_create):
            net_find.return_value = network
            subnet_find.return_value = subnet_models
            subnet_create.return_value = models.Subnet(
                network=models.Network(),
                cidr="192.168.1.1/24")
            yield subnet_create

    def test_create_subnet_overlapping_true(self):
        cfg.CONF.set_override('allow_overlapping_ips', True)
        with self._stubs() as subnet_create:
            s = dict(subnet=dict(
                gateway_ip=neutron_attrs.ATTR_NOT_SPECIFIED,
                dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                cidr="192.168.1.1/8",
                network_id=1))
            self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)

    def test_create_subnet_overlapping_false(self):
        cfg.CONF.set_override('allow_overlapping_ips', False)
        with self._stubs() as subnet_create:
            s = dict(subnet=dict(
                gateway_ip=neutron_attrs.ATTR_NOT_SPECIFIED,
                dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                cidr="192.168.1.1/8",
                network_id=1))
            self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)

    def test_create_subnet_overlapping_conflict(self):
        cfg.CONF.set_override('allow_overlapping_ips', False)
        with self._stubs(subnets=[dict(cidr="192.168.10.1/24")]):
            with self.assertRaises(exceptions.InvalidInput):
                s = dict(subnet=dict(cidr="192.168.1.1/8",
                                     network_id=1))
                self.plugin.create_subnet(self.context, s)


class TestQuarkCreateSubnetAllocationPools(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnet):
        s = models.Subnet(network=models.Network(id=1, subnets=[]))
        s.update(subnet)

        with contextlib.nested(
            mock.patch("quark.db.api.network_find"),
            mock.patch("quark.db.api.subnet_find"),
            mock.patch("quark.db.api.subnet_create"),
        ) as (net_find, subnet_find, subnet_create):
            net_find.return_value = s["network"]
            subnet_find.return_value = []
            subnet_create.return_value = s
            yield subnet_create

    def setUp(self):
        super(TestQuarkCreateSubnetAllocationPools, self).setUp()

    def tearDown(self):
        super(TestQuarkCreateSubnetAllocationPools, self).tearDown()

    def test_create_subnet_allocation_pools_zero(self):
        s = dict(subnet=dict(
            cidr="192.168.1.1/24",
            network_id=1))
        with self._stubs(s["subnet"]) as (subnet_create):
            resp = self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(resp["allocation_pools"],
                             [dict(start="192.168.1.2", end="192.168.1.254")])

    def test_create_subnet_allocation_pools_one(self):
        pools = [dict(start="192.168.1.10", end="192.168.1.20")]
        s = dict(subnet=dict(
            allocation_pools=pools,
            cidr="192.168.1.1/24",
            network_id=1))
        with self._stubs(s["subnet"]) as (subnet_create):
            resp = self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(resp["allocation_pools"], pools)

    def test_create_subnet_allocation_pools_two(self):
        pools = [dict(start="192.168.1.10", end="192.168.1.20"),
                 dict(start="192.168.1.40", end="192.168.1.50")]
        s = dict(subnet=dict(
            allocation_pools=pools,
            cidr="192.168.1.1/24",
            network_id=1))
        with self._stubs(s["subnet"]) as (subnet_create):
            resp = self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(resp["allocation_pools"], pools)

    def test_create_subnet_allocation_pools_empty_list(self):
        pools = []
        s = dict(subnet=dict(
            allocation_pools=pools,
            cidr="192.168.1.1/24",
            network_id=1))
        with self._stubs(s["subnet"]) as (subnet_create):
            resp = self.plugin.create_subnet(self.context, s)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(resp["allocation_pools"], pools)


# TODO(amir): Refactor the tests to test individual subnet attributes.
# * copy.deepcopy was necessary to maintain tests on keys, which is a bit ugly.
# * workaround is also in place for lame ATTR_NOT_SPECIFIED object()
class TestQuarkCreateSubnet(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnet=None, network=None, routes=None, dns=None):

        if network:
            net = models.Network()
            net.update(network)
            network = net
        subnet_mod = models.Subnet(network=models.Network())
        dns_ips = subnet.pop("dns_nameservers", [])
        host_routes = subnet.pop("host_routes", [])
        subnet_mod.update(subnet)

        subnet["dns_nameservers"] = dns_ips
        subnet["host_routes"] = host_routes
        routes = routes or []
        dns = dns or []
        route_models = [models.Route(**r) for r in routes]
        dns_models = [models.DNSNameserver(**d) for d in dns]

        with contextlib.nested(
            mock.patch("quark.db.api.subnet_create"),
            mock.patch("quark.db.api.network_find"),
            mock.patch("quark.db.api.dns_create"),
            mock.patch("quark.db.api.route_create"),
        ) as (subnet_create, net_find, dns_create, route_create):
            subnet_create.return_value = subnet_mod
            net_find.return_value = network
            route_create.side_effect = route_models
            dns_create.side_effect = dns_models
            yield subnet_create, dns_create, route_create

    def test_create_subnet(self):
        routes = [dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24", gateway_ip="0.0.0.0",
                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                        host_routes=neutron_attrs.ATTR_NOT_SPECIFIED,
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes
        ) as (subnet_create, dns_create, route_create):
            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
            host_routes = subnet["subnet"].pop("host_routes")
            subnet_request = copy.deepcopy(subnet)
            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
            subnet_request["subnet"]["host_routes"] = host_routes
            res = self.plugin.create_subnet(self.context,
                                            subnet_request)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            for key in subnet["subnet"].keys():
                if key == "host_routes":
                    self.assertEqual(res[key][0]["destination"], "0.0.0.0/0")
                    self.assertEqual(res[key][0]["nexthop"], "0.0.0.0")
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])

    def test_create_subnet_no_network_fails(self):
        subnet = dict(subnet=dict(network_id=1))
        with self._stubs(subnet=dict(), network=None):
            with self.assertRaises(exceptions.NetworkNotFound):
                self.plugin.create_subnet(self.context, subnet)

    def test_create_subnet_no_gateway_ip_defaults(self):
        routes = [dict(cidr="0.0.0.0/0", gateway="172.16.0.1")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24",
                        gateway_ip=neutron_attrs.ATTR_NOT_SPECIFIED,
                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes
        ) as (subnet_create, dns_create, route_create):
            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
            gateway_ip = subnet["subnet"].pop("gateway_ip")
            subnet_request = copy.deepcopy(subnet)
            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
            subnet_request["subnet"]["gateway_ip"] = gateway_ip
            res = self.plugin.create_subnet(self.context, subnet_request)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            for key in subnet["subnet"].keys():
                if key == "gateway_ip":
                    self.assertEqual(res[key], "172.16.0.1")
                elif key == "host_routes":
                    self.assertEqual(res[key][0]["destination"], "0.0.0.0/0")
                    self.assertEqual(res[key][0]["nexthop"], "172.16.0.1")
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])

    def test_create_subnet_dns_nameservers(self):
        routes = [dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
        dns_ns = [dict(ip="4.2.2.1"), dict(ip="4.2.2.2")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24", gateway_ip="0.0.0.0",
                        dns_nameservers=["4.2.2.1", "4.2.2.2"],
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes,
            dns=dns_ns
        ) as (subnet_create, dns_create, route_create):
            res = self.plugin.create_subnet(self.context,
                                            copy.deepcopy(subnet))
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 2)
            self.assertEqual(route_create.call_count, 1)
            for key in subnet["subnet"].keys():
                if key == "host_routes":
                    self.assertEqual(res[key][0]["destination"], "0.0.0.0/0")
                    self.assertEqual(res[key][0]["nexthop"], "0.0.0.0")
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])

    def test_create_subnet_routes(self):
        routes = [dict(cidr="1.1.1.1/8", gateway="172.16.0.4"),
                  dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24", gateway_ip="0.0.0.0",
                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                        host_routes=[{"destination": "1.1.1.1/8",
                                      "nexthop": "172.16.0.4"}],
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes
        ) as (subnet_create, dns_create, route_create):
            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
            subnet_request = copy.deepcopy(subnet)
            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
            res = self.plugin.create_subnet(self.context, subnet_request)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 2)
            for key in subnet["subnet"].keys():
                if key == "host_routes":
                    res_tuples = [(r["destination"], r["nexthop"])
                                  for r in res[key]]
                    self.assertIn(("1.1.1.1/8", "172.16.0.4"), res_tuples)
                    self.assertIn(("0.0.0.0/0", "0.0.0.0"), res_tuples)
                    self.assertEqual(2, len(res_tuples))
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])

    def test_create_subnet_default_route(self):
        routes = [dict(cidr="0.0.0.0/0", gateway="172.16.0.4")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24",
                        gateway_ip=neutron_attrs.ATTR_NOT_SPECIFIED,
                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                        host_routes=[{"destination": "0.0.0.0/0",
                                      "nexthop": "172.16.0.4"}],
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes
        ) as (subnet_create, dns_create, route_create):
            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
            gateway_ip = subnet["subnet"].pop("gateway_ip")
            subnet_request = copy.deepcopy(subnet)
            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
            subnet_request["subnet"]["gateway_ip"] = gateway_ip
            res = self.plugin.create_subnet(self.context, subnet_request)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            for key in subnet["subnet"].keys():
                if key == "host_routes":
                    res_tuples = [(r["destination"], r["nexthop"])
                                  for r in res[key]]
                    self.assertEqual([("0.0.0.0/0", "172.16.0.4")], res_tuples)
                elif key == "gateway_ip":
                    self.assertEqual(res[key], "172.16.0.4")
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])

    def test_create_subnet_default_route_gateway_ip(self):
        """If default route (host_routes) and gateway_ip are both provided,
        then host_route takes precedence.
        """
        routes = [dict(cidr="0.0.0.0/0", gateway="172.16.0.4")]
        subnet = dict(
            subnet=dict(network_id=1,
                        tenant_id=self.context.tenant_id, ip_version=4,
                        cidr="172.16.0.0/24",
                        gateway_ip="172.16.0.3",
                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
                        host_routes=[{"destination": "0.0.0.0/0",
                                      "nexthop": "172.16.0.4"}],
                        enable_dhcp=None))
        network = dict(network_id=1)
        with self._stubs(
            subnet=subnet["subnet"],
            network=network,
            routes=routes
        ) as (subnet_create, dns_create, route_create):
            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
            subnet_request = copy.deepcopy(subnet)
            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
            res = self.plugin.create_subnet(self.context, subnet_request)
            self.assertEqual(subnet_create.call_count, 1)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            for key in subnet["subnet"].keys():
                if key == "host_routes":
                    res_tuples = [(r["destination"], r["nexthop"])
                                  for r in res[key]]
                    self.assertEqual([("0.0.0.0/0", "172.16.0.4")], res_tuples)
                elif key == "gateway_ip":
                    self.assertEqual(res[key], "172.16.0.4")
                else:
                    self.assertEqual(res[key], subnet["subnet"][key])


class TestQuarkUpdateSubnet(test_quark_plugin.TestQuarkPlugin):
    DEFAULT_ROUTE = [dict(destination="0.0.0.0/0",
                          nexthop="172.16.0.1")]

    @contextlib.contextmanager
    def _stubs(self, host_routes=None, new_routes=None, find_routes=True,
               new_dns_servers=None):
        if host_routes is None:
            host_routes = []
        if new_routes:
            new_routes = [models.Route(cidr=r["destination"],
                                       gateway=r["nexthop"],
                                       subnet_id=1)
                          for r in new_routes]
        if new_dns_servers:
            new_dns_servers = [models.DNSNameserver(
                ip=ip,
                subnet_id=1) for ip in new_dns_servers]

        subnet = dict(
            id=1,
            network_id=1,
            tenant_id=self.context.tenant_id, ip_version=4,
            cidr="172.16.0.0/24",
            host_routes=host_routes,
            dns_nameservers=["4.2.2.1", "4.2.2.2"],
            enable_dhcp=None)

        dns_ips = subnet.pop("dns_nameservers", [])
        host_routes = subnet.pop("host_routes", [])
        subnet_mod = models.Subnet()

        subnet_mod.update(subnet)

        subnet_mod["dns_nameservers"] = [models.DNSNameserver(ip=ip)
                                         for ip in dns_ips]
        subnet_mod["routes"] = [models.Route(cidr=r["destination"],
                                             gateway=r["nexthop"],
                                             subnet_id=subnet_mod["id"])
                                for r in host_routes]
        with contextlib.nested(
            mock.patch("quark.db.api.subnet_find"),
            mock.patch("quark.db.api.subnet_update"),
            mock.patch("quark.db.api.dns_create"),
            mock.patch("quark.db.api.route_find"),
            mock.patch("quark.db.api.route_update"),
            mock.patch("quark.db.api.route_create"),
        ) as (subnet_find, subnet_update,
              dns_create,
              route_find, route_update, route_create):
            subnet_find.return_value = subnet_mod
            route_find.return_value = subnet_mod["routes"][0] \
                if subnet_mod["routes"] and find_routes else None
            new_subnet_mod = models.Subnet(network=models.Network())
            new_subnet_mod.update(subnet_mod)
            if new_routes:
                new_subnet_mod["routes"] = new_routes
            if new_dns_servers:
                new_subnet_mod["dns_nameservers"] = new_dns_servers
            subnet_update.return_value = new_subnet_mod
            yield dns_create, route_update, route_create

    def test_update_subnet_not_found(self):
        with self.assertRaises(exceptions.SubnetNotFound):
            self.plugin.update_subnet(self.context, 1, {})

    def test_update_subnet_dns_nameservers(self):
        new_dns_servers = ["1.1.1.2"]
        with self._stubs(
            host_routes=self.DEFAULT_ROUTE,
            new_dns_servers=new_dns_servers
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(dns_nameservers=new_dns_servers))
            res = self.plugin.update_subnet(self.context,
                                            1,
                                            req)
            self.assertEqual(dns_create.call_count, 1)
            self.assertEqual(route_create.call_count, 0)
            self.assertEqual(res["dns_nameservers"], new_dns_servers)

    def test_update_subnet_routes(self):
        new_routes = [dict(destination="10.0.0.0/24",
                           nexthop="1.1.1.1")]
        with self._stubs(
            host_routes=self.DEFAULT_ROUTE,
            new_routes=new_routes
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(
                host_routes=new_routes))
            res = self.plugin.update_subnet(self.context, 1, req)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            self.assertEqual(len(res["host_routes"]), 1)
            self.assertEqual(res["host_routes"][0]["destination"],
                             "10.0.0.0/24")
            self.assertEqual(res["host_routes"][0]["nexthop"],
                             "1.1.1.1")
            self.assertIsNone(res["gateway_ip"])

    def test_update_subnet_gateway_ip_with_default_route_in_db(self):
        with self._stubs(
            host_routes=self.DEFAULT_ROUTE,
            new_routes=[dict(destination="0.0.0.0/0", nexthop="1.2.3.4")]
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(gateway_ip="1.2.3.4"))
            res = self.plugin.update_subnet(self.context, 1, req)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 0)
            self.assertEqual(route_update.call_count, 1)
            self.assertEqual(len(res["host_routes"]), 1)
            self.assertEqual(res["host_routes"][0]["destination"],
                             "0.0.0.0/0")
            self.assertEqual(res["host_routes"][0]["nexthop"],
                             "1.2.3.4")
            self.assertEqual(res["gateway_ip"], "1.2.3.4")

    def test_update_subnet_gateway_ip_with_non_default_route_in_db(self):
        with self._stubs(
            host_routes=[dict(destination="1.1.1.1/8", nexthop="9.9.9.9")],
            find_routes=False,
            new_routes=[dict(destination="1.1.1.1/8", nexthop="9.9.9.9"),
                        dict(destination="0.0.0.0/0", nexthop="1.2.3.4")]
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(gateway_ip="1.2.3.4"))
            res = self.plugin.update_subnet(self.context, 1, req)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)

            self.assertEqual(res["gateway_ip"], "1.2.3.4")

            self.assertEqual(len(res["host_routes"]), 2)
            res_tuples = [(r["destination"], r["nexthop"])
                          for r in res["host_routes"]]
            self.assertIn(("0.0.0.0/0", "1.2.3.4"), res_tuples)
            self.assertIn(("1.1.1.1/8", "9.9.9.9"), res_tuples)

    def test_update_subnet_gateway_ip_without_default_route_in_db(self):
        with self._stubs(
            host_routes=None,
            new_routes=[dict(destination="0.0.0.0/0", nexthop="1.2.3.4")]
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(gateway_ip="1.2.3.4"))
            res = self.plugin.update_subnet(self.context, 1, req)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            self.assertEqual(len(res["host_routes"]), 1)
            self.assertEqual(res["host_routes"][0]["destination"],
                             "0.0.0.0/0")
            self.assertEqual(res["host_routes"][0]["nexthop"],
                             "1.2.3.4")
            self.assertEqual(res["gateway_ip"], "1.2.3.4")

    def test_update_subnet_gateway_ip_with_default_route_in_args(self):
        new_routes = [dict(destination="0.0.0.0/0",
                           nexthop="4.3.2.1")]
        with self._stubs(
            host_routes=self.DEFAULT_ROUTE,
            new_routes=new_routes
        ) as (dns_create, route_update, route_create):
            req = dict(subnet=dict(
                host_routes=new_routes,
                gateway_ip="1.2.3.4"))
            res = self.plugin.update_subnet(self.context, 1, req)
            self.assertEqual(dns_create.call_count, 0)
            self.assertEqual(route_create.call_count, 1)
            self.assertEqual(len(res["host_routes"]), 1)
            self.assertEqual(res["host_routes"][0]["destination"],
                             "0.0.0.0/0")
            self.assertEqual(res["host_routes"][0]["nexthop"],
                             "4.3.2.1")
            self.assertEqual(res["gateway_ip"], "4.3.2.1")


class TestQuarkDeleteSubnet(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnet, ips):
        ip_mods = []
        subnet_mod = None
        if subnet:
            subnet_mod = models.Subnet()
            subnet_mod.update(subnet)
        for ip in ips:
            ip_mod = models.IPAddress()
            ip_mod.update(ip)
            ip_mods.append(ip_mod)

        db_mod = "quark.db.api"
        with contextlib.nested(
            mock.patch("%s.subnet_find" % db_mod),
            mock.patch("%s.subnet_delete" % db_mod)
        ) as (sub_find, sub_delete):
            if subnet_mod:
                subnet_mod.allocated_ips = ip_mods
            sub_find.return_value = subnet_mod
            yield sub_delete

    def test_delete_subnet(self):
        subnet = dict(id=1)
        with self._stubs(subnet=subnet, ips=[]) as sub_delete:
            self.plugin.delete_subnet(self.context, 1)
            self.assertTrue(sub_delete.called)

    def test_delete_subnet_no_subnet_fails(self):
        with self._stubs(subnet=None, ips=[]):
            with self.assertRaises(exceptions.SubnetNotFound):
                self.plugin.delete_subnet(self.context, 1)

    def test_delete_subnet_has_allocated_ips_fails(self):
        subnet = dict(id=1)
        with self._stubs(subnet=subnet, ips=[{}]):
            with self.assertRaises(exceptions.SubnetInUse):
                self.plugin.delete_subnet(self.context, 1)


class TestSubnetsNotification(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, s, deleted_at=None):
        class FakeContext(object):
            def __enter__(*args, **kwargs):
                pass

            def __exit__(*args, **kwargs):
                pass

        self.context.session.begin = FakeContext

        s["network"] = models.Network()
        s["network"]["created_at"] = s["created_at"]
        subnet = models.Subnet(**s)
        db_mod = "quark.db.api"
        api_mod = "neutron.openstack.common.notifier.api"
        time_mod = "neutron.openstack.common.timeutils"
        with contextlib.nested(
            mock.patch("%s.subnet_find" % db_mod),
            mock.patch("%s.network_find" % db_mod),
            mock.patch("%s.subnet_create" % db_mod),
            mock.patch("%s.ip_policy_create" % db_mod),
            mock.patch("%s.subnet_delete" % db_mod),
            mock.patch("%s.notify" % api_mod),
            mock.patch("%s.utcnow" % time_mod)
        ) as (sub_find, net_find, sub_create, pol_cre, sub_del, notify,
              time_func):
            sub_create.return_value = subnet
            sub_find.return_value = subnet
            time_func.return_value = deleted_at
            yield notify

    def test_create_subnet_notification(self):
        s = dict(network_id=1, cidr="192.168.10.0/24",
                 tenant_id=1, id=1, created_at="123")
        with self._stubs(s) as notify:
            self.plugin.create_subnet(self.context, dict(subnet=s))
            notify.assert_called_once_with(
                self.context,
                notifier_api.publisher_id("network"),
                "ip_block.create",
                notifier_api.CONF.default_notification_level,
                dict(tenant_id=s["tenant_id"],
                     ip_block_id=s["id"],
                     created_at=s["created_at"]))

    def test_delete_subnet_notification(self):
        now = time.strftime('%Y-%m-%d %H:%M:%S')
        later = time.strftime('%Y-%m-%d %H:%M:%S')
        s = dict(tenant_id=1, id=1, created_at=now)
        with self._stubs(s, deleted_at=later) as notify:
            self.plugin.delete_subnet(self.context, 1)
            notify.assert_called_once_with(
                self.context,
                notifier_api.publisher_id("network"),
                "ip_block.delete",
                notifier_api.CONF.default_notification_level,
                dict(tenant_id=s["tenant_id"],
                     created_at=s["created_at"],
                     ip_block_id=s["id"],
                     deleted_at=later))


class TestQuarkDiagnoseSubnets(test_quark_plugin.TestQuarkPlugin):
    @contextlib.contextmanager
    def _stubs(self, subnets=None, routes=None):
        if routes is None:
            routes = []
        route_models = []
        for route in routes:
            r = models.Route()
            r.update(route)
            route_models.append(r)

        if isinstance(subnets, list):
            subnet_models = []
            for subnet in subnets:
                s_dict = subnet.copy()
                s_dict["routes"] = route_models
                s = models.Subnet(network=models.Network())
                s.update(s_dict)
                subnet_models.append(s)
        elif subnets:
            mod = models.Subnet(network=models.Network())
            mod.update(subnets)
            mod["routes"] = route_models
            subnet_models = mod
        else:
            subnet_models = None

        with mock.patch("quark.db.api.subnet_find") as subnet_find:
            subnet_find.return_value = subnet_models
            yield

    def test_diagnose_subnet_with_wildcard_id_no_existing_subnets(self):
        with self._stubs(subnets=[], routes=[]):
            expected = {'subnets': []}
            actual = self.plugin.diagnose_subnet(self.context, "*", None)
            self.assertEqual(expected, actual)

    def test_diagnose_subnet_with_wildcard_with_existing_subnets(self):
        subnet_id = str(uuid.uuid4())
        route = dict(id=1, cidr="0.0.0.0/0", gateway="192.168.0.1")

        subnet = dict(id=subnet_id, network_id=1, name=subnet_id,
                      tenant_id=self.context.tenant_id, ip_version=4,
                      cidr="192.168.0.0/24", gateway_ip="192.168.0.1",
                      dns_nameservers=[],
                      enable_dhcp=None)

        with self._stubs(subnets=[subnet], routes=[route]):
            actual = self.plugin.diagnose_subnet(self.context, "*", None)
            self.maxDiff = None
            self.assertEqual(subnet["id"], actual["subnets"][0]["id"])

    def test_diagnose_subnet_with_regular_id(self):
        subnet_id = "12345"
        route = dict(id=1, cidr="0.0.0.0/0", gateway="192.168.0.1")

        subnet = dict(id=subnet_id, network_id=1, name=subnet_id,
                      tenant_id=self.context.tenant_id, ip_version=4,
                      cidr="192.168.0.0/24", gateway_ip="192.168.0.1",
                      dns_nameservers=[],
                      enable_dhcp=None)

        with self._stubs(subnets=subnet, routes=[route]):
            actual = self.plugin.diagnose_subnet(self.context, subnet_id, None)
            self.assertEqual(subnet["id"], actual["subnets"]["id"])
