[ml2_cisco_apic]
-# Hostname for the APIC controller
-# apic_host=1.1.1.1
+# Hostname:port list of APIC controllers
+# apic_hosts=1.1.1.1:80, 1.1.1.2:8080, 1.1.1.3:80
# Username for the APIC controller
# apic_username=user
# Password for the APIC controller
# apic_password=password
-# Port for the APIC Controller
-# apic_port=80
-
# Names for APIC objects used by Neutron
# Note: When deploying multiple clouds against one APIC,
# these names must be unique between the clouds.
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# @author: Henry Gessau, Cisco Systems
-
-import collections
-import time
-
-import requests
-import requests.exceptions
-
-from neutron.openstack.common import jsonutils
-from neutron.openstack.common import log as logging
-from neutron.plugins.ml2.drivers.cisco.apic import exceptions as cexc
-
-
-LOG = logging.getLogger(__name__)
-
-APIC_CODE_FORBIDDEN = str(requests.codes.forbidden)
-
-
-# Info about a Managed Object's relative name (RN) and container.
-class ManagedObjectName(collections.namedtuple(
- 'MoPath', ['container', 'rn_fmt', 'can_create'])):
- def __new__(cls, container, rn_fmt, can_create=True):
- return super(ManagedObjectName, cls).__new__(cls, container, rn_fmt,
- can_create)
-
-
-class ManagedObjectClass(object):
-
- """Information about a Managed Object (MO) class.
-
- Constructs and keeps track of the distinguished name (DN) and relative
- name (RN) of a managed object (MO) class. The DN is the RN of the MO
- appended to the recursive RNs of its containers, i.e.:
- DN = uni/container-RN/.../container-RN/object-RN
-
- Also keeps track of whether the MO can be created in the APIC, as some
- MOs are read-only or used for specifying relationships.
- """
-
- supported_mos = {
- 'fvTenant': ManagedObjectName(None, 'tn-%s'),
- 'fvBD': ManagedObjectName('fvTenant', 'BD-%s'),
- 'fvRsBd': ManagedObjectName('fvAEPg', 'rsbd'),
- 'fvSubnet': ManagedObjectName('fvBD', 'subnet-[%s]'),
- 'fvCtx': ManagedObjectName('fvTenant', 'ctx-%s'),
- 'fvRsCtx': ManagedObjectName('fvBD', 'rsctx'),
- 'fvAp': ManagedObjectName('fvTenant', 'ap-%s'),
- 'fvAEPg': ManagedObjectName('fvAp', 'epg-%s'),
- 'fvRsProv': ManagedObjectName('fvAEPg', 'rsprov-%s'),
- 'fvRsCons': ManagedObjectName('fvAEPg', 'rscons-%s'),
- 'fvRsConsIf': ManagedObjectName('fvAEPg', 'rsconsif-%s'),
- 'fvRsDomAtt': ManagedObjectName('fvAEPg', 'rsdomAtt-[%s]'),
- 'fvRsPathAtt': ManagedObjectName('fvAEPg', 'rspathAtt-[%s]'),
-
- 'vzBrCP': ManagedObjectName('fvTenant', 'brc-%s'),
- 'vzSubj': ManagedObjectName('vzBrCP', 'subj-%s'),
- 'vzFilter': ManagedObjectName('fvTenant', 'flt-%s'),
- 'vzRsFiltAtt': ManagedObjectName('vzSubj', 'rsfiltAtt-%s'),
- 'vzEntry': ManagedObjectName('vzFilter', 'e-%s'),
- 'vzInTerm': ManagedObjectName('vzSubj', 'intmnl'),
- 'vzRsFiltAtt__In': ManagedObjectName('vzInTerm', 'rsfiltAtt-%s'),
- 'vzOutTerm': ManagedObjectName('vzSubj', 'outtmnl'),
- 'vzRsFiltAtt__Out': ManagedObjectName('vzOutTerm', 'rsfiltAtt-%s'),
- 'vzCPIf': ManagedObjectName('fvTenant', 'cif-%s'),
- 'vzRsIf': ManagedObjectName('vzCPIf', 'rsif'),
-
- 'vmmProvP': ManagedObjectName(None, 'vmmp-%s', False),
- 'vmmDomP': ManagedObjectName('vmmProvP', 'dom-%s'),
- 'vmmEpPD': ManagedObjectName('vmmDomP', 'eppd-[%s]'),
-
- 'physDomP': ManagedObjectName(None, 'phys-%s'),
-
- 'infra': ManagedObjectName(None, 'infra'),
- 'infraNodeP': ManagedObjectName('infra', 'nprof-%s'),
- 'infraLeafS': ManagedObjectName('infraNodeP', 'leaves-%s-typ-%s'),
- 'infraNodeBlk': ManagedObjectName('infraLeafS', 'nodeblk-%s'),
- 'infraRsAccPortP': ManagedObjectName('infraNodeP', 'rsaccPortP-[%s]'),
- 'infraAccPortP': ManagedObjectName('infra', 'accportprof-%s'),
- 'infraHPortS': ManagedObjectName('infraAccPortP', 'hports-%s-typ-%s'),
- 'infraPortBlk': ManagedObjectName('infraHPortS', 'portblk-%s'),
- 'infraRsAccBaseGrp': ManagedObjectName('infraHPortS', 'rsaccBaseGrp'),
- 'infraFuncP': ManagedObjectName('infra', 'funcprof'),
- 'infraAccPortGrp': ManagedObjectName('infraFuncP', 'accportgrp-%s'),
- 'infraRsAttEntP': ManagedObjectName('infraAccPortGrp', 'rsattEntP'),
- 'infraAttEntityP': ManagedObjectName('infra', 'attentp-%s'),
- 'infraRsDomP': ManagedObjectName('infraAttEntityP', 'rsdomP-[%s]'),
- 'infraRsVlanNs__phys': ManagedObjectName('physDomP', 'rsvlanNs'),
- 'infraRsVlanNs__vmm': ManagedObjectName('vmmDomP', 'rsvlanNs'),
-
- 'fvnsVlanInstP': ManagedObjectName('infra', 'vlanns-%s-%s'),
- 'fvnsEncapBlk__vlan': ManagedObjectName('fvnsVlanInstP',
- 'from-%s-to-%s'),
- 'fvnsVxlanInstP': ManagedObjectName('infra', 'vxlanns-%s'),
- 'fvnsEncapBlk__vxlan': ManagedObjectName('fvnsVxlanInstP',
- 'from-%s-to-%s'),
-
- # Read-only
- 'fabricTopology': ManagedObjectName(None, 'topology', False),
- 'fabricPod': ManagedObjectName('fabricTopology', 'pod-%s', False),
- 'fabricPathEpCont': ManagedObjectName('fabricPod', 'paths-%s', False),
- 'fabricPathEp': ManagedObjectName('fabricPathEpCont', 'pathep-%s',
- False),
- }
-
- # Note(Henry): The use of a mutable default argument _inst_cache is
- # intentional. It persists for the life of MoClass to cache instances.
- # noinspection PyDefaultArgument
- def __new__(cls, mo_class, _inst_cache={}):
- """Ensure we create only one instance per mo_class."""
- try:
- return _inst_cache[mo_class]
- except KeyError:
- new_inst = super(ManagedObjectClass, cls).__new__(cls)
- new_inst.__init__(mo_class)
- _inst_cache[mo_class] = new_inst
- return new_inst
-
- def __init__(self, mo_class):
- self.klass = mo_class
- self.klass_name = mo_class.split('__')[0]
- mo = self.supported_mos[mo_class]
- self.container = mo.container
- self.rn_fmt = mo.rn_fmt
- self.dn_fmt, self.args = self._dn_fmt()
- self.arg_count = self.dn_fmt.count('%s')
- rn_has_arg = self.rn_fmt.count('%s')
- self.can_create = rn_has_arg and mo.can_create
-
- def _dn_fmt(self):
- """Build the distinguished name format using container and RN.
-
- DN = uni/container-RN/.../container-RN/object-RN
-
- Also make a list of the required name arguments.
- Note: Call this method only once at init.
- """
- arg = [self.klass] if '%s' in self.rn_fmt else []
- if self.container:
- container = ManagedObjectClass(self.container)
- dn_fmt = '%s/%s' % (container.dn_fmt, self.rn_fmt)
- args = container.args + arg
- return dn_fmt, args
- return 'uni/%s' % self.rn_fmt, arg
-
- def dn(self, *args):
- """Return the distinguished name for a managed object."""
- return self.dn_fmt % args
-
-
-class ApicSession(object):
-
- """Manages a session with the APIC."""
-
- def __init__(self, host, port, usr, pwd, ssl):
- protocol = ssl and 'https' or 'http'
- self.api_base = '%s://%s:%s/api' % (protocol, host, port)
- self.session = requests.Session()
- self.session_deadline = 0
- self.session_timeout = 0
- self.cookie = {}
-
- # Log in
- self.authentication = None
- self.username = None
- self.password = None
- if usr and pwd:
- self.login(usr, pwd)
-
- @staticmethod
- def _make_data(key, **attrs):
- """Build the body for a msg out of a key and some attributes."""
- return jsonutils.dumps({key: {'attributes': attrs}})
-
- def _api_url(self, api):
- """Create the URL for a generic API."""
- return '%s/%s.json' % (self.api_base, api)
-
- def _mo_url(self, mo, *args):
- """Create a URL for a MO lookup by DN."""
- dn = mo.dn(*args)
- return '%s/mo/%s.json' % (self.api_base, dn)
-
- def _qry_url(self, mo):
- """Create a URL for a query lookup by MO class."""
- return '%s/class/%s.json' % (self.api_base, mo.klass_name)
-
- def _check_session(self):
- """Check that we are logged in and ensure the session is active."""
- if not self.authentication:
- raise cexc.ApicSessionNotLoggedIn
- if time.time() > self.session_deadline:
- self.refresh()
-
- def _send(self, request, url, data=None, refreshed=None):
- """Send a request and process the response."""
- if data is None:
- response = request(url, cookies=self.cookie)
- else:
- response = request(url, data=data, cookies=self.cookie)
- if response is None:
- raise cexc.ApicHostNoResponse(url=url)
- # Every request refreshes the timeout
- self.session_deadline = time.time() + self.session_timeout
- if data is None:
- request_str = url
- else:
- request_str = '%s, data=%s' % (url, data)
- LOG.debug(_("data = %s"), data)
- # imdata is where the APIC returns the useful information
- imdata = response.json().get('imdata')
- LOG.debug(_("Response: %s"), imdata)
- if response.status_code != requests.codes.ok:
- try:
- err_code = imdata[0]['error']['attributes']['code']
- err_text = imdata[0]['error']['attributes']['text']
- except (IndexError, KeyError):
- err_code = '[code for APIC error not found]'
- err_text = '[text for APIC error not found]'
- # If invalid token then re-login and retry once
- if (not refreshed and err_code == APIC_CODE_FORBIDDEN and
- err_text.lower().startswith('token was invalid')):
- self.login()
- return self._send(request, url, data=data, refreshed=True)
- raise cexc.ApicResponseNotOk(request=request_str,
- status=response.status_code,
- reason=response.reason,
- err_text=err_text, err_code=err_code)
- return imdata
-
- # REST requests
-
- def get_data(self, request):
- """Retrieve generic data from the server."""
- self._check_session()
- url = self._api_url(request)
- return self._send(self.session.get, url)
-
- def get_mo(self, mo, *args):
- """Retrieve a managed object by its distinguished name."""
- self._check_session()
- url = self._mo_url(mo, *args) + '?query-target=self'
- return self._send(self.session.get, url)
-
- def list_mo(self, mo):
- """Retrieve the list of managed objects for a class."""
- self._check_session()
- url = self._qry_url(mo)
- return self._send(self.session.get, url)
-
- def post_data(self, request, data):
- """Post generic data to the server."""
- self._check_session()
- url = self._api_url(request)
- return self._send(self.session.post, url, data=data)
-
- def post_mo(self, mo, *args, **kwargs):
- """Post data for a managed object to the server."""
- self._check_session()
- url = self._mo_url(mo, *args)
- data = self._make_data(mo.klass_name, **kwargs)
- return self._send(self.session.post, url, data=data)
-
- # Session management
-
- def _save_cookie(self, request, response):
- """Save the session cookie and its expiration time."""
- imdata = response.json().get('imdata')
- if response.status_code == requests.codes.ok:
- attributes = imdata[0]['aaaLogin']['attributes']
- try:
- self.cookie = {'APIC-Cookie': attributes['token']}
- except KeyError:
- raise cexc.ApicResponseNoCookie(request=request)
- timeout = int(attributes['refreshTimeoutSeconds'])
- LOG.debug(_("APIC session will expire in %d seconds"), timeout)
- # Give ourselves a few seconds to refresh before timing out
- self.session_timeout = timeout - 5
- self.session_deadline = time.time() + self.session_timeout
- else:
- attributes = imdata[0]['error']['attributes']
- return attributes
-
- def login(self, usr=None, pwd=None):
- """Log in to controller. Save user name and authentication."""
- usr = usr or self.username
- pwd = pwd or self.password
- name_pwd = self._make_data('aaaUser', name=usr, pwd=pwd)
- url = self._api_url('aaaLogin')
- try:
- response = self.session.post(url, data=name_pwd, timeout=10.0)
- except requests.exceptions.Timeout:
- raise cexc.ApicHostNoResponse(url=url)
- attributes = self._save_cookie('aaaLogin', response)
- if response.status_code == requests.codes.ok:
- self.username = usr
- self.password = pwd
- self.authentication = attributes
- else:
- self.authentication = None
- raise cexc.ApicResponseNotOk(request=url,
- status=response.status_code,
- reason=response.reason,
- err_text=attributes['text'],
- err_code=attributes['code'])
-
- def refresh(self):
- """Called when a session has timed out or almost timed out."""
- url = self._api_url('aaaRefresh')
- response = self.session.get(url, cookies=self.cookie)
- attributes = self._save_cookie('aaaRefresh', response)
- if response.status_code == requests.codes.ok:
- # We refreshed before the session timed out.
- self.authentication = attributes
- else:
- err_code = attributes['code']
- err_text = attributes['text']
- if (err_code == APIC_CODE_FORBIDDEN and
- err_text.lower().startswith('token was invalid')):
- # This means the token timed out, so log in again.
- LOG.debug(_("APIC session timed-out, logging in again."))
- self.login()
- else:
- self.authentication = None
- raise cexc.ApicResponseNotOk(request=url,
- status=response.status_code,
- reason=response.reason,
- err_text=err_text,
- err_code=err_code)
-
- def logout(self):
- """End session with controller."""
- if not self.username:
- self.authentication = None
- if self.authentication:
- data = self._make_data('aaaUser', name=self.username)
- self.post_data('aaaLogout', data=data)
- self.authentication = None
-
-
-class ManagedObjectAccess(object):
-
- """CRUD operations on APIC Managed Objects."""
-
- def __init__(self, session, mo_class):
- self.session = session
- self.mo = ManagedObjectClass(mo_class)
-
- def _create_container(self, *args):
- """Recursively create all container objects."""
- if self.mo.container:
- container = ManagedObjectAccess(self.session, self.mo.container)
- if container.mo.can_create:
- container_args = args[0: container.mo.arg_count]
- container._create_container(*container_args)
- container.session.post_mo(container.mo, *container_args)
-
- def create(self, *args, **kwargs):
- self._create_container(*args)
- if self.mo.can_create and 'status' not in kwargs:
- kwargs['status'] = 'created'
- return self.session.post_mo(self.mo, *args, **kwargs)
-
- def _mo_attributes(self, obj_data):
- if (self.mo.klass_name in obj_data and
- 'attributes' in obj_data[self.mo.klass_name]):
- return obj_data[self.mo.klass_name]['attributes']
-
- def get(self, *args):
- """Return a dict of the MO's attributes, or None."""
- imdata = self.session.get_mo(self.mo, *args)
- if imdata:
- return self._mo_attributes(imdata[0])
-
- def list_all(self):
- imdata = self.session.list_mo(self.mo)
- return filter(None, [self._mo_attributes(obj) for obj in imdata])
-
- def list_names(self):
- return [obj['name'] for obj in self.list_all()]
-
- def update(self, *args, **kwargs):
- return self.session.post_mo(self.mo, *args, **kwargs)
-
- def delete(self, *args):
- return self.session.post_mo(self.mo, *args, status='deleted')
-
-
-class RestClient(ApicSession):
-
- """APIC REST client for OpenStack Neutron."""
-
- def __init__(self, host, port=80, usr=None, pwd=None, ssl=False):
- """Establish a session with the APIC."""
- super(RestClient, self).__init__(host, port, usr, pwd, ssl)
-
- def __getattr__(self, mo_class):
- """Add supported MOs as properties on demand."""
- if mo_class not in ManagedObjectClass.supported_mos:
- raise cexc.ApicManagedObjectNotSupported(mo_class=mo_class)
- self.__dict__[mo_class] = ManagedObjectAccess(self, mo_class)
- return self.__dict__[mo_class]
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems Inc.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# @author: Arvind Somya (asomya@cisco.com), Cisco Systems Inc.
-
-import itertools
-import uuid
-
-from oslo.config import cfg
-
-from neutron.openstack.common import excutils
-from neutron.plugins.ml2.drivers.cisco.apic import apic_client
-from neutron.plugins.ml2.drivers.cisco.apic import apic_model
-from neutron.plugins.ml2.drivers.cisco.apic import config
-from neutron.plugins.ml2.drivers.cisco.apic import exceptions as cexc
-
-AP_NAME = 'openstack'
-CONTEXT_ENFORCED = '1'
-CONTEXT_UNENFORCED = '2'
-CONTEXT_DEFAULT = 'default'
-DN_KEY = 'dn'
-PORT_DN_PATH = 'topology/pod-1/paths-%s/pathep-[eth%s]'
-SCOPE_GLOBAL = 'global'
-SCOPE_TENANT = 'tenant'
-TENANT_COMMON = 'common'
-
-
-def group_by_ranges(i):
- """Group a list of numbers into tuples representing contiguous ranges."""
- for a, b in itertools.groupby(enumerate(sorted(i)), lambda (x, y): y - x):
- b = list(b)
- yield b[0][1], b[-1][1]
-
-
-class APICManager(object):
- """Class to manage APIC translations and workflow.
-
- This class manages translation from Neutron objects to APIC
- managed objects and contains workflows to implement these
- translations.
- """
- def __init__(self):
- self.db = apic_model.ApicDbModel()
-
- apic_conf = cfg.CONF.ml2_cisco_apic
- self.switch_dict = config.create_switch_dictionary()
-
- # Connect to the APIC
- self.apic = apic_client.RestClient(
- apic_conf.apic_host,
- apic_conf.apic_port,
- apic_conf.apic_username,
- apic_conf.apic_password
- )
-
- self.port_profiles = {}
- self.vmm_domain = None
- self.phys_domain = None
- self.vlan_ns = None
- self.node_profiles = {}
- self.entity_profile = None
- self.function_profile = None
- self.clear_node_profiles = apic_conf.apic_clear_node_profiles
-
- def ensure_infra_created_on_apic(self):
- """Ensure the infrastructure is setup.
-
- Loop over the switch dictionary from the config and
- setup profiles for switches, modules and ports
- """
- # Loop over switches
- for switch in self.switch_dict:
- # Create a node profile for this switch
- self.ensure_node_profile_created_for_switch(switch)
-
- # Check if a port profile exists for this node
- ppname = self.check_infra_port_profiles(switch)
-
- # Gather port ranges for this switch
- modules = self.gather_infra_module_ports(switch)
-
- # Setup each module and port range
- for module in modules:
- profile = self.db.get_profile_for_module(switch, ppname,
- module)
- if not profile:
- # Create host port selector for this module
- hname = uuid.uuid4()
- try:
- self.apic.infraHPortS.create(ppname, hname, 'range')
- # Add relation to the function profile
- fpdn = self.function_profile[DN_KEY]
- self.apic.infraRsAccBaseGrp.create(ppname, hname,
- 'range', tDn=fpdn)
- modules[module].sort()
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- self.apic.infraHPortS.delete(ppname, hname,
- 'range')
- else:
- hname = profile.hpselc_id
-
- ranges = group_by_ranges(modules[module])
- # Add this module and ports to the profile
- for prange in ranges:
- # Check if this port block is already added to the profile
- if not self.db.get_profile_for_module_and_ports(
- switch, ppname, module, prange[0], prange[-1]):
- # Create port block for this port range
- pbname = uuid.uuid4()
- self.apic.infraPortBlk.create(ppname, hname, 'range',
- pbname, fromCard=module,
- toCard=module,
- fromPort=str(prange[0]),
- toPort=str(prange[-1]))
- # Add DB row
- self.db.add_profile_for_module_and_ports(
- switch, ppname, hname, module,
- prange[0], prange[-1])
-
- def check_infra_port_profiles(self, switch):
- """Check and create infra port profiles for a node."""
- sprofile = self.db.get_port_profile_for_node(switch)
- ppname = None
- if not sprofile:
- # Generate uuid for port profile name
- ppname = uuid.uuid4()
- try:
- # Create port profile for this switch
- pprofile = self.ensure_port_profile_created_on_apic(ppname)
- # Add port profile to node profile
- ppdn = pprofile[DN_KEY]
- self.apic.infraRsAccPortP.create(switch, ppdn)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete port profile
- self.apic.infraAccPortP.delete(ppname)
- else:
- ppname = sprofile.profile_id
-
- return ppname
-
- def gather_infra_module_ports(self, switch):
- """Build modules and ports per module dictionary."""
- ports = self.switch_dict[switch]
- # Gather common modules
- modules = {}
- for port in ports:
- module, sw_port = port.split('/')
- if module not in modules:
- modules[module] = []
- modules[module].append(int(sw_port))
-
- return modules
-
- def ensure_context_unenforced(self, tenant_id=TENANT_COMMON,
- name=CONTEXT_DEFAULT):
- """Set the specified tenant's context to unenforced."""
- ctx = self.apic.fvCtx.get(tenant_id, name)
- if not ctx:
- self.apic.fvCtx.create(tenant_id, name,
- pcEnfPref=CONTEXT_UNENFORCED)
- elif ctx['pcEnfPref'] != CONTEXT_UNENFORCED:
- self.apic.fvCtx.update(tenant_id, name,
- pcEnfPref=CONTEXT_UNENFORCED)
-
- def ensure_context_enforced(self, tenant_id=TENANT_COMMON,
- name=CONTEXT_DEFAULT):
- """Set the specified tenant's context to enforced."""
- ctx = self.apic.fvCtx.get(tenant_id, name)
- if not ctx:
- self.apic.fvCtx.create(tenant_id, name, pcEnfPref=CONTEXT_ENFORCED)
- elif ctx['pcEnfPref'] != CONTEXT_ENFORCED:
- self.apic.fvCtx.update(tenant_id, name, pcEnfPref=CONTEXT_ENFORCED)
-
- def ensure_entity_profile_created_on_apic(self, name):
- """Create the infrastructure entity profile."""
- if self.clear_node_profiles:
- self.apic.infraAttEntityP.delete(name)
- self.entity_profile = self.apic.infraAttEntityP.get(name)
- if not self.entity_profile:
- try:
- phys_dn = self.phys_domain[DN_KEY]
- self.apic.infraAttEntityP.create(name)
- # Attach phys domain to entity profile
- self.apic.infraRsDomP.create(name, phys_dn)
- self.entity_profile = self.apic.infraAttEntityP.get(name)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the created entity profile
- self.apic.infraAttEntityP.delete(name)
-
- def ensure_function_profile_created_on_apic(self, name):
- """Create the infrastructure function profile."""
- if self.clear_node_profiles:
- self.apic.infraAccPortGrp.delete(name)
- self.function_profile = self.apic.infraAccPortGrp.get(name)
- if not self.function_profile:
- try:
- self.apic.infraAccPortGrp.create(name)
- # Attach entity profile to function profile
- entp_dn = self.entity_profile[DN_KEY]
- self.apic.infraRsAttEntP.create(name, tDn=entp_dn)
- self.function_profile = self.apic.infraAccPortGrp.get(name)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the created function profile
- self.apic.infraAccPortGrp.delete(name)
-
- def ensure_node_profile_created_for_switch(self, switch_id):
- """Creates a switch node profile.
-
- Create a node profile for a switch and add a switch
- to the leaf node selector
- """
- if self.clear_node_profiles:
- self.apic.infraNodeP.delete(switch_id)
- self.db.delete_profile_for_node(switch_id)
- sobj = self.apic.infraNodeP.get(switch_id)
- if not sobj:
- try:
- # Create Node profile
- self.apic.infraNodeP.create(switch_id)
- # Create leaf selector
- lswitch_id = uuid.uuid4()
- self.apic.infraLeafS.create(switch_id, lswitch_id, 'range')
- # Add leaf nodes to the selector
- name = uuid.uuid4()
- self.apic.infraNodeBlk.create(switch_id, lswitch_id, 'range',
- name, from_=switch_id,
- to_=switch_id)
- sobj = self.apic.infraNodeP.get(switch_id)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Remove the node profile
- self.apic.infraNodeP.delete(switch_id)
-
- self.node_profiles[switch_id] = {
- 'object': sobj
- }
-
- def ensure_port_profile_created_on_apic(self, name):
- """Create a port profile."""
- try:
- self.apic.infraAccPortP.create(name)
- return self.apic.infraAccPortP.get(name)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- self.apic.infraAccPortP.delete(name)
-
- def ensure_vmm_domain_created_on_apic(self, vmm_name,
- vlan_ns=None, vxlan_ns=None):
- """Create Virtual Machine Manager domain.
-
- Creates the VMM domain on the APIC and adds a VLAN or VXLAN
- namespace to that VMM domain.
- TODO (asomya): Add VXLAN support
- """
- provider = 'VMware'
- if self.clear_node_profiles:
- self.apic.vmmDomP.delete(provider, vmm_name)
- self.vmm_domain = self.apic.vmmDomP.get(provider, vmm_name)
- if not self.vmm_domain:
- try:
- self.apic.vmmDomP.create(provider, vmm_name)
- if vlan_ns:
- vlan_ns_dn = vlan_ns[DN_KEY]
- self.apic.infraRsVlanNs__vmm.create(provider, vmm_name,
- tDn=vlan_ns_dn)
- self.vmm_domain = self.apic.vmmDomP.get(provider, vmm_name)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the VMM domain
- self.apic.vmmDomP.delete(provider, vmm_name)
-
- def ensure_phys_domain_created_on_apic(self, phys_name,
- vlan_ns=None):
- """Create Virtual Machine Manager domain.
-
- Creates the VMM domain on the APIC and adds a VLAN or VXLAN
- namespace to that VMM domain.
- TODO (asomya): Add VXLAN support
- """
- if self.clear_node_profiles:
- self.apic.physDomP.delete(phys_name)
- self.phys_domain = self.apic.physDomP.get(phys_name)
- if not self.phys_domain:
- try:
- self.apic.physDomP.create(phys_name)
- if vlan_ns:
- vlan_ns_dn = vlan_ns[DN_KEY]
- self.apic.infraRsVlanNs__phys.create(phys_name,
- tDn=vlan_ns_dn)
- self.phys_domain = self.apic.physDomP.get(phys_name)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the physical domain
- self.apic.physDomP.delete(phys_name)
-
- def ensure_vlan_ns_created_on_apic(self, name, vlan_min, vlan_max):
- """Creates a static VLAN namespace with the given vlan range."""
- ns_args = name, 'static'
- if self.clear_node_profiles:
- self.apic.fvnsVlanInstP.delete(name, 'dynamic')
- self.apic.fvnsVlanInstP.delete(*ns_args)
- self.vlan_ns = self.apic.fvnsVlanInstP.get(*ns_args)
- if not self.vlan_ns:
- try:
- self.apic.fvnsVlanInstP.create(*ns_args)
- vlan_min = 'vlan-' + vlan_min
- vlan_max = 'vlan-' + vlan_max
- ns_blk_args = name, 'static', vlan_min, vlan_max
- vlan_encap = self.apic.fvnsEncapBlk__vlan.get(*ns_blk_args)
- if not vlan_encap:
- ns_kw_args = {
- 'name': 'encap',
- 'from': vlan_min,
- 'to': vlan_max
- }
- self.apic.fvnsEncapBlk__vlan.create(*ns_blk_args,
- **ns_kw_args)
- self.vlan_ns = self.apic.fvnsVlanInstP.get(*ns_args)
- return self.vlan_ns
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the vlan namespace
- self.apic.fvnsVlanInstP.delete(*ns_args)
-
- def ensure_tenant_created_on_apic(self, tenant_id):
- """Make sure a tenant exists on the APIC."""
- if not self.apic.fvTenant.get(tenant_id):
- self.apic.fvTenant.create(tenant_id)
-
- def ensure_bd_created_on_apic(self, tenant_id, bd_id):
- """Creates a Bridge Domain on the APIC."""
- if not self.apic.fvBD.get(tenant_id, bd_id):
- try:
- self.apic.fvBD.create(tenant_id, bd_id)
- # Add default context to the BD
- self.ensure_context_enforced()
- self.apic.fvRsCtx.create(tenant_id, bd_id,
- tnFvCtxName=CONTEXT_DEFAULT)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the bridge domain
- self.apic.fvBD.delete(tenant_id, bd_id)
-
- def delete_bd_on_apic(self, tenant_id, bd_id):
- """Deletes a Bridge Domain from the APIC."""
- self.apic.fvBD.delete(tenant_id, bd_id)
-
- def ensure_subnet_created_on_apic(self, tenant_id, bd_id, gw_ip):
- """Creates a subnet on the APIC
-
- The gateway ip (gw_ip) should be specified as a CIDR
- e.g. 10.0.0.1/24
- """
- if not self.apic.fvSubnet.get(tenant_id, bd_id, gw_ip):
- self.apic.fvSubnet.create(tenant_id, bd_id, gw_ip)
-
- def ensure_filter_created_on_apic(self, tenant_id, filter_id):
- """Create a filter on the APIC."""
- if not self.apic.vzFilter.get(tenant_id, filter_id):
- self.apic.vzFilter.create(tenant_id, filter_id)
-
- def ensure_epg_created_for_network(self, tenant_id, network_id, net_name):
- """Creates an End Point Group on the APIC.
-
- Create a new EPG on the APIC for the network spcified. This information
- is also tracked in the local DB and associate the bridge domain for the
- network with the EPG created.
- """
- # Check if an EPG is already present for this network
- epg = self.db.get_epg_for_network(network_id)
- if epg:
- return epg
-
- # Create a new EPG on the APIC
- epg_uid = '-'.join([str(net_name), str(uuid.uuid4())])
- try:
- self.apic.fvAEPg.create(tenant_id, AP_NAME, epg_uid)
-
- # Add bd to EPG
- bd = self.apic.fvBD.get(tenant_id, network_id)
- bd_name = bd['name']
-
- # Create fvRsBd
- self.apic.fvRsBd.create(tenant_id, AP_NAME, epg_uid,
- tnFvBDName=bd_name)
-
- # Add EPG to physical domain
- phys_dn = self.phys_domain[DN_KEY]
- self.apic.fvRsDomAtt.create(tenant_id, AP_NAME, epg_uid, phys_dn)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete the EPG
- self.apic.fvAEPg.delete(tenant_id, AP_NAME, epg_uid)
-
- # Stick it in the DB
- epg = self.db.write_epg_for_network(network_id, epg_uid)
-
- return epg
-
- def delete_epg_for_network(self, tenant_id, network_id):
- """Deletes the EPG from the APIC and removes it from the DB."""
- # Check if an EPG is already present for this network
- epg = self.db.get_epg_for_network(network_id)
- if not epg:
- return False
-
- # Delete this epg
- self.apic.fvAEPg.delete(tenant_id, AP_NAME, epg.epg_id)
- # Remove DB row
- self.db.delete_epg(epg)
-
- def create_tenant_filter(self, tenant_id):
- """Creates a tenant filter and a generic entry under it."""
- fuuid = uuid.uuid4()
- try:
- # Create a new tenant filter
- self.apic.vzFilter.create(tenant_id, fuuid)
- # Create a new entry
- euuid = uuid.uuid4()
- self.apic.vzEntry.create(tenant_id, fuuid, euuid)
- return fuuid
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- self.apic.vzFilter.delete(tenant_id, fuuid)
-
- def set_contract_for_epg(self, tenant_id, epg_id,
- contract_id, provider=False):
- """Set the contract for an EPG.
-
- By default EPGs are consumers to a contract. Set provider flag
- for a single EPG to act as a contract provider.
- """
- if provider:
- try:
- self.apic.fvRsProv.create(tenant_id, AP_NAME,
- epg_id, contract_id)
- self.db.set_provider_contract(epg_id)
- self.make_tenant_contract_global(tenant_id)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- self.make_tenant_contract_local(tenant_id)
- self.apic.fvRsProv.delete(tenant_id, AP_NAME,
- epg_id, contract_id)
- else:
- self.apic.fvRsCons.create(tenant_id, AP_NAME, epg_id, contract_id)
-
- def delete_contract_for_epg(self, tenant_id, epg_id,
- contract_id, provider=False):
- """Delete the contract for an End Point Group.
-
- Check if the EPG was a provider and attempt to grab another contract
- consumer from the DB and set that as the new contract provider.
- """
- if provider:
- self.apic.fvRsProv.delete(tenant_id, AP_NAME, epg_id, contract_id)
- self.db.unset_provider_contract(epg_id)
- # Pick out another EPG to set as contract provider
- epg = self.db.get_an_epg(epg_id)
- self.update_contract_for_epg(tenant_id, epg.epg_id,
- contract_id, True)
- else:
- self.apic.fvRsCons.delete(tenant_id, AP_NAME, epg_id, contract_id)
-
- def update_contract_for_epg(self, tenant_id, epg_id,
- contract_id, provider=False):
- """Updates the contract for an End Point Group."""
- self.apic.fvRsCons.delete(tenant_id, AP_NAME, epg_id, contract_id)
- self.set_contract_for_epg(tenant_id, epg_id, contract_id, provider)
-
- def create_tenant_contract(self, tenant_id):
- """Creates a tenant contract.
-
- Create a tenant contract if one doesn't exist. Also create a
- subject, filter and entry and set the filters to allow all
- protocol traffic on all ports
- """
- contract = self.db.get_contract_for_tenant(tenant_id)
- if not contract:
- cuuid = uuid.uuid4()
- try:
- # Create contract
- self.apic.vzBrCP.create(tenant_id, cuuid, scope=SCOPE_TENANT)
- acontract = self.apic.vzBrCP.get(tenant_id, cuuid)
- # Create subject
- suuid = uuid.uuid4()
- self.apic.vzSubj.create(tenant_id, cuuid, suuid)
- # Create filter and entry
- tfilter = self.create_tenant_filter(tenant_id)
- # Create interm and outterm
- self.apic.vzInTerm.create(tenant_id, cuuid, suuid)
- self.apic.vzRsFiltAtt__In.create(tenant_id, cuuid,
- suuid, tfilter)
- self.apic.vzOutTerm.create(tenant_id, cuuid, suuid)
- self.apic.vzRsFiltAtt__Out.create(tenant_id, cuuid,
- suuid, tfilter)
- # Create contract interface
- iuuid = uuid.uuid4()
- self.apic.vzCPIf.create(tenant_id, iuuid)
- self.apic.vzRsIf.create(tenant_id, iuuid,
- tDn=acontract[DN_KEY])
- # Store contract in DB
- contract = self.db.write_contract_for_tenant(tenant_id,
- cuuid, tfilter)
- except (cexc.ApicResponseNotOk, KeyError):
- with excutils.save_and_reraise_exception():
- # Delete tenant contract
- self.apic.vzBrCP.delete(tenant_id, cuuid)
-
- return contract
-
- def make_tenant_contract_global(self, tenant_id):
- """Mark the tenant contract's scope to global."""
- contract = self.db.get_contract_for_tenant(tenant_id)
- self.apic.vzBrCP.update(tenant_id, contract.contract_id,
- scope=SCOPE_GLOBAL)
-
- def make_tenant_contract_local(self, tenant_id):
- """Mark the tenant contract's scope to tenant."""
- contract = self.db.get_contract_for_tenant(tenant_id)
- self.apic.vzBrCP.update(tenant_id, contract.contract_id,
- scope=SCOPE_TENANT)
-
- def ensure_path_created_for_port(self, tenant_id, network_id,
- host_id, encap, net_name):
- """Create path attribute for an End Point Group."""
- encap = 'vlan-' + str(encap)
- epg = self.ensure_epg_created_for_network(tenant_id, network_id,
- net_name)
- eid = epg.epg_id
-
- # Get attached switch and port for this host
- host_config = config.get_switch_and_port_for_host(host_id)
- if not host_config:
- raise cexc.ApicHostNotConfigured(host=host_id)
- switch, port = host_config
- pdn = PORT_DN_PATH % (switch, port)
-
- # Check if exists
- patt = self.apic.fvRsPathAtt.get(tenant_id, AP_NAME, eid, pdn)
- if not patt:
- self.apic.fvRsPathAtt.create(tenant_id, AP_NAME, eid, pdn,
- encap=encap, mode="regular",
- instrImedcy="immediate")
apic_opts = [
- cfg.StrOpt('apic_host',
- help=_("Host name or IP Address of the APIC controller")),
+ cfg.ListOpt('apic_hosts',
+ default=[],
+ help=_("An ordered list of host names or IP addresses of "
+ "the APIC controller(s).")),
cfg.StrOpt('apic_username',
help=_("Username for the APIC controller")),
cfg.StrOpt('apic_password',
help=_("Password for the APIC controller"), secret=True),
- cfg.StrOpt('apic_port',
- help=_("Communication port for the APIC controller")),
+ cfg.BoolOpt('apic_use_ssl', default=True,
+ help=_("Use SSL to connect to the APIC controller")),
cfg.StrOpt('apic_vmm_provider', default='VMware',
help=_("Name for the VMM domain provider")),
cfg.StrOpt('apic_vmm_domain', default='openstack',
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# @author: Henry Gessau, Cisco Systems
-
-"""Exceptions used by Cisco APIC ML2 mechanism driver."""
-
-from neutron.common import exceptions
-
-
-class ApicHostNoResponse(exceptions.NotFound):
- """No response from the APIC via the specified URL."""
- message = _("No response from APIC at %(url)s")
-
-
-class ApicResponseNotOk(exceptions.NeutronException):
- """A response from the APIC was not HTTP OK."""
- message = _("APIC responded with HTTP status %(status)s: %(reason)s, "
- "Request: '%(request)s', "
- "APIC error code %(err_code)s: %(err_text)s")
-
-
-class ApicResponseNoCookie(exceptions.NeutronException):
- """A response from the APIC did not contain an expected cookie."""
- message = _("APIC failed to provide cookie for %(request)s request")
-
-
-class ApicSessionNotLoggedIn(exceptions.NotAuthorized):
- """Attempted APIC operation while not logged in to APIC."""
- message = _("Authorized APIC session not established")
-
-
-class ApicHostNotConfigured(exceptions.NotAuthorized):
- """The switch and port for the specified host are not configured."""
- message = _("The switch and port for host '%(host)s' are not configured")
-
-
-class ApicManagedObjectNotSupported(exceptions.NeutronException):
- """Attempted to use an unsupported Managed Object."""
- message = _("Managed Object '%(mo_class)s' is not supported")
-
-
-class ApicMultipleVlanRanges(exceptions.NeutronException):
- """Multiple VLAN ranges specified."""
- message = _("Multiple VLAN ranges are not supported in the APIC plugin. "
- "Please specify a single VLAN range. "
- "Current config: '%(vlan_ranges)s'")
#
# @author: Arvind Somya (asomya@cisco.com), Cisco Systems Inc.
+from apicapi import apic_manager
import netaddr
from oslo.config import cfg
from neutron.openstack.common import log
from neutron.plugins.common import constants
from neutron.plugins.ml2 import driver_api as api
-from neutron.plugins.ml2.drivers.cisco.apic import apic_manager
-from neutron.plugins.ml2.drivers.cisco.apic import exceptions as apic_exc
+from neutron.plugins.ml2.drivers.cisco.apic import apic_model
+from neutron.plugins.ml2.drivers.cisco.apic import config
LOG = log.getLogger(__name__)
class APICMechanismDriver(api.MechanismDriver):
+ @staticmethod
+ def get_apic_manager():
+ apic_config = cfg.CONF.ml2_cisco_apic
+ network_config = {
+ 'vlan_ranges': cfg.CONF.ml2_type_vlan.network_vlan_ranges,
+ 'switch_dict': config.create_switch_dictionary(),
+ }
+ return apic_manager.APICManager(apic_model.ApicDbModel(), log,
+ network_config, apic_config)
+
def initialize(self):
- self.apic_manager = apic_manager.APICManager()
-
- # Create a Phys domain and VLAN namespace
- # Get vlan ns name
- ns_name = cfg.CONF.ml2_cisco_apic.apic_vlan_ns_name
-
- # Grab vlan ranges
- if len(cfg.CONF.ml2_type_vlan.network_vlan_ranges) != 1:
- raise apic_exc.ApicMultipleVlanRanges(
- cfg.CONF.ml2_type_vlan.network_vlan_ranges)
- vlan_ranges = cfg.CONF.ml2_type_vlan.network_vlan_ranges[0]
- if ',' in vlan_ranges:
- raise apic_exc.ApicMultipleVlanRanges(vlan_ranges)
- (vlan_min, vlan_max) = vlan_ranges.split(':')[-2:]
-
- # Create VLAN namespace
- vlan_ns = self.apic_manager.ensure_vlan_ns_created_on_apic(ns_name,
- vlan_min,
- vlan_max)
- phys_name = cfg.CONF.ml2_cisco_apic.apic_vmm_domain
- # Create Physical domain
- self.apic_manager.ensure_phys_domain_created_on_apic(phys_name,
- vlan_ns)
-
- # Create entity profile
- ent_name = cfg.CONF.ml2_cisco_apic.apic_entity_profile
- self.apic_manager.ensure_entity_profile_created_on_apic(ent_name)
-
- # Create function profile
- func_name = cfg.CONF.ml2_cisco_apic.apic_function_profile
- self.apic_manager.ensure_function_profile_created_on_apic(func_name)
-
- # Create infrastructure on apic
+ self.apic_manager = APICMechanismDriver.get_apic_manager()
self.apic_manager.ensure_infra_created_on_apic()
def _perform_port_operations(self, context):
# Get network
network = context.network.current['id']
- net_name = context.network.current['name']
# Get port
port = context.current
for dhcp_host in dhcp_hosts:
self.apic_manager.ensure_path_created_for_port(tenant_id,
network,
- dhcp_host, seg,
- net_name)
+ dhcp_host, seg)
if host not in dhcp_hosts:
self.apic_manager.ensure_path_created_for_port(tenant_id, network,
- host, seg, net_name)
+ host, seg)
def create_port_postcommit(self, context):
self._perform_port_operations(context)
from neutron.openstack.common import excutils
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants
-from neutron.plugins.ml2.drivers.cisco.apic import apic_manager
+from neutron.plugins.ml2.drivers.cisco.apic import mechanism_apic as ma
LOG = logging.getLogger(__name__)
def __init__(self):
super(ApicL3ServicePlugin, self).__init__()
- self.manager = apic_manager.APICManager()
+ self.manager = ma.APICMechanismDriver.get_apic_manager()
@staticmethod
def get_plugin_type():
def remove_router_interface(self, context, router_id, interface_info):
"""Detach a subnet from a router."""
tenant_id = context.tenant_id
- subnet_id = interface_info['subnet_id']
- LOG.debug("Detaching subnet %(subnet_id)s from "
- "router %(router_id)s" % {'subnet_id': subnet_id,
- 'router_id': router_id})
+ if 'subnet_id' in interface_info:
+ subnet = self.get_subnet(context, interface_info['subnet_id'])
+ network_id = subnet['network_id']
+ else:
+ port = self.get_port(context, interface_info['port_id'])
+ network_id = port['network_id']
# Get network for this subnet
- subnet = self.get_subnet(context, subnet_id)
- network_id = subnet['network_id']
network = self.get_network(context, network_id)
contract = self.manager.create_tenant_contract(tenant_id)
return super(ApicL3ServicePlugin, self).remove_router_interface(
context, router_id, interface_info)
except Exception:
- LOG.error(_("Error detaching subnet %(subnet_id)s from "
- "router %(router_id)s") % {'subnet_id': subnet_id,
- 'router_id': router_id})
with excutils.save_and_reraise_exception():
self._add_epg_to_contract(tenant_id, epg, contract)
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# @author: Henry Gessau, Cisco Systems
-
-import mock
-import requests
-import requests.exceptions
-
-from neutron.plugins.ml2.drivers.cisco.apic import apic_client as apic
-from neutron.plugins.ml2.drivers.cisco.apic import exceptions as cexc
-from neutron.tests import base
-from neutron.tests.unit.ml2.drivers.cisco.apic import (
- test_cisco_apic_common as mocked)
-
-
-class TestCiscoApicClient(base.BaseTestCase, mocked.ControllerMixin):
-
- def setUp(self):
- super(TestCiscoApicClient, self).setUp()
- self.set_up_mocks()
- self.apic = apic.RestClient(mocked.APIC_HOST)
- self.addCleanup(mock.patch.stopall)
-
- def _mock_authenticate(self, timeout=300):
- self.reset_reponses()
- self.mock_apic_manager_login_responses(timeout=timeout)
- self.apic.login(mocked.APIC_USR, mocked.APIC_PWD)
-
- def test_login_by_instantiation(self):
- self.reset_reponses()
- self.mock_apic_manager_login_responses()
- apic2 = apic.RestClient(mocked.APIC_HOST,
- usr=mocked.APIC_USR, pwd=mocked.APIC_PWD)
- self.assertIsNotNone(apic2.authentication)
- self.assertEqual(apic2.username, mocked.APIC_USR)
-
- def test_client_session_login_ok(self):
- self._mock_authenticate()
- self.assertEqual(
- self.apic.authentication['userName'], mocked.APIC_USR)
- self.assertTrue(self.apic.api_base.startswith('http://'))
- self.assertEqual(self.apic.username, mocked.APIC_USR)
- self.assertIsNotNone(self.apic.authentication)
- self.apic = apic.RestClient(mocked.APIC_HOST, mocked.APIC_PORT,
- ssl=True)
- self.assertTrue(self.apic.api_base.startswith('https://'))
-
- def test_client_session_login_fail(self):
- self.mock_error_post_response(requests.codes.unauthorized,
- code='599',
- text=u'Fake error')
- self.assertRaises(cexc.ApicResponseNotOk, self.apic.login,
- mocked.APIC_USR, mocked.APIC_PWD)
-
- def test_client_session_login_timeout(self):
- self.response['post'].append(requests.exceptions.Timeout)
- self.assertRaises(cexc.ApicHostNoResponse, self.apic.login,
- mocked.APIC_USR, mocked.APIC_PWD)
-
- def test_client_session_logout_ok(self):
- self.mock_response_for_post('aaaLogout')
- self.apic.logout()
- self.assertIsNone(self.apic.authentication)
- # Multiple signouts should not cause an error
- self.apic.logout()
- self.assertIsNone(self.apic.authentication)
-
- def test_client_session_logout_fail(self):
- self._mock_authenticate()
- self.mock_error_post_response(requests.codes.timeout,
- code='123', text='failed')
- self.assertRaises(cexc.ApicResponseNotOk, self.apic.logout)
-
- def test_query_not_logged_in(self):
- self.apic.authentication = None
- self.assertRaises(cexc.ApicSessionNotLoggedIn,
- self.apic.fvTenant.get, mocked.APIC_TENANT)
-
- def test_query_no_response(self):
- self._mock_authenticate()
- requests.Session.get = mock.Mock(return_value=None)
- self.assertRaises(cexc.ApicHostNoResponse,
- self.apic.fvTenant.get, mocked.APIC_TENANT)
-
- def test_query_error_response_no_data(self):
- self._mock_authenticate()
- self.mock_error_get_response(requests.codes.bad) # No error attrs.
- self.assertRaises(cexc.ApicResponseNotOk,
- self.apic.fvTenant.get, mocked.APIC_TENANT)
-
- def test_generic_get_data(self):
- self._mock_authenticate()
- self.mock_response_for_get('topSystem', name='ifc1')
- top_system = self.apic.get_data('class/topSystem')
- self.assertIsNotNone(top_system)
- name = top_system[0]['topSystem']['attributes']['name']
- self.assertEqual(name, 'ifc1')
-
- def test_session_timeout_refresh_ok(self):
- self._mock_authenticate(timeout=-1)
- # Client will do refresh before getting tenant
- self.mock_response_for_get('aaaLogin', token='ok',
- refreshTimeoutSeconds=300)
- self.mock_response_for_get('fvTenant', name=mocked.APIC_TENANT)
- tenant = self.apic.fvTenant.get(mocked.APIC_TENANT)
- self.assertEqual(tenant['name'], mocked.APIC_TENANT)
-
- def test_session_timeout_refresh_no_cookie(self):
- self._mock_authenticate(timeout=-1)
- # Client will do refresh before getting tenant
- self.mock_response_for_get('aaaLogin', notoken='test')
- self.assertRaises(cexc.ApicResponseNoCookie,
- self.apic.fvTenant.get, mocked.APIC_TENANT)
-
- def test_session_timeout_refresh_error(self):
- self._mock_authenticate(timeout=-1)
- self.mock_error_get_response(requests.codes.timeout,
- code='503', text=u'timed out')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.apic.fvTenant.get, mocked.APIC_TENANT)
-
- def test_session_timeout_refresh_timeout_error(self):
- self._mock_authenticate(timeout=-1)
- # Client will try to get refresh, we fake a refresh error.
- self.mock_error_get_response(requests.codes.bad_request,
- code='403',
- text=u'Token was invalid. Expired.')
- # Client will then try to re-login.
- self.mock_apic_manager_login_responses()
- # Finally the client will try to get the tenant.
- self.mock_response_for_get('fvTenant', name=mocked.APIC_TENANT)
- tenant = self.apic.fvTenant.get(mocked.APIC_TENANT)
- self.assertEqual(tenant['name'], mocked.APIC_TENANT)
-
- def test_lookup_mo_bad_token_retry(self):
- self._mock_authenticate()
- # For the first get request we mock a bad token.
- self.mock_error_get_response(requests.codes.bad_request,
- code='403',
- text=u'Token was invalid. Expired.')
- # Client will then try to re-login.
- self.mock_apic_manager_login_responses()
- # Then the client will retry to get the tenant.
- self.mock_response_for_get('fvTenant', name=mocked.APIC_TENANT)
- tenant = self.apic.fvTenant.get(mocked.APIC_TENANT)
- self.assertEqual(tenant['name'], mocked.APIC_TENANT)
-
- def test_use_unsupported_managed_object(self):
- self._mock_authenticate()
- # unittest.assertRaises cannot catch exceptions raised in
- # __getattr__, so we need to defer the evaluation using lambda.
- self.assertRaises(cexc.ApicManagedObjectNotSupported,
- lambda: self.apic.nonexistentObject)
-
- def test_lookup_nonexistant_mo(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvTenant')
- self.assertIsNone(self.apic.fvTenant.get(mocked.APIC_TENANT))
-
- def test_lookup_existing_mo(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvTenant', name='infra')
- tenant = self.apic.fvTenant.get('infra')
- self.assertEqual(tenant['name'], 'infra')
-
- def test_list_mos_ok(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvTenant', name='t1')
- self.mock_append_to_response('fvTenant', name='t2')
- tlist = self.apic.fvTenant.list_all()
- self.assertIsNotNone(tlist)
- self.assertEqual(len(tlist), 2)
- self.assertIn({'name': 't1'}, tlist)
- self.assertIn({'name': 't2'}, tlist)
-
- def test_list_mo_names_ok(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvTenant', name='t1')
- self.mock_append_to_response('fvTenant', name='t2')
- tnlist = self.apic.fvTenant.list_names()
- self.assertIsNotNone(tnlist)
- self.assertEqual(len(tnlist), 2)
- self.assertIn('t1', tnlist)
- self.assertIn('t2', tnlist)
-
- def test_list_mos_split_class_fail(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvnsEncapBlk', name='Blk1')
- encap_blks = self.apic.fvnsEncapBlk__vlan.list_all()
- self.assertEqual(len(encap_blks), 1)
-
- def test_delete_mo_ok(self):
- self._mock_authenticate()
- self.mock_response_for_post('fvTenant')
- self.assertTrue(self.apic.fvTenant.delete(mocked.APIC_TENANT))
-
- def test_create_mo_ok(self):
- self._mock_authenticate()
- self.mock_response_for_post('fvTenant', name=mocked.APIC_TENANT)
- self.mock_response_for_get('fvTenant', name=mocked.APIC_TENANT)
- self.apic.fvTenant.create(mocked.APIC_TENANT)
- tenant = self.apic.fvTenant.get(mocked.APIC_TENANT)
- self.assertEqual(tenant['name'], mocked.APIC_TENANT)
-
- def test_create_mo_already_exists(self):
- self._mock_authenticate()
- self.mock_error_post_response(requests.codes.bad_request,
- code='103',
- text=u'Fake 103 error')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.apic.vmmProvP.create, mocked.APIC_VMMP)
-
- def test_create_mo_with_prereq(self):
- self._mock_authenticate()
- self.mock_response_for_post('fvTenant', name=mocked.APIC_TENANT)
- self.mock_response_for_post('fvBD', name=mocked.APIC_NETWORK)
- self.mock_response_for_get('fvBD', name=mocked.APIC_NETWORK)
- bd_args = mocked.APIC_TENANT, mocked.APIC_NETWORK
- self.apic.fvBD.create(*bd_args)
- network = self.apic.fvBD.get(*bd_args)
- self.assertEqual(network['name'], mocked.APIC_NETWORK)
-
- def test_create_mo_prereq_exists(self):
- self._mock_authenticate()
- self.mock_response_for_post('vmmDomP', name=mocked.APIC_DOMAIN)
- self.mock_response_for_get('vmmDomP', name=mocked.APIC_DOMAIN)
- self.apic.vmmDomP.create(mocked.APIC_VMMP, mocked.APIC_DOMAIN)
- dom = self.apic.vmmDomP.get(mocked.APIC_VMMP, mocked.APIC_DOMAIN)
- self.assertEqual(dom['name'], mocked.APIC_DOMAIN)
-
- def test_create_mo_fails(self):
- self._mock_authenticate()
- self.mock_response_for_post('fvTenant', name=mocked.APIC_TENANT)
- self.mock_error_post_response(requests.codes.bad_request,
- code='not103',
- text=u'Fake not103 error')
- bd_args = mocked.APIC_TENANT, mocked.APIC_NETWORK
- self.assertRaises(cexc.ApicResponseNotOk,
- self.apic.fvBD.create, *bd_args)
-
- def test_update_mo(self):
- self._mock_authenticate()
- self.mock_response_for_post('fvTenant', name=mocked.APIC_TENANT)
- self.mock_response_for_get('fvTenant', name=mocked.APIC_TENANT,
- more='extra')
- self.apic.fvTenant.update(mocked.APIC_TENANT, more='extra')
- tenant = self.apic.fvTenant.get(mocked.APIC_TENANT)
- self.assertEqual(tenant['name'], mocked.APIC_TENANT)
- self.assertEqual(tenant['more'], 'extra')
-
- def test_attr_fail_empty_list(self):
- self._mock_authenticate()
- self.mock_response_for_get('fvTenant') # No attrs for tenant.
- self.assertIsNone(self.apic.fvTenant.get(mocked.APIC_TENANT))
-
- def test_attr_fail_other_obj(self):
- self._mock_authenticate()
- self.mock_response_for_get('other', name=mocked.APIC_TENANT)
- self.assertIsNone(self.apic.fvTenant.get(mocked.APIC_TENANT))
from neutron.common import config as neutron_config
from neutron.plugins.ml2 import config as ml2_config
-from neutron.plugins.ml2.drivers.cisco.apic import apic_client as apic
from neutron.tests import base
OK = requests.codes.ok
-APIC_HOST = 'fake.controller.local'
+APIC_HOSTS = ['fake.controller.local']
APIC_PORT = 7580
APIC_USR = 'notadmin'
APIC_PWD = 'topsecret'
attrs['debug_mo'] = mo # useful for debugging
self._stage_mocked_response('post', OK, mo, **attrs)
- def mock_response_for_get(self, mo, **attrs):
- self._stage_mocked_response('get', OK, mo, **attrs)
-
- def mock_append_to_response(self, mo, **attrs):
- # Append a MO to the last get response.
- mo_attrs = attrs and {mo: {'attributes': attrs}} or {}
- self.response['get'][-1].json.return_value['imdata'].append(mo_attrs)
-
- def mock_error_post_response(self, status, **attrs):
- self._stage_mocked_response('post', status, 'error', **attrs)
-
- def mock_error_get_response(self, status, **attrs):
- self._stage_mocked_response('get', status, 'error', **attrs)
-
def _stage_mocked_response(self, req, mock_status, mo, **attrs):
response = mock.MagicMock()
response.status_code = mock_status
response.json.return_value = {'imdata': mo_attrs}
self.response[req].append(response)
- def mock_responses_for_create(self, obj):
- self._mock_container_responses_for_create(
- apic.ManagedObjectClass(obj).container)
- name = '-'.join([obj, 'name']) # useful for debugging
- self._stage_mocked_response('post', OK, obj, name=name)
-
- def _mock_container_responses_for_create(self, obj):
- # Recursively generate responses for creating obj's containers.
- if obj:
- mo = apic.ManagedObjectClass(obj)
- if mo.can_create:
- if mo.container:
- self._mock_container_responses_for_create(mo.container)
- name = '-'.join([obj, 'name']) # useful for debugging
- self._stage_mocked_response('post', OK, obj, debug_name=name)
-
def mock_apic_manager_login_responses(self, timeout=300):
# APIC Manager tests are based on authenticated session
self.mock_response_for_post('aaaLogin', userName=APIC_USR,
token='ok', refreshTimeoutSeconds=timeout)
- def assert_responses_drained(self, req=None):
- """Fail if all the expected responses have not been consumed."""
- request = {'post': self.session.post, 'get': self.session.get}
- reqs = req and [req] or ['post', 'get'] # Both if none specified.
- for req in reqs:
- try:
- request[req]('some url')
- except StopIteration:
- pass
- else:
- # User-friendly error message
- msg = req + ' response queue not drained'
- self.fail(msg=msg)
-
class ConfigMixin(object):
# Configure the Cisco APIC mechanism driver
apic_test_config = {
- 'apic_host': APIC_HOST,
+ 'apic_hosts': APIC_HOSTS,
'apic_username': APIC_USR,
'apic_password': APIC_PWD,
- 'apic_port': APIC_PORT,
'apic_vmm_domain': APIC_DOMAIN,
'apic_vlan_ns_name': APIC_VLAN_NAME,
'apic_vlan_range': '%d:%d' % (APIC_VLANID_FROM, APIC_VLANID_TO),
'MultiConfigParser').start()
self.mocked_parser.return_value.read.return_value = [apic_switch_cfg]
self.mocked_parser.return_value.parsed = [apic_switch_cfg]
-
-
-class DbModelMixin(object):
-
- """Mock the DB models for the APIC driver and service unit tests."""
-
- def __init__(self):
- self.mocked_session = None
-
- def set_up_mocks(self):
- self.mocked_session = mock.Mock()
- get_session = mock.patch('neutron.db.api.get_session').start()
- get_session.return_value = self.mocked_session
-
- def mock_db_query_filterby_first_return(self, value):
- """Mock db.session.query().filterby().first() to return value."""
- query = self.mocked_session.query.return_value
- query.filter_by.return_value.first.return_value = value
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# @author: Henry Gessau, Cisco Systems
-
-import mock
-from webob import exc as wexc
-
-from neutron.openstack.common import uuidutils
-
-from neutron.plugins.ml2.drivers.cisco.apic import apic_manager
-from neutron.plugins.ml2.drivers.cisco.apic import exceptions as cexc
-from neutron.tests import base
-from neutron.tests.unit.ml2.drivers.cisco.apic import (
- test_cisco_apic_common as mocked)
-
-
-class TestCiscoApicManager(base.BaseTestCase,
- mocked.ControllerMixin,
- mocked.ConfigMixin,
- mocked.DbModelMixin):
-
- def setUp(self):
- super(TestCiscoApicManager, self).setUp()
- mocked.ControllerMixin.set_up_mocks(self)
- mocked.ConfigMixin.set_up_mocks(self)
- mocked.DbModelMixin.set_up_mocks(self)
-
- self.mock_apic_manager_login_responses()
- self.mgr = apic_manager.APICManager()
- self.session = self.mgr.apic.session
- self.assert_responses_drained()
- self.reset_reponses()
-
- def test_mgr_session_login(self):
- login = self.mgr.apic.authentication
- self.assertEqual(login['userName'], mocked.APIC_USR)
-
- def test_mgr_session_logout(self):
- self.mock_response_for_post('aaaLogout')
- self.mgr.apic.logout()
- self.assert_responses_drained()
- self.assertIsNone(self.mgr.apic.authentication)
-
- def test_to_range(self):
- port_list = [4, 2, 3, 1, 7, 8, 10, 20, 6, 22, 21]
- expected_ranges = [(1, 4), (6, 8), (10, 10), (20, 22)]
- port_ranges = [r for r in apic_manager.group_by_ranges(port_list)]
- self.assertEqual(port_ranges, expected_ranges)
-
- def test_get_profiles(self):
- self.mock_db_query_filterby_first_return('faked')
- self.assertEqual(
- self.mgr.db.get_port_profile_for_node('node'),
- 'faked'
- )
- self.assertEqual(
- self.mgr.db.get_profile_for_module('node', 'prof', 'module'),
- 'faked'
- )
- self.assertEqual(
- self.mgr.db.get_profile_for_module_and_ports(
- 'node', 'prof', 'module', 'from', 'to'
- ),
- 'faked'
- )
-
- def test_add_profile(self):
- self.mgr.db.add_profile_for_module_and_ports(
- 'node', 'prof', 'hpselc', 'module', 'from', 'to')
- self.assertTrue(self.mocked_session.add.called)
- self.assertTrue(self.mocked_session.flush.called)
-
- def test_ensure_port_profile_created(self):
- port_name = mocked.APIC_PORT
- self.mock_responses_for_create('infraAccPortP')
- self.mock_response_for_get('infraAccPortP', name=port_name)
- port = self.mgr.ensure_port_profile_created_on_apic(port_name)
- self.assert_responses_drained()
- self.assertEqual(port['name'], port_name)
-
- def test_ensure_port_profile_created_exc(self):
- port_name = mocked.APIC_PORT
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('infraAccPortP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_port_profile_created_on_apic,
- port_name)
- self.assert_responses_drained()
-
- def test_ensure_node_profile_created_for_switch_old(self):
- old_switch = mocked.APIC_NODE_PROF
- self.mock_response_for_get('infraNodeP', name=old_switch)
- self.mgr.ensure_node_profile_created_for_switch(old_switch)
- self.assert_responses_drained()
- old_name = self.mgr.node_profiles[old_switch]['object']['name']
- self.assertEqual(old_name, old_switch)
-
- def test_ensure_node_profile_created_for_switch_new(self):
- new_switch = mocked.APIC_NODE_PROF
- self.mock_response_for_get('infraNodeP')
- self.mock_responses_for_create('infraNodeP')
- self.mock_responses_for_create('infraLeafS')
- self.mock_responses_for_create('infraNodeBlk')
- self.mock_response_for_get('infraNodeP', name=new_switch)
- self.mgr.ensure_node_profile_created_for_switch(new_switch)
- self.assert_responses_drained()
- new_name = self.mgr.node_profiles[new_switch]['object']['name']
- self.assertEqual(new_name, new_switch)
-
- def test_ensure_node_profile_created_for_switch_new_exc(self):
- new_switch = mocked.APIC_NODE_PROF
- self.mock_response_for_get('infraNodeP')
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('infraNodeP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_node_profile_created_for_switch,
- new_switch)
- self.assert_responses_drained()
-
- def test_ensure_vmm_domain_created_old(self):
- dom = mocked.APIC_DOMAIN
- self.mock_response_for_get('vmmDomP', name=dom)
- self.mgr.ensure_vmm_domain_created_on_apic(dom)
- self.assert_responses_drained()
- old_dom = self.mgr.vmm_domain['name']
- self.assertEqual(old_dom, dom)
-
- def _mock_new_vmm_dom_responses(self, dom, seg_type=None):
- vmm = mocked.APIC_VMMP
- dn = self.mgr.apic.vmmDomP.mo.dn(vmm, dom)
- self.mock_response_for_get('vmmDomP')
- self.mock_responses_for_create('vmmDomP')
- if seg_type:
- self.mock_responses_for_create(seg_type)
- self.mock_response_for_get('vmmDomP', name=dom, dn=dn)
-
- def test_ensure_vmm_domain_created_new_no_vlan_ns(self):
- dom = mocked.APIC_DOMAIN
- self._mock_new_vmm_dom_responses(dom)
- self.mgr.ensure_vmm_domain_created_on_apic(dom)
- self.assert_responses_drained()
- new_dom = self.mgr.vmm_domain['name']
- self.assertEqual(new_dom, dom)
-
- def test_ensure_vmm_domain_created_new_no_vlan_ns_exc(self):
- dom = mocked.APIC_DOMAIN
- self.mock_response_for_get('vmmDomP')
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('vmmDomP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_vmm_domain_created_on_apic, dom)
- self.assert_responses_drained()
-
- def test_ensure_vmm_domain_created_new_with_vlan_ns(self):
- dom = mocked.APIC_DOMAIN
- self._mock_new_vmm_dom_responses(dom, seg_type='infraRsVlanNs__vmm')
- ns = {'dn': 'test_vlan_ns'}
- self.mgr.ensure_vmm_domain_created_on_apic(dom, vlan_ns=ns)
- self.assert_responses_drained()
- new_dom = self.mgr.vmm_domain['name']
- self.assertEqual(new_dom, dom)
-
- def test_ensure_vmm_domain_created_new_with_vxlan_ns(self):
- dom = mocked.APIC_DOMAIN
- # TODO(Henry): mock seg_type vxlan when vxlan is ready
- self._mock_new_vmm_dom_responses(dom, seg_type=None)
- ns = {'dn': 'test_vxlan_ns'}
- self.mgr.ensure_vmm_domain_created_on_apic(dom, vxlan_ns=ns)
- self.assert_responses_drained()
- new_dom = self.mgr.vmm_domain['name']
- self.assertEqual(new_dom, dom)
-
- def test_ensure_infra_created_no_infra(self):
- self.mgr.switch_dict = {}
- self.mgr.ensure_infra_created_on_apic()
-
- def _ensure_infra_created_seq1_setup(self):
- am = 'neutron.plugins.ml2.drivers.cisco.apic.apic_manager.APICManager'
- np_create_for_switch = mock.patch(
- am + '.ensure_node_profile_created_for_switch').start()
- self.mock_db_query_filterby_first_return(None)
- pp_create_for_switch = mock.patch(
- am + '.ensure_port_profile_created_on_apic').start()
- pp_create_for_switch.return_value = {'dn': 'port_profile_dn'}
- return np_create_for_switch, pp_create_for_switch
-
- def test_ensure_infra_created_seq1(self):
- np_create_for_switch, pp_create_for_switch = (
- self._ensure_infra_created_seq1_setup())
-
- def _profile_for_module(aswitch, ppn, module):
- profile = mock.Mock()
- profile.ppn = ppn
- profile.hpselc_id = '-'.join([aswitch, module, 'hpselc_id'])
- return profile
-
- self.mgr.db.get_profile_for_module = mock.Mock(
- side_effect=_profile_for_module)
- self.mgr.db.get_profile_for_module_and_ports = mock.Mock(
- return_value=None)
- self.mgr.db.add_profile_for_module_and_ports = mock.Mock()
-
- num_switches = len(self.mgr.switch_dict)
- for loop in range(num_switches):
- self.mock_responses_for_create('infraRsAccPortP')
- self.mock_responses_for_create('infraPortBlk')
-
- self.mgr.ensure_infra_created_on_apic()
- self.assert_responses_drained()
- self.assertEqual(np_create_for_switch.call_count, num_switches)
- self.assertEqual(pp_create_for_switch.call_count, num_switches)
- for switch in self.mgr.switch_dict:
- np_create_for_switch.assert_any_call(switch)
-
- def test_ensure_infra_created_seq1_exc(self):
- np_create_for_switch, __ = self._ensure_infra_created_seq1_setup()
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('infraAccPortP')
-
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_infra_created_on_apic)
- self.assert_responses_drained()
- self.assertTrue(np_create_for_switch.called)
- self.assertEqual(np_create_for_switch.call_count, 1)
-
- def _ensure_infra_created_seq2_setup(self):
- am = 'neutron.plugins.ml2.drivers.cisco.apic.apic_manager.APICManager'
- np_create_for_switch = mock.patch(
- am + '.ensure_node_profile_created_for_switch').start()
-
- def _profile_for_node(aswitch):
- profile = mock.Mock()
- profile.profile_id = '-'.join([aswitch, 'profile_id'])
- return profile
-
- self.mgr.db.get_port_profile_for_node = mock.Mock(
- side_effect=_profile_for_node)
- self.mgr.db.get_profile_for_module = mock.Mock(
- return_value=None)
- self.mgr.function_profile = {'dn': 'dn'}
- self.mgr.db.get_profile_for_module_and_ports = mock.Mock(
- return_value=True)
-
- return np_create_for_switch
-
- def test_ensure_infra_created_seq2(self):
- np_create_for_switch = self._ensure_infra_created_seq2_setup()
-
- num_switches = len(self.mgr.switch_dict)
- for loop in range(num_switches):
- self.mock_responses_for_create('infraHPortS')
- self.mock_responses_for_create('infraRsAccBaseGrp')
-
- self.mgr.ensure_infra_created_on_apic()
- self.assert_responses_drained()
- self.assertEqual(np_create_for_switch.call_count, num_switches)
- for switch in self.mgr.switch_dict:
- np_create_for_switch.assert_any_call(switch)
-
- def test_ensure_infra_created_seq2_exc(self):
- np_create_for_switch = self._ensure_infra_created_seq2_setup()
-
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('infraHPortS')
-
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_infra_created_on_apic)
- self.assert_responses_drained()
- self.assertTrue(np_create_for_switch.called)
- self.assertEqual(np_create_for_switch.call_count, 1)
-
- def test_ensure_context_unenforced_new_ctx(self):
- self.mock_response_for_get('fvCtx')
- self.mock_responses_for_create('fvCtx')
- self.mgr.ensure_context_unenforced()
- self.assert_responses_drained()
-
- def test_ensure_context_unenforced_pref1(self):
- self.mock_response_for_get('fvCtx', pcEnfPref='1')
- self.mock_response_for_post('fvCtx')
- self.mgr.ensure_context_unenforced()
- self.assert_responses_drained()
-
- def test_ensure_context_unenforced_pref2(self):
- self.mock_response_for_get('fvCtx', pcEnfPref='2')
- self.mgr.ensure_context_unenforced()
- self.assert_responses_drained()
-
- def _mock_vmm_dom_prereq(self, dom):
- self._mock_new_vmm_dom_responses(dom)
- self.mgr.ensure_vmm_domain_created_on_apic(dom)
-
- def _mock_new_phys_dom_responses(self, dom, seg_type=None):
- dn = self.mgr.apic.physDomP.mo.dn(dom)
- self.mock_response_for_get('physDomP')
- self.mock_responses_for_create('physDomP')
- if seg_type:
- self.mock_responses_for_create(seg_type)
- self.mock_response_for_get('physDomP', name=dom, dn=dn)
-
- def _mock_phys_dom_prereq(self, dom):
- self._mock_new_phys_dom_responses(dom)
- self.mgr.ensure_phys_domain_created_on_apic(dom)
-
- def test_ensure_entity_profile_created_old(self):
- ep = mocked.APIC_ATT_ENT_PROF
- self.mock_response_for_get('infraAttEntityP', name=ep)
- self.mgr.ensure_entity_profile_created_on_apic(ep)
- self.assert_responses_drained()
-
- def _mock_new_entity_profile(self, exc=None):
- self.mock_response_for_get('infraAttEntityP')
- self.mock_responses_for_create('infraAttEntityP')
- self.mock_responses_for_create('infraRsDomP')
- if exc:
- self.mock_error_get_response(exc, code='103', text=u'Fail')
- else:
- self.mock_response_for_get('infraAttEntityP')
-
- def test_ensure_entity_profile_created_new(self):
- self._mock_phys_dom_prereq(mocked.APIC_PDOM)
- ep = mocked.APIC_ATT_ENT_PROF
- self._mock_new_entity_profile()
- self.mgr.ensure_entity_profile_created_on_apic(ep)
- self.assert_responses_drained()
-
- def test_ensure_entity_profile_created_new_exc(self):
- self._mock_phys_dom_prereq(mocked.APIC_PDOM)
- ep = mocked.APIC_ATT_ENT_PROF
- self._mock_new_entity_profile(exc=wexc.HTTPBadRequest)
- self.mock_response_for_post('infraAttEntityP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_entity_profile_created_on_apic, ep)
- self.assert_responses_drained()
-
- def _mock_entity_profile_preqreq(self):
- self._mock_phys_dom_prereq(mocked.APIC_PDOM)
- ep = mocked.APIC_ATT_ENT_PROF
- self._mock_new_entity_profile()
- self.mgr.ensure_entity_profile_created_on_apic(ep)
-
- def test_ensure_function_profile_created_old(self):
- self._mock_entity_profile_preqreq()
- fp = mocked.APIC_FUNC_PROF
- self.mock_response_for_get('infraAccPortGrp', name=fp)
- self.mgr.ensure_function_profile_created_on_apic(fp)
- self.assert_responses_drained()
- old_fp = self.mgr.function_profile['name']
- self.assertEqual(old_fp, fp)
-
- def _mock_new_function_profile(self, fp):
- dn = self.mgr.apic.infraAttEntityP.mo.dn(fp)
- self.mock_responses_for_create('infraAccPortGrp')
- self.mock_responses_for_create('infraRsAttEntP')
- self.mock_response_for_get('infraAccPortGrp', name=fp, dn=dn)
-
- def test_ensure_function_profile_created_new(self):
- fp = mocked.APIC_FUNC_PROF
- dn = self.mgr.apic.infraAttEntityP.mo.dn(fp)
- self.mgr.entity_profile = {'dn': dn}
- self.mock_response_for_get('infraAccPortGrp')
- self.mock_responses_for_create('infraAccPortGrp')
- self.mock_responses_for_create('infraRsAttEntP')
- self.mock_response_for_get('infraAccPortGrp', name=fp, dn=dn)
- self.mgr.ensure_function_profile_created_on_apic(fp)
- self.assert_responses_drained()
- new_fp = self.mgr.function_profile['name']
- self.assertEqual(new_fp, fp)
-
- def test_ensure_function_profile_created_new_exc(self):
- fp = mocked.APIC_FUNC_PROF
- dn = self.mgr.apic.infraAttEntityP.mo.dn(fp)
- self.mgr.entity_profile = {'dn': dn}
- self.mock_response_for_get('infraAccPortGrp')
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('infraAccPortGrp')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_function_profile_created_on_apic, fp)
- self.assert_responses_drained()
-
- def test_ensure_vlan_ns_created_old(self):
- ns = mocked.APIC_VLAN_NAME
- mode = mocked.APIC_VLAN_MODE
- self.mock_response_for_get('fvnsVlanInstP', name=ns, mode=mode)
- new_ns = self.mgr.ensure_vlan_ns_created_on_apic(ns, '100', '199')
- self.assert_responses_drained()
- self.assertIsNone(new_ns)
-
- def _mock_new_vlan_instance(self, ns, vlan_encap=None):
- self.mock_responses_for_create('fvnsVlanInstP')
- if vlan_encap:
- self.mock_response_for_get('fvnsEncapBlk', **vlan_encap)
- else:
- self.mock_response_for_get('fvnsEncapBlk')
- self.mock_responses_for_create('fvnsEncapBlk__vlan')
- self.mock_response_for_get('fvnsVlanInstP', name=ns)
-
- def test_ensure_vlan_ns_created_new_no_encap(self):
- ns = mocked.APIC_VLAN_NAME
- self.mock_response_for_get('fvnsVlanInstP')
- self._mock_new_vlan_instance(ns)
- new_ns = self.mgr.ensure_vlan_ns_created_on_apic(ns, '200', '299')
- self.assert_responses_drained()
- self.assertEqual(new_ns['name'], ns)
-
- def test_ensure_vlan_ns_created_new_exc(self):
- ns = mocked.APIC_VLAN_NAME
- self.mock_response_for_get('fvnsVlanInstP')
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('fvnsVlanInstP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_vlan_ns_created_on_apic,
- ns, '200', '299')
- self.assert_responses_drained()
-
- def test_ensure_vlan_ns_created_new_with_encap(self):
- ns = mocked.APIC_VLAN_NAME
- self.mock_response_for_get('fvnsVlanInstP')
- ns_args = {'name': 'encap', 'from': '300', 'to': '399'}
- self._mock_new_vlan_instance(ns, vlan_encap=ns_args)
- new_ns = self.mgr.ensure_vlan_ns_created_on_apic(ns, '300', '399')
- self.assert_responses_drained()
- self.assertEqual(new_ns['name'], ns)
-
- def test_ensure_tenant_created_on_apic(self):
- self.mock_response_for_get('fvTenant', name='any')
- self.mgr.ensure_tenant_created_on_apic('two')
- self.mock_response_for_get('fvTenant')
- self.mock_responses_for_create('fvTenant')
- self.mgr.ensure_tenant_created_on_apic('four')
- self.assert_responses_drained()
-
- def test_ensure_bd_created_existing_bd(self):
- self.mock_response_for_get('fvBD', name='BD')
- self.mgr.ensure_bd_created_on_apic('t1', 'two')
- self.assert_responses_drained()
-
- def test_ensure_bd_created_not_ctx(self):
- self.mock_response_for_get('fvBD')
- self.mock_responses_for_create('fvBD')
- self.mock_response_for_get('fvCtx')
- self.mock_responses_for_create('fvCtx')
- self.mock_responses_for_create('fvRsCtx')
- self.mgr.ensure_bd_created_on_apic('t2', 'three')
- self.assert_responses_drained()
-
- def test_ensure_bd_created_exc(self):
- self.mock_response_for_get('fvBD')
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('fvBD')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_bd_created_on_apic, 't2', 'three')
- self.assert_responses_drained()
-
- def test_ensure_bd_created_ctx_pref1(self):
- self.mock_response_for_get('fvBD')
- self.mock_responses_for_create('fvBD')
- self.mock_response_for_get('fvCtx', pcEnfPref='1')
- self.mock_responses_for_create('fvRsCtx')
- self.mgr.ensure_bd_created_on_apic('t3', 'four')
- self.assert_responses_drained()
-
- def test_ensure_bd_created_ctx_pref2(self):
- self.mock_response_for_get('fvBD')
- self.mock_responses_for_create('fvBD')
- self.mock_response_for_get('fvCtx', pcEnfPref='2')
- self.mock_response_for_post('fvCtx')
- self.mock_responses_for_create('fvRsCtx')
- self.mgr.ensure_bd_created_on_apic('t3', 'four')
- self.assert_responses_drained()
-
- def test_delete_bd(self):
- self.mock_response_for_post('fvBD')
- self.mgr.delete_bd_on_apic('t1', 'bd')
- self.assert_responses_drained()
-
- def test_ensure_subnet_created(self):
- self.mock_response_for_get('fvSubnet', name='sn1')
- self.mgr.ensure_subnet_created_on_apic('t0', 'bd1', '2.2.2.2/8')
- self.mock_response_for_get('fvSubnet')
- self.mock_responses_for_create('fvSubnet')
- self.mgr.ensure_subnet_created_on_apic('t2', 'bd3', '4.4.4.4/16')
- self.assert_responses_drained()
-
- def test_ensure_filter_created(self):
- self.mock_response_for_get('vzFilter', name='f1')
- self.mgr.ensure_filter_created_on_apic('t1', 'two')
- self.mock_response_for_get('vzFilter')
- self.mock_responses_for_create('vzFilter')
- self.mgr.ensure_filter_created_on_apic('t2', 'four')
- self.assert_responses_drained()
-
- def test_ensure_epg_created_for_network_old(self):
- self.mock_db_query_filterby_first_return('faked')
- epg = self.mgr.ensure_epg_created_for_network('X', 'Y', 'Z')
- self.assertEqual(epg, 'faked')
-
- def test_ensure_epg_created_for_network_new(self):
- tenant = mocked.APIC_TENANT
- network = mocked.APIC_NETWORK
- netname = mocked.APIC_NETNAME
- self._mock_phys_dom_prereq(mocked.APIC_PDOM)
- self.mock_db_query_filterby_first_return(None)
- self.mock_responses_for_create('fvAEPg')
- self.mock_response_for_get('fvBD', name=network)
- self.mock_responses_for_create('fvRsBd')
- self.mock_responses_for_create('fvRsDomAtt')
- new_epg = self.mgr.ensure_epg_created_for_network(tenant,
- network, netname)
- self.assert_responses_drained()
- self.assertEqual(new_epg.network_id, network)
- self.assertTrue(self.mocked_session.add.called)
- self.assertTrue(self.mocked_session.flush.called)
-
- def test_ensure_epg_created_for_network_exc(self):
- tenant = mocked.APIC_TENANT
- network = mocked.APIC_NETWORK
- netname = mocked.APIC_NETNAME
- self.mock_db_query_filterby_first_return(None)
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('fvAEPg')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.ensure_epg_created_for_network,
- tenant, network, netname)
- self.assert_responses_drained()
-
- def test_delete_epg_for_network_no_epg(self):
- self.mock_db_query_filterby_first_return(None)
- self.mgr.delete_epg_for_network('tenant', 'network')
-
- def test_delete_epg_for_network(self):
- epg = mock.Mock()
- epg.epg_id = mocked.APIC_EPG
- self.mock_db_query_filterby_first_return(epg)
- self.mock_response_for_post('fvAEPg')
- self.mgr.delete_epg_for_network('tenant', 'network')
- self.assertTrue(self.mocked_session.delete.called)
- self.assertTrue(self.mocked_session.flush.called)
-
- def test_ensure_path_created_for_port(self):
- epg = mock.Mock()
- epg.epg_id = 'epg01'
- eepg = mock.Mock(return_value=epg)
- apic_manager.APICManager.ensure_epg_created_for_network = eepg
- self.mock_response_for_get('fvRsPathAtt', tDn='foo')
- self.mgr.ensure_path_created_for_port('tenant', 'network', 'rhel01',
- 'static', 'netname')
- self.assert_responses_drained()
-
- def test_ensure_path_created_for_port_no_path_att(self):
- epg = mock.Mock()
- epg.epg_id = 'epg2'
- eepg = mock.Mock(return_value=epg)
- self.mgr.ensure_epg_created_for_network = eepg
- self.mock_response_for_get('fvRsPathAtt')
- self.mock_responses_for_create('fvRsPathAtt')
- self.mgr.ensure_path_created_for_port('tenant', 'network', 'ubuntu2',
- 'static', 'netname')
- self.assert_responses_drained()
-
- def test_ensure_path_created_for_port_unknown_host(self):
- epg = mock.Mock()
- epg.epg_id = 'epg3'
- eepg = mock.Mock(return_value=epg)
- apic_manager.APICManager.ensure_epg_created_for_network = eepg
- self.mock_response_for_get('fvRsPathAtt', tDn='foo')
- self.assertRaises(cexc.ApicHostNotConfigured,
- self.mgr.ensure_path_created_for_port,
- 'tenant', 'network', 'cirros3', 'static', 'netname')
-
- def test_create_tenant_filter(self):
- tenant = mocked.APIC_TENANT
- self.mock_responses_for_create('vzFilter')
- self.mock_responses_for_create('vzEntry')
- filter_id = self.mgr.create_tenant_filter(tenant)
- self.assert_responses_drained()
- self.assertTrue(uuidutils.is_uuid_like(str(filter_id)))
-
- def test_create_tenant_filter_exc(self):
- tenant = mocked.APIC_TENANT
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('vzFilter')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.create_tenant_filter, tenant)
- self.assert_responses_drained()
-
- def test_set_contract_for_epg_consumer(self):
- tenant = mocked.APIC_TENANT
- epg = mocked.APIC_EPG
- contract = mocked.APIC_CONTRACT
- self.mock_responses_for_create('fvRsCons')
- self.mgr.set_contract_for_epg(tenant, epg, contract)
- self.assert_responses_drained()
-
- def test_set_contract_for_epg_provider(self):
- tenant = mocked.APIC_TENANT
- epg = mocked.APIC_EPG
- contract = mocked.APIC_CONTRACT
- epg_obj = mock.Mock()
- epg_obj.epg_id = epg
- epg_obj.provider = False
- self.mock_db_query_filterby_first_return(epg_obj)
- self.mock_responses_for_create('fvRsProv')
- self.mock_response_for_post('vzBrCP')
- self.mgr.set_contract_for_epg(tenant, epg, contract, provider=True)
- self.assert_responses_drained()
- self.assertTrue(self.mocked_session.merge.called)
- self.assertTrue(self.mocked_session.flush.called)
- self.assertTrue(epg_obj.provider)
-
- def test_set_contract_for_epg_provider_exc(self):
- tenant = mocked.APIC_TENANT
- epg = mocked.APIC_EPG
- contract = mocked.APIC_CONTRACT
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('vzBrCP')
- self.mock_response_for_post('fvRsProv')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.set_contract_for_epg,
- tenant, epg, contract, provider=True)
- self.assert_responses_drained()
-
- def test_delete_contract_for_epg_consumer(self):
- tenant = mocked.APIC_TENANT
- epg = mocked.APIC_EPG
- contract = mocked.APIC_CONTRACT
- self.mock_response_for_post('fvRsCons')
- self.mgr.delete_contract_for_epg(tenant, epg, contract)
- self.assert_responses_drained()
-
- def test_delete_contract_for_epg_provider(self):
- tenant = mocked.APIC_TENANT
- epg = mocked.APIC_EPG
- contract = mocked.APIC_CONTRACT
- epg_obj = mock.Mock()
- epg_obj.epg_id = epg + '-other'
- epg_obj.provider = False
- self.mock_db_query_filterby_first_return(epg_obj)
- self.mock_response_for_post('fvRsProv')
- self.mock_response_for_post('fvRsCons')
- self.mock_responses_for_create('fvRsProv')
- self.mock_response_for_post('vzBrCP')
- self.mgr.delete_contract_for_epg(tenant, epg, contract, provider=True)
- self.assert_responses_drained()
- self.assertTrue(self.mocked_session.merge.called)
- self.assertTrue(self.mocked_session.flush.called)
- self.assertTrue(epg_obj.provider)
-
- def test_create_tenant_contract_existing(self):
- tenant = mocked.APIC_TENANT
- contract = mocked.APIC_CONTRACT
- self.mock_db_query_filterby_first_return(contract)
- new_contract = self.mgr.create_tenant_contract(tenant)
- self.assertEqual(new_contract, contract)
-
- def test_create_tenant_contract_new(self):
- tenant = mocked.APIC_TENANT
- contract = mocked.APIC_CONTRACT
- dn = self.mgr.apic.vzBrCP.mo.dn(tenant, contract)
- self.mock_db_query_filterby_first_return(None)
- self.mock_responses_for_create('vzBrCP')
- self.mock_response_for_get('vzBrCP', dn=dn)
- self.mock_responses_for_create('vzSubj')
- self.mock_responses_for_create('vzFilter')
- self.mock_responses_for_create('vzEntry')
- self.mock_responses_for_create('vzInTerm')
- self.mock_responses_for_create('vzRsFiltAtt__In')
- self.mock_responses_for_create('vzOutTerm')
- self.mock_responses_for_create('vzRsFiltAtt__Out')
- self.mock_responses_for_create('vzCPIf')
- self.mock_responses_for_create('vzRsIf')
- new_contract = self.mgr.create_tenant_contract(tenant)
- self.assert_responses_drained()
- self.assertTrue(self.mocked_session.add.called)
- self.assertTrue(self.mocked_session.flush.called)
- self.assertEqual(new_contract['tenant_id'], tenant)
-
- def test_create_tenant_contract_exc(self):
- tenant = mocked.APIC_TENANT
- self.mock_db_query_filterby_first_return(None)
- self.mock_error_post_response(wexc.HTTPBadRequest)
- self.mock_response_for_post('vzBrCP')
- self.assertRaises(cexc.ApicResponseNotOk,
- self.mgr.create_tenant_contract, tenant)
- self.assert_responses_drained()
# @author: Henry Gessau, Cisco Systems
import mock
+import sys
-from oslo.config import cfg
+sys.modules["apicapi"] = mock.Mock()
from neutron.extensions import portbindings
from neutron.plugins.ml2.drivers.cisco.apic import mechanism_apic as md
-from neutron.plugins.ml2.drivers import type_vlan # noqa
from neutron.tests import base
from neutron.tests.unit.ml2.drivers.cisco.apic import (
test_cisco_apic_common as mocked)
class TestCiscoApicMechDriver(base.BaseTestCase,
mocked.ControllerMixin,
- mocked.ConfigMixin,
- mocked.DbModelMixin):
+ mocked.ConfigMixin):
def setUp(self):
super(TestCiscoApicMechDriver, self).setUp()
mocked.ControllerMixin.set_up_mocks(self)
mocked.ConfigMixin.set_up_mocks(self)
- mocked.DbModelMixin.set_up_mocks(self)
self.mock_apic_manager_login_responses()
self.driver = md.APICMechanismDriver()
self.driver.cap_port_filter = 'test-cap_port_filter'
def test_initialize(self):
- cfg.CONF.set_override('network_vlan_ranges', ['physnet1:100:199'],
- 'ml2_type_vlan')
- ns = mocked.APIC_VLAN_NAME
- mode = mocked.APIC_VLAN_MODE
- self.mock_response_for_get('fvnsVlanInstP', name=ns, mode=mode)
- self.mock_response_for_get('physDomP', name=mocked.APIC_DOMAIN)
- self.mock_response_for_get('infraAttEntityP',
- name=mocked.APIC_ATT_ENT_PROF)
- self.mock_response_for_get('infraAccPortGrp',
- name=mocked.APIC_ACC_PORT_GRP)
- mock.patch('neutron.plugins.ml2.drivers.cisco.apic.apic_manager.'
- 'APICManager.ensure_infra_created_on_apic').start()
+ mgr = self.driver.apic_manager = mock.Mock()
self.driver.initialize()
- self.session = self.driver.apic_manager.apic.session
- self.assert_responses_drained()
+ mgr.ensure_infra_created_on_apic.assert_called_once()
def test_update_port_postcommit(self):
net_ctx = self._get_network_context(mocked.APIC_TENANT,
mocked.APIC_TENANT)
mgr.ensure_path_created_for_port.assert_called_once_with(
mocked.APIC_TENANT, mocked.APIC_NETWORK, HOST_ID1,
- ENCAP, mocked.APIC_NETWORK + '-name')
+ ENCAP)
def test_create_network_postcommit(self):
ctx = self._get_network_context(mocked.APIC_TENANT,
# @author: Arvind Somya (asomya@cisco.com), Cisco Systems
import mock
+import sys
+
+sys.modules["apicapi"] = mock.Mock()
from neutron.services.l3_router import l3_apic
from neutron.tests.unit import testlib_api
def setUp(self):
super(TestCiscoApicL3Plugin, self).setUp()
- mock.patch('neutron.plugins.ml2.drivers.cisco.apic.apic_manager.'
- 'APICManager').start()
self.plugin = l3_apic.ApicL3ServicePlugin()
self.context = FakeContext()
self.context.tenant_id = TENANT
Code Review / openstack-build / neutron-build.git / commitdiff