openstack/charms.openstack
Code Issues Proposed changes

Added ovs, odl and pci modules

Browse Source 
Added three new modules for managing OVS, ODL and PCI devices. They
exist in the new charms_openstack/{sdn,devices} directories.

The modules have come from the
git@github.com:openstack/charm-openvswitch-odl.git charm and are
largely unaltered apart from new Doc strings and unit tests.

Change-Id: I3e44afeb866a7551883da04a699af3807853a83e
This commit is contained in:
Liam Young
2016-09-12 14:03:39 +00:00
parent 396a9c629f
commit 759530dabe
15 changed files with 1958 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
charms_openstack/devices/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
# Copyright 2016 Canonical Ltd
#
# 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.

474
charms_openstack/devices/pci.py Normal file
View File

@@ -0,0 +1,474 @@
import re
import os
import glob
import shlex
import subprocess
import charmhelpers.core.decorators as decorators
import charmhelpers.core.hookenv as hookenv
def format_pci_addr(pci_addr):
"""Pad a PCI address eg 0:0:1.1 becomes 0000:00:01.1
:param pci_addr: str
:return pci_addr: str
"""
domain, bus, slot_func = pci_addr.split(':')
slot, func = slot_func.split('.')
return '{}:{}:{}.{}'.format(domain.zfill(4), bus.zfill(2), slot.zfill(2),
func)
class VPECLIException(Exception):
def __init__(self, code, message):
self.code = code
self.message = message
class PCINetDevice(object):
def __init__(self, pci_address):
"""Class representing a PCI device
:param pci_addr: str PCI address of device
"""
self.pci_address = pci_address
self.update_attributes()
def update_attributes(self):
"""Query the underlying system and update attributes of this device
"""
self.update_modalias_kmod()
self.update_interface_info()
@property
def loaded_kmod(self):
"""Return Kernel module this device is using
:returns str: Kernel module
"""
cmd = ['lspci', '-ks', self.pci_address]
lspci_output = subprocess.check_output(cmd)
kdrive = None
for line in lspci_output.split('\n'):
if 'Kernel driver' in line:
kdrive = line.split(':')[1].strip()
hookenv.log('Loaded kmod for {} is {}'.format(
self.pci_address, kdrive))
return kdrive
def update_modalias_kmod(self):
"""Set the default kernel module for this device
If a device is orphaned it has no kernel module loaded to support it
so look up the device in modules.alias and set the kernel module
it needs"""
cmd = ['lspci', '-ns', self.pci_address]
lspci_output = subprocess.check_output(cmd).split()
vendor_device = lspci_output[2]
vendor, device = vendor_device.split(':')
pci_string = 'pci:v{}d{}'.format(vendor.zfill(8), device.zfill(8))
kernel_name = self.get_kernel_name()
alias_files = '/lib/modules/{}/modules.alias'.format(kernel_name)
kmod = None
with open(alias_files, 'r') as f:
for line in f.readlines():
if pci_string in line:
kmod = line.split()[-1]
hookenv.log('module.alias kmod for {} is {}'.format(
self.pci_address, kmod))
self.modalias_kmod = kmod
def update_interface_info(self):
"""Set the interface name, mac address and state properties of this
object"""
if self.loaded_kmod:
if self.loaded_kmod == 'igb_uio':
return self.update_interface_info_vpe()
else:
return self.update_interface_info_eth()
else:
self.interface_name = None
self.mac_address = None
self.state = 'unbound'
def get_kernel_name(self):
"""Return the kernel release of the running kernel
:returns str: Kernel release
"""
return subprocess.check_output(['uname', '-r']).strip()
def pci_rescan(self):
"""Rescan of all PCI buses in the system, and
re-discover previously removed devices."""
rescan_file = '/sys/bus/pci/rescan'
with open(rescan_file, 'w') as f:
f.write('1')
def bind(self, kmod):
"""Write PCI address to the bind file to cause the driver to attempt to
bind to the device found at the PCI address. This is useful for
overriding default bindings."""
bind_file = '/sys/bus/pci/drivers/{}/bind'.format(kmod)
hookenv.log('Binding {} to {}'.format(self.pci_address, bind_file))
with open(bind_file, 'w') as f:
f.write(self.pci_address)
self.pci_rescan()
self.update_attributes()
def unbind(self):
"""Write PCI address to the unbind file to cause the driver to attempt
to unbind the device found at at the PCI address."""
if not self.loaded_kmod:
return
unbind_file = '/sys/bus/pci/drivers/{}/unbind'.format(self.loaded_kmod)
hookenv.log('Unbinding {} from {}'.format(
self.pci_address, unbind_file))
with open(unbind_file, 'w') as f:
f.write(self.pci_address)
self.pci_rescan()
self.update_attributes()
def update_interface_info_vpe(self):
"""Query VPE CLI to set the interface name, mac address and state
properties of this device"""
vpe_devices = self.get_vpe_interfaces_and_macs()
device_info = {}
for interface in vpe_devices:
if self.pci_address == interface['pci_address']:
device_info['interface'] = interface['interface']
device_info['macAddress'] = interface['macAddress']
if device_info:
self.interface_name = device_info['interface']
self.mac_address = device_info['macAddress']
self.state = 'vpebound'
else:
self.interface_name = None
self.mac_address = None
self.state = None
@decorators.retry_on_exception(5, base_delay=10,
exc_type=subprocess.CalledProcessError)
def get_vpe_cli_out(self):
"""Query VPE CLI and dump interface information
:returns str: confd_cli output"""
echo_cmd = [
'echo', '-e', 'show interfaces-state interface phys-address\nexit']
cli_cmd = ['/opt/cisco/vpe/bin/confd_cli', '-N', '-C', '-u', 'system']
echo = subprocess.Popen(echo_cmd, stdout=subprocess.PIPE)
cli_output = subprocess.check_output(cli_cmd, stdin=echo.stdout)
echo.wait()
echo.terminate
hookenv.log('confd_cli: ' + cli_output)
return cli_output
def get_vpe_interfaces_and_macs(self):
"""Parse output from VPE CLI and retrun list of interface data dicts
:returns list: list of dicts of interface data
eg [
{
'interface': 'TenGigabitEthernet6/0/0',
'macAddress': '84:b8:02:2a:5f:c3',
'pci_address': '0000:06:00.0'
},
{
'interface': 'TenGigabitEthernet7/0/0',
'macAddress': '84:b8:02:2a:5f:c4',
'pci_address': '0000:07:00.0'
},
]
"""
cli_output = self.get_vpe_cli_out()
vpe_devs = []
if 'local0' not in cli_output:
msg = ('local0 missing from confd_cli output, assuming things '
'went wrong')
raise VPECLIException(1, msg)
for line in cli_output.split('\n'):
if re.search(r'([0-9A-F]{2}[:-]){5}([0-9A-F]{2})', line, re.I):
interface, mac = line.split()
pci_addr = self.extract_pci_addr_from_vpe_interface(interface)
vpe_devs.append({
'interface': interface,
'macAddress': mac,
'pci_address': pci_addr,
})
return vpe_devs
def extract_pci_addr_from_vpe_interface(self, nic):
"""Convert a str from nic postfix format to padded format
:returns list: list of dicts of interface data
eg 6/1/2 -> 0000:06:01.2"""
hookenv.log('Extracting pci address from {}'.format(nic))
addr = re.sub(r'^.*Ethernet', '', nic, re.IGNORECASE)
bus, slot, func = addr.split('/')
domain = '0000'
pci_addr = format_pci_addr(
'{}:{}:{}.{}'.format(domain, bus, slot, func))
hookenv.log('pci address for {} is {}'.format(nic, pci_addr))
return pci_addr
def update_interface_info_eth(self):
"""Set the interface name, mac address and state
properties of this device if device is in sys fs"""
net_devices = self.get_sysnet_interfaces_and_macs()
for interface in net_devices:
if self.pci_address == interface['pci_address']:
self.interface_name = interface['interface']
self.mac_address = interface['macAddress']
self.state = interface['state']
def get_sysnet_interfaces_and_macs(self):
"""Query sys fs and retrun list of interface data dicts
eg [
{
'interface': 'eth2',
'macAddress': 'a8:9d:21:cf:93:fc',
'pci_address': '0000:10:00.0',
'state': 'up'
},
{
'interface': 'eth3',
'macAddress': 'a8:9d:21:cf:93:fd',
'pci_address': '0000:10:00.1',
'state': 'down'
}
]
"""
net_devs = []
for sdir in glob.glob('/sys/class/net/*'):
sym_link = sdir + "/device"
if os.path.islink(sym_link):
fq_path = os.path.realpath(sym_link)
path = fq_path.split('/')
if 'virtio' in path[-1]:
pci_address = path[-2]
else:
pci_address = path[-1]
net_devs.append({
'interface': self.get_sysnet_interface(sdir),
'macAddress': self.get_sysnet_mac(sdir),
'pci_address': pci_address,
'state': self.get_sysnet_device_state(sdir),
})
return net_devs
def get_sysnet_mac(self, sysdir):
"""Extract MAC address from sys device file
:returns str: mac address"""
mac_addr_file = sysdir + '/address'
with open(mac_addr_file, 'r') as f:
read_data = f.read()
mac = read_data.strip()
hookenv.log('mac from {} is {}'.format(mac_addr_file, mac))
return mac
def get_sysnet_device_state(self, sysdir):
"""Extract device state from sys device file
:returns str: device state"""
state_file = sysdir + '/operstate'
with open(state_file, 'r') as f:
read_data = f.read()
state = read_data.strip()
hookenv.log('state from {} is {}'.format(state_file, state))
return state
def get_sysnet_interface(self, sysdir):
"""Extract device file from FQ path
:returns str: interface name"""
return sysdir.split('/')[-1]
class PCINetDevices(object):
"""PCINetDevices represents a collection of PCI Network devices on the
running system"""
def __init__(self):
"""Initialise a collection of PCINetDevice"""
pci_addresses = self.get_pci_ethernet_addresses()
self.pci_devices = [PCINetDevice(dev) for dev in pci_addresses]
def get_pci_ethernet_addresses(self):
"""Query lspci to retrieve a list of PCI address for devices of type
'Ethernet controller'
:returns list: List of PCI addresses of Ethernet controllers"""
cmd = ['lspci', '-m', '-D']
lspci_output = subprocess.check_output(cmd)
pci_addresses = []
for line in lspci_output.split('\n'):
columns = shlex.split(line)
if len(columns) > 1 and columns[1] == 'Ethernet controller':
pci_address = columns[0]
pci_addresses.append(format_pci_addr(pci_address))
return pci_addresses
def update_devices(self):
"""Update attributes of each device in collection"""
for pcidev in self.pci_devices:
pcidev.update_attributes()
def get_macs(self):
"""MAC addresses of all devices in collection
:returns list: List of MAC addresses"""
macs = []
for pcidev in self.pci_devices:
if pcidev.mac_address:
macs.append(pcidev.mac_address)
return macs
def get_device_from_mac(self, mac):
"""Given a MAC address return the corresponding PCINetDevice
:returns PCINetDevice"""
for pcidev in self.pci_devices:
if pcidev.mac_address == mac:
return pcidev
def get_device_from_pci_address(self, pci_addr):
"""Given a PCI address return the corresponding PCINetDevice
:returns PCINetDevice"""
for pcidev in self.pci_devices:
if pcidev.pci_address == pci_addr:
return pcidev
def rebind_orphans(self):
"""Unbind orphaned devices from the kernel module they are currently
using and then bind it with its default kernel module"""
self.unbind_orphans()
self.bind_orphans()
def unbind_orphans(self):
"""Unbind orphaned devices from the kernel module they are currently
using"""
for orphan in self.get_orphans():
orphan.unbind()
self.update_devices()
def bind_orphans(self):
"""Bind orphans with their default kernel module"""
for orphan in self.get_orphans():
orphan.bind(orphan.modalias_kmod)
self.update_devices()
def get_orphans(self):
"""An 'orphan' is a device which is not fully setup. It may not be
associated with a kernel module or may lay a name or MAC address.
:returns list: List of PCINetDevice"""
orphans = []
for pcidev in self.pci_devices:
if not pcidev.loaded_kmod or pcidev.loaded_kmod == 'igb_uio':
if not pcidev.interface_name and not pcidev.mac_address:
orphans.append(pcidev)
return orphans
class PCIInfo(object):
def __init__(self):
"""Inspect the charm config option 'mac-network-map' against the MAC
addresses on the running system.
Attributes:
user_requested_config dict Dictionary of MAC addresses and the
networks they are associated with.
local_macs list MAC addresses on local machine
pci_addresses list PCI Addresses of network devices on
local machine
vpe_dev_string str String containing PCI addresse in
format used by vpe.conf
local_mac_nets dict Dictionary of list of dicts with
interface and netork information
keyed on MAC address eg
{
'mac1': [{'interface': 'eth0', 'net': 'net1'},
{'interface': 'eth0', 'net': 'net2'}],
'mac2': [{'interface': 'eth1', 'net': 'net1'}],}
"""
self.user_requested_config = self.get_user_requested_config()
net_devices = PCINetDevices()
self.local_macs = net_devices.get_macs()
self.pci_addresses = []
self.local_mac_nets = {}
for mac in self.user_requested_config.keys():
hookenv.log('Checking if {} is on this host'.format(mac))
if mac in self.local_macs:
hookenv.log('{} is on this host'.format(mac))
device = net_devices.get_device_from_mac(mac)
hookenv.log('{} is {} and is currently {}'.format(mac,
device.pci_address, device.interface_name))
if device.state == 'up':
hookenv.log('Refusing to add {} to device list as it is '
'{}'.format(device.pci_address, device.state))
else:
self.pci_addresses.append(device.pci_address)
self.local_mac_nets[mac] = []
for conf in self.user_requested_config[mac]:
self.local_mac_nets[mac].append({
'net': conf.get('net'),
'interface': device.interface_name,
})
if self.pci_addresses:
self.pci_addresses.sort()
self.vpe_dev_string = 'dev ' + ' dev '.join(self.pci_addresses)
else:
self.vpe_dev_string = 'no-pci'
hookenv.log('vpe_dev_string {}'.format(self.vpe_dev_string))
def parse_mmap_entry(self, conf):
"""Extract mac and net pairs from list in the form
['mac=mac1', 'net=net1']
:returns tuple: (mac, net)
"""
entry = {a.split('=')[0]: a.split('=')[1] for a in conf}
return entry['mac'], entry['net']
def get_user_requested_config(self):
''' Parse the user requested config str
mac=<mac>;net=<net> and return a dict keyed on mac address
:returns dict: Dictionary of MAC addresses and the networks they are
associated with. eg
mac-network-map set to 'mac=mac1;net=net1
mac=mac1;net=net2
mac=mac2;net=net1'
returns:
{
'mac1': [{'net': 'net1'}, {'net': 'net2'}],
'mac2': [{'net': 'net1'}]}
}
'''
mac_net_config = {}
mac_map = hookenv.config('mac-network-map')
if mac_map:
for conf_group in mac_map.split():
try:
mac, net = self.parse_mmap_entry(conf_group.split(';'))
# Ignore bad config entries
except IndexError:
hookenv.log('Ignoring bad config entry {} in'
'mac-network-map'.format(conf_group))
continue
except KeyError:
hookenv.log('Ignoring bad config entry {} in'
'mac-network-map'.format(conf_group))
continue
try:
mac_net_config[mac].append({'net': net})
except KeyError:
mac_net_config[mac] = [{'net': net}]
return mac_net_config

