[Kimchi-devel] [PATCH 5/5] Reorganize the network tests

It creates a new file (test_network.py) with the network test cases. As MockModel does not override any Model function related to network, only Model tests were added using the REST API. Signed-off-by: Aline Manera <alinefm(a)linux.vnet.ibm.com&gt; --- tests/test_model.py | 65 ---------------------- tests/test_network.py | 147 ++++++++++++++++++++++++++++++++++++++++++++++++++ tests/test_rest.py | 64 ---------------------- 3 files changed, 147 insertions(+), 129 deletions(-) create mode 100644 tests/test_network.py diff --git a/tests/test_model.py b/tests/test_model.py index a8d35dc..7e28aed 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..8949789 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,147 @@ +# -*- coding: utf-8 -*- +# +# Project Kimchi +# +# Copyright IBM, Corp. 2014 +# +# 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') + + +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): + def _do_test(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) + + # 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').read()) + if len(interfaces) > 0: + iface = interfaces[0]['name'] + networks.append({'name': u'bridge-network', 'connection': 'bridge', + 'interface': iface}) + networks.append({'name': u'vlan-network', 'connection': 'bridge', + 'interface': iface, 'vlan_id': 999}) + + for net in networks: + _do_test(net) diff --git a/tests/test_rest.py b/tests/test_rest.py index fa15ef6..8eac0f9 100644 --- a/tests/test_rest.py +++ b/tests/test_rest.py @@ -1606,70 +1606,6 @@ class RestTests(unittest.TestCase): 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) - def _task_lookup(self, taskid): return json.loads(self.request('/tasks/%s' % taskid).read()) -- 2.1.0