From: Salvatore Orlando Date: Fri, 1 Jul 2011 11:20:31 +0000 (+0100) Subject: Adding unit test X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=a93c1b1acbffc264ba21f3c0859af5f33afc9a95;p=openstack-build%2Fneutron-build.git Adding unit test Applying pep8 fixes Note: teardown does not yet work properly. Thinking about replacing it with something that simply deletes the temporary sql lite db. --- diff --git a/quantum/api/ports.py b/quantum/api/ports.py index be74cc9e7..2a38fbe39 100644 --- a/quantum/api/ports.py +++ b/quantum/api/ports.py @@ -133,7 +133,8 @@ class Controller(common.QuantumController): def get_resource(self, request, tenant_id, network_id, id): try: result = self.network_manager.get_port_details( - tenant_id, network_id, id).get('attachment-id', None) + tenant_id, network_id, id).get('attachment-id', + None) return dict(attachment=result) except exception.NetworkNotFound as e: return faults.Fault(faults.NetworkNotFound(e)) diff --git a/quantum/db/models.py b/quantum/db/models.py index af9189df0..2a0845386 100644 --- a/quantum/db/models.py +++ b/quantum/db/models.py @@ -96,10 +96,9 @@ class Port(BASE, QuantumBase): self.network_id = network_id self.state = "DOWN" - def __repr__(self): return "" % (self.uuid, self.network_id, - self.state,self.interface_id) + self.state, self.interface_id) class Network(BASE, QuantumBase): diff --git a/quantum/manager.py b/quantum/manager.py index 71a13fb9c..a42f15c68 100644 --- a/quantum/manager.py +++ b/quantum/manager.py @@ -35,6 +35,7 @@ from quantum_plugin_base import QuantumPluginBase CONFIG_FILE = "plugins.ini" LOG = logging.getLogger('quantum.manager') + def find_config(basepath): for root, dirs, files in os.walk(basepath): if CONFIG_FILE in files: diff --git a/quantum/plugins/SamplePlugin.py b/quantum/plugins/SamplePlugin.py index aba0c391d..623042a12 100644 --- a/quantum/plugins/SamplePlugin.py +++ b/quantum/plugins/SamplePlugin.py @@ -23,6 +23,7 @@ from quantum.db import api as db LOG = logging.getLogger('quantum.plugins.SamplePlugin') + class QuantumEchoPlugin(object): """ @@ -234,7 +235,7 @@ class FakePlugin(object): def __init__(self): db_options = {"sql_connection": "sqlite:///fake_plugin.sqllite"} - db.configure_db(db_options) + db.configure_db(db_options) FakePlugin._net_counter = 0 def _get_network(self, tenant_id, network_id): @@ -245,9 +246,12 @@ class FakePlugin(object): def _get_port(self, tenant_id, network_id, port_id): net = self._get_network(tenant_id, network_id) - port = db.port_get(port_id) + try: + port = db.port_get(port_id) + except: + raise exc.PortNotFound(net_id=network_id, port_id=port_id) # Port must exist and belong to the appropriate network. - if not port or port['network_id']!=net['uuid']: + if port['network_id'] != net['uuid']: raise exc.PortNotFound(net_id=network_id, port_id=port_id) return port @@ -274,10 +278,10 @@ class FakePlugin(object): LOG.debug("FakePlugin.get_all_networks() called") nets = [] for net in db.network_list(tenant_id): - net_item = {'net-id':str(net.uuid), - 'net-name':net.name} + net_item = {'net-id': str(net.uuid), + 'net-name': net.name} nets.append(net_item) - return nets + return nets def get_network_details(self, tenant_id, net_id): """ @@ -286,8 +290,8 @@ class FakePlugin(object): """ LOG.debug("FakePlugin.get_network_details() called") net = self._get_network(tenant_id, net_id) - return {'net-id':str(net.uuid), - 'net-name':net.name} + return {'net-id': str(net.uuid), + 'net-name': net.name} def create_network(self, tenant_id, net_name): """ @@ -309,7 +313,7 @@ class FakePlugin(object): # Verify that no attachments are plugged into the network if net: for port in db.port_list(net_id): - if port['interface-id']: + if port['interface_id']: raise exc.NetworkInUse(net_id=net_id) db.network_destroy(net_id) return net @@ -335,7 +339,7 @@ class FakePlugin(object): port_ids = [] ports = db.port_list(net_id) for x in ports: - d = {'port-id':str(x.uuid)} + d = {'port-id': str(x.uuid)} port_ids.append(d) return port_ids @@ -346,10 +350,9 @@ class FakePlugin(object): """ LOG.debug("FakePlugin.get_port_details() called") port = self._get_port(tenant_id, net_id, port_id) - return {'port-id':str(port.uuid), - 'attachment-id':port.interface_id, - 'port-state':port.state} - + return {'port-id': str(port.uuid), + 'attachment-id': port.interface_id, + 'port-state': port.state} def create_port(self, tenant_id, net_id, port_state=None): """ @@ -357,7 +360,7 @@ class FakePlugin(object): """ LOG.debug("FakePlugin.create_port() called") port = db.port_create(net_id) - port_item = {'port-id':str(port.uuid)} + port_item = {'port-id': str(port.uuid)} return port_item def update_port(self, tenant_id, net_id, port_id, new_state): @@ -366,8 +369,8 @@ class FakePlugin(object): """ LOG.debug("FakePlugin.update_port() called") self._validate_port_state(new_state) - db.port_set_state(port_id,new_state) - return + db.port_set_state(port_id, new_state) + return def delete_port(self, tenant_id, net_id, port_id): """ diff --git a/run_tests.py b/run_tests.py index 633a76040..d63cc34a4 100644 --- a/run_tests.py +++ b/run_tests.py @@ -63,7 +63,7 @@ To run a single functional test module:: """ import gettext -import logging +import logging import os import unittest import sys @@ -289,7 +289,7 @@ if __name__ == '__main__': hdlr.setFormatter(formatter) logger.addHandler(hdlr) logger.setLevel(logging.DEBUG) - + working_dir = os.path.abspath("tests") c = config.Config(stream=sys.stdout, env=os.environ, diff --git a/tests/functional/test_service.py b/tests/functional/test_service.py index a3f8985c6..e897843e0 100644 --- a/tests/functional/test_service.py +++ b/tests/functional/test_service.py @@ -17,8 +17,6 @@ # under the License. import gettext -import json -import sys import unittest gettext.install('quantum', unicode=1) @@ -161,7 +159,8 @@ class QuantumTest(unittest.TestCase): #self.assertEqual(res.status, 200, "Bad response: %s" % output) #resdict = json.loads(output) #attachment = resdict["attachment"] - #self.assertEqual(attachment, "fudd", "Attachment: %s" % attachment) + #self.assertEqual(attachment, "fudd", "Attachment: %s" + #% attachment) #def test_renameNetwork(self): #self.create_network(test_network1_data) diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index 8f349674e..6e4fe1960 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -15,14 +15,14 @@ # License for the specific language governing permissions and limitations # under the License. # @author: Brad Hall, Nicira Networks -# @author: Salvatore Orlando, Citrix Systems +# @author: Salvatore Orlando, Citrix Systems import logging import unittest import tests.unit.testlib as testlib -from quantum import api as server +from quantum import api as server from quantum.common.wsgi import Serializer LOG = logging.getLogger('quantum.tests.test_api') @@ -37,16 +37,15 @@ class APIPortsTest(unittest.TestCase): def tearDown(self): """Clear the test environment""" # Remove all the networks. - network_req = testlib.create_list_networks_request(self.tenant_id) + network_req = testlib.create_network_list_request(self.tenant_id) network_res = network_req.get_response(self.api) - network_data = Serializer().deserialize(network_res.body,"application/xml") + network_data = Serializer().deserialize(network_res.body, + "application/xml") for network in network_data["networks"].values(): network_delete_req = testlib. \ - create_network_delete_request(self.tenant_id,network['id']) - network_delete_req.get_response(self.api) - - - + create_network_delete_request(self.tenant_id, network['id']) + network_delete_req.get_response(self.api) + # Fault names copied here for reference # # _fault_names = { @@ -61,58 +60,51 @@ class APIPortsTest(unittest.TestCase): # 470: "serviceUnavailable", # 471: "pluginFault"} - def _test_delete_port(self, format): + LOG.debug("_test_delete_port - format:%s - START", format) content_type = "application/" + format port_state = "ACTIVE" + LOG.debug("Creating network and port") network_req = testlib.create_new_network_request(self.tenant_id, - self.network_name, + self.network_name, format) network_res = network_req.get_response(self.api) self.assertEqual(network_res.status_int, 200) network_data = Serializer().deserialize(network_res.body, content_type) - network_id = network_data["networks"]["network"]["id"] - port_req = testlib.create_new_port_request(self.tenant_id, + network_id = network_data['networks']['network']['id'] + port_req = testlib.create_new_port_request(self.tenant_id, network_id, port_state, format) port_res = port_req.get_response(self.api) self.assertEqual(port_res.status_int, 200) port_data = Serializer().deserialize(port_res.body, content_type) - port_id = port_data["ports"]["port"]["id"] + port_id = port_data['ports']['port']['id'] LOG.debug("Deleting port %(port_id)s for network %(network_id)s"\ " of tenant %(tenant_id)s", locals()) delete_port_req = testlib.create_port_delete_request(self.tenant_id, network_id, - port_id, + port_id, format) delete_port_res = delete_port_req.get_response(self.api) self.assertEqual(delete_port_res.status_int, 202) - - def test_deletePort_xml(self): + list_port_req = testlib.create_port_list_request(self.tenant_id, + network_id, + format) + list_port_res = list_port_req.get_response(self.api) + port_list_data = Serializer().deserialize(list_port_res.body, + content_type) + port_count = len(port_list_data['ports']) + self.assertEqual(port_count, 0) + LOG.debug("_test_delete_port - format:%s - END", format) + + def test_delete_port_xml(self): self._test_delete_port('xml') - def test_deletePort_json(self): + def test_delete_port_json(self): self._test_delete_port('json') - - #def test_deletePortNegative(self): - # tenant = "tenant1" - # network = "test1" - - # Check for network not found - #rv = self.port.delete("", tenant, network, 2) - #self.assertEqual(rv.wrapped_exc.status_int, 420) - - # Create a network to put the port on - #req = testlib.create_network_request(tenant, network) - #network_obj = self.network.create(req, tenant) - #network_id = network_obj["networks"]["network"]["id"] - - # Test for portnotfound - #rv = self.port.delete("", tenant, network_id, 2) - #self.assertEqual(rv.wrapped_exc.status_int, 430) - + def _test_delete_port_in_use(self): # Test for portinuse #rv = self.port.create(req, tenant, network_id) #port_id = rv["ports"]["port"]["id"] @@ -122,4 +114,39 @@ class APIPortsTest(unittest.TestCase): #self.assertEqual(rv.status_int, 202) #rv = self.port.delete("", tenant, network_id, port_id) #self.assertEqual(rv.wrapped_exc.status_int, 432) -# \ No newline at end of file + pass + + + def _test_delete_port_with_bad_id(self,format): + LOG.debug("_test_delete_port - format:%s - START", format) + content_type = "application/" + format + port_state = "ACTIVE" + LOG.debug("Creating network and port") + network_req = testlib.create_new_network_request(self.tenant_id, + self.network_name, + format) + network_res = network_req.get_response(self.api) + self.assertEqual(network_res.status_int, 200) + network_data = Serializer().deserialize(network_res.body, + content_type) + network_id = network_data['networks']['network']['id'] + port_req = testlib.create_new_port_request(self.tenant_id, + network_id, port_state, + format) + port_res = port_req.get_response(self.api) + self.assertEqual(port_res.status_int, 200) + + # Test for portnotfound + delete_port_req = testlib.create_port_delete_request(self.tenant_id, + network_id, + "A_BAD_ID", + format) + delete_port_res = delete_port_req.get_response(self.api) + self.assertEqual(delete_port_res.status_int, 430) + + def test_delete_port_with_bad_id_xml(self): + self._test_delete_port_wth_bad_id('xml') + + def test_delete_port_with_bad_id_json(self): + self._test_delete_port_with_bad_id('json') + diff --git a/tests/unit/testlib.py b/tests/unit/testlib.py index 15bb00811..024b8202a 100644 --- a/tests/unit/testlib.py +++ b/tests/unit/testlib.py @@ -3,7 +3,7 @@ import webob from quantum.common.wsgi import Serializer -def create_request(path, body, content_type, method= 'GET'): +def create_request(path, body, content_type, method='GET'): req = webob.Request.blank(path) req.method = method req.headers = {} @@ -11,10 +11,11 @@ def create_request(path, body, content_type, method= 'GET'): req.body = body return req -def create_list_networks_request(tenant_id, format='xml'): + +def create_network_list_request(tenant_id, format='xml'): method = 'GET' path = "/tenants/%(tenant_id)s/networks.%(format)s" % locals() - content_type = "application/" + format + content_type = "application/" + format return create_request(path, None, content_type, method) @@ -22,7 +23,7 @@ def create_new_network_request(tenant_id, network_name, format='xml'): method = 'POST' path = "/tenants/%(tenant_id)s/networks.%(format)s" % locals() data = {'network': {'net-name': '%s' % network_name}} - content_type = "application/" + format + content_type = "application/" + format body = Serializer().serialize(data, content_type) return create_request(path, body, content_type, method) @@ -35,6 +36,14 @@ def create_network_delete_request(tenant_id, network_id, format='xml'): return create_request(path, None, content_type, method) +def create_port_list_request(tenant_id, network_id, format='xml'): + method = 'GET' + path = "/tenants/%(tenant_id)s/networks/" \ + "%(network_id)s/ports.%(format)s" % locals() + content_type = "application/" + format + return create_request(path, None, content_type, method) + + def create_new_port_request(tenant_id, network_id, port_state, format='xml'): method = 'POST' path = "/tenants/%(tenant_id)s/networks/" \ @@ -44,6 +53,7 @@ def create_new_port_request(tenant_id, network_id, port_state, format='xml'): body = Serializer().serialize(data, content_type) return create_request(path, body, content_type, method) + def create_port_delete_request(tenant_id, network_id, port_id, format='xml'): method = 'DELETE' path = "/tenants/%(tenant_id)s/networks/" \