13
charms_openstack/sdn/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
# Copyright 2016 Canonical Ltd
#
# 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.

273
charms_openstack/sdn/odl.py Normal file
View File

@@ -0,0 +1,273 @@
'''ODL Controller API integration'''
import requests
from jinja2 import Environment, FileSystemLoader
import charmhelpers.core.hookenv as hookenv
from charmhelpers.core.decorators import retry_on_exception
TEMPLATE_DIR = 'charms_openstack/sdn/templates'
class ODLInteractionFatalError(Exception):
''' Generic exception for failures in interaction with ODL '''
pass
class ODLConfig(requests.Session):
"""Class used for interacting with an ODL controller"""
def __init__(self, username, password, host, port='8181'):
"""Setup attributes for contacting ODLs http API"""
super(ODLConfig, self).__init__()
self.mount("http://", requests.adapters.HTTPAdapter(max_retries=5))
self.base_url = 'http://{}:{}'.format(host, port)
self.auth = (username, password)
self.proxies = {}
self.timeout = 10
self.conf_url = self.base_url + '/restconf/config'
self.oper_url = self.base_url + '/restconf/operational'
self.netmap_url = self.conf_url + '/neutron-device-map:neutron_net_map'
self.node_query_url = self.oper_url + '/opendaylight-inventory:nodes/'
yang_mod_path = ('/opendaylight-inventory:nodes/node/'
'controller-config/yang-ext:mount/config:modules')
self.node_mount_url = self.conf_url + yang_mod_path
@retry_on_exception(5, base_delay=30,
exc_type=requests.exceptions.ConnectionError)
def contact_odl(self, request_type, url, headers=None, data=None,
whitelist_rcs=None, retry_rcs=None):
"""Send request to ODL controller and return the response
:param request_type: str HTTP Request Methods (GET, POST, DELETE etc)
:param url: str URL to issue request against
:param headers: str HTTP Header to be sent in request
:param data: str Data to be sent in request
:param whitelist_rcs: List List of acceptable return codes.
:param retry_rcs: List List of return codes which should trigger a
retry
:returns requests.Response: Response from request
"""
response = self.request(request_type, url, data=data, headers=headers)
ok_codes = [requests.codes.ok, requests.codes.no_content]
retry_codes = [requests.codes.service_unavailable]
if whitelist_rcs:
ok_codes.extend(whitelist_rcs)
if retry_rcs:
retry_codes.extend(retry_rcs)
if response.status_code not in ok_codes:
if response.status_code in retry_codes:
msg = "Recieved {} from ODL on {}".format(response.status_code,
url)
raise requests.exceptions.ConnectionError(msg)
else:
msg = "Contact failed status_code={}, {}".format(
response.status_code, url)
raise ODLInteractionFatalError(msg)
return response
def get_networks(self):
"""Query ODL for map of networks and physical hardware
:returns dict: neutron_net_map eg:
{
"physicalNetwork": [
{
"name": "net_d12",
"device": [
{
"device-name": "C240-M4-6",
"device-type": "vhostuser",
"interface": [
{
"interface-name": "TenGigabitEthernet6/0/0",
"macAddress": "84:b8:02:2a:5f:c3"
}
]
}
]
},
{
"name": "net_d11",
"device": [
{
"device-name": "C240-M4-6",
"device-type": "vhostuser",
"interface": [
{
"interface-name": "TenGigabitEthernet7/0/0",
"macAddress": "84:b8:02:2a:5f:c4"
}
]
}
]
}
}
"""
hookenv.log('Querying macs registered with odl')
# No netmap may have been registered yet, so 404 is ok
odl_req = self.contact_odl(
'GET', self.netmap_url, whitelist_rcs=[requests.codes.not_found])
if not odl_req:
hookenv.log('neutron_net_map not found in ODL')
return {}
odl_json = odl_req.json()
if odl_json.get('neutron_net_map'):
hookenv.log('neutron_net_map returned by ODL')
return odl_json['neutron_net_map']
else:
hookenv.log('neutron_net_map NOT returned by ODL')
return {}
def delete_net_device_entry(self, net, device_name):
"""Delete device from network
:param net: str Netork name that device should be deleted from
:param device_name: str Name of device to be deleted from network
"""
obj_url = self.netmap_url + \
'physicalNetwork/{}/device/{}'.format(net, device_name)
self.contact_odl('DELETE', obj_url)
def get_odl_registered_nodes(self):
"""Query ODL to retieve a list of registered servers
:return List: List of registered servers
"""
hookenv.log('Querying nodes registered with odl')
odl_req = self.contact_odl('GET', self.node_query_url)
odl_json = odl_req.json()
odl_node_ids = []
if odl_json.get('nodes'):
odl_nodes = odl_json['nodes'].get('node', [])
odl_node_ids = [entry['id'] for entry in odl_nodes]
hookenv.log(
'Following nodes are registered: ' + ' '.join(odl_node_ids))
return odl_node_ids
def odl_register_node(self, device_name, ip):
"""Register server with ODL
:param device_name: str
:param ip: str
"""
hookenv.log('Registering node {} ({}) with ODL'.format(
device_name, ip))
payload = self.render_node_xml(device_name, ip)
headers = {'Content-Type': 'application/xml'}
# Strictly a client should not retry on recipt of a bad_request (400)
# but ODL return 400s while it is initialising
self.contact_odl(
'POST', self.node_mount_url, headers=headers, data=payload,
retry_rcs=[requests.codes.bad_request])
def odl_register_macs(self, device_name, network, interface, mac,
device_type='vhostuser'):
"""Register a device as part of a network
:param device_name: str Name of server device that has the device
:param interface: str Name of the device
:param mac: str MAC address of the device
:param device_type: str Device type
"""
hookenv.log('Registering {} and {} on {}'.format(
network, interface, mac))
payload = self.render_mac_xml(device_name, network, interface, mac,
device_type)
headers = {'Content-Type': 'application/json'}
self.contact_odl(
'POST', self.netmap_url, headers=headers, data=payload)
def get_macs_networks(self, mac):
"""List of networks a MAC address is registered with
:returns str: List of Network names address is registered with
"""
registered_networks = self.get_networks()
nets = []
phy_nets = registered_networks.get('physicalNetwork')
if phy_nets:
for network in phy_nets:
for device in network.get('device', []):
for interface in device['interface']:
if interface['macAddress'] == mac:
nets.append(network['name'])
return nets
def is_device_registered(self, device_name):
"""Is device registered in ODL
:returns boolean:
"""
return device_name in self.get_odl_registered_nodes()
def is_net_device_registered(self, net_name, device_name, interface_name,
mac, device_type='vhostuser'):
"""Is device registered as part of a given network
:param net_name,: str Name of network
:param device_name: str Name of server device that has the device
:param interface_name: str Name of the device
:param mac: str MAC address of the device
:param device_type: str Device type
:returns boolean:
"""
networks = self.get_networks()
phy_nets = networks.get('physicalNetwork')
if phy_nets:
for net in phy_nets:
if net_name == net['name']:
for dev in net.get('device', []):
if device_name == dev['device-name'] \
and dev['device-type'] == device_type:
for interface in dev['interface']:
if (interface_name ==
interface['interface-name'] and
mac == interface['macAddress']):
return True
return False
def render_node_xml(self, device_name, ip, user='admin', password='admin'):
"""Return XML for rendering a node
:param device_name: str Name of server to be registered
:param ip: str IP on server to be registered
:param user: str username for ODL controller to use to talk back to
server
:param password: str password for ODL controller to use to talk back to
server
:returns str: XML for rendering a node
"""
env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
template = env.get_template('odl_node_registration')
node_xml = template.render(
host=device_name,
ip=ip,
username=user,
password=password,
)
return node_xml
def render_mac_xml(self, device_name, network, interface, mac,
device_type='vhostuser'):
"""Register a device as part of a network
:param device_name: str Name of server device that has the device
:param network: str Name of the network for device to be registered
against
:param interface: str Name of the device
:param mac: str MAC address of the device
:param device_type: str Device type
"""
env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
template = env.get_template('odl_mac_registration')
mac_xml = template.render(
host=device_name,
network=network,
interface=interface,
mac=mac,
device_type=device_type,
)
return mac_xml

