[Kimchi-devel] [PATCH 5/5] Reorganize the network tests
Aline Manera
alinefm at linux.vnet.ibm.com
Wed Jan 7 19:04:49 UTC 2015
It creates 2 new files: test_network.py and test_mock_network.py with
the network test cases.
The only test case on test_mock_network.py is related to VLAN tagging
bridge as it is hard to test on Model environment because it needs an
active interface to create the VLAN tagging bridge.
MockModel was also changed to return the system interfaces as the
libvirt Test driver needs a valid one to create the VLAN tagging bridge
but it does not need to be active.
Signed-off-by: Aline Manera <alinefm at linux.vnet.ibm.com>
---
src/kimchi/mockmodel.py | 22 +------
tests/test_mock_network.py | 71 ++++++++++++++++++++++
tests/test_model.py | 65 --------------------
tests/test_network.py | 144 +++++++++++++++++++++++++++++++++++++++++++++
tests/test_rest.py | 82 +-------------------------
5 files changed, 219 insertions(+), 165 deletions(-)
create mode 100644 tests/test_mock_network.py
create mode 100644 tests/test_network.py
diff --git a/src/kimchi/mockmodel.py b/src/kimchi/mockmodel.py
index 34c7b22..f8e317a 100644
--- a/src/kimchi/mockmodel.py
+++ b/src/kimchi/mockmodel.py
@@ -1,7 +1,7 @@
#
# Project Kimchi
#
-# Copyright IBM, Corp. 2013-2014
+# Copyright IBM, Corp. 2013-2015
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -64,7 +64,6 @@ class MockModel(Model):
osinfo.defaults = dict(defaults)
self._mock_devices = MockDevices()
- self._mock_interfaces = MockInterfaces()
self._mock_storagevolumes = MockStorageVolumes()
self._mock_swupdate = MockSoftwareUpdate()
self._mock_repositories = MockRepositories()
@@ -284,12 +283,6 @@ class MockModel(Model):
return self._model_storagevolume_lookup(pool, vol)
- def _mock_interfaces_get_list(self):
- return self._mock_interfaces.ifaces.keys()
-
- def _mock_interface_lookup(self, name):
- return self._mock_interfaces.ifaces[name]
-
def _mock_devices_get_list(self, _cap=None, _passthrough=None,
_passthrough_affected_by=None):
if _cap is None:
@@ -429,19 +422,6 @@ class MockStorageVolumes(object):
'ref_cnt': 0}}
-class MockInterfaces(object):
- def __init__(self):
- self.ifaces = {}
- params = {"eth1": "nic", "bond0": "bonding",
- "eth1.10": "vlan", "bridge0": "bridge"}
- for i, name in enumerate(params.iterkeys()):
- info = {'name': name, 'type': params[name],
- 'ipaddr': '192.168.%s.101' % (i + 1),
- 'netmask': '255.255.255.0', 'status': 'active'}
- self.ifaces[name] = info
- self.ifaces['eth1']['ipaddr'] = '192.168.0.101'
-
-
class MockDevices(object):
def __init__(self):
self.devices = {
diff --git a/tests/test_mock_network.py b/tests/test_mock_network.py
new file mode 100644
index 0000000..e018d11
--- /dev/null
+++ b/tests/test_mock_network.py
@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+#
+# Project Kimchi
+#
+# Copyright IBM, Corp. 2015
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+import json
+import os
+import unittest
+
+from functools import partial
+
+from kimchi.mockmodel import MockModel
+from test_network import _do_network_test
+from utils import get_free_port, patch_auth, request, run_server
+
+
+model = None
+test_server = None
+host = None
+port = None
+ssl_port = None
+cherrypy_port = None
+
+
+def setUpModule():
+ global test_server, model, host, port, ssl_port, cherrypy_port
+
+ patch_auth()
+ model = MockModel('/tmp/obj-store-test')
+ host = '127.0.0.1'
+ port = get_free_port('http')
+ ssl_port = get_free_port('https')
+ cherrypy_port = get_free_port('cherrypy_port')
+ test_server = run_server(host, port, ssl_port, test_mode=True,
+ cherrypy_port=cherrypy_port, model=model)
+
+
+def tearDownModule():
+ test_server.stop()
+ os.unlink('/tmp/obj-store-test')
+
+
+class MockNetworkTests(unittest.TestCase):
+ def setUp(self):
+ self.request = partial(request, host, ssl_port)
+ model.reset()
+
+ def test_vlan_tag_bridge(self):
+ # Verify the current system has at least one interface to create a
+ # bridged network
+ interfaces = json.loads(self.request('/interfaces?type=nic').read())
+ if len(interfaces) > 0:
+ iface = interfaces[0]['name']
+ _do_network_test(self, model, {'name': u'bridge-network',
+ 'connection': 'bridge',
+ 'interface': iface, 'vlan_id': 987})
diff --git a/tests/test_model.py b/tests/test_model.py
index 2d818ae..8181f7c 100644
--- a/tests/test_model.py
+++ b/tests/test_model.py
@@ -1064,69 +1064,6 @@ class ModelTests(unittest.TestCase):
self.assertEquals([], inst.vm_lookup(u'пeω-∨м')['users'])
self.assertEquals([], inst.vm_lookup(u'пeω-∨м')['groups'])
- @unittest.skipUnless(utils.running_as_root(), 'Must be run as root')
- def test_network(self):
- inst = model.Model(None, self.tmp_store)
-
- with RollbackContext() as rollback:
-
- # Regression test:
- # Kimchi fails creating new network #318
- name = 'test-network-no-subnet'
-
- networks = inst.networks_get_list()
- num = len(networks) + 1
- args = {'name': name,
- 'connection': 'nat',
- 'subnet': ''}
- inst.networks_create(args)
- rollback.prependDefer(inst.network_delete, name)
-
- networks = inst.networks_get_list()
- self.assertEquals(num, len(networks))
- networkinfo = inst.network_lookup(name)
- self.assertNotEqual(args['subnet'], networkinfo['subnet'])
- self.assertEqual(args['connection'], networkinfo['connection'])
- self.assertEquals('inactive', networkinfo['state'])
- self.assertEquals([], networkinfo['vms'])
- self.assertTrue(networkinfo['autostart'])
- self.assertTrue(networkinfo['persistent'])
-
- inst.network_activate(name)
- rollback.prependDefer(inst.network_deactivate, name)
-
- networkinfo = inst.network_lookup(name)
- self.assertEquals('active', networkinfo['state'])
-
- # test network creation with subnet passed
- name = 'test-network-subnet'
-
- networks = inst.networks_get_list()
- num = len(networks) + 1
- args = {'name': name,
- 'connection': 'nat',
- 'subnet': '127.0.100.0/24'}
- inst.networks_create(args)
- rollback.prependDefer(inst.network_delete, name)
-
- networks = inst.networks_get_list()
- self.assertEquals(num, len(networks))
- networkinfo = inst.network_lookup(name)
- self.assertEqual(args['subnet'], networkinfo['subnet'])
- self.assertEqual(args['connection'], networkinfo['connection'])
- self.assertEquals('inactive', networkinfo['state'])
- self.assertEquals([], networkinfo['vms'])
- self.assertTrue(networkinfo['autostart'])
-
- inst.network_activate(name)
- rollback.prependDefer(inst.network_deactivate, name)
-
- networkinfo = inst.network_lookup(name)
- self.assertEquals('active', networkinfo['state'])
-
- networks = inst.networks_get_list()
- self.assertEquals((num - 2), len(networks))
-
def test_multithreaded_connection(self):
def worker():
for i in xrange(100):
@@ -1296,8 +1233,6 @@ class ModelTests(unittest.TestCase):
inst.task_wait(taskid, timeout=10)
self.assertEquals('finished', inst.task_lookup(taskid)['status'])
-
-
@unittest.skipUnless(utils.running_as_root(), 'Must be run as root')
def test_delete_running_vm(self):
inst = model.Model(objstore_loc=self.tmp_store)
diff --git a/tests/test_network.py b/tests/test_network.py
new file mode 100644
index 0000000..5dbe54d
--- /dev/null
+++ b/tests/test_network.py
@@ -0,0 +1,144 @@
+# -*- coding: utf-8 -*-
+#
+# Project Kimchi
+#
+# Copyright IBM, Corp. 2015
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+import json
+import os
+import unittest
+
+from functools import partial
+
+from kimchi.model.model import Model
+from kimchi.rollbackcontext import RollbackContext
+from utils import get_free_port, patch_auth, request, rollback_wrapper
+from utils import run_server
+
+
+model = None
+test_server = None
+host = None
+port = None
+ssl_port = None
+cherrypy_port = None
+
+
+def setUpModule():
+ global test_server, model, host, port, ssl_port, cherrypy_port
+
+ patch_auth()
+ model = Model(None, '/tmp/obj-store-test')
+ host = '127.0.0.1'
+ port = get_free_port('http')
+ ssl_port = get_free_port('https')
+ cherrypy_port = get_free_port('cherrypy_port')
+ test_server = run_server(host, port, ssl_port, test_mode=True,
+ cherrypy_port=cherrypy_port, model=model)
+
+
+def tearDownModule():
+ test_server.stop()
+ os.unlink('/tmp/obj-store-test')
+
+
+def _do_network_test(self, model, params):
+ with RollbackContext() as rollback:
+ net_name = params['name']
+ uri = '/networks/%s' % net_name.encode('utf-8')
+
+ # Create a network
+ req = json.dumps(params)
+ resp = self.request('/networks', req, 'POST')
+ rollback.prependDefer(rollback_wrapper, model.network_delete,
+ net_name)
+ self.assertEquals(201, resp.status)
+
+ # Verify the network
+ resp = self.request(uri)
+ network = json.loads(resp.read())
+ self.assertEquals('inactive', network['state'])
+ self.assertTrue(network['persistent'])
+
+ # activate the network
+ resp = self.request(uri + '/activate', '{}', 'POST')
+ rollback.prependDefer(rollback_wrapper,
+ model.network_deactivate, net_name)
+ self.assertEquals(200, resp.status)
+ resp = self.request(uri)
+ network = json.loads(resp.read())
+ self.assertEquals('active', network['state'])
+
+ # Deactivate the network
+ resp = self.request(uri + '/deactivate', '{}', 'POST')
+ self.assertEquals(200, resp.status)
+ resp = self.request(uri)
+ network = json.loads(resp.read())
+ self.assertEquals('inactive', network['state'])
+
+ # Delete the network
+ resp = self.request(uri, '{}', 'DELETE')
+ self.assertEquals(204, resp.status)
+
+
+class NetworkTests(unittest.TestCase):
+ def setUp(self):
+ self.request = partial(request, host, ssl_port)
+
+ def test_get_networks(self):
+ networks = json.loads(self.request('/networks').read())
+ self.assertIn('default', [net['name'] for net in networks])
+
+ with RollbackContext() as rollback:
+ # Now add a couple of Networks to the mock model
+ for i in xrange(5):
+ name = 'network-%i' % i
+ req = json.dumps({'name': name,
+ 'connection': 'nat',
+ 'subnet': '127.0.10%i.0/24' % i})
+
+ resp = self.request('/networks', req, 'POST')
+ rollback.prependDefer(model.network_delete, name)
+ self.assertEquals(201, resp.status)
+ network = json.loads(resp.read())
+ self.assertEquals([], network["vms"])
+
+ nets = json.loads(self.request('/networks').read())
+ self.assertEquals(len(networks) + 5, len(nets))
+
+ network = json.loads(self.request('/networks/network-1').read())
+ keys = [u'name', u'connection', u'interface', u'subnet', u'dhcp',
+ u'vms', u'in_use', u'autostart', u'state', u'persistent']
+ self.assertEquals(sorted(keys), sorted(network.keys()))
+
+ def test_network_lifecycle(self):
+ # Verify all the supported network type
+ networks = [{'name': u'kīмсhī-пet', 'connection': 'isolated'},
+ {'name': u'nat-network', 'connection': 'nat'},
+ {'name': u'subnet-network', 'connection': 'nat',
+ 'subnet': '127.0.100.0/24'}]
+
+ # Verify the current system has at least one interface to create a
+ # bridged network
+ interfaces = json.loads(self.request('/interfaces?type=nic').read())
+ if len(interfaces) > 0:
+ iface = interfaces[0]['name']
+ networks.append({'name': u'bridge-network', 'connection': 'bridge',
+ 'interface': iface})
+
+ for net in networks:
+ _do_network_test(self, model, net)
diff --git a/tests/test_rest.py b/tests/test_rest.py
index fa15ef6..0c36efb 100644
--- a/tests/test_rest.py
+++ b/tests/test_rest.py
@@ -2,7 +2,7 @@
#
# Project Kimchi
#
-# Copyright IBM, Corp. 2013-2014
+# Copyright IBM, Corp. 2013-2015
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -1590,85 +1590,9 @@ class RestTests(unittest.TestCase):
resp = self.request('/interfaces').read()
self.assertIn('name', resp)
interfaces = json.loads(resp)
+ keys = ['name', 'type', 'ipaddr', 'netmask', 'status']
for interface in interfaces:
- self.assertIn('name', interface)
- self.assertIn('type', interface)
- self.assertIn('ipaddr', interface)
- self.assertIn('netmask', interface)
- self.assertIn('status', interface)
-
- ident = "eth1"
- resp = self.request('/interfaces/%s' % ident).read()
- interface = json.loads(resp)
- self.assertEquals(interface['name'], ident)
- self.assertEquals(interface['type'], "nic")
- self.assertEquals(interface['ipaddr'], "192.168.0.101")
- self.assertEquals(interface['netmask'], "255.255.255.0")
- self.assertEquals(interface['status'], "active")
-
- def test_get_networks(self):
- networks = json.loads(request(host, ssl_port, '/networks').read())
- self.assertEquals(1, len(networks))
- self.assertEquals('default', networks[0]['name'])
- self.assertEquals([], networks[0]['vms'])
-
- # Now add a couple of Networks to the mock model
- for i in xrange(5):
- name = 'network-%i' % i
- req = json.dumps({'name': name,
- 'connection': 'nat',
- 'subnet': '127.0.10%i.0/24' % i})
-
- resp = request(host, ssl_port, '/networks', req, 'POST')
- self.assertEquals(201, resp.status)
- network = json.loads(resp.read())
- self.assertEquals([], network["vms"])
-
- networks = json.loads(request(host, ssl_port, '/networks').read())
- self.assertEquals(6, len(networks))
-
- network = json.loads(request(host, ssl_port,
- '/networks/network-1').read())
- self.assertEquals('network-1', network['name'])
- self.assertEquals('inactive', network['state'])
- # Delete the network
- for i in xrange(5):
- resp = request(host, ssl_port, '/networks/network-%i' % i,
- '{}', 'DELETE')
- self.assertEquals(204, resp.status)
-
- def test_network_action(self):
- # Create a network
- req = json.dumps({'name': 'test-network',
- 'connection': 'nat',
- 'net': '127.0.1.0/24'})
- resp = request(host, ssl_port, '/networks', req, 'POST')
- self.assertEquals(201, resp.status)
-
- # Verify the network
- network = json.loads(request(host, ssl_port,
- '/networks/test-network').read())
- self.assertEquals('inactive', network['state'])
- self.assertTrue(network['persistent'])
-
- # activate the network
- resp = request(host, ssl_port,
- '/networks/test-network/activate', '{}', 'POST')
- network = json.loads(request(host, ssl_port,
- '/networks/test-network').read())
- self.assertEquals('active', network['state'])
-
- # Deactivate the network
- resp = request(host, ssl_port,
- '/networks/test-network/deactivate', '{}', 'POST')
- network = json.loads(request(host, ssl_port,
- '/networks/test-network').read())
- self.assertEquals('inactive', network['state'])
-
- # Delete the network
- resp = request(host, ssl_port, '/networks/test-network', '{}',
- 'DELETE')
- self.assertEquals(204, resp.status)
+ self.assertEquals(sorted(keys), sorted(interface.keys()))
def _task_lookup(self, taskid):
return json.loads(self.request('/tasks/%s' % taskid).read())
--
2.1.0
More information about the Kimchi-devel
mailing list