compass-core/compass/tests/hdsdiscovery/test_hdsdiscovery.py
Xicheng Chang 3d4e5d0af6 Changed ansible directory structure
Moved roles to the parent directory
so that different adapters don't have
to have duplicated roles in their
respective paths.

patch2: fix some code to follow new pep8 standards.
Change-Id: I748c1730be8045c1cb83f91eaa9f0f551cd20a6f
2015-05-22 17:06:47 -07:00

247 lines
8.9 KiB
Python
Executable File

#!/usr/bin/python
#
# Copyright 2014 Huawei Technologies Co. 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.
"""test hdsdiscovery module."""
import os
import unittest2
from mock import Mock
from mock import patch
os.environ['COMPASS_IGNORE_SETTING'] = 'true'
from compass.utils import setting_wrapper as setting
reload(setting)
from compass.hdsdiscovery.hdmanager import HDManager
from compass.hdsdiscovery.vendors.huawei.huawei import Huawei
from compass.hdsdiscovery.vendors.huawei.plugins.mac import Mac
from compass.hdsdiscovery.vendors.ovswitch.plugins.mac import Mac as OVSMac
from compass.utils import flags
from compass.utils import logsetting
SNMP_V2_CREDENTIALS = {'version': '2c',
'community': 'public'}
class HuaweiTest(unittest2.TestCase):
"""test huawei switch snmp get."""
def setUp(self):
super(HuaweiTest, self).setUp()
logsetting.init()
self.huawei = Huawei()
self.sys_info = 'Huawei Technologies'
def tearDown(self):
del self.huawei
super(HuaweiTest, self).tearDown()
def test_is_this_vendor(self):
"""test device vendor is haiwei."""
# Incorrect system information
incorrect_sys_info = "xxx"
self.assertFalse(
self.huawei.is_this_vendor(incorrect_sys_info))
# Correct vendor
self.assertTrue(
self.huawei.is_this_vendor(self.sys_info))
class HuaweiMacTest(unittest2.TestCase):
"""test get mac from huawei device."""
def setUp(self):
super(HuaweiMacTest, self).setUp()
logsetting.init()
host = '192.168.1.1'
credential = {'version': '2c', 'community': 'public'}
self.mac_plugin = Mac(host, credential)
def tearDown(self):
del self.mac_plugin
super(HuaweiMacTest, self).tearDown()
@patch("compass.hdsdiscovery.utils.snmpwalk_by_cl")
def test_process_data(self, mock_snmpwalk):
"""get progress data function."""
# GET operation haven't been implemeneted.
self.assertIsNone(self.mac_plugin.process_data('GET'))
# SNMP Walk Timeout
# utils.snmpwalk_by_cl = Mock(return_value=None)
mock_snmpwalk.return_value = None
self.assertIsNone(self.mac_plugin.process_data())
# Successfully get MAC addresses from the switch
mock_snmp_walk_result = [
{"iid": "40.110.212.77.198.190.88.1.48", "value": "10"},
{"iid": "40.110.212.100.199.74.88.1.48", "value": "11"},
{"iid": "0.12.41.53.220.2.88.1.48", "value": "12"}
]
expected_mac_info = [
{"mac": "28:6e:d4:4d:c6:be", "port": "1", "vlan": "88"},
{"mac": "28:6e:d4:64:c7:4a", "port": "2", "vlan": "88"},
{"mac": "00:0c:29:35:dc:02", "port": "3", "vlan": "88"}
]
# utils.snmpwalk_by_cl = Mock(return_value=mock_snmp_walk_result)
mock_snmpwalk.return_value = mock_snmp_walk_result
self.mac_plugin.get_port = Mock()
self.mac_plugin.get_port.side_effect = ["1", "2", "3"]
result = self.mac_plugin.process_data()
self.assertEqual(expected_mac_info, result)
class OVSMacTest(unittest2.TestCase):
"""ovs switch test."""
def setUp(self):
super(OVSMacTest, self).setUp()
logsetting.init()
self.host = '10.145.88.160'
self.credential = {'username': 'root', 'password': 'huawei'}
@patch('compass.hdsdiscovery.utils.ssh_remote_execute')
def test_scan(self, ovs_mock):
"""test scan ovs switch."""
ovs_mock.return_value = []
mac_instance = OVSMac(self.host, self.credential)
self.assertIsNone(mac_instance.scan())
del mac_instance
ovs_mock.return_value = ['\n', '\n', '\n']
mac_instance = OVSMac(self.host, self.credential)
self.assertEqual([], mac_instance.scan())
del mac_instance
class HDManagerTest(unittest2.TestCase):
"""test HDManager."""
def setUp(self):
super(HDManagerTest, self).setUp()
logsetting.init()
self.manager = HDManager()
self.correct_host = '33.33.33.1'
self.correct_host_2 = '127.0.0.1'
self.correct_credential = {'version': '2c', 'community': 'public'}
def tearDown(self):
del self.manager
super(HDManagerTest, self).tearDown()
@patch('compass.hdsdiscovery.hdmanager.HDManager.get_sys_info')
def test_get_vendor(self, sys_info_mock):
"""test get_vendor."""
# Incorrect ip
vendor, state, err = self.manager.get_vendor('1234.1.1.1',
self.correct_credential)
self.assertIsNone(vendor)
self.assertEqual('error', state)
# Incorrect credential
incorr_cred = {'version': '1v', 'community': 'private'}
vendor, state, err = self.manager.get_vendor(self.correct_host,
incorr_cred)
self.assertIsNone(vendor)
self.assertEqual('error', state)
# SNMP get system description Timeout
excepted_err_msg = 'Timeout: No Response from 127.0.0.1.'
sys_info_mock.return_value = (None, excepted_err_msg)
result, state, err = self.manager.get_vendor(self.correct_host,
self.correct_credential)
self.assertIsNone(result)
self.assertEqual(state, 'unreachable')
self.assertEqual(err, excepted_err_msg)
# No vendor plugin supported
excepted_err_msg = 'Not supported switch vendor!'
sys_info_mock.return_value = ('xxxxxx', excepted_err_msg)
result, state, err = self.manager.get_vendor(self.correct_host,
self.correct_credential)
self.assertIsNone(result)
self.assertEqual(state, 'notsupported')
self.assertEqual(err, excepted_err_msg)
# Found the correct vendor
sys_info = ['Huawei Versatile Routing Platform Software',
'ProCurve J9089A Switch 2610-48-PWR, revision R.11.25',
'Pica8 XorPlus Platform Software']
expected_vendor_names = ['huawei', 'hp', 'pica8', 'appliance']
for info, expected_vendor in zip(sys_info, expected_vendor_names):
sys_info_mock.return_value = (info, '')
# the result is a tuple ($vendor, $state, $error_message)
result = self.manager.get_vendor(self.correct_host,
self.correct_credential)
self.assertEqual(result[0], expected_vendor)
@patch('compass.hdsdiscovery.hdmanager.HDManager.get_sys_info')
def test_is_valid_vendor(self, sys_info_mock):
"""test is_valid_vendor."""
# non-exsiting vendor under vendors directory
self.assertFalse(
self.manager.is_valid_vendor(self.correct_host,
self.correct_credential,
'xxxx')
)
# No system description retrieved
sys_info_mock.return_value = (None, 'TIMEOUT')
self.assertFalse(
self.manager.is_valid_vendor(self.correct_host,
self.correct_credential,
'pica8')
)
# Incorrect vendor name
sys_info = 'Pica8 XorPlus Platform Software'
sys_info_mock.return_value = (sys_info, '')
self.assertFalse(
self.manager.is_valid_vendor(self.correct_host,
self.correct_credential,
'huawei')
)
# Correct vendor name
self.assertTrue(
self.manager.is_valid_vendor(self.correct_host,
self.correct_credential,
'pica8')
)
def test_learn(self):
"""test learn."""
# non-exsiting plugin
self.assertIsNone(self.manager.learn(self.correct_host,
self.correct_credential,
'huawei', 'xxx'))
# non-existing vendor
self.assertIsNone(self.manager.learn(self.correct_host,
self.correct_credential,
'xxxx', 'mac'))
if __name__ == '__main__':
flags.init()
logsetting.init()
unittest2.main()