33
charms_openstack/sdn/ovs.py Normal file
View File

@@ -0,0 +1,33 @@
import subprocess
import charmhelpers.core.hookenv as hookenv
def set_manager(connection_url):
"""Configure the OVSDB manager for the switch
:param connection_url: str URL for OVS manager
"""
subprocess.check_call(['ovs-vsctl', 'set-manager', connection_url])
@hookenv.cached
def _get_ovstbl():
ovstbl = subprocess.check_output(['ovs-vsctl', 'get',
'Open_vSwitch', '.',
'_uuid']).strip()
return ovstbl
def set_config(key, value, table='other_config'):
"""Set key value pairs in a table
:param key: str
:param value: str
:param table: str Table to apply setting to
"""
subprocess.check_call(
['ovs-vsctl', 'set',
'Open_vSwitch', _get_ovstbl(),
'{}:{}={}'.format(table, key, value)]
)

13
charms_openstack/sdn/templates/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
# Copyright 2016 Canonical Ltd
#
# 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.

17
charms_openstack/sdn/templates/odl_mac_registration Normal file
View File

@@ -0,0 +1,17 @@
{
"neutron-device-map:physicalNetwork": {
"name": "{{ network }}",
"device": [
{
"interface": [
{
"macAddress": "{{ mac }}",
"interface-name": "{{ interface }}"
}
],
"device-name": "{{ host }}",
"device-type": "vhostuser"
}
]
}
}

29
charms_openstack/sdn/templates/odl_node_registration Normal file
View File

@@ -0,0 +1,29 @@
<module xmlns="urn:opendaylight:params:xml:ns:yang:controller:config">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">prefix:sal-netconf-connector</type>
<name>{{ host }}</name>
<address xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">{{ ip }}</address>
<port xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">2022</port>
<username xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">{{ username }}</username>
<password xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">{{ password }}</password>
<tcp-only xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">false</tcp-only>
<event-executor xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-event-executor</type>
<name>global-event-executor</name>
</event-executor>
<binding-registry xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding">prefix:binding-broker-osgi-registry</type>
<name>binding-osgi-broker</name>
</binding-registry>
<dom-registry xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom">prefix:dom-broker-osgi-registry</type>
<name>dom-broker</name>
</dom-registry>
<client-dispatcher xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:config:netconf">prefix:netconf-client-dispatcher</type>
<name>global-netconf-dispatcher</name>
</client-dispatcher>
<processing-executor xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:connector:netconf">
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:threadpool">prefix:threadpool</type>
<name>global-netconf-processing-executor</name>
</processing-executor>
</module>

4
test-requirements.txt
View File

@@ -1,3 +1,7 @@
simplejson
requests
httpretty
pep8
flake8>=2.2.4,<=2.4.1
os-testr>=0.4.1
paramiko<2.0

21
unit_tests/__init__.py
View File

@@ -21,6 +21,7 @@ charmhelpers = mock.MagicMock()
sys.modules['apt_pkg'] = apt_pkg
sys.modules['charmhelpers'] = charmhelpers
sys.modules['charmhelpers.core'] = charmhelpers.core
sys.modules['charmhelpers.core.decorators'] = charmhelpers.core.decorators
sys.modules['charmhelpers.core.hookenv'] = charmhelpers.core.hookenv
sys.modules['charmhelpers.core.host'] = charmhelpers.core.host
sys.modules['charmhelpers.core.templating'] = charmhelpers.core.templating
@@ -39,3 +40,23 @@ sys.modules['charmhelpers.cli'] = charmhelpers.cli
sys.modules['charmhelpers.contrib.hahelpers'] = charmhelpers.contrib.hahelpers
sys.modules['charmhelpers.contrib.hahelpers.cluster'] = (
charmhelpers.contrib.hahelpers.cluster)
def _fake_retry(num_retries, base_delay=0, exc_type=Exception):
def _retry_on_exception_inner_1(f):
def _retry_on_exception_inner_2(*args, **kwargs):
return f(*args, **kwargs)
return _retry_on_exception_inner_2
return _retry_on_exception_inner_1
mock.patch(
'charmhelpers.core.decorators.retry_on_exception',
_fake_retry).start()
def _fake_cached(f):
return f
mock.patch(
'charmhelpers.core.hookenv.cached',
_fake_cached).start()

