From cf2b9c152677f4e0f4130c3b1a96011a9bc472ab Mon Sep 17 00:00:00 2001 From: rohitagarwalla Date: Sat, 3 Sep 2011 03:42:37 -0700 Subject: [PATCH] made general exception handling messages consistent removed LOG pylint errors cleanup in tests --- tests/unit/database_stubs.py | 48 ++++++++++++++++++------------------ tests/unit/test_database.py | 44 ++++++++------------------------- 2 files changed, 34 insertions(+), 58 deletions(-) diff --git a/tests/unit/database_stubs.py b/tests/unit/database_stubs.py index 66ddc1006..5b9a9c81e 100644 --- a/tests/unit/database_stubs.py +++ b/tests/unit/database_stubs.py @@ -35,14 +35,14 @@ class QuantumDB(object): nets = [] try: for net in db.network_list(tenant_id): - LOG.debug("Getting network: %s" % net.uuid) + LOG.debug("Getting network: %s", net.uuid) net_dict = {} net_dict["tenant-id"] = net.tenant_id net_dict["net-id"] = str(net.uuid) net_dict["net-name"] = net.name nets.append(net_dict) except Exception, exc: - LOG.error("Failed to get all networks: %s" % str(exc)) + LOG.error("Failed to get all networks: %s", str(exc)) return nets def get_network(self, network_id): @@ -50,14 +50,14 @@ class QuantumDB(object): net = [] try: for net in db.network_get(network_id): - LOG.debug("Getting network: %s" % net.uuid) + LOG.debug("Getting network: %s", net.uuid) net_dict = {} net_dict["tenant-id"] = net.tenant_id net_dict["net-id"] = str(net.uuid) net_dict["net-name"] = net.name net.append(net_dict) except Exception, exc: - LOG.error("Failed to get network: %s" % str(exc)) + LOG.error("Failed to get network: %s", str(exc)) return net def create_network(self, tenant_id, net_name): @@ -65,43 +65,43 @@ class QuantumDB(object): net_dict = {} try: res = db.network_create(tenant_id, net_name) - LOG.debug("Created network: %s" % res.uuid) + LOG.debug("Created network: %s", res.uuid) net_dict["tenant-id"] = res.tenant_id net_dict["net-id"] = str(res.uuid) net_dict["net-name"] = res.name return net_dict except Exception, exc: - LOG.error("Failed to create network: %s" % str(exc)) + LOG.error("Failed to create network: %s", str(exc)) def delete_network(self, net_id): """Delete a network""" try: net = db.network_destroy(net_id) - LOG.debug("Deleted network: %s" % net.uuid) + LOG.debug("Deleted network: %s", net.uuid) net_dict = {} net_dict["net-id"] = str(net.uuid) return net_dict except Exception, exc: - raise Exception("Failed to delete port: %s" % str(exc)) + LOG.error("Failed to delete network: %s", str(exc)) def rename_network(self, tenant_id, net_id, new_name): """Rename a network""" try: net = db.network_rename(net_id, tenant_id, new_name) - LOG.debug("Renamed network: %s" % net.uuid) + LOG.debug("Renamed network: %s", net.uuid) net_dict = {} net_dict["net-id"] = str(net.uuid) net_dict["net-name"] = net.name return net_dict except Exception, exc: - raise Exception("Failed to rename network: %s" % str(exc)) + LOG.error("Failed to rename network: %s", str(exc)) def get_all_ports(self, net_id): """Get all ports""" ports = [] try: for port in db.port_list(net_id): - LOG.debug("Getting port: %s" % port.uuid) + LOG.debug("Getting port: %s", port.uuid) port_dict = {} port_dict["port-id"] = str(port.uuid) port_dict["net-id"] = str(port.network_id) @@ -110,14 +110,14 @@ class QuantumDB(object): ports.append(port_dict) return ports except Exception, exc: - LOG.error("Failed to get all ports: %s" % str(exc)) + LOG.error("Failed to get all ports: %s", str(exc)) def get_port(self, net_id, port_id): """Get a port""" port_list = [] port = db.port_get(port_id, net_id) try: - LOG.debug("Getting port: %s" % port.uuid) + LOG.debug("Getting port: %s", port.uuid) port_dict = {} port_dict["port-id"] = str(port.uuid) port_dict["net-id"] = str(port.network_id) @@ -126,38 +126,38 @@ class QuantumDB(object): port_list.append(port_dict) return port_list except Exception, exc: - LOG.error("Failed to get port: %s" % str(exc)) + LOG.error("Failed to get port: %s", str(exc)) def create_port(self, net_id): """Add a port""" port_dict = {} try: port = db.port_create(net_id) - LOG.debug("Creating port %s" % port.uuid) + LOG.debug("Creating port %s", port.uuid) port_dict["port-id"] = str(port.uuid) port_dict["net-id"] = str(port.network_id) port_dict["int-id"] = port.interface_id port_dict["state"] = port.state return port_dict except Exception, exc: - LOG.error("Failed to create port: %s" % str(exc)) + LOG.error("Failed to create port: %s", str(exc)) def delete_port(self, net_id, port_id): """Delete a port""" try: port = db.port_destroy(port_id, net_id) - LOG.debug("Deleted port %s" % port.uuid) + LOG.debug("Deleted port %s", port.uuid) port_dict = {} port_dict["port-id"] = str(port.uuid) return port_dict except Exception, exc: - raise Exception("Failed to delete port: %s" % str(exc)) + LOG.error("Failed to delete port: %s", str(exc)) def update_port(self, net_id, port_id, port_state): """Update a port""" try: port = db.port_set_state(net_id, port_id, port_state) - LOG.debug("Updated port %s" % port.uuid) + LOG.debug("Updated port %s", port.uuid) port_dict = {} port_dict["port-id"] = str(port.uuid) port_dict["net-id"] = str(port.network_id) @@ -165,13 +165,13 @@ class QuantumDB(object): port_dict["state"] = port.state return port_dict except Exception, exc: - raise Exception("Failed to update port state: %s" % str(exc)) + LOG.error("Failed to update port state: %s", str(exc)) def plug_interface(self, net_id, port_id, int_id): """Plug interface to a port""" try: port = db.port_set_attachment(port_id, net_id, int_id) - LOG.debug("Attached interface to port %s" % port.uuid) + LOG.debug("Attached interface to port %s", port.uuid) port_dict = {} port_dict["port-id"] = str(port.uuid) port_dict["net-id"] = str(port.network_id) @@ -179,12 +179,12 @@ class QuantumDB(object): port_dict["state"] = port.state return port_dict except Exception, exc: - raise Exception("Failed to plug interface: %s" % str(exc)) + LOG.error("Failed to plug interface: %s", str(exc)) def unplug_interface(self, net_id, port_id): """Unplug interface to a port""" try: db.port_unset_attachment(port_id, net_id) - LOG.debug("Detached interface from port %s" % port_id) + LOG.debug("Detached interface from port %s", port_id) except Exception, exc: - raise Exception("Failed to unplug interface: %s" % str(exc)) + LOG.error("Failed to unplug interface: %s", str(exc)) diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py index ffd1694f7..8d7840f39 100644 --- a/tests/unit/test_database.py +++ b/tests/unit/test_database.py @@ -31,7 +31,7 @@ LOG = logging.getLogger('quantum.tests.test_database') class QuantumDBTest(unittest.TestCase): - """Class conisting of Quantum DB unit tests""" + """Class consisting of Quantum DB unit tests""" def setUp(self): """Setup for tests""" db.configure_db({'sql_connection': 'sqlite:///:memory:'}) @@ -47,7 +47,6 @@ class QuantumDBTest(unittest.TestCase): """test to create network""" net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1") self.assertTrue(net1["net-name"] == "plugin_test1") - self.teardown_network_port() def testb_get_networks(self): """test to get all networks""" @@ -61,7 +60,6 @@ class QuantumDBTest(unittest.TestCase): if "plugin_test" in net["net-name"]: count += 1 self.assertTrue(count == 2) - self.teardown_network_port() def testc_delete_network(self): """test to delete network""" @@ -69,12 +67,8 @@ class QuantumDBTest(unittest.TestCase): self.assertTrue(net1["net-name"] == "plugin_test1") self.dbtest.delete_network(net1["net-id"]) nets = self.dbtest.get_all_networks(self.tenant_id) - count = 0 - for net in nets: - if "plugin_test1" in net["net-name"]: - count += 1 + count = len(nets) self.assertTrue(count == 0) - self.teardown_network_port() def testd_rename_network(self): """test to rename network""" @@ -83,19 +77,21 @@ class QuantumDBTest(unittest.TestCase): net = self.dbtest.rename_network(self.tenant_id, net1["net-id"], "plugin_test1_renamed") self.assertTrue(net["net-name"] == "plugin_test1_renamed") - self.teardown_network_port() def teste_create_port(self): """test to create port""" net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1") port = self.dbtest.create_port(net1["net-id"]) self.assertTrue(port["net-id"] == net1["net-id"]) + + def testf_get_ports(self): + """test to get ports""" + net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1") + port = self.dbtest.create_port(net1["net-id"]) + self.assertTrue(port["net-id"] == net1["net-id"]) ports = self.dbtest.get_all_ports(net1["net-id"]) - count = 0 - for por in ports: - count += 1 + count = len(ports) self.assertTrue(count == 1) - self.teardown_network_port() def testf_delete_port(self): """test to delete port""" @@ -103,18 +99,11 @@ class QuantumDBTest(unittest.TestCase): port = self.dbtest.create_port(net1["net-id"]) self.assertTrue(port["net-id"] == net1["net-id"]) ports = self.dbtest.get_all_ports(net1["net-id"]) - count = 0 - for por in ports: - count += 1 - self.assertTrue(count == 1) for por in ports: self.dbtest.delete_port(net1["net-id"], por["port-id"]) ports = self.dbtest.get_all_ports(net1["net-id"]) - count = 0 - for por in ports: - count += 1 + count = len(ports) self.assertTrue(count == 0) - self.teardown_network_port() def testg_plug_unplug_interface(self): """test to plug/unplug interface""" @@ -126,16 +115,3 @@ class QuantumDBTest(unittest.TestCase): self.dbtest.unplug_interface(net1["net-id"], port1["port-id"]) port = self.dbtest.get_port(net1["net-id"], port1["port-id"]) self.assertTrue(port[0]["int-id"] == None) - self.teardown_network_port() - - def teardown_network_port(self): - """tearDown for Network and Port table""" - networks = self.dbtest.get_all_networks(self.tenant_id) - for net in networks: - netid = net["net-id"] - name = net["net-name"] - if "plugin_test" in name: - ports = self.dbtest.get_all_ports(netid) - for por in ports: - self.dbtest.delete_port(netid, por["port-id"]) - self.dbtest.delete_network(netid) -- 2.45.2