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

29 Dec 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@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

    