74
unit_tests/odl_responses.py Normal file
View File

@@ -0,0 +1,74 @@
NEUTRON_NET_MAP = """
{
"neutron_net_map": {
"physicalNetwork": [
{
"name": "net_d12",
"device": [
{
"device-name": "C240-M4-6",
"device-type": "vhostuser",
"interface": [
{
"interface-name": "TenGigabitEthernet6/0/0",
"macAddress": "84:b8:02:2a:5f:c3"
}
]
}
]
},
{
"name": "net_d11",
"device": [
{
"device-name": "C240-M4-6",
"device-type": "vhostuser",
"interface": [
{
"interface-name": "TenGigabitEthernet7/0/0",
"macAddress": "84:b8:02:2a:5f:c4"
}
]
}
]
},
{
"name": "net_d10",
"device": [
{
"device-name": "C240-M4-6",
"device-type": "vhostuser",
"interface": [
{
"interface-name": "TenGigabitEthernet6/0/0",
"macAddress": "84:b8:02:2a:5f:c3"
}
]
}
]
}
]
}
}"""
NEUTRON_NET_MAP_EMPTY = """
{
"neutron_net_map": {
"physicalNetwork": [
]
}
}"""
ODL_REGISTERED_NODES = """
{
"nodes": {
"node": [
{
"id": "C240-M4-6"
},
{
"id": "controller-config"
}
]
}
}"""

198
unit_tests/pci_responses.py Normal file
View File

@@ -0,0 +1,198 @@
import copy
# flake8: noqa
LSPCI = """
0000:00:00.0 "Host bridge" "Intel Corporation" "Haswell-E DMI2" -r02 "Intel Corporation" "Device 0000"
0000:00:03.0 "PCI bridge" "Intel Corporation" "Haswell-E PCI Express Root Port 3" -r02 "" ""
0000:00:03.2 "PCI bridge" "Intel Corporation" "Haswell-E PCI Express Root Port 3" -r02 "" ""
0000:00:05.0 "System peripheral" "Intel Corporation" "Haswell-E Address Map, VTd_Misc, System Management" -r02 "" ""
0000:00:05.1 "System peripheral" "Intel Corporation" "Haswell-E Hot Plug" -r02 "" ""
0000:00:05.2 "System peripheral" "Intel Corporation" "Haswell-E RAS, Control Status and Global Errors" -r02 "" ""
0000:00:05.4 "PIC" "Intel Corporation" "Haswell-E I/O Apic" -r02 -p20 "Intel Corporation" "Device 0000"
0000:00:11.0 "Unassigned class [ff00]" "Intel Corporation" "Wellsburg SPSR" -r05 "Intel Corporation" "Device 7270"
0000:00:11.4 "SATA controller" "Intel Corporation" "Wellsburg sSATA Controller [AHCI mode]" -r05 -p01 "Cisco Systems Inc" "Device 0067"
0000:00:16.0 "Communication controller" "Intel Corporation" "Wellsburg MEI Controller #1" -r05 "Intel Corporation" "Device 7270"
0000:00:16.1 "Communication controller" "Intel Corporation" "Wellsburg MEI Controller #2" -r05 "Intel Corporation" "Device 7270"
0000:00:1a.0 "USB controller" "Intel Corporation" "Wellsburg USB Enhanced Host Controller #2" -r05 -p20 "Intel Corporation" "Device 7270"
0000:00:1c.0 "PCI bridge" "Intel Corporation" "Wellsburg PCI Express Root Port #1" -rd5 "" ""
0000:00:1c.3 "PCI bridge" "Intel Corporation" "Wellsburg PCI Express Root Port #4" -rd5 "" ""
0000:00:1c.4 "PCI bridge" "Intel Corporation" "Wellsburg PCI Express Root Port #5" -rd5 "" ""
0000:00:1d.0 "USB controller" "Intel Corporation" "Wellsburg USB Enhanced Host Controller #1" -r05 -p20 "Intel Corporation" "Device 7270"
0000:00:1f.0 "ISA bridge" "Intel Corporation" "Wellsburg LPC Controller" -r05 "Intel Corporation" "Device 7270"
0000:00:1f.2 "SATA controller" "Intel Corporation" "Wellsburg 6-Port SATA Controller [AHCI mode]" -r05 -p01 "Cisco Systems Inc" "Device 0067"
0000:01:00.0 "PCI bridge" "Cisco Systems Inc" "VIC 82 PCIe Upstream Port" -r01 "" ""
0000:02:00.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:02:01.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:03:00.0 "Unclassified device [00ff]" "Cisco Systems Inc" "VIC Management Controller" -ra2 "Cisco Systems Inc" "Device 012e"
0000:04:00.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Upstream Port" -ra2 "" ""
0000:05:00.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:05:01.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:05:02.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:05:03.0 "PCI bridge" "Cisco Systems Inc" "VIC PCIe Downstream Port" -ra2 "" ""
0000:06:00.0 "Ethernet controller" "Cisco Systems Inc" "VIC Ethernet NIC" -ra2 "Cisco Systems Inc" "Device 012e"
0000:07:00.0 "Ethernet controller" "Cisco Systems Inc" "VIC Ethernet NIC" -ra2 "Cisco Systems Inc" "Device 012e"
0000:08:00.0 "Fibre Channel" "Cisco Systems Inc" "VIC FCoE HBA" -ra2 "Cisco Systems Inc" "Device 012e"
0000:09:00.0 "Fibre Channel" "Cisco Systems Inc" "VIC FCoE HBA" -ra2 "Cisco Systems Inc" "Device 012e"
0000:0b:00.0 "RAID bus controller" "LSI Logic / Symbios Logic" "MegaRAID SAS-3 3108 [Invader]" -r02 "Cisco Systems Inc" "Device 00db"
0000:0f:00.0 "VGA compatible controller" "Matrox Electronics Systems Ltd." "MGA G200e [Pilot] ServerEngines (SEP1)" -r02 "Cisco Systems Inc" "Device 0101"
0000:10:00.0 "Ethernet controller" "Intel Corporation" "I350 Gigabit Network Connection" -r01 "Cisco Systems Inc" "Device 00d6"
0000:10:00.1 "Ethernet controller" "Intel Corporation" "I350 Gigabit Network Connection" -r01 "Cisco Systems Inc" "Device 00d6"
0000:7f:08.0 "System peripheral" "Intel Corporation" "Haswell-E QPI Link 0" -r02 "Intel Corporation" "Haswell-E QPI Link 0"
"""
CONFD_CLI = """
NAME PHYS ADDRESS
--------------------------------------------
TenGigabitEthernet6/0/0 84:b8:02:2a:5f:c3
TenGigabitEthernet7/0/0 84:b8:02:2a:5f:c4
local0 -
"""
CONFD_CLI_ONE_MISSING = """
NAME PHYS ADDRESS
--------------------------------------------
TenGigabitEthernet6/0/0 84:b8:02:2a:5f:c3
local0 -
"""
CONFD_CLI_INVMAC = """
NAME PHYS ADDRESS
--------------------------------------------
TenGigabitEthernet6/0/0 no:ta:va:li:dm:ac
TenGigabitEthernet7/0/0 84:b8:02:2a:5f:c4
local0 -
"""
CONFD_CLI_NODEVS = """
NAME PHYS ADDRESS
--------------------------------------------
local0 -
"""
CONFD_CLI_NOLOCAL = """
NAME PHYS ADDRESS
--------------------------------------------
"""
SYS_TREE = {
'/sys/class/net/eth2': '../../devices/pci0000:00/0000:00:1c.4/0000:10:00.0/net/eth2',
'/sys/class/net/eth3': '../../devices/pci0000:00/0000:00:1c.4/0000:10:00.1/net/eth3',
'/sys/class/net/juju-br0': '../../devices/virtual/net/juju-br0',
'/sys/class/net/lo': '../../devices/virtual/net/lo',
'/sys/class/net/lxcbr0': '../../devices/virtual/net/lxcbr0',
'/sys/class/net/veth1GVRCF': '../../devices/virtual/net/veth1GVRCF',
'/sys/class/net/veth7AXEUK': '../../devices/virtual/net/veth7AXEUK',
'/sys/class/net/vethACOIJJ': '../../devices/virtual/net/vethACOIJJ',
'/sys/class/net/vethMQ819H': '../../devices/virtual/net/vethMQ819H',
'/sys/class/net/virbr0': '../../devices/virtual/net/virbr0',
'/sys/class/net/virbr0-nic': '../../devices/virtual/net/virbr0-nic',
'/sys/devices/pci0000:00/0000:00:1c.4/0000:10:00.0/net/eth2/device': '../../../0000:10:00.0',
'/sys/devices/pci0000:00/0000:00:1c.4/0000:10:00.1/net/eth3/device': '../../../0000:10:00.1',
}
LSPCI_KS_IGB_UNBOUND = """
{} Ethernet controller: Intel Corporation I350 Gigabit Network Connection (rev 01)
Subsystem: Cisco Systems Inc Device 00d6
"""
LSPCI_KS_IGB_BOUND = """
{} Ethernet controller: Intel Corporation I350 Gigabit Network Connection (rev 01)
Subsystem: Cisco Systems Inc Device 00d6
Kernel driver in use: igb
"""
LSPCI_KS_IGBUIO_BOUND = """
{} Ethernet controller: Cisco Systems Inc VIC Ethernet NIC (rev a2)
Subsystem: Cisco Systems Inc VIC 1240 MLOM Ethernet NIC
Kernel driver in use: igb_uio
"""
LSPCI_KS = {
'0000:06:00.0': LSPCI_KS_IGBUIO_BOUND.format('06:00.0'),
'0000:10:00.0': LSPCI_KS_IGB_BOUND.format('10:00.0'),
}
MODALIAS = """
alias pci:v00001137d00000071sv*sd*bc*sc*i* enic
alias pci:v00001137d00000044sv*sd*bc*sc*i* enic
alias pci:v00001137d00000043sv*sd*bc*sc*i* enic
alias pci:v00008086d000010D6sv*sd*bc*sc*i* igb
alias pci:v00008086d000010A9sv*sd*bc*sc*i* igb
alias pci:v00008086d00001522sv*sd*bc*sc*i* igb
alias pci:v00008086d00001521sv*sd*bc*sc*i* igb
alias pci:v00008086d0000157Csv*sd*bc*sc*i* igb
"""
LSPCI_NS = {
'0000:06:00.0': "06:00.0 0200: 1137:0043 (rev a2)",
'0000:07:00.0': "07:00.0 0200: 1137:0043 (rev a2)",
'0000:10:00.0': "10:00.0 0200: 8086:1521 (rev 01)",
'0000:10:00.1': "10:00.1 0200: 8086:1521 (rev 01)",
}
FILE_CONTENTS = {
'/sys/class/net/eth2/address': 'a8:9d:21:cf:93:fc',
'/sys/class/net/eth3/address': 'a8:9d:21:cf:93:fd',
'/sys/class/net/eth2/operstate': 'up',
'/sys/class/net/eth3/operstate': 'down',
'/lib/modules/3.13.0-35-generic/modules.alias': MODALIAS,
}
COMMANDS = {
'LSPCI_MD': ['lspci', '-m', '-D'],
'LSPCI_KS': ['lspci', '-ks'],
'LSPCI_NS': ['lspci', '-ns'],
'UNAME_R': ['uname', '-r'],
'CONFD_CLI': ['/opt/cisco/vpe/bin/confd_cli', '-N', '-C', '-u', 'system'],
}
NET_SETUP = {
'LSPCI_MD': LSPCI,
'UNAME_R': '3.13.0-35-generic',
'CONFD_CLI': CONFD_CLI,
'0000:06:00.0': {
'LSPCI_KS': LSPCI_KS_IGBUIO_BOUND.format('06:00.0'),
'LSPCI_NS': "06:00.0 0200: 1137:0043 (rev a2)",
},
'0000:07:00.0': {
'LSPCI_KS': LSPCI_KS_IGBUIO_BOUND.format('07:00.0'),
'LSPCI_NS': "07:00.0 0200: 1137:0043 (rev a2)",
},
'0000:10:00.0': {
'LSPCI_KS': LSPCI_KS_IGB_BOUND.format('10:00.0'),
'LSPCI_NS': "10:00.0 0200: 8086:1521 (rev 01)",
},
'0000:10:00.1': {
'LSPCI_KS': LSPCI_KS_IGB_BOUND.format('10:00.1'),
'LSPCI_NS': "10:00.1 0200: 8086:1521 (rev 01)",
},
}
NET_SETUP_ORPHAN = copy.deepcopy(NET_SETUP)
NET_SETUP_ORPHAN['CONFD_CLI'] = CONFD_CLI_ONE_MISSING
NET_SETUP_ORPHAN['0000:07:00.0']['LSPCI_KS'] = LSPCI_KS_IGB_UNBOUND.format('07:00.0')
QN_CONF = """
lc_procs = { svm_cleanup vpe confd orca }
install_root = "/cisco"
svm_cleanup = {
pgm = "$(install_root)/bin/svm_cleanup",
run_once = "yes",
max_synchronous_wait = "5.0",
console_output = "yes"
}
vpe = {
pgm = "$(install_root)/bin/vpe",
args = "unix { nodaemon log /tmp/vpe.log cli-listen localhost:5002 full-coredump } api-trace { on } dpdk { socket-mem 1024 dev 0000:00:06.0 }",
max_cpu_percent = "111.0",
console_output = "yes",
crash_reset_all="yes"
}
confd = {
pgm = "$(install_root)/bin/confd",
args = "--foreground -c $(install_root)/etc/confd/confd.conf",
max_cpu_percent = "111.0",
crash_reset_all="yes"
}
orca = {
pgm = "$(install_root)/bin/orca",
args = "unix { nodaemon log /tmp/orca.log cli-listen localhost:5003 }",
console_output = "yes",
max_cpu_percent = "111.0",
crash_reset_all="yes"
"""
DPKG_L = """
ii net-tools 1.60-25ubuntu2.1 amd64 The NET-3 networking toolkit
ii netbase 5.2 all Basic TCP/IP networking system
ii netcat-openbsd 1.105-7ubuntu1 amd64 TCP/IP swiss army knife
ii nova-common 1:2014.1.4-0ubuntu2.1.1~ppa201506221720 all OpenStack Compute - common files
"""

