SEGMENTS = 'segments'
SEGMENT = 'segment'
BRIDGE_DOMAIN_SUFFIX = '_bd'
+LOGICAL_NETWORK_SUFFIX = '_log_net'
ENCAPSULATION_PROFILE_SUFFIX = '_profile'
UUID_LENGTH = 36
kwargs["multicast_ip_range"] = network_profile[
"multicast_ip_range"]
kwargs["segment_range"] = network_profile["segment_range"]
+ kwargs["sub_type"] = network_profile["sub_type"]
elif network_profile["segment_type"] == c_const.NETWORK_TYPE_TRUNK:
kwargs["sub_type"] = network_profile["sub_type"]
net_profile = n1kv_models_v2.NetworkProfile(**kwargs)
EXTENDED_ATTRIBUTES_2_0 = {
'networks': {
- PROFILE_ID: {'allow_post': True, 'allow_put': True,
+ PROFILE_ID: {'allow_post': True, 'allow_put': False,
'validate': {'type:regex': attributes.UUID_PATTERN},
'default': attributes.ATTR_NOT_SPECIFIED,
'is_visible': True},
'is_visible': True},
},
'ports': {
- PROFILE_ID: {'allow_post': True, 'allow_put': True,
+ PROFILE_ID: {'allow_post': True, 'allow_put': False,
'validate': {'type:regex': attributes.UUID_PATTERN},
'default': attributes.ATTR_NOT_SPECIFIED,
'is_visible': True}
'is_visible': True},
'name': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': ''},
- 'segment_type': {'allow_post': True, 'allow_put': True,
+ 'segment_type': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': ''},
- 'sub_type': {'allow_post': True, 'allow_put': True,
+ 'sub_type': {'allow_post': True, 'allow_put': False,
'is_visible': True,
'default': attributes.ATTR_NOT_SPECIFIED},
'segment_range': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': ''},
- 'sub_type': {'allow_post': True, 'allow_put': True,
- 'is_visible': True, 'default': ''},
'multicast_ip_range': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': '0.0.0.0'},
'multicast_ip_index': {'allow_post': False, 'allow_put': False,
'is_visible': False, 'default': '0'},
- 'physical_network': {'allow_post': True, 'allow_put': True,
+ 'physical_network': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': ''},
'tenant_id': {'allow_post': True, 'allow_put': False,
'is_visible': False, 'default': ''},
# Define paths for the URI where the client connects for HTTP requests.
port_profiles_path = "/virtual-port-profile"
- network_segments_path = "/network-segment"
network_segment_path = "/network-segment/%s"
- network_segment_pools_path = "/network-segment-pool"
network_segment_pool_path = "/network-segment-pool/%s"
- ip_pools_path = "/ip-pool-template"
ip_pool_path = "/ip-pool-template/%s"
ports_path = "/kvm/vm-network/%s/ports"
port_path = "/kvm/vm-network/%s/ports/%s"
vm_network_path = "/kvm/vm-network/%s"
bridge_domains_path = "/kvm/bridge-domain"
bridge_domain_path = "/kvm/bridge-domain/%s"
- logical_networks_path = "/logical-network"
logical_network_path = "/logical-network/%s"
events_path = "/kvm/events"
clusters_path = "/cluster"
:param network: network dict
:param overlay_subtype: string representing subtype of overlay network
"""
- body = {'name': network['name'] + c_const.BRIDGE_DOMAIN_SUFFIX,
+ body = {'name': network['id'] + c_const.BRIDGE_DOMAIN_SUFFIX,
'segmentId': network[providernet.SEGMENTATION_ID],
- 'subType': overlay_subtype}
+ 'subType': overlay_subtype,
+ 'tenantId': network['tenant_id']}
if overlay_subtype == c_const.NETWORK_SUBTYPE_NATIVE_VXLAN:
body['groupIp'] = network[n1kv_profile.MULTICAST_IP]
return self._post(self.bridge_domains_path,
:param name: name of the bridge domain to be deleted
"""
- return self._delete(self.bridge_domain_path % (name))
+ return self._delete(self.bridge_domain_path % name)
def create_network_segment(self, network, network_profile):
"""
:param network: network dict
:param network_profile: network profile dict
"""
- body = {'name': network['name'],
+ body = {'publishName': network['name'],
+ 'description': network['name'],
'id': network['id'],
- 'mode': 'access',
- 'networkSegmentPool': network_profile['name'], }
+ 'tenantId': network['tenant_id'],
+ 'networkSegmentPool': network_profile['id'], }
if network[providernet.NETWORK_TYPE] == c_const.NETWORK_TYPE_VLAN:
body['vlan'] = network[providernet.SEGMENTATION_ID]
elif network[providernet.NETWORK_TYPE] == c_const.NETWORK_TYPE_OVERLAY:
- body['bridgeDomain'] = (network['name'] +
+ body['bridgeDomain'] = (network['id'] +
c_const.BRIDGE_DOMAIN_SUFFIX)
if network_profile['segment_type'] == c_const.NETWORK_TYPE_TRUNK:
body['mode'] = c_const.NETWORK_TYPE_TRUNK
body['addSegments'] = network['add_segment_list']
body['delSegments'] = network['del_segment_list']
else:
- body['encapProfile'] = (network['name'] +
+ body['encapProfile'] = (network['id'] +
c_const.ENCAPSULATION_PROFILE_SUFFIX)
else:
body['mode'] = 'access'
body['segmentType'] = network_profile['segment_type']
- return self._post(self.network_segments_path,
+ return self._post(self.network_segment_path % network['id'],
body=body)
- def update_network_segment(self, network_segment_name, body):
+ def update_network_segment(self, network_segment_id, body):
"""
Update a network segment on the VSM.
Network segment on VSM can be updated to associate it with an ip-pool
or update its description and segment id.
- :param network_segment_name: name of the network segment
+ :param network_segment_id: UUID representing the network segment
:param body: dict of arguments to be updated
"""
- return self._post(self.network_segment_path % (network_segment_name),
+ return self._post(self.network_segment_path % network_segment_id,
body=body)
- def delete_network_segment(self, network_segment_name):
+ def delete_network_segment(self, network_segment_id):
"""
Delete a network segment on the VSM.
- :param network_segment_name: name of the network segment
+ :param network_segment_id: UUID representing the network segment
"""
- return self._delete(self.network_segment_path % (network_segment_name))
+ return self._delete(self.network_segment_path % network_segment_id)
- def create_logical_network(self, network_profile):
+ def create_logical_network(self, network_profile, tenant_id):
"""
Create a logical network on the VSM.
:param network_profile: network profile dict
+ :param tenant_id: UUID representing the tenant
"""
LOG.debug(_("Logical network"))
- body = {'name': network_profile['name']}
- return self._post(self.logical_networks_path,
+ body = {'description': network_profile['name'],
+ 'tenantId': tenant_id}
+ logical_network_name = (network_profile['id'] +
+ c_const.LOGICAL_NETWORK_SUFFIX)
+ return self._post(self.logical_network_path % logical_network_name,
body=body)
- def delete_logical_network(self, network_profile):
+ def delete_logical_network(self, logical_network_name):
"""
Delete a logical network on VSM.
- :param network_profile: network profile dict
+ :param logical_network_name: string representing name of the logical
+ network
"""
return self._delete(
- self.logical_network_path % (network_profile['name']))
+ self.logical_network_path % logical_network_name)
- def create_network_segment_pool(self, network_profile):
+ def create_network_segment_pool(self, network_profile, tenant_id):
"""
Create a network segment pool on the VSM.
:param network_profile: network profile dict
+ :param tenant_id: UUID representing the tenant
"""
LOG.debug(_("network_segment_pool"))
+ logical_network_name = (network_profile['id'] +
+ c_const.LOGICAL_NETWORK_SUFFIX)
body = {'name': network_profile['name'],
+ 'description': network_profile['name'],
'id': network_profile['id'],
- 'logicalNetwork': network_profile['name']}
- return self._post(self.network_segment_pools_path,
- body=body)
+ 'logicalNetwork': logical_network_name,
+ 'tenantId': tenant_id}
+ return self._post(
+ self.network_segment_pool_path % network_profile['id'],
+ body=body)
- def update_network_segment_pool(self, network_segment_pool, body):
+ def update_network_segment_pool(self, network_profile):
"""
Update a network segment pool on the VSM.
- :param network_segment_pool: string representing the name of network
- segment pool to be updated
- :param body: dictionary representing key values of network segment
- pool which need to be updated
+ :param network_profile: network profile dict
"""
+ body = {'name': network_profile['name'],
+ 'description': network_profile['name']}
return self._post(self.network_segment_pool_path %
- (network_segment_pool), body=body)
+ network_profile['id'], body=body)
- def delete_network_segment_pool(self, network_segment_pool_name):
+ def delete_network_segment_pool(self, network_segment_pool_id):
"""
Delete a network segment pool on the VSM.
- :param network_segment_pool_name: name of the network segment pool
+ :param network_segment_pool_id: UUID representing the network
+ segment pool
"""
return self._delete(self.network_segment_pool_path %
- (network_segment_pool_name))
+ network_segment_pool_id)
def create_ip_pool(self, subnet):
"""
body = {'addressRangeStart': address_range_start,
'addressRangeEnd': address_range_end,
'ipAddressSubnet': netmask,
- 'name': subnet['name'],
+ 'description': subnet['name'],
'gateway': subnet['gateway_ip'],
- 'networkAddress': network_address}
- return self._post(self.ip_pools_path,
+ 'networkAddress': network_address,
+ 'tenantId': subnet['tenant_id']}
+ return self._post(self.ip_pool_path % subnet['id'],
body=body)
- def delete_ip_pool(self, subnet_name):
+ def delete_ip_pool(self, subnet_id):
"""
Delete an ip-pool on the VSM.
- :param subnet_name: name of the subnet
+ :param subnet_id: UUID representing the subnet
"""
- return self._delete(self.ip_pool_path % (subnet_name))
+ return self._delete(self.ip_pool_path % subnet_id)
def create_vm_network(self,
port,
vm_network_name,
- policy_profile,
- network_name):
+ policy_profile):
"""
Create a VM network on the VSM.
:param port: port dict
:param vm_network_name: name of the VM network
:param policy_profile: policy profile dict
- :param network_name: string representing the name of the network
"""
body = {'name': vm_network_name,
'networkSegmentId': port['network_id'],
- 'networkSegment': network_name,
+ 'networkSegment': port['network_id'],
'portProfile': policy_profile['name'],
'portProfileId': policy_profile['id'],
+ 'tenantId': port['tenant_id'],
}
return self._post(self.vm_networks_path,
body=body)
:param vm_network_name: name of the VM network
"""
- return self._delete(self.vm_network_path % (vm_network_name))
+ return self._delete(self.vm_network_path % vm_network_name)
def create_n1kv_port(self, port, vm_network_name):
"""
"""
body = {'id': port['id'],
'macAddress': port['mac_address']}
- return self._post(self.ports_path % (vm_network_name),
+ if port.get('fixed_ips'):
+ body['ipAddress'] = port['fixed_ips'][0]['ip_address']
+ return self._post(self.ports_path % vm_network_name,
body=body)
def update_n1kv_port(self, vm_network_name, port_id, body):
:param port_id: UUID of the port
:param body: dict of the arguments to be updated
"""
- return self._post(self.port_path % ((vm_network_name), (port_id)),
+ return self._post(self.port_path % (vm_network_name, port_id),
body=body)
def delete_n1kv_port(self, vm_network_name, port_id):
:param vm_network_name: name of the VM network which imports this port
:param port_id: UUID of the port
"""
- return self._delete(self.port_path % ((vm_network_name), (port_id)))
+ return self._delete(self.port_path % (vm_network_name, port_id))
def _do_request(self, method, action, body=None,
headers=None):
"""
if not format:
format = self.format
- return "application/%s" % (format)
+ return "application/%s" % format
def _delete(self, action, body=None, headers=None):
return self._do_request("DELETE", action, body=body,
:param body: mapping dictionary
"""
return self._post(self.encap_profile_path
- % (profile_name), body=body)
+ % profile_name, body=body)
def delete_encapsulation_profile(self, name):
"""
:param name: name of the encapsulation profile to be deleted
"""
- return self._delete(self.encap_profile_path % (name))
+ return self._delete(self.encap_profile_path % name)
return profile_id
- def _send_create_logical_network_request(self, network_profile):
+ def _send_create_logical_network_request(self, network_profile, tenant_id):
"""
Send create logical network request to VSM.
:param network_profile: network profile dictionary
+ :param tenant_id: UUID representing the tenant
"""
LOG.debug(_('_send_create_logical_network'))
n1kvclient = n1kv_client.Client()
- n1kvclient.create_logical_network(network_profile)
+ n1kvclient.create_logical_network(network_profile, tenant_id)
def _send_delete_logical_network_request(self, network_profile):
"""
"""
LOG.debug('_send_delete_logical_network')
n1kvclient = n1kv_client.Client()
- n1kvclient.delete_logical_network(network_profile)
+ logical_network_name = (network_profile['id'] +
+ c_const.LOGICAL_NETWORK_SUFFIX)
+ n1kvclient.delete_logical_network(logical_network_name)
def _send_create_network_profile_request(self, context, profile):
"""
"""
LOG.debug(_('_send_create_network_profile_request: %s'), profile['id'])
n1kvclient = n1kv_client.Client()
- n1kvclient.create_network_segment_pool(profile)
+ n1kvclient.create_network_segment_pool(profile, context.tenant_id)
+
+ def _send_update_network_profile_request(self, profile):
+ """
+ Send update network profile request to VSM.
+
+ :param profile: network profile dictionary
+ """
+ LOG.debug(_('_send_update_network_profile_request: %s'), profile['id'])
+ n1kvclient = n1kv_client.Client()
+ n1kvclient.update_network_segment_pool(profile)
def _send_delete_network_profile_request(self, profile):
"""
LOG.debug(_('_send_delete_network_profile_request: %s'),
profile['name'])
n1kvclient = n1kv_client.Client()
- n1kvclient.delete_network_segment_pool(profile['name'])
+ n1kvclient.delete_network_segment_pool(profile['id'])
def _send_create_network_request(self, context, network, segment_pairs):
"""
profile = n1kv_db_v2.get_network_profile(
db_session, network[n1kv_profile.PROFILE_ID])
n1kvclient = n1kv_client.Client()
- body = {'name': network['name'],
+ body = {'publishName': network['name'],
'id': network['id'],
- 'networkDefinition': profile['name'],
+ 'networkSegmentPool': profile['id'],
'vlan': network[providernet.SEGMENTATION_ID],
'mode': 'access',
'segmentType': profile['segment_type'],
LOG.debug(_('add_segments=%s'), body['addSegments'])
LOG.debug(_('del_segments=%s'), body['delSegments'])
if profile['sub_type'] == c_const.NETWORK_TYPE_OVERLAY:
- encap_profile = (network['name'] +
+ encap_profile = (network['id'] +
c_const.ENCAPSULATION_PROFILE_SUFFIX)
encap_dict = {'name': encap_profile,
'addMappings': (
del_segments))}
n1kvclient.update_encapsulation_profile(context, encap_profile,
encap_dict)
- n1kvclient.update_network_segment(network['name'], body)
+ n1kvclient.update_network_segment(network['id'], body)
def _send_delete_network_request(self, context, network):
"""
n1kvclient = n1kv_client.Client()
session = context.session
if network[providernet.NETWORK_TYPE] == c_const.NETWORK_TYPE_OVERLAY:
- name = network['name'] + c_const.BRIDGE_DOMAIN_SUFFIX
+ name = network['id'] + c_const.BRIDGE_DOMAIN_SUFFIX
n1kvclient.delete_bridge_domain(name)
elif network[providernet.NETWORK_TYPE] == c_const.NETWORK_TYPE_TRUNK:
profile = self.get_network_profile(
context, network[n1kv_profile.PROFILE_ID])
if profile['sub_type'] == c_const.NETWORK_TYPE_OVERLAY:
- profile_name = (network['name'] +
+ profile_name = (network['id'] +
c_const.ENCAPSULATION_PROFILE_SUFFIX)
n1kvclient.delete_encapsulation_profile(profile_name)
elif (network[providernet.NETWORK_TYPE] ==
profile_dict['delSegments'].append(mapping_dict)
n1kvclient.update_encapsulation_profile(context, profile,
profile_dict)
- n1kvclient.delete_network_segment(network['name'])
+ n1kvclient.delete_network_segment(network['id'])
def _send_create_subnet_request(self, context, subnet):
"""
:param subnet: subnet dictionary
"""
LOG.debug(_('_send_create_subnet_request: %s'), subnet['id'])
- network = self.get_network(context, subnet['network_id'])
n1kvclient = n1kv_client.Client()
n1kvclient.create_ip_pool(subnet)
- body = {'ipPoolName': subnet['name']}
- n1kvclient.update_network_segment(network['name'], body=body)
+ body = {'ipPool': subnet['id']}
+ n1kvclient.update_network_segment(subnet['network_id'], body=body)
def _send_delete_subnet_request(self, context, subnet):
"""
:param subnet: subnet dictionary
"""
LOG.debug(_('_send_delete_subnet_request: %s'), subnet['name'])
- network = self.get_network(context, subnet['network_id'])
- body = {'ipPoolName': subnet['name'], 'deleteSubnet': True}
+ body = {'ipPool': subnet['id'], 'deleteSubnet': True}
n1kvclient = n1kv_client.Client()
- n1kvclient.update_network_segment(network['name'], body=body)
- n1kvclient.delete_ip_pool(subnet['name'])
+ n1kvclient.update_network_segment(subnet['network_id'], body=body)
+ n1kvclient.delete_ip_pool(subnet['id'])
def _send_create_port_request(self, context, port):
"""
except cisco_exceptions.VMNetworkNotFound:
policy_profile = n1kv_db_v2.get_policy_profile(
context.session, port[n1kv_profile.PROFILE_ID])
- network = self.get_network(context, port['network_id'])
vm_network_name = (c_const.VM_NETWORK_NAME_PREFIX +
str(port[n1kv_profile.PROFILE_ID]) +
"_" + str(port['network_id']))
n1kvclient = n1kv_client.Client()
n1kvclient.create_vm_network(port,
vm_network_name,
- policy_profile,
- network['name'])
+ policy_profile)
n1kvclient.create_n1kv_port(port, vm_network_name)
else:
vm_network_name = vm_network['name']
(network_type, physical_network,
segmentation_id) = self._process_provider_create(context,
network['network'])
- self._add_dummy_profile_only_if_testing(network)
profile_id = self._process_network_profile(context, network['network'])
segment_pairs = None
LOG.debug(_('Create network: profile_id=%s'), profile_id)
:param port: port dictionary
:returns: port object
"""
- self._add_dummy_profile_only_if_testing(port)
-
if ('device_id' in port['port'] and port['port']['device_owner'] in
['network:dhcp', 'network:router_interface']):
p_profile_name = c_conf.CISCO_N1K.network_node_policy_profile
LOG.debug(_("Created port: %s"), pt)
return pt
- def _add_dummy_profile_only_if_testing(self, obj):
- """
- Method to be patched by the test_n1kv_plugin module to
- inject n1kv:profile_id into the network/port object, since the plugin
- tests for its existence. This method does not affect
- the plugin code in any way.
- """
- pass
-
def update_port(self, context, id, port):
"""
Update port parameters.
n1kv_db_v2.sync_vxlan_allocations(context.session,
self.vxlan_id_ranges)
try:
- self._send_create_logical_network_request(_network_profile)
+ self._send_create_logical_network_request(_network_profile,
+ context.tenant_id)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
super(N1kvNeutronPluginV2, self).delete_network_profile(
n1kv_db_v2.delete_vxlan_allocations(context.session,
self.delete_vxlan_ranges)
self._send_delete_network_profile_request(_network_profile)
+
+ def update_network_profile(self, context, net_profile_id, network_profile):
+ """
+ Update a network profile.
+
+ :param context: neutron api request context
+ :param net_profile_id: UUID of the network profile to update
+ :param network_profile: dictionary containing network profile object
+ """
+ session = context.session
+ with session.begin(subtransactions=True):
+ net_p = (super(N1kvNeutronPluginV2, self).
+ update_network_profile(context,
+ net_profile_id,
+ network_profile))
+ self._send_update_network_profile_request(net_p)
+ return net_p
# @author: Abhishek Raut, Cisco Systems Inc.
from mock import patch
+import os
+from oslo.config import cfg
+from neutron.api.v2 import attributes
+from neutron.common.test_lib import test_config
from neutron import context
import neutron.db.api as db
from neutron.plugins.cisco.db import n1kv_db_v2
-from neutron.plugins.cisco.db import n1kv_models_v2
from neutron.plugins.cisco.db import network_db_v2 as cdb
+from neutron.plugins.cisco import extensions
from neutron.plugins.cisco.extensions import n1kv_profile
+from neutron.plugins.cisco.extensions import network_profile
from neutron.plugins.cisco.n1kv import n1kv_client
from neutron.plugins.cisco.n1kv import n1kv_neutron_plugin
from neutron.tests import base
+from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_db_plugin as test_plugin
return self.buffer
-def _fake_add_dummy_profile_for_test(self, obj):
- """
- Replacement for a function in the N1KV neutron plugin module.
-
- Since VSM is not available at the time of tests, we have no
- policy profiles. Hence we inject a dummy policy/network profile into the
- port/network object.
- """
- dummy_profile_name = "dummy_profile"
- dummy_tenant_id = "test-tenant"
- db_session = db.get_session()
- if 'port' in obj:
- dummy_profile_id = "00000000-1111-1111-1111-000000000000"
- self._add_policy_profile(dummy_profile_name,
- dummy_profile_id,
- dummy_tenant_id)
- obj['port'][n1kv_profile.PROFILE_ID] = dummy_profile_id
- elif 'network' in obj:
- profile = {'name': 'dummy_profile',
- 'segment_type': 'vlan',
- 'physical_network': 'phsy1',
- 'segment_range': '3968-4047'}
- self.network_vlan_ranges = {profile[
- 'physical_network']: [(3968, 4047)]}
- n1kv_db_v2.sync_vlan_allocations(db_session, self.network_vlan_ranges)
- np = n1kv_db_v2.create_network_profile(db_session, profile)
- obj['network'][n1kv_profile.PROFILE_ID] = np.id
-
-
def _fake_setup_vsm(self):
"""Fake establish Communication with Cisco Nexus1000V VSM."""
self.agent_vsm = True
self._poll_policies(event_type="port_profile")
+class NetworkProfileTestExtensionManager(object):
+
+ def get_resources(self):
+ # Add the resources to the global attribute map
+ # This is done here as the setup process won't
+ # initialize the main API router which extends
+ # the global attribute map
+ attributes.RESOURCE_ATTRIBUTE_MAP.update(
+ network_profile.RESOURCE_ATTRIBUTE_MAP)
+ return network_profile.Network_profile.get_resources()
+
+ def get_actions(self):
+ return []
+
+ def get_request_extensions(self):
+ return []
+
+
class N1kvPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
_plugin_name = ('neutron.plugins.cisco.n1kv.'
DEFAULT_RESP_BODY = ""
DEFAULT_RESP_CODE = 200
DEFAULT_CONTENT_TYPE = ""
+ fmt = "json"
+
+ def _make_test_policy_profile(self, name='service_profile'):
+ """
+ Create a policy profile record for testing purpose.
- def _make_test_policy_profile(self, id):
- """Create a policy profile record for testing purpose."""
- profile = {'id': id,
- 'name': 'TestGrizzlyPP'}
- profile_obj = n1kv_db_v2.create_policy_profile(profile)
- return profile_obj
-
- def _make_test_profile(self):
- """Create a profile record for testing purposes."""
- alloc_obj = n1kv_models_v2.N1kvVlanAllocation(physical_network='foo',
- vlan_id=123)
- alloc_obj.allocated = False
- segment_range = "100-900"
- segment_type = 'vlan'
- physical_network = 'phys1'
- profile_obj = n1kv_models_v2.NetworkProfile(
- name="test_np",
- segment_type=segment_type,
- segment_range=segment_range,
- physical_network=physical_network)
- session = db.get_session()
- session.add(profile_obj)
- session.flush()
- return profile_obj
+ :param name: string representing the name of the policy profile to
+ create. Default argument value chosen to correspond to the
+ default name specified in config.py file.
+ """
+ uuid = test_api_v2._uuid()
+ profile = {'id': uuid,
+ 'name': name}
+ return n1kv_db_v2.create_policy_profile(profile)
+
+ def _make_test_profile(self, name='default_network_profile'):
+ """
+ Create a profile record for testing purposes.
+
+ :param name: string representing the name of the network profile to
+ create. Default argument value chosen to correspond to the
+ default name specified in config.py file.
+ """
+ db_session = db.get_session()
+ profile = {'name': name,
+ 'segment_type': 'vlan',
+ 'physical_network': 'phsy1',
+ 'segment_range': '3968-4047'}
+ self.network_vlan_ranges = {profile[
+ 'physical_network']: [(3968, 4047)]}
+ n1kv_db_v2.sync_vlan_allocations(db_session, self.network_vlan_ranges)
+ return n1kv_db_v2.create_network_profile(db_session, profile)
def setUp(self):
"""
fake_get_cred_name.return_value = {"user_name": "admin",
"password": "admin_password"}
- # Patch a dummy profile creation into the N1K plugin code. The original
- # function in the plugin is a noop for production, but during test, we
- # need it to return a dummy network profile.
- (n1kv_neutron_plugin.N1kvNeutronPluginV2.
- _add_dummy_profile_only_if_testing) = _fake_add_dummy_profile_for_test
-
n1kv_neutron_plugin.N1kvNeutronPluginV2._setup_vsm = _fake_setup_vsm
+ test_config['plugin_name_v2'] = self._plugin_name
+ cfg.CONF.set_override('api_extensions_path',
+ os.path.dirname(extensions.__file__))
+ self.addCleanup(cfg.CONF.reset)
+ ext_mgr = NetworkProfileTestExtensionManager()
+ test_config['extension_manager'] = ext_mgr
+ self.addCleanup(self.restore_test_config)
+
+ # Save the original RESOURCE_ATTRIBUTE_MAP
+ self.saved_attr_map = {}
+ for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.items():
+ self.saved_attr_map[resource] = attrs.copy()
+ # Update the RESOURCE_ATTRIBUTE_MAP with n1kv specific extended attrs.
+ attributes.RESOURCE_ATTRIBUTE_MAP["networks"].update(
+ n1kv_profile.EXTENDED_ATTRIBUTES_2_0["networks"])
+ attributes.RESOURCE_ATTRIBUTE_MAP["ports"].update(
+ n1kv_profile.EXTENDED_ATTRIBUTES_2_0["ports"])
+ self.addCleanup(self.restore_resource_attribute_map)
+ self.addCleanup(db.clear_db)
super(N1kvPluginTestCase, self).setUp(self._plugin_name)
# Create some of the database entries that we require.
- profile_obj = self._make_test_profile()
- policy_profile_obj = (self._make_test_policy_profile(
- '41548d21-7f89-4da0-9131-3d4fd4e8BBB8'))
- # Additional args for create_network(), create_port(), etc.
- self.more_args = {
- "network": {"n1kv:profile_id": profile_obj.id},
- "port": {"n1kv:profile_id": policy_profile_obj.id}
- }
+ self._make_test_profile()
+ self._make_test_policy_profile()
+
+ def restore_resource_attribute_map(self):
+ # Restore the original RESOURCE_ATTRIBUTE_MAP
+ attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
+
+ def restore_test_config(self):
+ # Restore the original test_config
+ del test_config['plugin_name_v2']
def test_plugin(self):
self._make_network('json',
self.assertIn('tenant_id', body['networks'][0])
+class TestN1kvNetworkProfiles(N1kvPluginTestCase):
+ def _prepare_net_profile_data(self, segment_type):
+ netp = {'network_profile': {'name': 'netp1',
+ 'segment_type': segment_type,
+ 'tenant_id': self.tenant_id}}
+ if segment_type == 'vlan':
+ netp['network_profile']['segment_range'] = '100-200'
+ netp['network_profile']['physical_network'] = 'phys1'
+ elif segment_type == 'overlay':
+ netp['network_profile']['segment_range'] = '10000-10010'
+ netp['network_profile']['sub_type'] = 'enhanced'
+ return netp
+
+ def test_create_network_profile_plugin(self):
+ data = self._prepare_net_profile_data('vlan')
+ net_p_req = self.new_create_request('network_profiles', data)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 201)
+
+ def test_update_network_profile_physical_network_fail(self):
+ net_p = self._make_test_profile(name='netp1')
+ data = {'network_profile': {'physical_network': 'some-phys-net'}}
+ net_p_req = self.new_update_request('network_profiles',
+ data,
+ net_p['id'])
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_update_network_profile_segment_type_fail(self):
+ net_p = self._make_test_profile(name='netp1')
+ data = {'network_profile': {'segment_type': 'overlay'}}
+ net_p_req = self.new_update_request('network_profiles',
+ data,
+ net_p['id'])
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_update_network_profile_sub_type_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ net_p_req = self.new_create_request('network_profiles', net_p_dict)
+ net_p = self.deserialize(self.fmt,
+ net_p_req.get_response(self.ext_api))
+ data = {'network_profile': {'sub_type': 'vlan'}}
+ update_req = self.new_update_request('network_profiles',
+ data,
+ net_p['network_profile']['id'])
+ update_res = update_req.get_response(self.ext_api)
+ self.assertEqual(update_res.status_int, 400)
+
+
class TestN1kvBasicGet(test_plugin.TestBasicGet,
N1kvPluginTestCase):
class TestN1kvPorts(test_plugin.TestPortsV2,
N1kvPluginTestCase):
- def _make_other_tenant_profile(self):
- """Underlying test uses other tenant Id for tests."""
- profile_obj = self._make_test_profile()
- policy_profile_obj = self._make_test_policy_profile(
- '41548d21-7f89-4da0-9131-3d4fd4e8BBB9')
- self.more_args = {
- "network": {"n1kv:profile_id": profile_obj.id},
- "port": {"n1kv:profile_id": policy_profile_obj.id}
- }
-
- def test_create_port_public_network(self):
- # The underlying test function needs a profile for a different tenant.
- self._make_other_tenant_profile()
- super(TestN1kvPorts, self).test_create_port_public_network()
-
- def test_create_port_public_network_with_ip(self):
- # The underlying test function needs a profile for a different tenant.
- self._make_other_tenant_profile()
- super(TestN1kvPorts, self).test_create_port_public_network_with_ip()
-
- def test_create_ports_bulk_emulated(self):
- # The underlying test function needs a profile for a different tenant.
- self._make_other_tenant_profile()
- super(TestN1kvPorts,
- self).test_create_ports_bulk_emulated()
-
- def test_create_ports_bulk_emulated_plugin_failure(self):
- # The underlying test function needs a profile for a different tenant.
- self._make_other_tenant_profile()
- super(TestN1kvPorts,
- self).test_create_ports_bulk_emulated_plugin_failure()
-
- def test_delete_port_public_network(self):
- self._make_other_tenant_profile()
- super(TestN1kvPorts, self).test_delete_port_public_network()
+ def test_create_port_with_default_n1kv_profile_id(self):
+ """Test port create without passing policy profile id."""
+ with self.port() as port:
+ db_session = db.get_session()
+ pp = n1kv_db_v2.get_policy_profile(
+ db_session, port['port'][n1kv_profile.PROFILE_ID])
+ self.assertEqual(pp['name'], 'service_profile')
+
+ def test_create_port_with_n1kv_profile_id(self):
+ """Test port create with policy profile id."""
+ profile_obj = self._make_test_policy_profile(name='test_profile')
+ with self.network() as network:
+ data = {'port': {n1kv_profile.PROFILE_ID: profile_obj.id,
+ 'tenant_id': self.tenant_id,
+ 'network_id': network['network']['id']}}
+ port_req = self.new_create_request('ports', data)
+ port = self.deserialize(self.fmt,
+ port_req.get_response(self.api))
+ self.assertEqual(port['port'][n1kv_profile.PROFILE_ID],
+ profile_obj.id)
+ self._delete('ports', port['port']['id'])
+
+ def test_update_port_with_n1kv_profile_id(self):
+ """Test port update failure while updating policy profile id."""
+ with self.port() as port:
+ data = {'port': {n1kv_profile.PROFILE_ID: 'some-profile-uuid'}}
+ port_req = self.new_update_request('ports',
+ data,
+ port['port']['id'])
+ res = port_req.get_response(self.api)
+ # Port update should fail to update policy profile id.
+ self.assertEqual(res.status_int, 400)
class TestN1kvNetworks(test_plugin.TestNetworksV2,
N1kvPluginTestCase):
- _default_tenant = "somebody_else" # Tenant-id determined by underlying
- # DB-plugin test cases. Need to use this
- # one for profile creation
-
- def test_update_network_set_not_shared_single_tenant(self):
- # The underlying test function needs a profile for a different tenant.
- profile_obj = self._make_test_profile()
- policy_profile_obj = self._make_test_policy_profile(
- '41548d21-7f89-4da0-9131-3d4fd4e8BBB9')
- self.more_args = {
- "network": {"n1kv:profile_id": profile_obj.id},
- "port": {"n1kv:profile_id": policy_profile_obj.id}
- }
- super(TestN1kvNetworks,
- self).test_update_network_set_not_shared_single_tenant()
+ def _prepare_net_data(self, net_profile_id):
+ return {'network': {'name': 'net1',
+ n1kv_profile.PROFILE_ID: net_profile_id,
+ 'tenant_id': self.tenant_id}}
+
+ def test_create_network_with_default_n1kv_profile_id(self):
+ """Test network create without passing network profile id."""
+ with self.network() as network:
+ db_session = db.get_session()
+ np = n1kv_db_v2.get_network_profile(
+ db_session, network['network'][n1kv_profile.PROFILE_ID])
+ self.assertEqual(np['name'], 'default_network_profile')
+
+ def test_create_network_with_n1kv_profile_id(self):
+ """Test network create with network profile id."""
+ profile_obj = self._make_test_profile(name='test_profile')
+ data = self._prepare_net_data(profile_obj.id)
+ network_req = self.new_create_request('networks', data)
+ network = self.deserialize(self.fmt,
+ network_req.get_response(self.api))
+ self.assertEqual(network['network'][n1kv_profile.PROFILE_ID],
+ profile_obj.id)
+
+ def test_update_network_with_n1kv_profile_id(self):
+ """Test network update failure while updating network profile id."""
+ with self.network() as network:
+ data = {'network': {n1kv_profile.PROFILE_ID: 'some-profile-uuid'}}
+ network_req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ res = network_req.get_response(self.api)
+ # Network update should fail to update network profile id.
+ self.assertEqual(res.status_int, 400)
class TestN1kvNonDbTest(base.BaseTestCase):
Code Review / openstack-build / neutron-build.git / commitdiff