[Kimchi-devel] [PATCH 5/5] Reorganize the network tests
Aline Manera
alinefm at linux.vnet.ibm.com
Mon Dec 29 14:53:14 UTC 2014
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 at linux.vnet.ibm.com>
---
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
More information about the Kimchi-devel
mailing list