558
unit_tests/test_charms_openstack_devices_pci.py Normal file
View File

@@ -0,0 +1,558 @@
# Copyright 2016 Canonical Ltd
#
# 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.
# Note that the unit_tests/__init__.py has the following lines to stop
# side effects from the imorts from charm helpers.
# sys.path.append('./lib')
# mock out some charmhelpers libraries as they have apt install side effects
# sys.modules['charmhelpers.contrib.openstack.utils'] = mock.MagicMock()
# sys.modules['charmhelpers.contrib.network.ip'] = mock.MagicMock()
from __future__ import absolute_import
import mock
import charms_openstack.devices.pci as pci
import unit_tests.pci_responses as pci_responses
import unit_tests.utils as utils
def mocked_subprocess(subproc_map=None):
def _subproc(cmd, stdin=None):
for key in pci_responses.COMMANDS.keys():
if pci_responses.COMMANDS[key] == cmd:
return subproc_map[key]
elif pci_responses.COMMANDS[key] == cmd[:-1]:
return subproc_map[cmd[-1]][key]
if not subproc_map:
subproc_map = pci_responses.NET_SETUP
return _subproc
class mocked_filehandle(object):
def _setfilename(self, fname, omode):
self.FILENAME = fname
def _getfilecontents_read(self):
return pci_responses.FILE_CONTENTS[self.FILENAME]
def _getfilecontents_readlines(self):
return pci_responses.FILE_CONTENTS[self.FILENAME].split('\n')
class PCIDevTest(utils.BaseTestCase):
def test_format_pci_addr(self):
self.assertEqual(pci.format_pci_addr('0:0:1.1'), '0000:00:01.1')
self.assertEqual(pci.format_pci_addr(
'0000:00:02.1'), '0000:00:02.1')
class PCINetDeviceTest(utils.BaseTestCase):
def test_init(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
a = pci.PCINetDevice('pciaddr')
self.update_attributes.assert_called_once_with()
self.assertEqual(a.pci_address, 'pciaddr')
def test_update_attributes(self):
self.patch_object(pci.PCINetDevice, '__init__')
self.patch_object(pci.PCINetDevice, 'loaded_kmod')
self.patch_object(pci.PCINetDevice, 'update_modalias_kmod')
self.patch_object(pci.PCINetDevice, 'update_interface_info')
a = pci.PCINetDevice('pciaddr')
a.update_attributes()
self.update_modalias_kmod.assert_called_once_with()
self.update_interface_info.assert_called_once_with()
def test_loaded_kmod(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = mocked_subprocess()
device = pci.PCINetDevice('0000:06:00.0')
self.assertEqual(device.loaded_kmod, 'igb_uio')
def test_update_modalias_kmod(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
device = pci.PCINetDevice('0000:07:00.0')
self.subprocess.check_output.side_effect = mocked_subprocess()
with utils.patch_open() as (_open, _file):
super_fh = mocked_filehandle()
_file.readlines = mock.MagicMock()
_open.side_effect = super_fh._setfilename
_file.read.side_effect = super_fh._getfilecontents_read
_file.readlines.side_effect = super_fh._getfilecontents_readlines
device.update_modalias_kmod()
self.assertEqual(device.modalias_kmod, 'enic')
def test_update_interface_info_call_vpeinfo(self):
self.patch_object(pci.PCINetDevice, 'update_interface_info_eth')
self.patch_object(pci.PCINetDevice, 'update_interface_info_vpe')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_kernel_name')
self.patch_object(pci.PCINetDevice, 'loaded_kmod', new='igb_uio')
self.patch_object(pci, 'subprocess')
self.get_kernel_name.return_value = '3.13.0-77-generic'
self.subprocess.check_output.side_effect = \
mocked_subprocess()
dev6 = pci.PCINetDevice('0000:06:00.0')
dev6.update_interface_info()
self.update_interface_info_vpe.assert_called_with()
self.assertFalse(self.update_interface_info_eth.called)
def test_update_interface_info_call_ethinfo(self):
self.patch_object(pci.PCINetDevice, 'update_interface_info_eth')
self.patch_object(pci.PCINetDevice, 'update_interface_info_vpe')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_kernel_name')
self.patch_object(pci.PCINetDevice, 'loaded_kmod', new='igb')
self.patch_object(pci, 'subprocess')
self.get_kernel_name.return_value = '3.13.0-77-generic'
self.subprocess.check_output.side_effect = \
mocked_subprocess()
dev = pci.PCINetDevice('0000:10:00.0')
dev.update_interface_info()
self.update_interface_info_eth.assert_called_with()
self.assertFalse(self.update_interface_info_vpe.called)
def test_test_update_interface_info_orphan(self):
self.patch_object(pci.PCINetDevice, 'update_interface_info_eth')
self.patch_object(pci.PCINetDevice, 'update_interface_info_vpe')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_kernel_name')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = \
mocked_subprocess(
subproc_map=pci_responses.NET_SETUP_ORPHAN)
dev = pci.PCINetDevice('0000:07:00.0')
dev.update_interface_info()
self.assertFalse(self.update_interface_info_vpe.called)
self.assertFalse(self.update_interface_info_eth.called)
self.assertEqual(dev.interface_name, None)
self.assertEqual(dev.mac_address, None)
def test_get_kernel_name(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
dev = pci.PCINetDevice('0000:07:00.0')
self.subprocess.check_output.return_value = '3.13.0-55-generic'
self.assertEqual(dev.get_kernel_name(), '3.13.0-55-generic')
def test_pci_rescan(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
dev = pci.PCINetDevice('0000:07:00.0')
with utils.patch_open() as (_open, _file):
dev.pci_rescan()
_open.assert_called_with('/sys/bus/pci/rescan', 'w')
_file.write.assert_called_with('1')
def test_bind(self):
self.patch_object(pci.PCINetDevice, 'pci_rescan')
self.patch_object(pci.PCINetDevice, 'update_attributes')
dev = pci.PCINetDevice('0000:07:00.0')
with utils.patch_open() as (_open, _file):
dev.bind('enic')
_open.assert_called_with('/sys/bus/pci/drivers/enic/bind', 'w')
_file.write.assert_called_with('0000:07:00.0')
self.pci_rescan.assert_called_with()
self.update_attributes.assert_called_with()
def test_unbind(self):
self.patch_object(pci.PCINetDevice, 'pci_rescan')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'loaded_kmod', new='igb_uio')
dev = pci.PCINetDevice('0000:07:00.0')
with utils.patch_open() as (_open, _file):
dev.unbind()
_open.assert_called_with(
'/sys/bus/pci/drivers/igb_uio/unbind', 'w')
_file.write.assert_called_with('0000:07:00.0')
self.pci_rescan.assert_called_with()
self.update_attributes.assert_called_with()
def test_update_interface_info_vpe(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_vpe_interfaces_and_macs')
self.get_vpe_interfaces_and_macs.return_value = [
{
'interface': 'TenGigabitEthernet6/0/0',
'macAddress': '84:b8:02:2a:5f:c3',
'pci_address': '0000:06:00.0'},
{
'interface': 'TenGigabitEthernet7/0/0',
'macAddress': '84:b8:02:2a:5f:c4',
'pci_address': '0000:07:00.0'}]
dev = pci.PCINetDevice('0000:07:00.0')
dev.update_interface_info_vpe()
self.assertEqual('TenGigabitEthernet7/0/0', dev.interface_name)
self.assertEqual('84:b8:02:2a:5f:c4', dev.mac_address)
self.assertEqual('vpebound', dev.state)
def test_update_interface_info_vpe_orphan(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_vpe_interfaces_and_macs')
self.get_vpe_interfaces_and_macs.return_value = [
{
'interface': 'TenGigabitEthernet6/0/0',
'macAddress': '84:b8:02:2a:5f:c3',
'pci_address': '0000:06:00.0'}]
dev = pci.PCINetDevice('0000:07:00.0')
dev.update_interface_info_vpe()
self.assertEqual(None, dev.interface_name)
self.assertEqual(None, dev.mac_address)
self.assertEqual(None, dev.state)
def test_get_vpe_cli_out(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = \
mocked_subprocess()
dev = pci.PCINetDevice('0000:07:00.0')
self.assertTrue('local0' in dev.get_vpe_cli_out())
def test_get_vpe_interfaces_and_macs(self):
self.patch_object(pci.PCINetDevice, 'get_vpe_cli_out')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = \
mocked_subprocess()
self.get_vpe_cli_out.return_value = pci_responses.CONFD_CLI
dev = pci.PCINetDevice('0000:07:00.0')
vpe_devs = dev.get_vpe_interfaces_and_macs()
expect = [
{
'interface': 'TenGigabitEthernet6/0/0',
'macAddress': '84:b8:02:2a:5f:c3',
'pci_address': '0000:06:00.0'
},
{
'interface': 'TenGigabitEthernet7/0/0',
'macAddress': '84:b8:02:2a:5f:c4',
'pci_address': '0000:07:00.0'
},
]
self.assertEqual(vpe_devs, expect)
def test_get_vpe_interfaces_and_macs_invalid_cli(self):
self.patch_object(pci.PCINetDevice, 'get_vpe_cli_out')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = \
mocked_subprocess()
dev = pci.PCINetDevice('0000:07:00.0')
self.get_vpe_cli_out.return_value = pci_responses.CONFD_CLI_NOLOCAL
with self.assertRaises(pci.VPECLIException):
dev.get_vpe_interfaces_and_macs()
def test_get_vpe_interfaces_and_macs_invmac(self):
self.patch_object(pci.PCINetDevice, 'get_vpe_cli_out')
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci, 'subprocess')
self.subprocess.check_output.side_effect = \
mocked_subprocess()
dev = pci.PCINetDevice('0000:07:00.0')
self.get_vpe_cli_out.return_value = pci_responses.CONFD_CLI_INVMAC
vpe_devs = dev.get_vpe_interfaces_and_macs()
expect = [
{
'interface': 'TenGigabitEthernet7/0/0',
'macAddress': '84:b8:02:2a:5f:c4',
'pci_address': '0000:07:00.0'
},
]
self.assertEqual(vpe_devs, expect)
def test_extract_pci_addr_from_vpe_interface(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
dev = pci.PCINetDevice('0000:07:00.0')
self.assertEqual(dev.extract_pci_addr_from_vpe_interface(
'TenGigabitEthernet1/1/1'), '0000:01:01.1')
self.assertEqual(dev.extract_pci_addr_from_vpe_interface(
'TenGigabitEtherneta/0/0'), '0000:0a:00.0')
self.assertEqual(dev.extract_pci_addr_from_vpe_interface(
'GigabitEthernet0/2/0'), '0000:00:02.0')
def test_update_interface_info_eth(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
self.patch_object(pci.PCINetDevice, 'get_sysnet_interfaces_and_macs')
dev = pci.PCINetDevice('0000:10:00.0')
self.get_sysnet_interfaces_and_macs.return_value = [
{
'interface': 'eth2',
'macAddress': 'a8:9d:21:cf:93:fc',
'pci_address': '0000:10:00.0',
'state': 'up'
},
{
'interface': 'eth3',
'macAddress': 'a8:9d:21:cf:93:fd',
'pci_address': '0000:10:00.1',
'state': 'down'
}
]
dev.update_interface_info_eth()
self.assertEqual(dev.interface_name, 'eth2')
def test_get_sysnet_interfaces_and_macs_virtio(self):
self.patch_object(pci.glob, 'glob')
self.patch_object(pci.os.path, 'islink')
self.patch_object(pci.os.path, 'realpath')
self.patch_object(pci.PCINetDevice, 'get_sysnet_device_state')
self.patch_object(pci.PCINetDevice, 'get_sysnet_mac')
self.patch_object(pci.PCINetDevice, 'get_sysnet_interface')
self.patch_object(pci.PCINetDevice, 'update_attributes')
dev = pci.PCINetDevice('0000:06:00.0')
self.glob.return_value = ['/sys/class/net/eth2']
self.get_sysnet_interface.return_value = 'eth2'
self.get_sysnet_mac.return_value = 'a8:9d:21:cf:93:fc'
self.get_sysnet_device_state.return_value = 'up'
self.realpath.return_value = ('/sys/devices/pci0000:00/0000:00:07.0/'
'virtio5')
self.islink.return_value = True
expect = {
'interface': 'eth2',
'macAddress': 'a8:9d:21:cf:93:fc',
'pci_address': '0000:00:07.0',
'state': 'up',
}
self.assertEqual(dev.get_sysnet_interfaces_and_macs(), [expect])
def test_get_sysnet_interfaces_and_macs(self):
self.patch_object(pci.glob, 'glob')
self.patch_object(pci.os.path, 'islink')
self.patch_object(pci.os.path, 'realpath')
self.patch_object(pci.PCINetDevice, 'get_sysnet_device_state')
self.patch_object(pci.PCINetDevice, 'get_sysnet_mac')
self.patch_object(pci.PCINetDevice, 'get_sysnet_interface')
self.patch_object(pci.PCINetDevice, 'update_attributes')
dev = pci.PCINetDevice('0000:06:00.0')
self.glob.return_value = ['/sys/class/net/eth2']
self.get_sysnet_interface.return_value = 'eth2'
self.get_sysnet_mac.return_value = 'a8:9d:21:cf:93:fc'
self.get_sysnet_device_state.return_value = 'up'
self.realpath.return_value = (
'/sys/devices/pci0000:00/0000:00:02.0/0000:02:00.0/0000:03:00.0/'
'0000:04:00.0/0000:05:01.0/0000:07:00.0')
self.islink.return_value = True
expect = {
'interface': 'eth2',
'macAddress': 'a8:9d:21:cf:93:fc',
'pci_address': '0000:07:00.0',
'state': 'up',
}
self.assertEqual(dev.get_sysnet_interfaces_and_macs(), [expect])
def test_get_sysnet_mac(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
device = pci.PCINetDevice('0000:10:00.1')
with utils.patch_open() as (_open, _file):
super_fh = mocked_filehandle()
_file.readlines = mock.MagicMock()
_open.side_effect = super_fh._setfilename
_file.read.side_effect = super_fh._getfilecontents_read
macaddr = device.get_sysnet_mac('/sys/class/net/eth3')
self.assertEqual(macaddr, 'a8:9d:21:cf:93:fd')
def test_get_sysnet_device_state(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
device = pci.PCINetDevice('0000:10:00.1')
with utils.patch_open() as (_open, _file):
super_fh = mocked_filehandle()
_file.readlines = mock.MagicMock()
_open.side_effect = super_fh._setfilename
_file.read.side_effect = super_fh._getfilecontents_read
state = device.get_sysnet_device_state('/sys/class/net/eth3')
self.assertEqual(state, 'down')
def test_get_sysnet_interface(self):
self.patch_object(pci.PCINetDevice, 'update_attributes')
device = pci.PCINetDevice('0000:10:00.1')
self.assertEqual(
device.get_sysnet_interface('/sys/class/net/eth3'), 'eth3')
class PCINetDevicesTest(utils.BaseTestCase):
def test_init(self):
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
pci.PCINetDevices()
self.PCINetDevice.assert_called_once_with('pciaddr')
def test_get_pci_ethernet_addresses(self):
self.patch_object(pci, 'subprocess')
self.patch_object(pci, 'PCINetDevice')
self.subprocess.check_output.side_effect = \
mocked_subprocess()
a = pci.PCINetDevices()
self.assertEqual(
a.get_pci_ethernet_addresses(),
['0000:06:00.0', '0000:07:00.0', '0000:10:00.0', '0000:10:00.1'])
def test_update_devices(self):
pcinetdev = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.PCINetDevice.return_value = pcinetdev
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
a = pci.PCINetDevices()
a.update_devices()
pcinetdev.update_attributes.assert_called_once_with()
def test_get_macs(self):
pcinetdev = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.PCINetDevice.return_value = pcinetdev
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
pcinetdev.mac_address = 'mac1'
a = pci.PCINetDevices()
self.assertEqual(a.get_macs(), ['mac1'])
def test_get_device_from_mac(self):
pcinetdev = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.PCINetDevice.return_value = pcinetdev
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
pcinetdev.mac_address = 'mac1'
a = pci.PCINetDevices()
self.assertEqual(a.get_device_from_mac('mac1'), pcinetdev)
def test_get_device_from_pci_address(self):
pcinetdev = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.PCINetDevice.return_value = pcinetdev
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
pcinetdev.pci_address = 'pciaddr'
a = pci.PCINetDevices()
self.assertEqual(a.get_device_from_pci_address('pciaddr'), pcinetdev)
def test_rebind_orphans(self):
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci.PCINetDevices, 'unbind_orphans')
self.patch_object(pci.PCINetDevices, 'bind_orphans')
self.patch_object(pci, 'PCINetDevice')
self.get_pci_ethernet_addresses.return_value = []
a = pci.PCINetDevices()
a.rebind_orphans()
self.unbind_orphans.assert_called_once_with()
self.bind_orphans.assert_called_once_with()
def test_unbind_orphans(self):
orphan = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
self.patch_object(pci.PCINetDevices, 'get_orphans')
self.patch_object(pci.PCINetDevices, 'update_devices')
self.patch_object(pci, 'PCINetDevice')
self.get_orphans.return_value = [orphan]
a = pci.PCINetDevices()
a.unbind_orphans()
orphan.unbind.assert_called_once_with()
self.update_devices.assert_called_once_with()
def test_bind_orphans(self):
orphan = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
self.patch_object(pci.PCINetDevices, 'get_orphans')
self.patch_object(pci.PCINetDevices, 'update_devices')
self.patch_object(pci, 'PCINetDevice')
self.get_orphans.return_value = [orphan]
orphan.modalias_kmod = 'kmod'
a = pci.PCINetDevices()
a.bind_orphans()
orphan.bind.assert_called_once_with('kmod')
self.update_devices.assert_called_once_with()
def test_get_orphans(self):
pcinetdev = mock.MagicMock()
self.patch_object(pci.PCINetDevices, 'get_pci_ethernet_addresses')
self.patch_object(pci, 'PCINetDevice')
self.PCINetDevice.return_value = pcinetdev
self.get_pci_ethernet_addresses.return_value = ['pciaddr']
pcinetdev.loaded_kmod = None
pcinetdev.interface_name = None
pcinetdev.mac_address = None
a = pci.PCINetDevices()
self.assertEqual(a.get_orphans(), [pcinetdev])
class PCIInfoTest(utils.BaseTestCase):
def dev_mock(self, state, pci_address, interface_name):
dev = mock.MagicMock()
dev.state = state
dev.pci_address = pci_address
dev.interface_name = interface_name
return dev
def test_init(self):
net_dev_mocks = {
'mac1': self.dev_mock('down', 'pciaddr0', 'eth0'),
'mac2': self.dev_mock('down', 'pciaddr1', 'eth1'),
'mac3': self.dev_mock('up', 'pciaddr3', 'eth2'),
}
net_devs = mock.MagicMock()
self.patch_object(pci.PCIInfo, 'get_user_requested_config')
self.patch_object(pci, 'PCINetDevices')
self.PCINetDevices.return_value = net_devs
net_devs.get_macs.return_value = net_dev_mocks.keys()
net_devs.get_device_from_mac.side_effect = lambda x: net_dev_mocks[x]
self.get_user_requested_config.return_value = {
'mac1': [{'net': 'net1'}, {'net': 'net2'}],
'mac2': [{'net': 'net1'}],
'mac3': [{'net': 'net1'}]}
a = pci.PCIInfo()
expect = {
'mac1': [{'interface': 'eth0', 'net': 'net1'},
{'interface': 'eth0', 'net': 'net2'}],
'mac2': [{'interface': 'eth1', 'net': 'net1'}]}
self.assertEqual(a.local_mac_nets, expect)
self.assertEqual(a.vpe_dev_string, 'dev pciaddr0 dev pciaddr1')
def test_get_user_requested_config(self):
self.patch_object(pci.PCIInfo, '__init__')
self.patch_object(pci.hookenv, 'config')
self.config.return_value = ('mac=mac1;net=net1 mac=mac1;net=net2'
' mac=mac2;net=net1')
a = pci.PCIInfo()
expect = {
'mac1': [{'net': 'net1'}, {'net': 'net2'}],
'mac2': [{'net': 'net1'}]}
self.assertEqual(a.get_user_requested_config(), expect)
def test_get_user_requested_invalid_entries(self):
self.patch_object(pci.PCIInfo, '__init__')
self.patch_object(pci.hookenv, 'config')
self.config.return_value = ('ac=mac1;net=net1 randomstuff'
' mac=mac2;net=net1')
a = pci.PCIInfo()
expect = {'mac2': [{'net': 'net1'}]}
self.assertEqual(a.get_user_requested_config(), expect)
def test_get_user_requested_config_empty(self):
self.patch_object(pci.PCIInfo, '__init__')
self.patch_object(pci.hookenv, 'config')
self.config.return_value = None
a = pci.PCIInfo()
expect = {}
self.assertEqual(a.get_user_requested_config(), expect)

187
unit_tests/test_charms_openstack_sdn_odl.py Normal file
View File

@@ -0,0 +1,187 @@
import httpretty
import requests
import simplejson
import unit_tests.odl_responses as odl_responses
import charms_openstack.sdn.odl as odl
import unit_tests.utils as utils
NOT_JSON = "Im not json"
class ODLTest(utils.BaseTestCase):
def setUp(self):
super(ODLTest, self).setUp()
self.odlc = odl.ODLConfig('bob', 'pword', '10.0.0.10', port='93')
self.patch_object(odl.hookenv, 'log')
def test_base(self):
self.assertEqual(self.odlc.auth, ('bob', 'pword'))
self.assertEqual(self.odlc.base_url, 'http://10.0.0.10:93')
@httpretty.activate
def test_contact_odl(self):
httpretty.register_uri(httpretty.GET, "http://10.0.0.10:93/geturl",
body='[{"title": "Test Data"}]',
content_type="application/json", status=200)
response = self.odlc.contact_odl('GET', 'http://10.0.0.10:93/geturl')
self.assertEqual(response.json(), [{"title": "Test Data"}])
@httpretty.activate
def test_contact_odl_empty(self):
url = 'http://10.0.0.10:93/puturl'
httpretty.register_uri(httpretty.PUT, url,
body='', status=204)
response = self.odlc.contact_odl('PUT', url)
self.assertEqual(response.status_code, 204)
@httpretty.activate
def test_contact_odl_notfound(self):
httpretty.register_uri(httpretty.GET, "http://10.0.0.10:93/geturl",
status=404)
with self.assertRaises(odl.ODLInteractionFatalError):
self.odlc.contact_odl('GET', 'http://10.0.0.10:93/geturl')
@httpretty.activate
def test_contact_odl_retry(self):
httpretty.register_uri(httpretty.GET, "http://10.0.0.10:93/geturl",
status=404)
with self.assertRaises(requests.exceptions.ConnectionError):
self.odlc.contact_odl(
'GET', 'http://10.0.0.10:93/geturl', retry_rcs=[404])
@httpretty.activate
def test_get_networks(self):
url = self.odlc.netmap_url
httpretty.register_uri(
httpretty.GET, url, status=200, body=odl_responses.NEUTRON_NET_MAP)
nets = self.odlc.get_networks()
self.assertTrue('physicalNetwork' in nets.keys())
self.assertEqual(len(nets['physicalNetwork']), 3)
net_names = [net['name'] for net in nets['physicalNetwork']]
for net in ['net_d10', 'net_d11', 'net_d12']:
self.assertTrue(net in net_names)
@httpretty.activate
def test_get_networks_nonets(self):
url = self.odlc.netmap_url
httpretty.register_uri(httpretty.GET, url, status=200, body="{}")
nets = self.odlc.get_networks()
self.assertEqual(nets, {})
@httpretty.activate
def test_get_networks_no_neutron_map(self):
url = self.odlc.netmap_url
httpretty.register_uri(httpretty.GET, url, status=404)
nets = self.odlc.get_networks()
self.assertEqual(nets, {})
@httpretty.activate
def test_get_networks_notjson(self):
url = self.odlc.netmap_url
httpretty.register_uri(httpretty.GET, url, status=200, body=NOT_JSON)
with self.assertRaises(simplejson.JSONDecodeError):
self.odlc.get_networks()
def test_delete_net_device_entry(self):
self.patch_object(odl.ODLConfig, 'contact_odl')
self.odlc.delete_net_device_entry('net_d10', 'mymachine')
url = self.odlc.netmap_url + 'physicalNetwork/net_d10/device/mymachine'
self.contact_odl.assert_called_with('DELETE', url)
@httpretty.activate
def test_get_odl_registered_nodes(self):
url = self.odlc.node_query_url
httpretty.register_uri(
httpretty.GET, url, status=200,
body=odl_responses.ODL_REGISTERED_NODES)
nodes = self.odlc.get_odl_registered_nodes()
self.assertEqual(nodes, ['C240-M4-6', 'controller-config'])
@httpretty.activate
def test_get_odl_registered_empty(self):
url = self.odlc.node_query_url
httpretty.register_uri(httpretty.GET, url, status=200, body="{}")
nodes = self.odlc.get_odl_registered_nodes()
self.assertEqual(nodes, [])
@httpretty.activate
def test_get_odl_registered_notjson(self):
url = self.odlc.node_query_url
httpretty.register_uri(httpretty.GET, url, status=200, body=NOT_JSON)
with self.assertRaises(simplejson.JSONDecodeError):
self.odlc.get_odl_registered_nodes()
def test_odl_register_node(self):
self.patch_object(odl.ODLConfig, 'contact_odl')
url = self.odlc.node_mount_url
self.odlc.odl_register_node('mymachine', '10.0.0.11')
reg_call = self.contact_odl.call_args_list[0]
self.assertTrue(reg_call[0], ('POST', url))
def test_odl_register_macs(self):
self.patch_object(odl.ODLConfig, 'contact_odl')
url = self.odlc.conf_url
self.odlc.odl_register_macs(
"C240-M4-6", "net_d1", "TenGigabitEthernet6/0/0",
"84:b8:02:2a:5f:c3")
reg_call = self.contact_odl.call_args_list[0]
self.assertTrue(reg_call[0], ('POST', url))
@httpretty.activate
def test_get_macs_networks(self):
url = self.odlc.netmap_url
httpretty.register_uri(
httpretty.GET, url, status=200, body=odl_responses.NEUTRON_NET_MAP)
nets = self.odlc.get_macs_networks('84:b8:02:2a:5f:c3')
self.assertEqual(nets, ['net_d12', 'net_d10'])
@httpretty.activate
def test_get_macs_networks_nomatch(self):
url = self.odlc.netmap_url
httpretty.register_uri(
httpretty.GET, url, status=200, body=odl_responses.NEUTRON_NET_MAP)
nets = self.odlc.get_macs_networks('04:08:02:0a:0f:03')
self.assertEqual(nets, [])
@httpretty.activate
def test_get_macs_networks_nonets(self):
url = self.odlc.netmap_url
httpretty.register_uri(httpretty.GET, url, status=200, body="{}")
nets = self.odlc.get_macs_networks('04:08:02:0a:0f:03')
self.assertEqual(nets, [])
@httpretty.activate
def test_is_device_registered(self):
url = self.odlc.node_query_url
httpretty.register_uri(
httpretty.GET, url, status=200,
body=odl_responses.ODL_REGISTERED_NODES)
self.assertTrue(self.odlc.is_device_registered('C240-M4-6'))
@httpretty.activate
def test_is_device_registered_false(self):
url = self.odlc.node_query_url
httpretty.register_uri(
httpretty.GET, url, status=200,
body=odl_responses.ODL_REGISTERED_NODES)
self.assertFalse(self.odlc.is_device_registered('B240-M4-7'))
@httpretty.activate
def test_is_net_device_registered(self):
url = self.odlc.netmap_url
httpretty.register_uri(
httpretty.GET, url, status=200, body=odl_responses.NEUTRON_NET_MAP)
self.assertTrue(self.odlc.is_net_device_registered(
'net_d10', 'C240-M4-6', 'TenGigabitEthernet6/0/0',
'84:b8:02:2a:5f:c3'))
@httpretty.activate
def test_is_net_device_registered_false(self):
url = self.odlc.netmap_url
httpretty.register_uri(
httpretty.GET, url, status=200, body=odl_responses.NEUTRON_NET_MAP)
self.assertFalse(self.odlc.is_net_device_registered(
'net_d510', 'C240-M4-6', 'TenGigabitEthernet6/0/0',
'84:b8:02:2a:5f:c3'))

51
unit_tests/test_charms_openstack_sdn_ovs.py Normal file
View File

@@ -0,0 +1,51 @@
# Copyright 2016 Canonical Ltd
#
# 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.
# Note that the unit_tests/__init__.py has the following lines to stop
# side effects from the imorts from charm helpers.
# sys.path.append('./lib')
# mock out some charmhelpers libraries as they have apt install side effects
# sys.modules['charmhelpers.contrib.openstack.utils'] = mock.MagicMock()
# sys.modules['charmhelpers.contrib.network.ip'] = mock.MagicMock()
from __future__ import absolute_import
import unit_tests.utils as utils
import charms_openstack.sdn.ovs as ovs
class TestCharmOpenStackSDNOVS(utils.BaseTestCase):
def test_set_manager(self):
self.patch_object(ovs, 'subprocess')
ovs.set_manager('myurl')
self.subprocess.check_call.assert_called_once_with(
['ovs-vsctl', 'set-manager', 'myurl'])
def test__get_ovstbl(self):
self.patch_object(ovs, 'subprocess')
self.subprocess.check_output.return_value = 'ovstbl'
self.assertEqual(ovs._get_ovstbl(), 'ovstbl')
self.subprocess.check_output.assert_called_once_with(
['ovs-vsctl', 'get', 'Open_vSwitch', '.', '_uuid'])
def test_set_config(self):
self.patch_object(ovs, 'subprocess')
self.patch_object(ovs, '_get_ovstbl')
self._get_ovstbl.return_value = 'a_uuid'
ovs.set_config('mykey', 'myvalue', 'mytable')
self.subprocess.check_call.assert_called_once_with(
['ovs-vsctl', 'set', 'Open_vSwitch', 'a_uuid',
'mytable:mykey=myvalue'])