"show_net": {
"func": cli_lib.show_net,
"args": ["tenant-id", "net-id"]},
- "rename_net": {
- "func": cli_lib.rename_net,
+ "update_net": {
+ "func": cli_lib.update_net,
"args": ["tenant-id", "net-id", "new-name"]},
"list_ports": {
"func": cli_lib.list_ports,
"delete_port": {
"func": cli_lib.delete_port,
"args": ["tenant-id", "net-id", "port-id"]},
- "set_port_state": {
- "func": cli_lib.set_port_state,
- "args": ["tenant-id", "net-id", "port-id", "new_state"]},
+ "update_port": {
+ "func": cli_lib.update_port,
+ "args": ["tenant-id", "net-id", "port-id", "params"]},
"show_port": {
"func": cli_lib.show_port,
"args": ["tenant-id", "net-id", "port-id"]},
"create_net": "Created a new Virtual Network with ID: " +
"%(network_id)s\n" +
"for Tenant: %(tenant_id)s",
- "rename_net": "Renamed Virtual Network with ID: %(network.id)s\n" +
- "for Tenant: %(tenant_id)s\n" +
- "new name is: %(network.name)s",
+ "update_net": "Updated Virtual Network with ID: %(network.id)s\n" +
+ "for Tenant: %(tenant_id)s\n",
"delete_net": "Deleted Virtual Network with ID: %(network_id)s\n" +
"for Tenant %(tenant_id)s",
"list_ports": "Ports on Virtual Network: %(network_id)s\n" +
"interface: %(port.attachment)s\n" +
"on Virtual Network: %(network_id)s\n" +
"for Tenant: %(tenant_id)s",
- "set_port_state": "Updated state for Logical Port " +
+ "update_port": "Updated Logical Port " +
"with ID: %(port.id)s\n" +
- "new state is: %(port.state)s\n" +
"on Virtual Network: %(network_id)s\n" +
"for tenant: %(tenant_id)s",
"delete_port": "Deleted Logical Port with ID: %(port_id)s\n" +
_handle_exception(ex)
-def rename_net(client, *args):
- tenant_id, network_id, name = args
- data = {'network': {'name': '%s' % name}}
+def update_net(client, *args):
+ tenant_id, network_id, param_data = args
+ data = {'network': {}}
+ for kv in param_data.split(","):
+ k, v = kv.split("=")
+ data['network'][k] = v
+ data['network']['id'] = network_id
try:
client.update_network(network_id, data)
LOG.debug("Operation 'update_network' executed.")
# Response has no body. Use data for populating output
- data['network']['id'] = network_id
- output = prepare_output("rename_net", tenant_id, data)
+ output = prepare_output("update_net", tenant_id, data)
print output
except Exception as ex:
_handle_exception(ex)
_handle_exception(ex)
-def set_port_state(client, *args):
- tenant_id, network_id, port_id, new_state = args
- data = {'port': {'state': '%s' % new_state}}
+def update_port(client, *args):
+ tenant_id, network_id, port_id, param_data = args
+ data = {'port': {}}
+ for kv in param_data.split(","):
+ k, v = kv.split("=")
+ data['port'][k] = v
+ data['network_id'] = network_id
+ data['port']['id'] = port_id
try:
- client.set_port_state(network_id, port_id, data)
- LOG.debug("Operation 'set_port_state' executed.")
+ client.update_port(network_id, port_id, data)
+ LOG.debug("Operation 'udpate_port' executed.")
# Response has no body. Use data for populating output
- data['network_id'] = network_id
- data['port']['id'] = port_id
- output = prepare_output("set_port_state", tenant_id, data)
+ output = prepare_output("update_port", tenant_id, data)
print output
except Exception as ex:
_handle_exception(ex)
exception_args={"net_id": network, "port_id": port})
@ApiCall
- def set_port_state(self, network, port, body=None):
+ def update_port(self, network, port, body=None):
"""
- Sets the state of the specified port
+ Sets the attributes of the specified port
"""
return self.do_request("PUT",
self.port_path % (network, port), body=body,
exception_args={"net_id": network,
- "port_id": port,
- "port_state": str(body)})
+ "port_id": port})
@ApiCall
def show_port_attachment(self, network, port):
LOG.debug("_test_delete_port - tenant:%s "\
"- format:%s - END", format, tenant)
- def _test_set_port_state(self, tenant=TENANT_1, format='json', status=200):
- LOG.debug("_test_set_port_state - tenant:%s "\
+ def _test_update_port(self, tenant=TENANT_1, format='json', status=200):
+ LOG.debug("_test_update_port - tenant:%s "\
"- format:%s - START", format, tenant)
- self._assert_sanity(self.client.set_port_state,
+ self._assert_sanity(self.client.update_port,
status,
"PUT",
"networks/001/ports/001",
{'port': {'state': 'ACTIVE'}}],
params={'tenant': tenant, 'format': format})
- LOG.debug("_test_set_port_state - tenant:%s "\
+ LOG.debug("_test_update_port - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_show_port_attachment(self,
def test_delete_port_error_432(self):
self._test_delete_port(status=432)
- def test_set_port_state_json(self):
- self._test_set_port_state(format='json')
+ def test_update_port_json(self):
+ self._test_update_port(format='json')
- def test_set_port_state_xml(self):
- self._test_set_port_state(format='xml')
+ def test_update_port_xml(self):
+ self._test_update_port(format='xml')
- def test_set_port_state_alt_tenant(self):
- self._test_set_port_state(tenant=TENANT_2)
+ def test_update_port_alt_tenant(self):
+ self._test_update_port(tenant=TENANT_2)
- def test_set_port_state_error_470(self):
- self._test_set_port_state(status=470)
+ def test_update_port_error_470(self):
+ self._test_update_port(status=470)
- def test_set_port_state_error_401(self):
- self._test_set_port_state(status=401)
+ def test_update_port_error_401(self):
+ self._test_update_port(status=401)
- def test_set_port_state_error_400(self):
- self._test_set_port_state(status=400)
+ def test_update_port_error_400(self):
+ self._test_update_port(status=400)
- def test_set_port_state_error_420(self):
- self._test_set_port_state(status=420)
+ def test_update_port_error_420(self):
+ self._test_update_port(status=420)
- def test_set_port_state_error_430(self):
- self._test_set_port_state(status=430)
+ def test_update_port_error_430(self):
+ self._test_update_port(status=430)
- def test_set_port_state_error_431(self):
- self._test_set_port_state(status=431)
+ def test_update_port_error_431(self):
+ self._test_update_port(status=431)
def test_show_port_attachment_json(self):
self._test_show_port_attachment(format='json')
raise q_exc.NetworkNotFound(net_id=net_id)
-def network_rename(tenant_id, net_id, new_name):
+def network_update(net_id, tenant_id, **kwargs):
session = get_session()
net = network_get(net_id)
- _check_duplicate_net_name(tenant_id, new_name)
- net.name = new_name
+ for key in kwargs.keys():
+ if key == "name":
+ _check_duplicate_net_name(tenant_id, kwargs[key])
+ net[key] = kwargs[key]
session.merge(net)
session.flush()
return net
raise q_exc.PortNotFound(net_id=net_id, port_id=port_id)
-def port_set_state(net_id, port_id, new_state):
- if new_state not in ('ACTIVE', 'DOWN'):
- raise q_exc.StateInvalid(port_state=new_state)
-
+def port_update(port_id, net_id, **kwargs):
# confirm network exists
network_get(net_id)
port = port_get(net_id, port_id)
session = get_session()
- port.state = new_state
+ for key in kwargs.keys():
+ if key == "state":
+ if kwargs[key] not in ('ACTIVE', 'DOWN'):
+ raise q_exc.StateInvalid(port_state=kwargs[key])
+ port[key] = kwargs[key]
session.merge(port)
session.flush()
return port
pass
@abstractmethod
- def rename_network(self, args):
+ def update_network(self, args):
"""
Returns a dictionary containing the first element as a device
IP address list. The model then invokes the device-specific plugin
pass
@abstractmethod
- def rename_network(self, tenant_id, net_id, new_name, **kwargs):
+ def update_network(self, tenant_id, net_id, name, **kwargs):
"""
:returns:
:raises:
pass
@abstractmethod
- def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
:returns:
:raises:
pass
@abstractmethod
- def rename_network(self, args):
+ def update_network(self, args):
"""
:returns:
:raises:
return new_networks_list
- def create_network(self, tenant_id, net_name):
+ def create_network(self, tenant_id, net_name, **kwargs):
"""
Creates a new Virtual Network, and assigns it
a symbolic name.
return new_network
- def rename_network(self, tenant_id, net_id, new_name):
+ def update_network(self, tenant_id, net_id, **kwargs):
"""
Updates the symbolic name belonging to a particular
Virtual Network.
"""
- LOG.debug("rename_network() called\n")
- network = db.network_rename(tenant_id, net_id, new_name)
+ LOG.debug("update_network() called\n")
+ network = db.network_update(net_id, tenant_id, **kwargs)
self._invoke_device_plugins(self._func_name(), [tenant_id, net_id,
- new_name])
+ kwargs])
net_dict = cutil.make_net_dict(network[const.UUID],
network[const.NETWORKNAME],
[])
return ports_on_net
- def create_port(self, tenant_id, net_id, port_state=None):
+ def create_port(self, tenant_id, net_id, port_state=None, **kwargs):
"""
Creates a port on the specified Virtual Network.
"""
raise exc.PortInUse(port_id=port_id, net_id=net_id,
att_id=attachment_id)
- def update_port(self, tenant_id, net_id, port_id, port_state):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
Updates the state of a port on the specified Virtual Network.
"""
LOG.debug("update_port() called\n")
network = db.network_get(net_id)
self._invoke_device_plugins(self._func_name(), [tenant_id, net_id,
- port_id, port_state])
- self._validate_port_state(port_state)
- db.port_set_state(net_id, port_id, port_state)
- new_port_dict = cutil.make_port_dict(port_id, port_state, net_id,
+ port_id, kwargs])
+ self._validate_port_state(kwargs["state"])
+ db.port_update(port_id, net_id, **kwargs)
+
+ new_port_dict = cutil.make_port_dict(port_id, kwargs["state"], net_id,
None)
return new_port_dict
def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
"""Invoke only the device plugin"""
+ # If the last param is a dict, add it to kwargs
+ if args and type(args[-1]) is dict:
+ kwargs.update(args.pop())
+
return getattr(self._plugins[plugin_key], function_name)(*args,
**kwargs)
"""Not implemented for this model"""
pass
- def rename_network(self, args):
+ def update_network(self, args):
"""Support for the Quantum core API call"""
output = []
ucs_output = self._invoke_plugin_per_device(const.UCS_PLUGIN,
def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
"""Invoke only the device plugin"""
+ # If the last param is a dict, add it to kwargs
+ if args and type(args[-1]) is dict:
+ kwargs.update(args.pop())
+
return getattr(self._plugins[plugin_key], function_name)(*args,
**kwargs)
"""Not implemented for this model"""
pass
- def rename_network(self, args):
+ def update_network(self, args):
"""Support for the Quantum core API call"""
self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(),
args)
network = self._get_network(tenant_id, net_id)
return network
- def rename_network(self, tenant_id, net_id, new_name, **kwargs):
+ def update_network(self, tenant_id, net_id, **kwargs):
"""
- Updates the symbolic name belonging to a particular
+ Updates the properties of a particular
Virtual Network.
"""
- LOG.debug("NexusPlugin:rename_network() called\n")
+ LOG.debug("NexusPlugin:update_network() called\n")
network = self._get_network(tenant_id, net_id)
- network[const.NET_NAME] = new_name
+ network[const.NET_NAME] = kwargs["name"]
return network
def get_all_ports(self, tenant_id, net_id, **kwargs):
LOG.debug("get_network_details() called\n")
return self._get_all_ucsms()
- def rename_network(self, args):
+ def update_network(self, args):
"""Return all UCSM IPs"""
- LOG.debug("rename_network() called\n")
+ LOG.debug("update_network() called\n")
return self._get_all_ucsms()
def get_all_ports(self, args):
return new_network
- def rename_network(self, tenant_id, net_id, new_name, **kwargs):
+ def update_network(self, tenant_id, net_id, **kwargs):
"""
Updates the symbolic name belonging to a particular
Virtual Network.
"""
- LOG.debug("UCSVICPlugin:rename_network() called\n")
+ LOG.debug("UCSVICPlugin:update_network() called\n")
self._set_ucsm(kwargs[const.DEVICE_IP])
network = db.network_get(net_id)
net_dict = cutil.make_net_dict(network[const.UUID],
blade_id, interface_dn)
return udb.remove_portbinding(port_id)
- def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
Updates the state of a port on the specified Virtual Network.
"""
except Exception, exc:
raise Exception("Failed to delete port: %s" % str(exc))
- def rename_network(self, tenant_id, net_id, new_name):
- """Rename a network"""
+ def update_network(self, tenant_id, net_id, **kwargs):
+ """Update a network"""
try:
- net = db.network_rename(tenant_id, net_id, new_name)
- LOG.debug("Renamed network: %s" % net.uuid)
+ 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 rename network: %s" % str(exc))
+ raise Exception("Failed to update network: %s" % str(exc))
def get_all_ports(self, net_id):
"""Get all ports"""
self.assertTrue(count == 0)
self.teardown_network_port()
- def testd_rename_network(self):
- """test to rename network"""
+ def testd_update_network(self):
+ """test to update (rename) network"""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["net-name"] == "plugin_test1")
- net = self.dbtest.rename_network(self.tenant_id, net1["net-id"],
- "plugin_test1_renamed")
+ net = self.dbtest.update_network(self.tenant_id, net1["net-id"],
+ name="plugin_test1_renamed")
self.assertTrue(net["net-name"] == "plugin_test1_renamed")
self.teardown_network_port()
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
tenant_id, net_id)
LOG.debug("test_show_network_not_found - END")
- def test_rename_network(self, net_tenant_id=None,
+ def test_update_network(self, net_tenant_id=None,
new_name='new_test_network'):
"""
Tests rename of a Virtual Network .
"""
- LOG.debug("test_rename_network - START")
+ LOG.debug("test_update_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
- rename_net_dict = self._l2network_plugin.rename_network(
- tenant_id, new_net_dict[const.NET_ID], new_name)
+ rename_net_dict = self._l2network_plugin.update_network(
+ tenant_id, new_net_dict[const.NET_ID],
+ name=new_name)
net = db.network_get(new_net_dict[const.NET_ID])
self.assertEqual(net[const.NETWORKNAME], new_name)
self.assertEqual(new_name, rename_net_dict[const.NET_NAME])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
- LOG.debug("test_rename_network - END")
+ LOG.debug("test_update_network - END")
- def test_rename_networkDNE(self, net_tenant_id=None,
+ def test_update_networkDNE(self, net_tenant_id=None,
net_id='0005', new_name='new_test_network'):
"""
- Tests rename of a Virtual Network when Network does not exist.
+ Tests update of a Virtual Network when Network does not exist.
"""
- LOG.debug("test_rename_network_not_found - START")
+ LOG.debug("test_update_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
- self._l2network_plugin.rename_network,
- tenant_id, net_id, new_name)
- LOG.debug("test_rename_network_not_found - END")
+ self._l2network_plugin.update_network,
+ tenant_id, net_id, name=new_name)
+ LOG.debug("test_update_network_not_found - END")
def test_list_networks(self, tenant_id='test_network'):
"""
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ self.state)
port_dict2 = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ self.state)
port_list = self._l2network_plugin.get_all_ports(
tenant_id, new_net_dict[const.NET_ID])
port_temp_list = [port_dict, port_dict2]
LOG.debug("test_list_ports - END")
def test_create_port(self, tenant_id='test_network',
- port_state=const.PORT_UP):
+ state=const.PORT_UP):
"""
Tests creation of Ports.
"""
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], port_state)
+ tenant_id, new_net_dict[const.NET_ID], state)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
- self.assertEqual(port_dict[const.PORT_STATE], port_state)
+ self.assertEqual(port_dict[const.PORT_STATE], state)
self.assertEqual(port_dict[const.NET_ID], new_net_dict[const.NET_ID])
- self.assertEqual(port[const.PORTSTATE], port_state)
+ self.assertEqual(port[const.PORTSTATE], state)
self.assertEqual(port[const.NETWORKID], new_net_dict[const.NET_ID])
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_create_port - END")
def test_create_port_network_DNE(
- self, net_tenant_id=None, net_id='0005', port_state=const.PORT_UP):
+ self, net_tenant_id=None, net_id='0005', state=const.PORT_UP):
"""
Tests creation of Ports when network does not exist.
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.create_port,
- tenant_id, net_id, port_state)
+ tenant_id, net_id, state)
LOG.debug("test_create_port_network_DNE - END:")
def test_delete_port(self, tenant_id='test_tenant',
- port_state=const.PORT_UP):
+ state=const.PORT_UP):
"""
Tests deletion of Ports
"""
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ state=state)
delete_port_dict = self._l2network_plugin.delete_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
- self.port_state)
+ self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
LOG.debug("test_delete_portInUse - END")
def test_update_port(self, tenant_id='test_tenant',
- port_state=const.PORT_DOWN):
+ state=const.PORT_DOWN):
"""
Tests updation of Ports.
"""
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ self.state)
update_port_dict = self._l2network_plugin.update_port(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ port_dict[const.PORT_ID], state=state)
new_port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
- self.assertEqual(new_port[const.PORTSTATE], port_state)
- self.assertEqual(update_port_dict[const.PORT_STATE], port_state)
+ self.assertEqual(new_port[const.PORTSTATE], state)
+ self.assertEqual(update_port_dict[const.PORT_STATE], state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_update_port - END")
LOG.debug("test_update_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_port, tenant_id,
- net_id, port_id, const.PORT_UP)
+ net_id, port_id, const.PORT_UP, state=const.PORT_UP)
LOG.debug("test_update_port_networkDNE - END")
def test_update_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
tenant_id, self.network_name)
self.assertRaises(
exc.PortNotFound, self._l2network_plugin.update_port, tenant_id,
- new_net_dict[const.NET_ID], port_id, const.PORT_UP)
+ new_net_dict[const.NET_ID], port_id, state=const.PORT_UP)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_update_portDNE - END")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID],
+ self.state)
get_port_dict = self._l2network_plugin.get_port_details(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
- self.assertEqual(port[const.PORTSTATE], self.port_state)
- self.assertEqual(get_port_dict[const.PORT_STATE], self.port_state)
+ self.assertEqual(port[const.PORTSTATE], self.state)
+ self.assertEqual(get_port_dict[const.PORT_STATE], self.state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_show_port - END")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
- self.port_state)
+ self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
new_net_dict = self._l2network_plugin.create_network(
tenant_id, 'test_network')
port_dict = self._l2network_plugin.create_port(
- tenant_id, new_net_dict[const.NET_ID], 'const.PORT_UP')
+ tenant_id, new_net_dict[const.NET_ID],
+ const.PORT_UP)
self._l2network_plugin.associate_portprofile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
- self.port_state)
+ self.state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
- self.port_state)
+ self.state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
self.assertEqual(result_vlan_name, expected_output)
LOG.debug("test_get_vlan_name - END")
- def test_validate_port_state(self, port_state=const.PORT_UP):
+ def test_validate_state(self, state=const.PORT_UP):
"""
Tests validate port state
"""
- LOG.debug("test_validate_port_state - START")
- result = self._l2network_plugin._validate_port_state(port_state)
+ LOG.debug("test_validate_state - START")
+ result = self._l2network_plugin._validate_state(state)
self.assertEqual(result, True)
- LOG.debug("test_validate_port_state - END")
+ LOG.debug("test_validate_state - END")
- def test_invalid_port_state(self, port_state="BADSTATE"):
+ def test_invalid_state(self, state="BADSTATE"):
"""
Tests invalidate port state
"""
- LOG.debug("test_validate_port_state - START")
+ LOG.debug("test_validate_state - START")
self.assertRaises(exc.StateInvalid,
- self._l2network_plugin._validate_port_state,
- port_state)
- LOG.debug("test_validate_port_state - END")
+ self._l2network_plugin._validate_state,
+ state)
+ LOG.debug("test_validate_state - END")
def setUp(self):
"""
self.network_name = "test_network"
self.profile_name = "test_tenant_port_profile"
self.qos = "test_qos"
- self.port_state = const.PORT_UP
+ self.state = const.PORT_UP
self.net_id = '00005'
self.port_id = 'p0005'
self.remote_interface = 'new_interface'
self._l2network_plugin = l2network_plugin.L2Network()
+ LOG.debug(self._l2network_plugin)
+ LOG.debug("asdfasdfasdfasdfasdf")
"""
Clean up functions after the tests
res[const.NET_PORTS] = ports
return res
- def _make_port_dict(self, port_id, port_state, net_id, attachment):
- res = {const.PORT_ID: port_id, const.PORT_STATE: port_state}
+ def _make_port_dict(self, port_id, state, net_id, attachment):
+ res = {const.PORT_ID: port_id, const.PORT_STATE: state}
res[const.NET_ID] = net_id
res[const.ATTACHMENT] = attachment
return res
LOG.debug("test_delete_networkDNE - END")
- def test_rename_network(self):
+ def test_update_network(self):
"""Support for the Quantum core API call"""
- LOG.debug("test_rename_network - START")
+ LOG.debug("test_update_network - START")
self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
+
self._l2network_multiblade.create_network([tenant_id,
net_name,
self.net_id,
vlan_name(self.net_id),
vlan_id])
- db.network_rename(tenant_id, self.net_id, new_net_name)
- networks = self._l2network_multiblade.rename_network([tenant_id,
+ db.network_update(self.net_id, tenant_id, name=new_net_name)
+ networks = self._l2network_multiblade.update_network([tenant_id,
self.net_id,
- new_net_name])
+ {'name': new_net_name}])
self.assertEqual(networks.__len__(), self.ucs_count)
for network in networks:
self.assertEqual(network[const.NET_ID], self.net_id)
self.assertEqual(network[const.NET_NAME], new_net_name)
- LOG.debug("test_rename_network - END")
+ LOG.debug("test_update_network - END")
- def test_rename_networkDNE(self):
+ def test_update_networkDNE(self):
"""Support for the Quantum core API call"""
- LOG.debug("test_rename_networkDNE - START")
+ LOG.debug("test_update_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
- self._l2network_multiblade.rename_network,
- [tenant_id, net_id, new_net_name])
- LOG.debug("test_rename_networkDNE - END")
+ self._l2network_multiblade.update_network,
+ [tenant_id, net_id, {'name': new_net_name}])
+ LOG.debug("test_update_networkDNE - END")
def test_get_all_networks(self):
"""Not implemented for this model"""
import unittest
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
+from quantum.plugins.cisco.common import cisco_credentials as creds
from quantum.plugins.cisco.db import l2network_db as cdb
from quantum.plugins.cisco.db import api as db
from quantum.plugins.cisco.common import cisco_credentials as cred
self.vlan_name = "q-" + str(self.net_id) + "vlan"
self.vlan_id = 267
self.port_id = "9"
+ db.configure_db({'sql_connection': 'sqlite:///:memory:'})
cdb.initialize()
cred.Store.initialize()
self._cisco_nexus_plugin = cisco_nexus_plugin.NexusPlugin()
LOG.debug("test_get_network_details_network_does_not_exist - END")
- def test_rename_network(self, new_name="new_network_name",
+ def test_update_network(self, new_name="new_network_name",
net_tenant_id=None, network_name=None):
"""
- Tests rename of a Virtual Network .
+ Tests update of a Virtual Network .
"""
- LOG.debug("test_rename_network - START")
+ LOG.debug("test_update_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
tenant_id, self.net_name, network_created["net-id"],
self.vlan_name, self.vlan_id)
rename_net_dict = self._cisco_nexus_plugin.rename_network(
- tenant_id, new_net_dict[const.NET_ID], new_name)
+ tenant_id, new_net_dict[const.NET_ID], name=new_name)
self.assertEqual(rename_net_dict[const.NET_NAME], new_name)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
- LOG.debug("test_rename_network - END")
+ LOG.debug("test_update_network - END")
- def test_rename_network_DNE(self, new_name="new_network_name",
+ def test_update_network_DNE(self, new_name="new_network_name",
net_tenant_id=None, network_id='0005'):
"""
- Tests rename of a Virtual Network when Network does not exist.
+ Tests update of a Virtual Network when Network does not exist.
"""
- LOG.debug("test_rename_network_DNE - START")
+ LOG.debug("test_update_network_DNE - START")
if net_tenant_id:
tenant_id = net_tenant_id
net_id = self.net_id
self.assertRaises(exc.NetworkNotFound,
- self._cisco_nexus_plugin.rename_network,
- new_name, tenant_id, net_id)
+ self._cisco_nexus_plugin.update_network,
+ tenant_id, net_id, name=new_name)
- LOG.debug("test_rename_network_DNE - END")
+ LOG.debug("test_update_network_DNE - END")
def test_list_all_networks(self, net_tenant_id=None):
"""
LOG.debug("test_%s - START", cmd)
net = self._l2network.create_network(tenant, net_name)
port = self._l2network.create_port(tenant, net[const.NET_ID],
- port_state)
+ port_state, state=port_state)
args = [tenant, net[const.NET_ID], port[const.PORT_ID]]
if params is not None:
"""Test that the UCS Inventory returns the correct devices to use"""
self._test_get_all_ucms('get_network_details')
- def test_rename_network(self):
+ def test_update_network(self):
"""Test that the UCS Inventory returns the correct devices to use"""
- self._test_get_all_ucms('rename_network')
+ self._test_get_all_ucms('update_network')
def test_get_all_ports(self):
"""Test that the UCS Inventory returns the correct devices to use"""
ports = self.get_all_ports(tenant_id, net_id)
return self._make_net_dict(str(net.uuid), net.name, ports)
- def rename_network(self, tenant_id, net_id, new_name):
- net = db.network_rename(net_id, tenant_id, new_name)
+ def update_network(self, tenant_id, net_id, **kwargs):
+ net = db.network_update(net_id, tenant_id, **kwargs)
return self._make_net_dict(str(net.uuid), net.name, None)
def _make_port_dict(self, port_id, port_state, net_id, attachment):
return self._make_port_dict(str(port.uuid), port.state,
port.network_id, port.interface_id)
- def update_port(self, tenant_id, net_id, port_id, port_state):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
Updates the state of a port on the specified Virtual Network.
"""
LOG.debug("update_port() called\n")
port = db.port_get(port_id, net_id)
- db.port_set_state(port_id, net_id, port_state)
+ db.port_update(port_id, net_id, **kwargs)
return self._make_port_dict(str(port.uuid), port.state,
port.network_id, port.interface_id)
"""
print("get_network_details() called\n")
- def rename_network(self, tenant_id, net_id, new_name):
- """
- Updates the symbolic name belonging to a particular
- Virtual Network.
- """
- print("rename_network() called\n")
+ def update_network(self, tenant_id, net_id, **kwargs):
+ print("update_network() called")
def get_all_ports(self, tenant_id, net_id):
"""
"""
print("delete_port() called\n")
- def update_port(self, tenant_id, net_id, port_id, port_state):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
- Updates the state of a port on the specified Virtual Network.
+ Updates the attributes of a port on the specified Virtual Network.
"""
print("update_port() called\n")
# Network not found
raise exc.NetworkNotFound(net_id=net_id)
- def rename_network(self, tenant_id, net_id, new_name):
+ def update_network(self, tenant_id, net_id, **kwargs):
"""
- Updates the symbolic name belonging to a particular
- Virtual Network.
+ Updates the attributes of a particular Virtual Network.
"""
- LOG.debug("FakePlugin.rename_network() called")
- db.network_rename(net_id, tenant_id, new_name)
- net = self._get_network(tenant_id, net_id)
+ LOG.debug("FakePlugin.update_network() called")
+ net = db.network_update(net_id, tenant_id, **kwargs)
return net
def get_all_ports(self, tenant_id, net_id):
port_item = {'port-id': str(port.uuid)}
return port_item
- def update_port(self, tenant_id, net_id, port_id, new_state):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
- Updates the state of a port on the specified Virtual Network.
+ Updates the attributes of a port on the specified Virtual Network.
"""
LOG.debug("FakePlugin.update_port() called")
#validate port and network ids
self._get_network(tenant_id, net_id)
self._get_port(tenant_id, net_id, port_id)
- self._validate_port_state(new_state)
- db.port_set_state(port_id, net_id, new_state)
+ port = db.port_update(port_id, net_id, **kwargs)
port_item = {'port-id': port_id,
- 'port-state': new_state}
+ 'port-state': port['state']}
return port_item
def delete_port(self, tenant_id, net_id, port_id):
except exc.HTTPError as e:
return faults.Fault(e)
try:
- self._plugin.rename_network(tenant_id, id,
- request_params['name'])
+ self._plugin.update_network(tenant_id, id,
+ **request_params)
return exc.HTTPNoContent()
except exception.NetworkNotFound as e:
return faults.Fault(faults.NetworkNotFound(e))
return faults.Fault(e)
try:
self._plugin.update_port(tenant_id, network_id, id,
- request_params['state'])
+ **request_params)
return exc.HTTPNoContent()
except exception.NetworkNotFound as e:
return faults.Fault(faults.NetworkNotFound(e))
raise q_exc.NetworkNotFound(net_id=net_id)
-def network_rename(net_id, tenant_id, new_name):
+def network_update(net_id, tenant_id, **kwargs):
session = get_session()
net = network_get(net_id)
- _check_duplicate_net_name(tenant_id, new_name)
- net.name = new_name
+ for key in kwargs.keys():
+ if key == "name":
+ _check_duplicate_net_name(tenant_id, kwargs[key])
+ net[key] = kwargs[key]
session.merge(net)
session.flush()
return net
raise q_exc.PortNotFound(net_id=net_id, port_id=port_id)
-def port_set_state(port_id, net_id, new_state):
- if new_state not in ('ACTIVE', 'DOWN'):
- raise q_exc.StateInvalid(port_state=new_state)
-
+def port_update(port_id, net_id, **kwargs):
# confirm network exists
network_get(net_id)
-
port = port_get(port_id, net_id)
session = get_session()
- port.state = new_state
+ for key in kwargs.keys():
+ if key == "state":
+ if kwargs[key] not in ('ACTIVE', 'DOWN'):
+ raise q_exc.StateInvalid(port_state=kwargs[key])
+ port[key] = kwargs[key]
session.merge(port)
session.flush()
return port
pass
@abstractmethod
- def rename_network(self, tenant_id, net_id, new_name):
+ def update_network(self, tenant_id, net_id, **kwargs):
"""
- Updates the symbolic name belonging to a particular
- Virtual Network.
+ Updates the attributes of a particular Virtual Network.
:returns: a sequence of mappings representing the new network
attributes, with the following signature:
pass
@abstractmethod
- def update_port(self, tenant_id, net_id, port_id, port_state):
+ def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""
- Updates the state of a specific port on the
+ Updates the attributes of a specific port on the
specified Virtual Network.
:returns: a mapping sequence with the following signature:
except Exception, exc:
LOG.error("Failed to delete network: %s", str(exc))
- def rename_network(self, tenant_id, net_id, new_name):
+ def update_network(self, tenant_id, net_id, param_data):
"""Rename a network"""
try:
- net = db.network_rename(net_id, tenant_id, new_name)
- LOG.debug("Renamed network: %s", net.uuid)
+ print param_data
+ net = db.network_update(net_id, tenant_id, **param_data)
+ LOG.debug("Updated network: %s", net.uuid)
net_dict = {}
net_dict["id"] = str(net.uuid)
net_dict["name"] = net.name
return net_dict
except Exception, exc:
- LOG.error("Failed to rename network: %s", str(exc))
+ LOG.error("Failed to update network: %s", str(exc))
def get_all_ports(self, net_id):
"""Get all ports"""
except Exception, exc:
LOG.error("Failed to delete port: %s", str(exc))
- def update_port(self, net_id, port_id, port_state):
+ def update_port(self, net_id, port_id, **kwargs):
"""Update a port"""
try:
- port = db.port_set_state(net_id, port_id, port_state)
+ port = db.port_set_state(net_id, port_id, **kwargs)
LOG.debug("Updated port %s", port.uuid)
port_dict = {}
port_dict["id"] = str(port.uuid)
self.assertEqual(show_network_res.status_int, 420)
LOG.debug("_test_show_network_not_found - format:%s - END", format)
- def _test_rename_network(self, format):
- LOG.debug("_test_rename_network - format:%s - START", format)
+ def _test_update_network(self, format):
+ LOG.debug("_test_update_network - format:%s - START", format)
content_type = "application/%s" % format
new_name = 'new_network_name'
network_id = self._create_network(format)
self.assertEqual({'id': network_id,
'name': new_name},
network_data['network'])
- LOG.debug("_test_rename_network - format:%s - END", format)
+ LOG.debug("_test_update_network - format:%s - END", format)
- def _test_rename_network_badrequest(self, format):
- LOG.debug("_test_rename_network_badrequest - format:%s - START",
+ def _test_update_network_badrequest(self, format):
+ LOG.debug("_test_update_network_badrequest - format:%s - START",
format)
network_id = self._create_network(format)
bad_body = {'network': {'bad-attribute': 'very-bad'}}
custom_req_body=bad_body)
update_network_res = update_network_req.get_response(self.api)
self.assertEqual(update_network_res.status_int, 400)
- LOG.debug("_test_rename_network_badrequest - format:%s - END",
+ LOG.debug("_test_update_network_badrequest - format:%s - END",
format)
- def _test_rename_network_not_found(self, format):
- LOG.debug("_test_rename_network_not_found - format:%s - START",
+ def _test_update_network_not_found(self, format):
+ LOG.debug("_test_update_network_not_found - format:%s - START",
format)
new_name = 'new_network_name'
update_network_req = testlib.update_network_request(self.tenant_id,
format)
update_network_res = update_network_req.get_response(self.api)
self.assertEqual(update_network_res.status_int, 420)
- LOG.debug("_test_rename_network_not_found - format:%s - END",
+ LOG.debug("_test_update_network_not_found - format:%s - END",
format)
def _test_delete_network(self, format):
def test_delete_network_xml(self):
self._test_delete_network('xml')
- def test_rename_network_json(self):
- self._test_rename_network('json')
+ def test_update_network_json(self):
+ self._test_update_network('json')
- def test_rename_network_xml(self):
- self._test_rename_network('xml')
+ def test_update_network_xml(self):
+ self._test_update_network('xml')
- def test_rename_network_badrequest_json(self):
- self._test_rename_network_badrequest('json')
+ def test_update_network_badrequest_json(self):
+ self._test_update_network_badrequest('json')
- def test_rename_network_badrequest_xml(self):
- self._test_rename_network_badrequest('xml')
+ def test_update_network_badrequest_xml(self):
+ self._test_update_network_badrequest('xml')
- def test_rename_network_not_found_json(self):
- self._test_rename_network_not_found('json')
+ def test_update_network_not_found_json(self):
+ self._test_update_network_not_found('json')
- def test_rename_network_not_found_xml(self):
- self._test_rename_network_not_found('xml')
+ def test_update_network_not_found_xml(self):
+ self._test_update_network_not_found('xml')
def test_delete_network_in_use_json(self):
self._test_delete_network_in_use('json')
# Must add newline at the end to match effect of print call
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
- def _verify_rename_network(self):
+ def _verify_update_network(self):
# Verification - get raw result from db
nw_list = db.network_list(self.tenant_id)
network_data = {'id': nw_list[0].uuid,
'name': nw_list[0].name}
# Fill CLI template
- output = cli.prepare_output('rename_net', self.tenant_id,
+ output = cli.prepare_output('update_net', self.tenant_id,
dict(network=network_data))
# Verify!
# Must add newline at the end to match effect of print call
# Must add newline at the end to match effect of print call
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
- def _verify_set_port_state(self, network_id, port_id):
+ def _verify_update_port(self, network_id, port_id):
# Verification - get raw result from db
port = db.port_get(port_id, network_id)
port_data = {'id': port.uuid, 'state': port.state}
# Fill CLI template
- output = cli.prepare_output('set_port_state', self.tenant_id,
+ output = cli.prepare_output('update_port', self.tenant_id,
dict(network_id=network_id,
port=port_data))
# Verify!
LOG.debug(self.fake_stdout.content)
self._verify_show_network()
- def test_rename_network(self):
+ def test_update_network(self):
try:
net = db.network_create(self.tenant_id, self.network_name_1)
network_id = net['uuid']
- cli.rename_net(self.client, self.tenant_id,
- network_id, self.network_name_2)
+ cli.update_net(self.client, self.tenant_id,
+ network_id, 'name=%s' % self.network_name_2)
except:
LOG.exception("Exception caught: %s", sys.exc_info())
- self.fail("test_rename_network failed due to an exception")
+ self.fail("test_update_network failed due to an exception")
LOG.debug("Operation completed. Verifying result")
LOG.debug(self.fake_stdout.content)
- self._verify_rename_network()
+ self._verify_update_network()
def test_list_ports(self):
try:
LOG.debug(self.fake_stdout.content)
self._verify_delete_port(network_id, port_id)
- def test_set_port_state(self):
+ def test_update_port(self):
try:
net = db.network_create(self.tenant_id, self.network_name_1)
network_id = net['uuid']
port = db.port_create(network_id)
port_id = port['uuid']
# Default state is DOWN - change to ACTIVE.
- cli.set_port_state(self.client, self.tenant_id, network_id,
- port_id, 'ACTIVE')
+ cli.update_port(self.client, self.tenant_id, network_id,
+ port_id, 'state=ACTIVE')
except:
LOG.exception("Exception caught: %s", sys.exc_info())
- self.fail("test_set_port_state failed due to an exception")
+ self.fail("test_update_port failed due to an exception")
LOG.debug("Operation completed. Verifying result")
LOG.debug(self.fake_stdout.content)
- self._verify_set_port_state(network_id, port_id)
+ self._verify_update_port(network_id, port_id)
def test_show_port_no_attach(self):
network_id = None
count = len(nets)
self.assertTrue(count == 0)
- def testd_rename_network(self):
+ def testd_update_network(self):
"""test to rename network"""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["name"] == "plugin_test1")
- net = self.dbtest.rename_network(self.tenant_id, net1["id"],
- "plugin_test1_renamed")
+ net = self.dbtest.update_network(self.tenant_id, net1["id"],
+ {'name': "plugin_test1_renamed"})
+ print net
self.assertTrue(net["name"] == "plugin_test1_renamed")
def teste_create_port(self):
Code Review / openstack-build / neutron-build.git / commitdiff