"""
import webob.dec
-from quantum.common import wsgi
+from quantum import wsgi
class Fault(webob.exc.HTTPException):
def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
"""Invoke only the device plugin"""
- # If the last param is a dict, add it to kwargs
- if args and type(args[-1]) is dict:
+ # If there are more args than needed, add them to kwargs
+ func = getattr(self._plugins[plugin_key], function_name)
+
+ if args.__len__() + 1 > inspect.getargspec(func).args.__len__():
kwargs.update(args.pop())
- return getattr(self._plugins[plugin_key], function_name)(*args,
- **kwargs)
+ return func(*args, **kwargs)
def get_all_networks(self, args):
"""Not implemented for this model"""
"""
import os
+from quantum.common.config import find_config_file
from quantum.plugins.cisco.common import cisco_configparser as confp
-CONF_FILE = "../conf/nexus.ini"
-
-CP = confp.CiscoConfigParser(os.path.dirname(os.path.realpath(__file__)) \
- + "/" + CONF_FILE)
+CP = confp.CiscoConfigParser(find_config_file({}, None, "nexus.ini"))
SECTION = CP['SWITCH']
NEXUS_IP_ADDRESS = SECTION['nexus_ip_address']
--- /dev/null
+#!/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 Cisco plugin
+
+export PLUGIN_DIR=plugins/cisco-plugin/lib/quantum/plugins/cisco/
+./run_tests.sh -N
+"""
+
+import os
+import unittest
+import sys
+from nose import config
+from nose import core
+
+sys.path.append(os.getcwd())
+
+import tools.source_environment
+from quantum.common.test_lib import run_tests
+
+import quantum.plugins.cisco.tests.unit
+
+
+def main():
+ c = config.Config(stream=sys.stdout,
+ env=os.environ,
+ verbosity=3,
+ includeExe=True,
+ traverseNamespace=True,
+ plugins=core.DefaultPluginManager())
+ c.configureWhere(quantum.plugins.cisco.tests.unit.__path__)
+ sys.exit(run_tests(c))
+
+if __name__ == '__main__':
+ main()
import os.path
import routes
from webtest import TestApp
-from extensions import credential
-from extensions import portprofile
-from extensions import novatenant
-from extensions import qos
-from extensions import multiport
+from quantum.extensions import credential
+from quantum.extensions import portprofile
+from quantum.extensions import novatenant
+from quantum.extensions import qos
+from quantum.extensions import multiport
from quantum.plugins.cisco.db import api as db
-from quantum.common import wsgi
+from quantum import wsgi
from quantum.common import config
from quantum.common import extensions
from quantum import api as server
from quantum.plugins.cisco.l2network_plugin import L2Network
-from tests.unit.extension_stubs import StubBaseAppController
+from quantum.tests.unit.extension_stubs import StubBaseAppController
from quantum.common.extensions import (PluginAwareExtensionManager,
ExtensionMiddleware)
from quantum.manager import QuantumManager
from quantum.plugins.cisco import l2network_plugin
-TEST_CONF_FILE = os.path.join(os.path.dirname(__file__), os.pardir,
- os.pardir, 'conf', 'quantum.conf.ciscoext')
+TEST_CONF_FILE = config.find_config_file({}, None, 'quantum.conf.ciscoext')
EXTENSIONS_PATH = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir,
os.pardir, os.pardir, os.pardir, "extensions")
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
- state=state)
+ state)
delete_port_dict = self._l2network_plugin.delete_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_update_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_port, tenant_id,
- net_id, port_id, const.PORT_UP, state=const.PORT_UP)
+ net_id, port_id, state=const.PORT_UP)
LOG.debug("test_update_port_networkDNE - END")
def test_update_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
self.assertEqual(result_vlan_name, expected_output)
LOG.debug("test_get_vlan_name - END")
- def test_validate_state(self, state=const.PORT_UP):
+ def test_validate_port_state(self, state=const.PORT_UP):
"""
Tests validate port state
"""
- LOG.debug("test_validate_state - START")
- result = self._l2network_plugin._validate_state(state)
+ LOG.debug("test_validate_port_state - START")
+ result = self._l2network_plugin._validate_port_state(state)
self.assertEqual(result, True)
- LOG.debug("test_validate_state - END")
+ LOG.debug("test_validate_port_state - END")
- def test_invalid_state(self, state="BADSTATE"):
+ def test_invalid_port_state(self, state="BADSTATE"):
"""
Tests invalidate port state
"""
- LOG.debug("test_validate_state - START")
+ LOG.debug("test_validate_port_state - START")
self.assertRaises(exc.StateInvalid,
- self._l2network_plugin._validate_state,
+ self._l2network_plugin._validate_port_state,
state)
- LOG.debug("test_validate_state - END")
+ LOG.debug("test_validate_port_state - END")
def setUp(self):
"""
self.port_id = 'p0005'
self.remote_interface = 'new_interface'
self._l2network_plugin = l2network_plugin.L2Network()
- LOG.debug(self._l2network_plugin)
- LOG.debug("asdfasdfasdfasdfasdf")
"""
Clean up functions after the tests
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, network_created["net-id"],
self.vlan_name, self.vlan_id)
- rename_net_dict = self._cisco_nexus_plugin.rename_network(
+ rename_net_dict = self._cisco_nexus_plugin.update_network(
tenant_id, new_net_dict[const.NET_ID], name=new_name)
self.assertEqual(rename_net_dict[const.NET_NAME], new_name)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
def _load_inventory(self):
"""Load the inventory from a config file"""
inventory = deepcopy(conf.INVENTORY)
- LOG.info("Loaded UCS inventory: %s\n" % inventory)
- LOG.info("Building UCS inventory state (this may take a while)...")
+ #LOG.info("Loaded UCS inventory: %s\n" % inventory)
+ #LOG.info("Building UCS inventory state (this may take a while)...")
for ucsm in inventory.keys():
ucsm_ip = inventory[ucsm][const.IP_ADDRESS]
ucsm_password)
blades_dict[blade_id] = blade_data
- LOG.debug("UCS Inventory state is: %s\n" % self._inventory_state)
+ #LOG.debug("UCS Inventory state is: %s\n" % self._inventory_state)
return True
def _get_host_name(self, ucsm_ip, chassis_id, blade_id):
blade_data = self._get_blade_state(chassis_id, blade_id, ucsm_ip,
ucsm_username, ucsm_password)
blade_intf_data = blade_data[const.BLADE_INTF_DATA]
+ #import sys
+ #sys.exit(ucsm_ip)
chassis_data = self._inventory_state[ucsm_ip][chassis_id]
old_blade_intf_data = chassis_data[blade_id][const.BLADE_INTF_DATA]
#!/usr/bin/env python
-import tools.source_nonplugin_environment
+import tools.source_environment
from quantum.run_tests import main as tests
tests()
from webob import exc
-from extensions import _credential_view as credential_view
+from quantum.extensions import _credential_view as credential_view
from quantum.api import api_common as common
from quantum.common import extensions
from quantum.manager import QuantumManager
"""
from webob import exc
-from extensions import _novatenant_view as novatenant_view
+from quantum.extensions import _novatenant_view as novatenant_view
from quantum.api import api_common as common
from quantum.common import exceptions as qexception
from quantum.common import extensions
from webob import exc
-from extensions import _pprofiles as pprofiles_view
+from quantum.extensions import _pprofiles as pprofiles_view
from quantum.api import api_common as common
from quantum.common import exceptions as qexception
from quantum.common import extensions
import logging
from webob import exc
-from extensions import _qos_view as qos_view
+from quantum.extensions import _qos_view as qos_view
from quantum.api import api_common as common
from quantum.common import extensions
from quantum.manager import QuantumManager