Allow tox to run plugin specific unit tests
Fixes bug 1041316 This patch covers linuxbridge, nec, nicira, and ryu plugins. Change-Id: I024a46bc9b49ffe60cdbe9fa275f30ac9d829a9a
This commit is contained in:
Salvatore Orlando
parent
af6bbaf344
commit
bfb1555120
@ -1,73 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 OpenStack, LLC
|
||||
# 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.
|
||||
|
||||
|
||||
"""Unittest runner for quantum Linux Bridge plugin
|
||||
|
||||
This file should be run from the top dir in the quantum directory
|
||||
|
||||
To run all tests::
|
||||
PLUGIN_DIR=quantum/plugins/linuxbridge ./run_tests.sh
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
from nose import config
|
||||
from nose import core
|
||||
|
||||
sys.path.append(os.getcwd())
|
||||
sys.path.append(os.path.dirname(__file__))
|
||||
|
||||
|
||||
from quantum.common.test_lib import run_tests, test_config
|
||||
import quantum.tests.unit
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit_status = False
|
||||
|
||||
# if a single test case was specified,
|
||||
# we should only invoked the tests once
|
||||
invoke_once = len(sys.argv) > 1
|
||||
|
||||
test_config['plugin_name_v2'] = "lb_quantum_plugin.LinuxBridgePluginV2"
|
||||
|
||||
cwd = os.getcwd()
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
includeExe=True,
|
||||
traverseNamespace=True,
|
||||
plugins=core.DefaultPluginManager())
|
||||
c.configureWhere(quantum.tests.unit.__path__)
|
||||
exit_status = run_tests(c)
|
||||
|
||||
if invoke_once:
|
||||
sys.exit(0)
|
||||
|
||||
os.chdir(cwd)
|
||||
|
||||
working_dir = os.path.abspath("quantum/plugins/linuxbridge")
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
workingDir=working_dir)
|
||||
exit_status = exit_status or run_tests(c)
|
||||
|
||||
sys.exit(exit_status)
|
||||
@ -1,485 +0,0 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Copyright 2012 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: Shweta Padubidri, Cisco Systems, Inc.
|
||||
|
||||
import ConfigParser
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
import quantum.db.api as db
|
||||
from quantum.plugins.linuxbridge.agent import (
|
||||
linuxbridge_quantum_agent as linux_agent,
|
||||
)
|
||||
from quantum.plugins.linuxbridge.common import constants as lconst
|
||||
from quantum.plugins.linuxbridge.db import l2network_db as cdb
|
||||
from quantum.plugins.linuxbridge import LinuxBridgePlugin
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LinuxBridgeAgentTest(unittest.TestCase):
|
||||
|
||||
def test_add_gateway_interface(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc',
|
||||
mac_address='fe:16:3e:51:60:dd'):
|
||||
|
||||
LOG.debug("test_tap_gateway_interface - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_bridge(bridge_name)
|
||||
device_name = self.gw_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_device(device_name, mac_address)
|
||||
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
|
||||
self._linuxbridge_quantum_agent.process_port_binding(
|
||||
new_network[lconst.NET_ID], device_name, str(vlan_id))
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
|
||||
self.assertTrue(device_name in list_interface)
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_add_gateway_interface - END")
|
||||
|
||||
def test_add_tap_interface(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc',
|
||||
mac_address='fe:16:3e:51:60:dd'):
|
||||
|
||||
LOG.debug("test_add_tap_interface - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_bridge(bridge_name)
|
||||
device_name = self.tap_name_prefix + interface_id[0:11]
|
||||
self.create_device(device_name, mac_address)
|
||||
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
|
||||
self._linuxbridge_quantum_agent.process_port_binding(
|
||||
new_network[lconst.NET_ID], interface_id, str(vlan_id))
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
|
||||
self.assertTrue(device_name in list_interface)
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_add_tap_interface -END")
|
||||
|
||||
def test_remove_interface(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc',
|
||||
mac_address='fe:16:3e:51:60:dd'):
|
||||
|
||||
LOG.debug("test_remove_interface - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_bridge(bridge_name)
|
||||
device_name = self.tap_name_prefix + interface_id[0:11]
|
||||
self.create_device(device_name, mac_address)
|
||||
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
|
||||
self._linuxbridge_quantum_agent.process_port_binding(
|
||||
new_network[lconst.NET_ID], interface_id, str(vlan_id))
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(bridge_name,
|
||||
device_name)
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self.assertFalse(device_name in list_interface)
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_device(device_name)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_remove_interface -END")
|
||||
|
||||
def test_ensure_vlan_bridge(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc'):
|
||||
|
||||
LOG.debug("test_ensure_vlan_bridge - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
vlan_subinterface = self.physical_interface + '.' + str(vlan_id)
|
||||
|
||||
self._linuxbridge_quantum_agent.linux_br.ensure_vlan_bridge(
|
||||
new_network[lconst.NET_ID], str(vlan_id))
|
||||
list_quantum_bridges = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_all_quantum_bridges())
|
||||
self.assertTrue(bridge_name in list_quantum_bridges)
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self.assertTrue(vlan_subinterface in list_interface)
|
||||
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_ensure_vlan_bridge -END")
|
||||
|
||||
def test_delete_vlan_bridge(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc'):
|
||||
|
||||
LOG.debug("test_delete_vlan_bridge - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
vlan_subinterface = self.physical_interface + '.' + str(vlan_id)
|
||||
|
||||
self._linuxbridge_quantum_agent.linux_br.ensure_vlan_bridge(
|
||||
new_network[lconst.NET_ID], str(vlan_id))
|
||||
self._linuxbridge_quantum_agent.linux_br.delete_vlan_bridge(
|
||||
bridge_name)
|
||||
|
||||
self.assertEquals(self.device_exists(vlan_subinterface), False)
|
||||
self.assertEquals(self.device_exists(bridge_name), False)
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_delete_vlan_bridge - END")
|
||||
|
||||
def test_process_deleted_networks(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc'):
|
||||
|
||||
LOG.debug("test_delete_vlan_bridge - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
vlan_bindings = {}
|
||||
vlan_bindings[new_network[lconst.NET_ID]] = (
|
||||
cdb.get_vlan_binding(new_network[lconst.NET_ID]))
|
||||
vlan_id = vlan_bindings[new_network[lconst.NET_ID]][lconst.VLANID]
|
||||
vlan_subinterface = self.physical_interface + '.' + str(vlan_id)
|
||||
|
||||
self._linuxbridge_quantum_agent.linux_br.ensure_vlan_bridge(
|
||||
new_network[lconst.NET_ID], str(vlan_id))
|
||||
self.tearDownUnplugInterface(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
vlan_bindings = {}
|
||||
self._linuxbridge_quantum_agent.process_deleted_networks(vlan_bindings)
|
||||
|
||||
self.assertEquals(self.device_exists(vlan_subinterface), False)
|
||||
self.assertEquals(self.device_exists(bridge_name), False)
|
||||
LOG.debug("test_delete_vlan_bridge - END")
|
||||
|
||||
def test_process_unplugged_tap_interface(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-'
|
||||
'42ea-b9e6-'
|
||||
'7313d1c522cc',
|
||||
mac_address='fe:16:3e:51:60:dd'):
|
||||
|
||||
LOG.debug("test_process_unplugged_tap_interface - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_bridge(bridge_name)
|
||||
device_name = self.tap_name_prefix + interface_id[0:11]
|
||||
self.create_device(device_name, mac_address)
|
||||
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
|
||||
self._linuxbridge_quantum_agent.process_port_binding(
|
||||
new_network[lconst.NET_ID], interface_id, str(vlan_id))
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self._linuxbridge_plugin.unplug_interface(tenant_id,
|
||||
new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
plugged_interface = []
|
||||
self._linuxbridge_quantum_agent.process_unplugged_interfaces(
|
||||
plugged_interface)
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self.assertFalse(device_name in list_interface)
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_device(device_name)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownNetworkPort(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_test_process_unplugged_tap_interface -END")
|
||||
|
||||
def test_process_unplugged_interface_empty(self, tenant_id="test_tenant",
|
||||
network_name="test_network"):
|
||||
""" test to unplug not plugged port. It should not raise exception
|
||||
"""
|
||||
LOG.debug("test_process_unplugged_interface_empty - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.unplug_interface(tenant_id,
|
||||
new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
self.tearDownNetworkPort(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_process_unplugged_interface_empty -END")
|
||||
|
||||
def test_process_unplugged_gw_interface(self, tenant_id="test_tenant",
|
||||
network_name="test_network",
|
||||
interface_id='fe701ddf-26a2-42ea-'
|
||||
'b9e6-7313d1c522cc',
|
||||
mac_address='fe:16:3e:51:60:dd'):
|
||||
|
||||
LOG.debug("test_process_unplugged_gw_interface - START")
|
||||
new_network = (
|
||||
self._linuxbridge_plugin.create_network(tenant_id, network_name))
|
||||
new_port = self._linuxbridge_plugin.create_port(
|
||||
tenant_id, new_network[lconst.NET_ID], lconst.PORT_UP)
|
||||
self._linuxbridge_plugin.plug_interface(
|
||||
tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID], interface_id)
|
||||
bridge_name = self.br_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_bridge(bridge_name)
|
||||
device_name = self.gw_name_prefix + new_network[lconst.NET_ID][0:11]
|
||||
self.create_device(device_name, mac_address)
|
||||
|
||||
vlan_bind = cdb.get_vlan_binding(new_network[lconst.NET_ID])
|
||||
vlan_id = vlan_bind[lconst.VLANID]
|
||||
|
||||
self._linuxbridge_quantum_agent.process_port_binding(
|
||||
new_network[lconst.NET_ID], interface_id, str(vlan_id))
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self._linuxbridge_plugin.unplug_interface(tenant_id,
|
||||
new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
plugged_interface = []
|
||||
self._linuxbridge_quantum_agent.process_unplugged_interfaces(
|
||||
plugged_interface)
|
||||
list_interface = (self._linuxbridge_quantum_agent.linux_br.
|
||||
get_interfaces_on_bridge(bridge_name))
|
||||
self.assertFalse(device_name in list_interface)
|
||||
for interface in list_interface:
|
||||
self._linuxbridge_quantum_agent.linux_br.remove_interface(
|
||||
bridge_name, interface)
|
||||
self.delete_device(interface)
|
||||
self.delete_device(device_name)
|
||||
self.delete_bridge(bridge_name)
|
||||
self.tearDownNetworkPort(tenant_id, new_network[lconst.NET_ID],
|
||||
new_port[lconst.PORT_ID])
|
||||
|
||||
LOG.debug("test_test_process_unplugged_gw_interface -END")
|
||||
|
||||
def create_bridge(self, bridge_name):
|
||||
"""
|
||||
Create a bridge
|
||||
"""
|
||||
self.run_cmd(['brctl', 'addbr', bridge_name])
|
||||
self.run_cmd(['brctl', 'setfd', bridge_name, str(0)])
|
||||
self.run_cmd(['brctl', 'stp', bridge_name, 'off'])
|
||||
self.run_cmd(['ip', 'link', 'set', bridge_name, 'up'])
|
||||
|
||||
def delete_bridge(self, bridge_name):
|
||||
"""
|
||||
Delete a bridge
|
||||
"""
|
||||
self.run_cmd(['ip', 'link', 'set', bridge_name, 'down'])
|
||||
self.run_cmd(['brctl', 'delbr', bridge_name])
|
||||
|
||||
def create_device(self, dev, mac_address):
|
||||
self.run_cmd(['ip', 'tuntap', 'add', dev, 'mode', 'tap'])
|
||||
self.run_cmd(['ip', 'link', 'set', dev, "address", mac_address])
|
||||
self.run_cmd(['ip', 'link', 'set', dev, 'up'])
|
||||
|
||||
def delete_device(self, dev):
|
||||
self.run_cmd(['ip', 'link', 'delete', dev])
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
Set up function
|
||||
"""
|
||||
self.tenant_id = "test_tenant"
|
||||
self.network_name = "test_network"
|
||||
self.config_file = os.path.join(os.path.dirname(__file__), os.pardir,
|
||||
os.pardir, os.pardir, os.pardir,
|
||||
os.pardir, "etc", "quantum",
|
||||
"plugins", "linuxbridge",
|
||||
"linuxbridge_conf.ini")
|
||||
|
||||
config = ConfigParser.ConfigParser()
|
||||
self.br_name_prefix = "brq"
|
||||
self.gw_name_prefix = "gw-"
|
||||
self.tap_name_prefix = "tap"
|
||||
self.v2 = True
|
||||
self.db_connection = 'sqlite://'
|
||||
self._linuxbridge_plugin = LinuxBridgePlugin.LinuxBridgePlugin()
|
||||
try:
|
||||
fh = open(self.config_file)
|
||||
fh.close()
|
||||
config.read(self.config_file)
|
||||
self.physical_interface = config.get("LINUX_BRIDGE",
|
||||
"physical_interface")
|
||||
self.polling_interval = config.get("AGENT", "polling_interval")
|
||||
self.reconnect_interval = config.get("DATABASE",
|
||||
"reconnect_interval")
|
||||
self.root_helper = config.get("AGENT", "root_helper")
|
||||
except Exception, e:
|
||||
LOG.error("Unable to parse config file \"%s\": \nException%s"
|
||||
% (self.config_file, str(e)))
|
||||
sys.exit(1)
|
||||
self._linuxbridge = linux_agent.LinuxBridge(self.br_name_prefix,
|
||||
self.physical_interface,
|
||||
self.root_helper)
|
||||
plugin = linux_agent.LinuxBridgeQuantumAgentDB(
|
||||
self.br_name_prefix,
|
||||
self.physical_interface,
|
||||
self.polling_interval,
|
||||
self.reconnect_interval,
|
||||
self.root_helper,
|
||||
self.v2,
|
||||
self.db_connection)
|
||||
self._linuxbridge_quantum_agent = plugin
|
||||
|
||||
def run_cmd(self, args):
|
||||
cmd = shlex.split(self.root_helper) + args
|
||||
LOG.debug("Running command: " + " ".join(cmd))
|
||||
p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
|
||||
retval = p.communicate()[0]
|
||||
if p.returncode == -(signal.SIGALRM):
|
||||
LOG.debug("Timeout running command: " + " ".join(args))
|
||||
if retval:
|
||||
LOG.debug("Command returned: %s" % retval)
|
||||
return retval
|
||||
|
||||
def device_exists(self, device):
|
||||
"""Check if ethernet device exists."""
|
||||
retval = self.run_cmd(['ip', 'link', 'show', 'dev', device])
|
||||
if retval:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
"""
|
||||
Clean up functions after the tests
|
||||
"""
|
||||
def tearDown(self):
|
||||
"""Clear the test environment(Clean the Database)"""
|
||||
db.clear_db()
|
||||
|
||||
def tearDownNetwork(self, tenant_id, network_dict_id):
|
||||
"""
|
||||
Tear down the Network
|
||||
"""
|
||||
self._linuxbridge_plugin.delete_network(tenant_id, network_dict_id)
|
||||
|
||||
def tearDownUnplugInterface(self, tenant_id, network_dict_id, port_id):
|
||||
"""
|
||||
Tear down the port
|
||||
"""
|
||||
self._linuxbridge_plugin.unplug_interface(tenant_id, network_dict_id,
|
||||
port_id)
|
||||
self.tearDownNetworkPort(tenant_id, network_dict_id, port_id)
|
||||
|
||||
def tearDownNetworkPort(self, tenant_id, network_dict_id, port_id):
|
||||
"""
|
||||
Tear down Network Port
|
||||
"""
|
||||
self._linuxbridge_plugin.delete_port(tenant_id, network_dict_id,
|
||||
port_id)
|
||||
self.tearDownNetwork(tenant_id, network_dict_id)
|
||||
@ -1,72 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 OpenStack, LLC
|
||||
# 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.
|
||||
|
||||
|
||||
"""Unittest runner for quantum NEC OpenFlow plugin
|
||||
|
||||
This file should be run from the top dir in the quantum directory
|
||||
|
||||
To run all tests::
|
||||
PLUGIN_DIR=quantum/plugins/nec ./run_tests.sh
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
from nose import config
|
||||
from nose import core
|
||||
|
||||
sys.path.append(os.getcwd())
|
||||
sys.path.append(os.path.dirname(__file__))
|
||||
|
||||
from quantum.common.test_lib import run_tests, test_config
|
||||
import quantum.tests.unit
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit_status = False
|
||||
|
||||
# if a single test case was specified,
|
||||
# we should only invoked the tests once
|
||||
invoke_once = len(sys.argv) > 1
|
||||
|
||||
test_config['plugin_name'] = "nec_plugin.NECPluginV2"
|
||||
|
||||
cwd = os.getcwd()
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
includeExe=True,
|
||||
traverseNamespace=True,
|
||||
plugins=core.DefaultPluginManager())
|
||||
c.configureWhere(quantum.tests.unit.__path__)
|
||||
exit_status = run_tests(c)
|
||||
|
||||
if invoke_once:
|
||||
sys.exit(0)
|
||||
|
||||
os.chdir(cwd)
|
||||
|
||||
working_dir = os.path.abspath("quantum/plugins/nec")
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
workingDir=working_dir)
|
||||
exit_status = exit_status or run_tests(c)
|
||||
|
||||
sys.exit(exit_status)
|
||||
@ -1,113 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 OpenStack, LLC
|
||||
# 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.
|
||||
|
||||
|
||||
"""Unittest runner for Nicira NVP plugin
|
||||
|
||||
This file should be run from the top dir in the quantum directory
|
||||
|
||||
To run all tests::
|
||||
PLUGIN_DIR=quantum/plugins/nicira ./run_tests.sh
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import mock
|
||||
from nose import config
|
||||
from nose import core
|
||||
|
||||
CONFIG_FILE_OPT = "--config-file"
|
||||
NICIRA_PATH = "quantum/plugins/nicira/nicira_nvp_plugin"
|
||||
|
||||
sys.path.append(os.getcwd())
|
||||
sys.path.append(os.path.dirname(__file__))
|
||||
sys.path.append(os.path.abspath(NICIRA_PATH))
|
||||
|
||||
from quantum.common.test_lib import run_tests, test_config
|
||||
from quantum.openstack.common import cfg
|
||||
import quantum.tests.unit
|
||||
from quantum import version
|
||||
|
||||
from tests import fake_nvpapiclient
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit_status = False
|
||||
do_mock = False
|
||||
# remove the value
|
||||
test_config['config_files'] = []
|
||||
|
||||
# if a single test case was specified,
|
||||
# we should only invoked the tests once
|
||||
invoke_once = len(sys.argv) > 1
|
||||
# this will allow us to pass --config-file to run_tests.sh for
|
||||
# running the unit tests against a real backend
|
||||
# if --config-file has been specified, remove it from sys.argv
|
||||
# otherwise nose will complain
|
||||
while CONFIG_FILE_OPT in sys.argv:
|
||||
test_config['config_files'].append(
|
||||
sys.argv.pop(sys.argv.index(CONFIG_FILE_OPT) + 1))
|
||||
# and the option itself
|
||||
sys.argv.remove(CONFIG_FILE_OPT)
|
||||
|
||||
# if no config file available, inject one for fake backend tests
|
||||
if not test_config.get('config_files'):
|
||||
do_mock = True
|
||||
test_config['config_files'] = [os.path.abspath('%s/tests/nvp.ini.test'
|
||||
% NICIRA_PATH)]
|
||||
|
||||
test_config['plugin_name_v2'] = "QuantumPlugin.NvpPluginV2"
|
||||
cwd = os.getcwd()
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
includeExe=True,
|
||||
traverseNamespace=True,
|
||||
plugins=core.DefaultPluginManager())
|
||||
c.configureWhere(quantum.tests.unit.__path__)
|
||||
|
||||
# patch nvpapi client if not running against "real" back end
|
||||
if do_mock:
|
||||
fc = fake_nvpapiclient.FakeClient(os.path.abspath('%s/tests'
|
||||
% NICIRA_PATH))
|
||||
mock_nvpapi = mock.patch('NvpApiClient.NVPApiHelper', autospec=True)
|
||||
instance = mock_nvpapi.start()
|
||||
instance.return_value.login.return_value = "the_cookie"
|
||||
|
||||
def _fake_request(*args, **kwargs):
|
||||
return fc.fake_request(*args, **kwargs)
|
||||
|
||||
instance.return_value.request.side_effect = _fake_request
|
||||
|
||||
exit_status = run_tests(c)
|
||||
if invoke_once:
|
||||
sys.exit(0)
|
||||
|
||||
os.chdir(cwd)
|
||||
|
||||
working_dir = os.path.abspath(NICIRA_PATH)
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
workingDir=working_dir)
|
||||
exit_status = exit_status or run_tests(c)
|
||||
|
||||
# restore original nvpapi client (probably pleonastic here)
|
||||
if do_mock:
|
||||
mock_nvpapi.stop()
|
||||
sys.exit(exit_status)
|
||||
@ -1,77 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 Isaku Yamahata <yamahata at private email ne jp>
|
||||
# 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.
|
||||
|
||||
|
||||
"""Unittest runner for quantum Ryu plugin
|
||||
|
||||
This file should be run from the top dir in the quantum directory
|
||||
|
||||
To run all tests::
|
||||
PLUGIN_DIR=quantum/plugins/ryu ./run_tests.sh
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
from nose import config
|
||||
from nose import core
|
||||
|
||||
sys.path.append(os.getcwd())
|
||||
sys.path.append(os.path.dirname(__file__))
|
||||
|
||||
from quantum.common.test_lib import run_tests, test_config
|
||||
from quantum.plugins.ryu.tests.unit.utils import patch_fake_ryu_client
|
||||
import quantum.tests.unit
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit_status = False
|
||||
|
||||
# if a single test case was specified,
|
||||
# we should only invoked the tests once
|
||||
invoke_once = len(sys.argv) > 1
|
||||
|
||||
test_config['plugin_name_v2'] = "ryu_quantum_plugin.RyuQuantumPluginV2"
|
||||
|
||||
cwd = os.getcwd()
|
||||
# patch modules for ryu.app.client and ryu.app.rest_nw_id
|
||||
# With those, plugin can be tested without ryu installed
|
||||
with patch_fake_ryu_client():
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
includeExe=True,
|
||||
traverseNamespace=True,
|
||||
plugins=core.DefaultPluginManager())
|
||||
c.configureWhere(quantum.tests.unit.__path__)
|
||||
|
||||
exit_status = run_tests(c)
|
||||
|
||||
if invoke_once:
|
||||
sys.exit(0)
|
||||
|
||||
os.chdir(cwd)
|
||||
|
||||
working_dir = os.path.abspath("quantum/plugins/ryu")
|
||||
c = config.Config(stream=sys.stdout,
|
||||
env=os.environ,
|
||||
verbosity=3,
|
||||
workingDir=working_dir)
|
||||
exit_status = exit_status or run_tests(c)
|
||||
|
||||
sys.exit(exit_status)
|
||||
@ -1,6 +1,7 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Copyright 2012 NEC Corporation. All rights reserved.
|
||||
|
||||
# Copyright 2012 OpenStack LLC.
|
||||
# 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
|
||||
@ -15,7 +15,6 @@
|
||||
# under the License.
|
||||
# @author: Ryota MIBU
|
||||
|
||||
import quantum.openstack.common.rpc
|
||||
from quantum.plugins.nec.common import config
|
||||
from quantum.tests.unit import test_db_plugin
|
||||
|
||||
@ -35,25 +34,13 @@ class NECPluginTestBase(object):
|
||||
|
||||
plugin = 'quantum.plugins.nec.nec_plugin.NECPluginV2'
|
||||
config.CONF.set_override('core_plugin', plugin)
|
||||
driver = "quantum.plugins.nec.tests.unit.stub_ofc_driver.StubOFCDriver"
|
||||
driver = "quantum.tests.unit.nec.stub_ofc_driver.StubOFCDriver"
|
||||
config.CONF.set_override('driver', driver, 'OFC')
|
||||
config.CONF.set_override('rpc_backend',
|
||||
'quantum.openstack.common.rpc.impl_fake')
|
||||
self.api = test_db_plugin.APIRouter()
|
||||
self._skip_native_bulk = False
|
||||
|
||||
|
||||
class TestPortsV2(NECPluginTestBase, test_db_plugin.TestPortsV2):
|
||||
pass
|
||||
|
||||
|
||||
class TestNetworksV2(NECPluginTestBase, test_db_plugin.TestNetworksV2):
|
||||
pass
|
||||
|
||||
|
||||
# NOTE: This plugin does not override methods for subnet.
|
||||
#class TestSubnetsV2(NECPluginTestBase, test_db_plugin.TestSubnetsV2):
|
||||
# pass
|
||||
super(NECPluginTestBase, self).setUp(plugin)
|
||||
|
||||
|
||||
# TODO(r-mibu): write UT for packet_filters.
|
||||
@ -28,7 +28,7 @@ class OFCManagerTest(unittest.TestCase):
|
||||
"""Class conisting of OFCManager unit tests"""
|
||||
|
||||
def setUp(self):
|
||||
driver = "quantum.plugins.nec.tests.unit.stub_ofc_driver.StubOFCDriver"
|
||||
driver = "quantum.tests.unit.nec.stub_ofc_driver.StubOFCDriver"
|
||||
config.CONF.set_override('driver', driver, 'OFC')
|
||||
ndb.initialize()
|
||||
self.ofc = ofc_manager.OFCManager()
|
||||
@ -1,4 +1,7 @@
|
||||
# Copyright 2012 Nicira Networks, Inc.
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 OpenStack LLC.
|
||||
# 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
|
||||
@ -11,7 +14,3 @@
|
||||
# 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: Brad Hall, Nicira Networks, Inc.
|
||||
#@author: Dave Lapsley, Nicira Networks, Inc.
|
||||
#@author: Aaron Rosen, Nicira Networks, Inc.
|
||||
@ -18,10 +18,9 @@ import os
|
||||
import mock
|
||||
|
||||
import quantum.common.test_lib as test_lib
|
||||
from quantum.plugins.nicira.nicira_nvp_plugin.tests import fake_nvpapiclient
|
||||
from quantum.tests.unit.nicira import fake_nvpapiclient
|
||||
import quantum.tests.unit.test_db_plugin as test_plugin
|
||||
|
||||
NICIRA_PATH = '../../plugins/nicira/nicira_nvp_plugin'
|
||||
NICIRA_PKG_PATH = 'quantum.plugins.nicira.nicira_nvp_plugin'
|
||||
|
||||
|
||||
@ -30,12 +29,11 @@ class NiciraPluginV2TestCase(test_plugin.QuantumDbPluginV2TestCase):
|
||||
_plugin_name = ('%s.QuantumPlugin.NvpPluginV2' % NICIRA_PKG_PATH)
|
||||
|
||||
def setUp(self):
|
||||
config_file_path = os.path.abspath('%s/tests/nvp.ini.test'
|
||||
% NICIRA_PATH)
|
||||
test_lib.test_config['config_files'] = [config_file_path]
|
||||
etc_path = os.path.join(os.path.dirname(__file__), 'etc')
|
||||
test_lib.test_config['config_files'] = [os.path.join(etc_path,
|
||||
'nvp.ini.test')]
|
||||
# mock nvp api client
|
||||
fc = fake_nvpapiclient.FakeClient(os.path.abspath('%s/tests'
|
||||
% NICIRA_PATH))
|
||||
fc = fake_nvpapiclient.FakeClient(etc_path)
|
||||
self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
|
||||
% NICIRA_PKG_PATH, autospec=True)
|
||||
instance = self.mock_nvpapi.start()
|
||||
16
quantum/tests/unit/ryu/__init__.py
Normal file
16
quantum/tests/unit/ryu/__init__.py
Normal file
@ -0,0 +1,16 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 OpenStack LLC.
|
||||
# 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.
|
||||
@ -56,14 +56,22 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
|
||||
|
||||
def setUp(self, plugin=None):
|
||||
super(QuantumDbPluginV2TestCase, self).setUp()
|
||||
|
||||
# NOTE(jkoelker) for a 'pluggable' framework, Quantum sure
|
||||
# doesn't like when the plugin changes ;)
|
||||
db._ENGINE = None
|
||||
db._MAKER = None
|
||||
# Make sure at each test a new instance of the plugin is returned
|
||||
QuantumManager._instance = None
|
||||
|
||||
# Save the attributes map in case the plugin will alter it
|
||||
# loading extensions
|
||||
# Note(salvatore-orlando): shallow copy is not good enough in
|
||||
# this case, but copy.deepcopy does not seem to work, since it
|
||||
# causes test failures
|
||||
self._attribute_map_bk = {}
|
||||
for item in attributes.RESOURCE_ATTRIBUTE_MAP:
|
||||
self._attribute_map_bk[item] = (attributes.
|
||||
RESOURCE_ATTRIBUTE_MAP[item].
|
||||
copy())
|
||||
self._tenant_id = 'test-tenant'
|
||||
|
||||
json_deserializer = JSONDeserializer()
|
||||
@ -105,6 +113,8 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
|
||||
db._ENGINE = None
|
||||
db._MAKER = None
|
||||
cfg.CONF.reset()
|
||||
# Restore the original attribute map
|
||||
attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk
|
||||
|
||||
def _req(self, method, resource, data=None, fmt='json',
|
||||
id=None, params=None, action=None):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user