Commit 1267db9e authored by Christos Stavrakakis's avatar Christos Stavrakakis
Browse files

cyclades: Add tests for IPv6 only networks

Add tests to check creation and usage of IPv6 only networks. Tests also
cover creation of other type of networks.
parent a0ad2412
# vim: set fileencoding=utf-8 :
# Copyright 2013 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
#
# The views and conclusions contained in the software and documentation are
# those of the authors and should not be interpreted as representing official
# policies, either expressed or implied, of GRNET S.A.
# Provides automated tests for logic module
from django.test import TestCase
from snf_django.lib.api import faults
from snf_django.utils.testing import mocked_quotaholder
from synnefo.logic import networks
from synnefo.db import models_factory as mfactory
from synnefo.db.models import BridgePoolTable, MacPrefixPoolTable
from synnefo import settings
from copy import copy
#@patch("synnefo.logic.rapi_pool.GanetiRapiClient")
class NetworkTest(TestCase):
def test_create(self):
kwargs = {
"name": "test",
"user_id": "user",
"subnet": "192.168.20.0/24",
"flavor": "CUSTOM",
}
# wrong gateway
kw = copy(kwargs)
kw["gateway"] = "192.168.3.1"
self.assertRaises(faults.BadRequest, networks.create, **kw)
# wrong subnet
kw = copy(kwargs)
kw["subnet"] = "192.168.2.0"
self.assertRaises(faults.OverLimit, networks.create, **kw)
kw["subnet"] = "192.168.0.0/16"
self.assertRaises(faults.OverLimit, networks.create, **kw)
kw["subnet"] = "192.168.0.3/24"
self.assertRaises(faults.BadRequest, networks.create, **kw)
# wrong flavor
kw = copy(kwargs)
kw["flavor"] = "UNKNOWN"
self.assertRaises(faults.BadRequest, networks.create, **kw)
# Test create objet
kwargs["gateway"] = "192.168.20.1"
kwargs["public"] = True
kwargs["dhcp"] = False
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.subnet, "192.168.20.0/24")
self.assertEqual(net.gateway, "192.168.20.1")
self.assertEqual(net.public, True)
self.assertEqual(net.flavor, "CUSTOM")
self.assertEqual(net.dhcp, False)
self.assertEqual(net.action, "CREATE")
self.assertEqual(net.state, "ACTIVE")
self.assertEqual(net.name, "test")
self.assertEqual(net.userid, "user")
# Test for each for flavor type
# MAC_FILTERED
kwargs["flavor"] = "MAC_FILTERED"
# Test exception if no rules exists
self.assertRaises(faults.ServiceUnavailable, networks.create, **kwargs)
mfactory.MacPrefixPoolTableFactory(base="aa:bb:0")
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.mode, "bridged")
self.assertEqual(net.mac_prefix, "aa:bb:1")
self.assertEqual(net.link, settings.DEFAULT_MAC_FILTERED_BRIDGE)
self.assertEqual(net.backend_tag, ["private-filtered"])
pool = MacPrefixPoolTable.get_pool()
self.assertFalse(pool.is_available("aa:bb:1"))
# PHYSICAL_VLAN
kwargs["flavor"] = "PHYSICAL_VLAN"
# Test exception if no rules exists
self.assertRaises(faults.ServiceUnavailable, networks.create, **kwargs)
mfactory.BridgePoolTableFactory(base="prv")
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.mode, "bridged")
self.assertEqual(net.mac_prefix, settings.DEFAULT_MAC_PREFIX)
self.assertEqual(net.link, "prv1")
self.assertEqual(net.backend_tag, ["physical-vlan"])
pool = BridgePoolTable.get_pool()
self.assertFalse(pool.is_available(net.link))
# IP_LESS_ROUTED
kwargs["flavor"] = "IP_LESS_ROUTED"
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.mode, "routed")
self.assertEqual(net.mac_prefix, settings.DEFAULT_MAC_PREFIX)
self.assertEqual(net.link, settings.DEFAULT_ROUTING_TABLE)
self.assertEqual(net.backend_tag, ["ip-less-routed"])
# CUSTOM
kwargs["flavor"] = "CUSTOM"
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.mode, "bridged")
self.assertEqual(net.mac_prefix, settings.DEFAULT_MAC_PREFIX)
self.assertEqual(net.link, settings.DEFAULT_BRIDGE)
self.assertEqual(net.backend_tag, [])
def test_create_network_ipv6(self):
kwargs = {
"name": "test",
"user_id": "user",
"flavor": "CUSTOM",
"subnet6": "2001:648:2ffc:1112::/64",
}
# Wrong subnet
kw = copy(kwargs)
kw["subnet6"] = "2001:64q:2ffc:1112::/64"
self.assertRaises(faults.BadRequest, networks.create, **kw)
# Wrong gateway
kw = copy(kwargs)
kw["gateway6"] = "2001:64q:2ffc:1119::1"
self.assertRaises(faults.BadRequest, networks.create, **kw)
# floating_ip_pools can not be ipv6 only
kw = copy(kwargs)
kw["floating_ip_pool"] = True
self.assertRaises(faults.BadRequest, networks.create, **kw)
kwargs["gateway6"] = "2001:648:2ffc:1112::1"
with mocked_quotaholder():
net = networks.create(**kwargs)
self.assertEqual(net.subnet6, "2001:648:2ffc:1112::/64")
self.assertEqual(net.gateway6, "2001:648:2ffc:1112::1")
self.assertRaises(Exception, net.get_pool)
# vim: set fileencoding=utf-8 :
# Copyright 2013 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
#
# The views and conclusions contained in the software and documentation are
# those of the authors and should not be interpreted as representing official
# policies, either expressed or implied, of GRNET S.A.
# Provides automated tests for logic module
from django.test import TestCase
#from snf_django.utils.testing import mocked_quotaholder
from synnefo.logic import servers
from synnefo.db import models_factory as mfactory
from mock import patch
@patch("synnefo.logic.rapi_pool.GanetiRapiClient")
class ServerTest(TestCase):
def test_connect_network(self, mrapi):
# Common connect
net = mfactory.NetworkFactory(subnet="192.168.2.0/24",
gateway="192.168.2.1",
state="ACTIVE",
dhcp=True,
flavor="CUSTOM")
vm = mfactory.VirtualMachineFactory()
mfactory.BackendNetworkFactory(network=net, backend=vm.backend)
mrapi().ModifyInstance.return_value = 42
servers.connect(vm, net)
pool = net.get_pool(with_lock=False)
self.assertFalse(pool.is_available("192.168.2.2"))
args, kwargs = mrapi().ModifyInstance.call_args
nics = kwargs["nics"][0]
self.assertEqual(args[0], vm.backend_vm_id)
self.assertEqual(nics[0], "add")
self.assertEqual(nics[1]["ip"], "192.168.2.2")
self.assertEqual(nics[1]["network"], net.backend_id)
# No dhcp
vm = mfactory.VirtualMachineFactory()
net = mfactory.NetworkFactory(subnet="192.168.2.0/24",
gateway="192.168.2.1",
state="ACTIVE",
dhcp=False)
mfactory.BackendNetworkFactory(network=net, backend=vm.backend)
servers.connect(vm, net)
pool = net.get_pool(with_lock=False)
self.assertTrue(pool.is_available("192.168.2.2"))
args, kwargs = mrapi().ModifyInstance.call_args
nics = kwargs["nics"][0]
self.assertEqual(args[0], vm.backend_vm_id)
self.assertEqual(nics[0], "add")
self.assertEqual(nics[1]["ip"], None)
self.assertEqual(nics[1]["network"], net.backend_id)
# Test connect to IPv6 only network
vm = mfactory.VirtualMachineFactory()
net = mfactory.NetworkFactory(subnet6="2000::/64",
state="ACTIVE",
gateway="2000::1")
mfactory.BackendNetworkFactory(network=net, backend=vm.backend)
servers.connect(vm, net)
args, kwargs = mrapi().ModifyInstance.call_args
nics = kwargs["nics"][0]
self.assertEqual(args[0], vm.backend_vm_id)
self.assertEqual(nics[0], "add")
self.assertEqual(nics[1]["ip"], None)
self.assertEqual(nics[1]["network"], net.backend_id)
......@@ -139,8 +139,8 @@ class ServerCommandTest(TestCase):
try:
servers.start(vm)
except:
m.resolve_commissions.assert_called_once_with('', [],
[vm.serial.serial])
m.resolve_commissions\
.assert_called_once_with('', [], [vm.serial.serial])
def test_task_after(self, mrapi):
return
......@@ -148,29 +148,26 @@ class ServerCommandTest(TestCase):
mrapi().StartupInstance.return_value = 1
mrapi().ShutdownInstance.return_value = 2
mrapi().RebootInstance.return_value = 2
with mocked_quotaholder() as m:
with mocked_quotaholder():
vm.task = None
vm.operstate = "STOPPED"
servers.start(vm)
self.assertEqual(vm.task, "START")
self.assertEqual(vm.task_job_id, 1)
with mocked_quotaholder() as m:
with mocked_quotaholder():
vm.task = None
vm.operstate = "STARTED"
servers.stop(vm)
self.assertEqual(vm.task, "STOP")
self.assertEqual(vm.task_job_id, 2)
with mocked_quotaholder() as m:
with mocked_quotaholder():
vm.task = None
servers.reboot(vm)
self.assertEqual(vm.task, "REBOOT")
self.assertEqual(vm.task_job_id, 3)
## Test Callbacks
@patch('synnefo.lib.amqp.AMQPClient')
class UpdateDBTest(TestCase):
def create_msg(self, **kwargs):
......@@ -272,9 +269,9 @@ class UpdateDBTest(TestCase):
fp1 = mfactory.FloatingIPFactory(machine=vm2, network=network)
fp2 = mfactory.FloatingIPFactory(machine=vm2, network=network)
mfactory.NetworkInterfaceFactory(machine=vm2, network=network,
ipv4=fp1.ipv4)
ipv4=fp1.ipv4)
mfactory.NetworkInterfaceFactory(machine=vm2, network=network,
ipv4=fp2.ipv4)
ipv4=fp2.ipv4)
pool = network.get_pool()
pool.reserve(fp1.ipv4)
pool.reserve(fp2.ipv4)
......@@ -461,7 +458,7 @@ class UpdateNetTest(TestCase):
def test_empty_nic(self, client):
vm = mfactory.VirtualMachineFactory(operstate='ERROR')
for public in [True, False]:
net = mfactory.NetworkFactory(public=public)
net = mfactory.NetworkFactory(public=public, subnet6=None)
msg = self.create_msg(nics=[{'network': net.backend_id}],
instance=vm.backend_vm_id)
update_db(client, msg)
......@@ -481,7 +478,7 @@ class UpdateNetTest(TestCase):
def test_full_nic(self, client):
vm = mfactory.VirtualMachineFactory(operstate='ERROR')
net = mfactory.NetworkFactory(subnet='10.0.0.0/24')
net = mfactory.NetworkFactory(subnet='10.0.0.0/24', subnet6=None)
pool = net.get_pool()
self.assertTrue(pool.is_available('10.0.0.22'))
pool.save()
......@@ -573,8 +570,8 @@ class UpdateNetworkTest(TestCase):
"""Test network creation when a backend is offline"""
net = mfactory.NetworkFactory(state='ACTIVE')
bn1 = mfactory.BackendNetworkFactory(network=net)
bn2 = mfactory.BackendNetworkFactory(network=net,
backend__offline=True)
mfactory.BackendNetworkFactory(network=net,
backend__offline=True)
msg = self.create_msg(operation='OP_NETWORK_CONNECT',
network=net.backend_id,
cluster=bn1.backend.clustername)
......@@ -889,8 +886,8 @@ class ReconciliationTest(TestCase):
"nic.networks": [],
"tags": []}]
self.reconciler.reconcile()
cmrapi.DeleteInstance.assert_called_once_with(
"%s22" % settings.BACKEND_PREFIX_ID)
cmrapi.DeleteInstance\
.assert_called_once_with("%s22" % settings.BACKEND_PREFIX_ID)
def test_unsynced_operstate(self, mrapi):
vm1 = mfactory.VirtualMachineFactory(backend=self.backend,
......@@ -972,3 +969,5 @@ class ReconciliationTest(TestCase):
from synnefo.logic.test.rapi_pool_tests import *
from synnefo.logic.test.utils_tests import *
from synnefo.logic.test.networks import *
from synnefo.logic.test.servers import *
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment