2812f9f664
The mock third party library was needed for mock support in py2 runtimes. Since we now only support py36 and later, we can use the standard lib unittest.mock module instead. Change-Id: I491ca9f482a00b7ca3fa44aa8c26ef73559c178f
557 lines
24 KiB
Python
557 lines
24 KiB
Python
# 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 unittest 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)
|