pass
def update_network(self, context, id, network):
- """Currently there is no processing required for the device plugins"""
- pass
+ """
+ Perform this operation in the context of the configured device
+ plugins.
+ """
+ n = network
+ vlan = cdb.get_vlan_binding(id)
+ args = [n['tenant_id'], id, {'vlan_id': vlan.vlan_id},
+ {'net_admin_state': n['admin_state_up']},
+ {'vlan_ids': ''}]
+ nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
+ self._func_name(),
+ args)
+ return nexus_output
def delete_network(self, context, id, kwargs):
"""
supported_extension_aliases = []
_plugins = {}
_inventory = {}
- _methods_to_delegate = ['update_network', 'get_network', 'get_networks',
+ _methods_to_delegate = ['get_network', 'get_networks',
'create_port', 'create_port_bulk', 'delete_port',
'update_port', 'get_port', 'get_ports',
'create_subnet', 'create_subnet_bulk',
raise
def update_network(self, context, id, network):
- """For this model this method will be delegated to vswitch plugin"""
- pass
+ """
+ Perform this operation in the context of the configured device
+ plugins.
+ """
+ LOG.debug("update_network() called\n")
+ args = [context, id, network]
+ ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
+ self._func_name(),
+ args)
+ vlan_id = odb.get_vlan(ovs_output[0]['id'])
+ vlan_ids = ','.join(str(vlan[0]) for vlan in odb.get_vlans())
+ args = [ovs_output[0]['tenant_id'], id, {'vlan_id': vlan_id},
+ {'net_admin_state': ovs_output[0]['admin_state_up']},
+ {'vlan_ids': vlan_ids}]
+ nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
+ self._func_name(),
+ args)
+ return ovs_output[0]
def delete_network(self, context, id):
"""
Virtual Network.
"""
LOG.debug("update_network() called\n")
- try:
- self._invoke_device_plugins(self._func_name(), [context, id,
- network])
- return super(PluginV2, self).update_network(context, id,
- network)
- except:
- raise
+ upd_net_dict = super(PluginV2, self).update_network(context, id,
+ network)
+ self._invoke_device_plugins(self._func_name(), [context, id,
+ upd_net_dict])
+ return upd_net_dict
def delete_network(self, context, id):
"""
if vlans == '':
vlans = 'none'
return vlans.strip(',')
+
+ def add_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
+ nexus_ports, nexus_ssh_port, vlan_ids=None):
+ """
+ Adds a vlan from interfaces on the Nexus switch given the VLAN ID
+ """
+ with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
+ nexus_password) as man:
+ if not vlan_ids:
+ vlan_ids = self.build_vlans_cmd()
+ for ports in nexus_ports:
+ self.enable_vlan_on_trunk_int(man, ports, vlan_ids)
+
+ def remove_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
+ nexus_ports, nexus_ssh_port):
+ """
+ Removes a vlan from interfaces on the Nexus switch given the VLAN ID
+ """
+ with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
+ nexus_password) as man:
+ for ports in nexus_ports:
+ self.disable_vlan_on_trunk_int(man, ports, vlan_id)
for key in kwargs:
if key == const.CONTEXT:
context = kwargs[const.CONTEXT]
- if key == const.BASE_PLUGIN_REF:
+ elif key == const.BASE_PLUGIN_REF:
base_plugin_ref = kwargs[const.BASE_PLUGIN_REF]
- if key == 'vlan_id':
+ elif key == 'vlan_id':
vlan_id = kwargs['vlan_id']
if vlan_id is None:
vlan_id = self._get_vlan_id_for_network(tenant_id, net_id,
Virtual Network.
"""
LOG.debug("NexusPlugin:update_network() called\n")
- network = self._get_network(tenant_id, net_id)
- network[const.NET_NAME] = kwargs["name"]
- return network
+ if 'net_admin_state' in kwargs:
+ net_admin_state = kwargs['net_admin_state']
+ vlan_id = kwargs['vlan_id']
+ vlan_ids = kwargs['vlan_ids']
+ if not net_admin_state:
+ self._client.remove_vlan_int(
+ str(vlan_id), self._nexus_ip,
+ self._nexus_username, self._nexus_password,
+ self._nexus_ports, self._nexus_ssh_port)
+ else:
+ self._client.add_vlan_int(
+ str(vlan_id), self._nexus_ip,
+ self._nexus_username, self._nexus_password,
+ self._nexus_ports, self._nexus_ssh_port,
+ vlan_ids)
+ return net_id
def get_all_ports(self, tenant_id, net_id, **kwargs):
"""
Builds a string with all the VLANs on the same Switch
"""
pass
+
+ def add_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
+ nexus_ports, nexus_ssh_port, vlan_ids=None):
+ """
+ Adds a vlan from interfaces on the Nexus switch given the VLAN ID
+ """
+ pass
+
+ def remove_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
+ nexus_ports, nexus_ssh_port):
+ """
+ Removes a vlan from interfaces on the Nexus switch given the VLAN ID
+ """
+ pass
from quantum.plugins.cisco.nexus import cisco_nexus_plugin_v2
-LOG = logging.getLogger(__name__)
-
-
NEXUS_IP_ADDRESS = '1.1.1.1'
NEXUS_USERNAME = 'username'
NEXUS_PASSWORD = 'password'
'__init__', new=new_nexus_init):
self._cisco_nexus_plugin = cisco_nexus_plugin_v2.NexusPlugin()
- def test_create_delete_network(self):
+ def test_a_create_delete_network(self):
"""
Tests creation of two new Virtual Network.
Tests deletion of one Virtual Network.
test_nexus_clear_vlan after this test to clean
up the second vlan created by this test.
"""
- LOG.debug("test_create_delete_network - START")
tenant_id = self.tenant_id
net_name = self.net_name
net_id = self.net_id
self.second_vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.second_vlan_id)
- netid = self._cisco_nexus_plugin.delete_network(
+ expected_net_id = self._cisco_nexus_plugin.delete_network(
tenant_id, net_id, vlan_id=str(vlan_id))
- self.assertEqual(netid, net_id)
-
- LOG.debug("test_create_delete_network - END")
+ self.assertEqual(expected_net_id, net_id)
- def test_nexus_clear_vlan(self):
+ def test_b_nexus_clear_vlan(self):
"""
Test to clean up second vlan of nexus device
created by test_create_delete_network. This
test will fail if it is run individually.
"""
- LOG.debug("test_nexus_clear_vlan - START")
tenant_id = self.tenant_id
second_net_id = self.second_net_id
second_vlan_id = self.second_vlan_id
+ expected_second_net_id = self._cisco_nexus_plugin.delete_network(
+ tenant_id, second_net_id,
+ vlan_id=str(second_vlan_id))
+
+ self.assertEqual(expected_second_net_id, second_net_id)
+
+ def test_c_update_network_False(self):
+ """
+ Test to update a network state to False
+ resulting in disabling a vlan corresponding to
+ that network from the configured nexus interfaces
+ """
+ tenant_id = self.tenant_id
+ net_name = self.net_name
+ net_id = self.net_id
+ vlan_name = self.vlan_name
+ vlan_id = self.vlan_id
+ second_net_name = self.second_net_name
+ second_net_id = self.second_net_id
+ second_vlan_name = self.second_vlan_name
+ second_vlan_id = self.second_vlan_id
+
+ new_net_dict = self._cisco_nexus_plugin.create_network(
+ tenant_id, net_name, net_id,
+ vlan_name, vlan_id, vlan_ids=str(vlan_id))
+
+ vlan_ids = str(vlan_id) + "," + str(second_vlan_id)
+ new_net_dict = self._cisco_nexus_plugin.create_network(
+ tenant_id, second_net_name, second_net_id,
+ second_vlan_name, second_vlan_id,
+ vlan_ids=vlan_ids)
+
+ expected_net_id = self._cisco_nexus_plugin.update_network(
+ tenant_id, net_id, net_admin_state=False,
+ vlan_id=vlan_id, vlan_ids=str(vlan_id))
+
+ self.assertEqual(expected_net_id, net_id)
+
+ def test_d_nexus_clean_vlan_update(self):
+ """
+ Cleans up vlans on the nexus for the two
+ created networks
+ """
+ tenant_id = self.tenant_id
+ net_id = self.net_id
+ vlan_id = self.vlan_id
+ second_net_id = self.second_net_id
+ second_vlan_id = self.second_vlan_id
+
netid = self._cisco_nexus_plugin.delete_network(
+ tenant_id, net_id, vlan_id=str(vlan_id))
+
+ self.assertEqual(netid, net_id)
+
+ expected_second_net_id = self._cisco_nexus_plugin.delete_network(
tenant_id, second_net_id,
vlan_id=str(second_vlan_id))
- self.assertEqual(netid, second_net_id)
+ self.assertEqual(expected_second_net_id, second_net_id)
+
+ def test_e_update_network_True(self):
+ """
+ Test to update a disabled network state to True
+ resulting in enabling a vlan corresponding to
+ that network to the configured nexus interfaces
+ """
+ tenant_id = self.tenant_id
+ net_name = self.net_name
+ net_id = self.net_id
+ vlan_name = self.vlan_name
+ vlan_id = self.vlan_id
+ second_vlan_id = self.second_vlan_id
+
+ self.test_c_update_network_False()
+
+ vlan_ids = str(vlan_id) + "," + str(second_vlan_id)
+ expected_net_id = self._cisco_nexus_plugin.update_network(
+ tenant_id, net_id, net_admin_state=True,
+ vlan_id=vlan_id, vlan_ids=str(vlan_ids))
+
+ self.assertEqual(expected_net_id, net_id)
- LOG.debug("test_nexus_clear_vlan - END")
+ self.test_d_nexus_clean_vlan_update()
def tearDown(self):
"""Clear the test environment"""
Code Review / openstack-build / neutron-build.git / commitdiff