# 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. from test_utils import CharmTestCase, patch_open from test_pci_helper import ( check_device, mocked_subprocess, mocked_filehandle, mocked_globs, mocked_islink, mocked_realpath, ) from unittest.mock import patch, MagicMock import pci TO_PATCH = [ 'glob', 'subprocess', ] NOT_JSON = "Im not json" class PCITest(CharmTestCase): def setUp(self): super(PCITest, self).setUp(pci, TO_PATCH) 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(CharmTestCase): def setUp(self): super(PCINetDeviceTest, self).setUp(pci, TO_PATCH) @patch('os.path.islink') @patch('os.path.realpath') def eth_int(self, pci_address, _osrealpath, _osislink, subproc_map=None): self.glob.glob.side_effect = mocked_globs _osislink.side_effect = mocked_islink _osrealpath.side_effect = mocked_realpath self.subprocess.check_output.side_effect = mocked_subprocess( subproc_map=subproc_map) with patch_open() as (_open, _file): super_fh = mocked_filehandle() _file.readlines = MagicMock() _open.side_effect = super_fh._setfilename _file.read.side_effect = super_fh._getfilecontents_read _file.readlines.side_effect = super_fh._getfilecontents_readlines netint = pci.PCINetDevice(pci_address) return netint def test_base_eth_device(self): net = self.eth_int('0000:10:00.0') expect = { 'interface_name': 'eth2', 'mac_address': 'a8:9d:21:cf:93:fc', 'pci_address': '0000:10:00.0', 'state': 'up', 'sriov': False, } self.assertTrue(check_device(net, expect)) @patch('pci.get_sysnet_interfaces_and_macs') @patch('pci.PCINetDevice.update_attributes') def test_update_interface_info(self, _update, _sysnet_ints): dev = pci.PCINetDevice('0000:10:00.0') _sysnet_ints.return_value = [ { 'interface': 'eth2', 'mac_address': 'a8:9d:21:cf:93:fc', 'pci_address': '0000:10:00.0', 'state': 'up', 'sriov': False, }, { 'interface': 'eth3', 'mac_address': 'a8:9d:21:cf:93:fd', 'pci_address': '0000:10:00.1', 'state': 'down', 'sriov': False, } ] dev.update_interface_info() self.assertEqual(dev.interface_name, 'eth2') @patch('os.path.islink') @patch('os.path.realpath') @patch('pci.is_sriov') @patch('pci.get_sysnet_device_state') @patch('pci.get_sysnet_mac') @patch('pci.get_sysnet_interface') @patch('pci.PCINetDevice.update_attributes') def test_get_sysnet_interfaces_and_macs(self, _update, _interface, _mac, _state, _sriov, _osrealpath, _osislink): self.glob.glob.return_value = ['/sys/class/net/eth2'] _interface.return_value = 'eth2' _mac.return_value = 'a8:9d:21:cf:93:fc' _state.return_value = 'up' _sriov.return_value = False _osrealpath.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') expect = { 'interface': 'eth2', 'mac_address': 'a8:9d:21:cf:93:fc', 'pci_address': '0000:07:00.0', 'state': 'up', 'sriov': False, } self.assertEqual(pci.get_sysnet_interfaces_and_macs(), [expect]) @patch('os.path.islink') @patch('os.path.realpath') @patch('pci.is_sriov') @patch('pci.get_sysnet_device_state') @patch('pci.get_sysnet_mac') @patch('pci.get_sysnet_interface') @patch('pci.PCINetDevice.update_attributes') def test_get_sysnet_interfaces_and_macs_virtio(self, _update, _interface, _mac, _state, _sriov, _osrealpath, _osislink): self.glob.glob.return_value = ['/sys/class/net/eth2'] _interface.return_value = 'eth2' _mac.return_value = 'a8:9d:21:cf:93:fc' _state.return_value = 'up' _sriov.return_value = False _osrealpath.return_value = ('/sys/devices/pci0000:00/0000:00:07.0/' 'virtio5') expect = { 'interface': 'eth2', 'mac_address': 'a8:9d:21:cf:93:fc', 'pci_address': '0000:00:07.0', 'state': 'up', 'sriov': False, } self.assertEqual(pci.get_sysnet_interfaces_and_macs(), [expect]) @patch('pci.PCINetDevice.update_attributes') def test_get_sysnet_mac(self, _update): with patch_open() as (_open, _file): super_fh = mocked_filehandle() _file.readlines = MagicMock() _open.side_effect = super_fh._setfilename _file.read.side_effect = super_fh._getfilecontents_read macaddr = pci.get_sysnet_mac('/sys/class/net/eth3') self.assertEqual(macaddr, 'a8:9d:21:cf:93:fd') @patch('pci.PCINetDevice.update_attributes') def test_get_sysnet_device_state(self, _update): with patch_open() as (_open, _file): super_fh = mocked_filehandle() _file.readlines = MagicMock() _open.side_effect = super_fh._setfilename _file.read.side_effect = super_fh._getfilecontents_read state = pci.get_sysnet_device_state('/sys/class/net/eth3') self.assertEqual(state, 'down') @patch('pci.PCINetDevice.update_attributes') def test_get_sysnet_interface(self, _update): self.assertEqual( pci.get_sysnet_interface('/sys/class/net/eth3'), 'eth3') class PCINetDevicesTest(CharmTestCase): def setUp(self): super(PCINetDevicesTest, self).setUp(pci, TO_PATCH) @patch('os.path.islink') def pci_devs(self, _osislink, subproc_map=None): self.glob.glob.side_effect = mocked_globs rp_patcher = patch('os.path.realpath') rp_mock = rp_patcher.start() rp_mock.side_effect = mocked_realpath _osislink.side_effect = mocked_islink self.subprocess.check_output.side_effect = mocked_subprocess( subproc_map=subproc_map) with patch_open() as (_open, _file): super_fh = mocked_filehandle() _file.readlines = MagicMock() _open.side_effect = super_fh._setfilename _file.read.side_effect = super_fh._getfilecontents_read _file.readlines.side_effect = super_fh._getfilecontents_readlines devices = pci.PCINetDevices() rp_patcher.stop() return devices def test_base(self): devices = self.pci_devs() self.assertEqual(len(devices.pci_devices), 2) expect = { '0000:10:00.0': { 'interface_name': 'eth2', 'mac_address': 'a8:9d:21:cf:93:fc', 'pci_address': '0000:10:00.0', 'state': 'up', 'sriov': False, }, '0000:10:00.1': { 'interface_name': 'eth3', 'mac_address': 'a8:9d:21:cf:93:fd', 'pci_address': '0000:10:00.1', 'state': 'down', 'sriov': False, }, } for device in devices.pci_devices: self.assertTrue(check_device(device, expect[device.pci_address])) def test_get_pci_ethernet_addresses(self): devices = self.pci_devs() expect = ['0000:10:00.0', '0000:10:00.1'] self.assertEqual(devices.get_pci_ethernet_addresses(), expect) @patch('pci.PCINetDevice.update_attributes') def test_update_devices(self, _update): devices = self.pci_devs() call_count = _update.call_count devices.update_devices() self.assertEqual(_update.call_count, call_count + 2) def test_get_macs(self): devices = self.pci_devs() expect = ['a8:9d:21:cf:93:fc', 'a8:9d:21:cf:93:fd'] self.assertEqual(devices.get_macs(), expect) def test_get_device_from_mac(self): devices = self.pci_devs() expect = { '0000:10:00.1': { 'interface_name': 'eth3', 'mac_address': 'a8:9d:21:cf:93:fd', 'pci_address': '0000:10:00.1', 'state': 'down', 'sriov': False, }, } self.assertTrue(check_device( devices.get_device_from_mac('a8:9d:21:cf:93:fd'), expect['0000:10:00.1'])) def test_get_device_from_pci_address(self): devices = self.pci_devs() expect = { '0000:10:00.1': { 'interface_name': 'eth3', 'mac_address': 'a8:9d:21:cf:93:fd', 'pci_address': '0000:10:00.1', 'state': 'down', }, } self.assertTrue(check_device( devices.get_device_from_pci_address('0000:10:00.1'), expect['0000:10:00.1']))