fix errors in database test cases.
fix errors in database test cases, add one test case to tets port update. Change-Id: I52108a55924d3ab750a8c453a80ed6abdd62bb23
This commit is contained in:
Yong Sheng Gong
parent
686982198c
commit
9a4403782d
|
|
@ -24,7 +24,6 @@ import logging
|
|||
import unittest
|
||||
|
||||
import quantum.db.api as db
|
||||
from quantum.plugins.linuxbridge.common import constants as const
|
||||
import quantum.plugins.linuxbridge.db.l2network_db as l2network_db
|
||||
|
||||
|
||||
|
|
@ -117,21 +116,6 @@ class QuantumDB(object):
|
|||
LOG.error("Failed to get all networks: %s" % str(exc))
|
||||
return nets
|
||||
|
||||
def get_network(self, network_id):
|
||||
"""Get a network"""
|
||||
net = []
|
||||
try:
|
||||
for net in db.network_get(network_id):
|
||||
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))
|
||||
return net
|
||||
|
||||
def create_network(self, tenant_id, net_name):
|
||||
"""Create a network"""
|
||||
net_dict = {}
|
||||
|
|
@ -158,118 +142,6 @@ class QuantumDB(object):
|
|||
except Exception, exc:
|
||||
raise Exception("Failed to delete port: %s" % str(exc))
|
||||
|
||||
def update_network(self, tenant_id, net_id, **kwargs):
|
||||
"""Update a network"""
|
||||
try:
|
||||
net = db.network_update(net_id, tenant_id, **kwargs)
|
||||
LOG.debug("Updated 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 update 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)
|
||||
port_dict = {}
|
||||
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
|
||||
port_dict["net"] = port.network
|
||||
ports.append(port_dict)
|
||||
return ports
|
||||
except Exception, 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(net_id, port_id)
|
||||
try:
|
||||
LOG.debug("Getting port: %s" % port.uuid)
|
||||
port_dict = {}
|
||||
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
|
||||
port_list.append(port_dict)
|
||||
return port_list
|
||||
except Exception, 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)
|
||||
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))
|
||||
|
||||
def delete_port(self, net_id, port_id):
|
||||
"""Delete a port"""
|
||||
try:
|
||||
port = db.port_destroy(net_id, port_id)
|
||||
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))
|
||||
|
||||
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)
|
||||
port_dict = {}
|
||||
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:
|
||||
raise Exception("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(net_id, port_id, int_id)
|
||||
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)
|
||||
port_dict["int-id"] = port.interface_id
|
||||
port_dict["state"] = port.state
|
||||
return port_dict
|
||||
except Exception, exc:
|
||||
raise Exception("Failed to plug interface: %s" % str(exc))
|
||||
|
||||
def unplug_interface(self, net_id, port_id):
|
||||
"""Unplug interface to a port"""
|
||||
try:
|
||||
port = db.port_unset_attachment(net_id, port_id)
|
||||
LOG.debug("Detached interface from port %s" % port.uuid)
|
||||
port_dict = {}
|
||||
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:
|
||||
raise Exception("Failed to unplug interface: %s" % str(exc))
|
||||
|
||||
|
||||
class L2networkDBTest(unittest.TestCase):
|
||||
"""Class conisting of L2network DB unit tests"""
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ stubs.py provides interface methods for
|
|||
the database test cases
|
||||
"""
|
||||
import logging
|
||||
import unittest
|
||||
|
||||
from quantum.db import api as db
|
||||
|
||||
|
|
@ -157,7 +156,7 @@ class QuantumDB(object):
|
|||
def update_port(self, net_id, port_id, **kwargs):
|
||||
"""Update a port"""
|
||||
try:
|
||||
port = db.port_set_state(net_id, port_id, **kwargs)
|
||||
port = db.port_update(port_id, net_id, **kwargs)
|
||||
LOG.debug("Updated port %s", port.uuid)
|
||||
port_dict = {}
|
||||
port_dict["id"] = str(port.uuid)
|
||||
|
|
|
|||
|
|
@ -89,6 +89,20 @@ class QuantumDBTest(unittest.TestCase):
|
|||
count = len(ports)
|
||||
self.assertTrue(count == 1)
|
||||
|
||||
def testf_update_port(self):
|
||||
"""test to update port"""
|
||||
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
|
||||
port = self.dbtest.create_port(net1["id"])
|
||||
self.dbtest.update_port(port["net-id"],
|
||||
port['id'],
|
||||
state='ACTIVE',
|
||||
interface_id='interface_id1')
|
||||
self.assertTrue(port["net-id"] == net1["id"])
|
||||
ports = self.dbtest.get_all_ports(net1["id"])
|
||||
new_port = ports[0]
|
||||
self.assertEqual('ACTIVE', new_port['state'])
|
||||
self.assertEqual('interface_id1', new_port['attachment'])
|
||||
|
||||
def testf_delete_port(self):
|
||||
"""test to delete port"""
|
||||
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
|
||||
|
|
|
|||
Loading…
Reference in New Issue