# which case, the credentials are already populated
pass
- @staticmethod
- def put_credential(cred_name, username, password):
- """Set the username and password."""
- cdb.add_credential(cred_name, username, password)
-
@staticmethod
def get_username(cred_name):
"""Get the username."""
"""Get the password."""
credential = cdb.get_credential_name(cred_name)
return credential[const.CREDENTIAL_PASSWORD]
-
- @staticmethod
- def get_credential(cred_name):
- """Get the username and password."""
- cdb.get_credential_name(cred_name)
- return {const.USERNAME: const.CREDENTIAL_USERNAME,
- const.PASSWORD: const.CREDENTIAL_PASSWORD}
-
- @staticmethod
- def delete_credential(cred_name):
- """Delete a credential."""
- cdb.remove_credential(cred_name)
--- /dev/null
+# Copyright (c) 2013 Cisco Systems Inc.
+#
+# 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.
+
+import mock
+from oslo.config import cfg
+
+from neutron.common import config as neutron_config
+from neutron.plugins.cisco.common import config as cisco_config
+from neutron.tests import base
+from neutron.tests.unit import test_api_v2
+
+
+class TestCiscoNexusPluginConfig(base.BaseTestCase):
+
+ def setUp(self):
+ # Point neutron config file to: neutron/tests/etc/neutron.conf.test
+ args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
+ neutron_config.parse(args=args)
+
+ super(TestCiscoNexusPluginConfig, self).setUp()
+
+ def test_config_parse_error(self):
+ """Check that config error is raised upon config parser failure."""
+ with mock.patch.object(cfg, 'MultiConfigParser') as parser:
+ parser.return_value.read.return_value = []
+ self.assertRaises(cfg.Error, cisco_config.CiscoConfigOptions)
+
+ def test_create_device_dictionary(self):
+ """Test creation of the device dictionary based on nexus config."""
+ test_config = {
+ 'NEXUS_SWITCH:1.1.1.1': {
+ 'username': ['admin'],
+ 'password': ['mySecretPassword'],
+ 'ssh_port': [22],
+ 'compute1': ['1/1'],
+ 'compute2': ['1/2'],
+ },
+ 'NEXUS_SWITCH:2.2.2.2': {
+ 'username': ['admin'],
+ 'password': ['mySecretPassword'],
+ 'ssh_port': [22],
+ 'compute3': ['1/1'],
+ 'compute4': ['1/2'],
+ },
+ }
+ expected_dev_dict = {
+ ('NEXUS_SWITCH', '1.1.1.1', 'username'): 'admin',
+ ('NEXUS_SWITCH', '1.1.1.1', 'password'): 'mySecretPassword',
+ ('NEXUS_SWITCH', '1.1.1.1', 'ssh_port'): 22,
+ ('NEXUS_SWITCH', '1.1.1.1', 'compute1'): '1/1',
+ ('NEXUS_SWITCH', '1.1.1.1', 'compute2'): '1/2',
+ ('NEXUS_SWITCH', '2.2.2.2', 'username'): 'admin',
+ ('NEXUS_SWITCH', '2.2.2.2', 'password'): 'mySecretPassword',
+ ('NEXUS_SWITCH', '2.2.2.2', 'ssh_port'): 22,
+ ('NEXUS_SWITCH', '2.2.2.2', 'compute3'): '1/1',
+ ('NEXUS_SWITCH', '2.2.2.2', 'compute4'): '1/2',
+ }
+ with mock.patch.object(cfg, 'MultiConfigParser') as parser:
+ parser.return_value.read.return_value = cfg.CONF.config_file
+ parser.return_value.parsed = [test_config]
+ cisco_config.CiscoConfigOptions()
+ self.assertEqual(cisco_config.device_dictionary,
+ expected_dev_dict)
import testtools
from neutron.db import api as db
+from neutron.plugins.cisco.common import cisco_constants
+from neutron.plugins.cisco.common import cisco_credentials_v2
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
+from neutron.plugins.cisco.common import config as config
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco import network_plugin
from neutron.tests import base
def setUp(self):
super(CiscoNetworkDbTest, self).setUp()
db.configure_db()
- self.session = db.get_session()
# The Cisco network plugin includes a thin layer of QoS and
# credential API methods which indirectly call Cisco QoS and
self.assertRaises(c_exc.CredentialNotFound,
self._network_plugin.get_credential_details,
"dummyCredentialId")
+
+
+class CiscoCredentialStoreTest(base.BaseTestCase):
+
+ """Cisco Credential Store unit tests."""
+
+ def setUp(self):
+ super(CiscoCredentialStoreTest, self).setUp()
+ db.configure_db()
+ self.addCleanup(db.clear_db)
+
+ def test_cred_store_init_duplicate_creds_ignored(self):
+ """Check that with multi store instances, dup creds are ignored."""
+ # Create a device dictionary containing credentials for 1 switch.
+ dev_dict = {
+ ('dev_id', '1.1.1.1', cisco_constants.USERNAME): 'user_1',
+ ('dev_id', '1.1.1.1', cisco_constants.PASSWORD): 'password_1',
+ ('dev_id', '1.1.1.1', 'host_a'): '1/1',
+ ('dev_id', '1.1.1.1', 'host_b'): '1/2',
+ ('dev_id', '1.1.1.1', 'host_c'): '1/3',
+ }
+ with mock.patch.object(config, 'get_device_dictionary',
+ return_value=dev_dict):
+ # Create and initialize 2 instances of credential store.
+ cisco_credentials_v2.Store().initialize()
+ cisco_credentials_v2.Store().initialize()
+ # There should be only 1 switch credential in the database.
+ self.assertEqual(len(cdb.get_all_credentials()), 1)