import ConfigParser
import logging as LOG
+ from optparse import OptionParser
import os
import sys
-import unittest
+ from quantum.common import exceptions as q_exc
from quantum.quantum_plugin_base import QuantumPluginBase
- from optparse import OptionParser
import quantum.db.api as db
import ovs_db
Updates the state of a port on the specified Virtual Network.
"""
LOG.debug("update_port() called\n")
- port = db.port_get(port_id)
- port['port-state'] = port_state
- return port
+ port = db.port_get(port_id, net_id)
+ db.port_set_state(port_id, net_id, port_state)
+ return self._make_port_dict(str(port.uuid), port.state, None, None)
def get_port_details(self, tenant_id, net_id, port_id):
- port = db.port_get(port_id)
- rv = {"port-id": port.uuid, "attachment": port.interface_id,
- "net-id": port.network_id, "port-state": "UP"}
- return rv
+ port = db.port_get(port_id, net_id)
+ return self._make_port_dict(str(port.uuid), port.state,
+ port.network_id, port.interface_id)
def plug_interface(self, tenant_id, net_id, port_id, remote_iface_id):
- db.port_set_attachment(port_id, remote_iface_id)
+ db.port_set_attachment(port_id, net_id, remote_iface_id)
def unplug_interface(self, tenant_id, net_id, port_id):
- db.port_set_attachment(port_id, "")
+ db.port_set_attachment(port_id, net_id, "")
def get_interface_details(self, tenant_id, net_id, port_id):
- res = db.port_get(port_id)
+ res = db.port_get(port_id, net_id)
return res.interface_id
-
-
-class VlanMapTest(unittest.TestCase):
-
- def setUp(self):
- self.vmap = VlanMap()
-
- def tearDown(self):
- pass
-
- def testAddVlan(self):
- vlan_id = self.vmap.acquire("foobar")
- self.assertTrue(vlan_id == 2)
-
- def testReleaseVlan(self):
- vlan_id = self.vmap.acquire("foobar")
- self.vmap.release("foobar")
- self.assertTrue(self.vmap.get(vlan_id) == None)
-
-
-if __name__ == "__main__":
- usagestr = "Usage: %prog [OPTIONS] <command> [args]"
- parser = OptionParser(usage=usagestr)
- parser.add_option("-v", "--verbose", dest="verbose",
- action="store_true", default=False, help="turn on verbose logging")
-
- options, args = parser.parse_args()
-
- if options.verbose:
- LOG.basicConfig(level=LOG.DEBUG)
- else:
- LOG.basicConfig(level=LOG.WARN)
-
- # Make sqlalchemy quieter
- LOG.getLogger('sqlalchemy.engine').setLevel(LOG.WARN)
- # Run the tests
- suite = unittest.TestLoader().loadTestsFromTestCase(VlanMapTest)
- unittest.TextTestRunner(verbosity=2).run(suite)