[Mellanox] Add more unit test coverage for platform API (#15842) (#16137)

- Why I did it
Increase UT coverage for Nvidia platform API code

Work item tracking
Microsoft ADO (number only):

- How I did it
Focus on low coverage file:
1. component.py
2. watchdog.py
3. pcie.py

- How to verify it
Run the unit test, the coverage has been changed from 70% to 90%

Co-authored-by: Junchao-Mellanox <57339448+Junchao-Mellanox@users.noreply.github.com>
This commit is contained in:
mssonicbld 2023-08-14 22:40:38 +08:00 committed by GitHub
parent 3119077744
commit 75b7ec361c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 941 additions and 53 deletions

View File

@ -536,8 +536,8 @@ class ComponentSSD(Component):
try:
reboot_required = self.get_firmware_update_notification(image_path) is not None
except RuntimeError as e:
return FW_AUTO_ERR_UNKNOWN
return FW_AUTO_ERR_UNKNOWN
# Update if no reboot needed
if not reboot_required:
self.update_firmware(image_path)
@ -810,9 +810,9 @@ class ComponentCPLD(Component):
# Install burn. Error if fail.
if not self.install_firmware(image_path):
return FW_AUTO_ERR_UNKNOWN
# Schedule refresh
return FW_AUTO_SCHEDULED
return FW_AUTO_SCHEDULED
def get_firmware_version(self):
part_number_file = self.CPLD_PART_NUMBER_FILE.format(self.idx)
@ -844,7 +844,7 @@ class ComponentCPLD(Component):
def get_firmware_update_notification(self, image_path):
name, ext = os.path.splitext(os.path.basename(image_path))
if ext == self.COMPONENT_FIRMWARE_EXTENSION:
if ext in self.COMPONENT_FIRMWARE_EXTENSION:
return "Power cycle (with 30 sec delay) or refresh image is required to complete {} firmware update".format(self.name)
return "Immediate power cycle is required to complete {} firmware update".format(self.name)

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2019-2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2019-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -198,18 +198,6 @@ def deinitialize_sdk_handle(sdk_handle):
return False
class SdkHandleContext(object):
def __init__(self):
self.sdk_handle = None
def __enter__(self):
self.sdk_handle = initialize_sdk_handle()
return self.sdk_handle
def __exit__(self, exc_type, exc_val, exc_tb):
deinitialize_sdk_handle(self.sdk_handle)
class NvidiaSFPCommon(SfpOptoeBase):
def __init__(self, sfp_index):
super(NvidiaSFPCommon, self).__init__()

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2019-2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2019-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -80,7 +80,7 @@ class WatchdogImplBase(WatchdogBase):
super(WatchdogImplBase, self).__init__()
self.watchdog_path = wd_device_path
self.watchdog = os.open(self.watchdog_path, os.O_WRONLY)
self.watchdog = self.open_handle()
# Opening a watchdog descriptor starts
# watchdog timer;
@ -90,6 +90,9 @@ class WatchdogImplBase(WatchdogBase):
self.timeout = self._gettimeout()
def open_handle(self):
return os.open(self.watchdog_path, os.O_WRONLY)
def _enablecard(self):
"""
Turn on the watchdog timer
@ -290,6 +293,7 @@ def get_watchdog():
for device in os.listdir("/dev/"):
if device.startswith("watchdog") and is_mlnx_wd_main(device):
watchdog_main_device_name = device
break
if watchdog_main_device_name is None:
return None

Binary file not shown.

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2021-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -326,3 +326,8 @@ class TestChassis:
exceptionRaised = True
assert exceptionRaised
def test_parse_dmi(self):
chassis = Chassis()
content = chassis._parse_dmi(os.path.join(test_path, 'dmi_file'))
assert content.get('Version') == 'A4'

View File

@ -0,0 +1,519 @@
#
# Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# 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 os
import pytest
import subprocess
import sys
if sys.version_info.major == 3:
from unittest import mock
else:
import mock
test_path = os.path.dirname(os.path.abspath(__file__))
modules_path = os.path.dirname(test_path)
sys.path.insert(0, modules_path)
from sonic_platform.chassis import Chassis
from sonic_platform.component import ComponentONIE, \
ComponentSSD, \
ComponentBIOS, \
ComponentBIOSSN2201, \
ComponentCPLD, \
ComponentCPLDSN2201, \
MPFAManager, \
ONIEUpdater, \
Component
from sonic_platform_base.component_base import FW_AUTO_INSTALLED, \
FW_AUTO_UPDATED, \
FW_AUTO_SCHEDULED, \
FW_AUTO_ERR_BOOT_TYPE, \
FW_AUTO_ERR_IMAGE, \
FW_AUTO_ERR_UNKNOWN
class TestComponent:
@mock.patch('sonic_platform.chassis.utils.is_host')
@mock.patch('sonic_platform.chassis.DeviceDataManager.get_cpld_component_list', mock.MagicMock(return_value=[]))
def test_chassis_component(self, mock_is_host):
mock_is_host.return_value = False
c = Chassis()
assert not c.get_all_components()
mock_is_host.return_value = True
component_list = c.get_all_components()
assert len(component_list) > 0
assert c.get_num_components() > 0
assert c.get_component(0) is not None
@mock.patch('sonic_platform.component.ComponentONIE._check_file_validity')
@mock.patch('sonic_platform.component.ONIEUpdater', mock.MagicMock())
def test_onie_component(self, mock_check_file):
c = ComponentONIE()
assert c.get_name() == 'ONIE'
assert c.get_description() == 'ONIE - Open Network Install Environment'
c.onie_updater.get_onie_version = mock.MagicMock(return_value='1.0')
assert c.get_firmware_version() == '1.0'
c.onie_updater.get_onie_firmware_info = mock.MagicMock(return_value={})
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
c.onie_updater.get_onie_firmware_info = mock.MagicMock(return_value={'image_version': '1.1'})
assert c.get_available_firmware_version('') == '1.1'
assert c.get_firmware_update_notification('') == \
'Immediate cold reboot is required to complete ONIE firmware update'
mock_check_file.return_value = False
assert not c.install_firmware('')
c.update_firmware('')
mock_check_file.return_value = True
c.onie_updater.update_firmware = mock.MagicMock()
assert c.install_firmware('')
c.onie_updater.update_firmware.side_effect = RuntimeError('')
assert not c.install_firmware('')
@mock.patch('sonic_platform.component.os.path.exists')
@mock.patch('sonic_platform.component.subprocess.check_call')
@mock.patch('sonic_platform.component.subprocess.check_output')
def test_ssd_component(self, mock_check_output, mock_check_call, mock_exists):
c = ComponentSSD()
firmware_info = [
'Firmware Version:1.0',
'Available Firmware Version:1.1',
'Upgrade Required:yes',
'Power Cycle Required:yes'
]
mock_check_output.return_value = '\n'.join(firmware_info)
assert c.get_firmware_version() == '1.0'
assert c.get_available_firmware_version('') == '1.1'
assert c.get_firmware_update_notification('') == \
'Immediate power cycle is required to complete SSD firmware update'
mock_check_output.return_value = ''
with pytest.raises(RuntimeError):
c.get_firmware_version()
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
mock_check_output.return_value = 'Upgrade Required:invalid'
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
with pytest.raises(RuntimeError):
c.get_firmware_update_notification('')
mock_check_output.return_value = 'Upgrade Required:no'
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
assert c.get_firmware_update_notification('') is None
mock_check_output.return_value = 'Upgrade Required:yes'
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
firmware_info = [
'Power Cycle Required:invalid',
'Upgrade Required:yes'
]
mock_check_output.return_value = '\n'.join(firmware_info)
with pytest.raises(RuntimeError):
c.get_firmware_update_notification('')
firmware_info = [
'Firmware Version:1.0',
'Upgrade Required:yes'
]
mock_check_output.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
c.get_firmware_version()
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
with pytest.raises(RuntimeError):
c.get_firmware_update_notification('')
# install firmware
c._check_file_validity = mock.MagicMock(return_value=False)
assert not c.install_firmware('')
c.update_firmware('')
c._check_file_validity = mock.MagicMock(return_value=True)
assert c.install_firmware('')
mock_check_call.assert_called_with(c.SSD_FIRMWARE_UPDATE_COMMAND, universal_newlines=True)
assert c.install_firmware('', False)
mock_check_call.assert_called_with(c.SSD_FIRMWARE_INSTALL_COMMAND, universal_newlines=True)
mock_check_call.side_effect = subprocess.CalledProcessError(1, None)
assert not c.install_firmware('')
# auto update firmware
mock_exists.return_value = False
assert c.auto_update_firmware('', '') == FW_AUTO_ERR_IMAGE
c.get_firmware_update_notification = mock.MagicMock(side_effect=RuntimeError(''))
mock_exists.return_value = True
assert c.auto_update_firmware('', '') == FW_AUTO_ERR_UNKNOWN
c.update_firmware = mock.MagicMock()
c.get_firmware_update_notification = mock.MagicMock(return_value=None)
assert c.auto_update_firmware('', '') == FW_AUTO_UPDATED
c.get_firmware_update_notification = mock.MagicMock(return_value='yes')
assert c.auto_update_firmware('', '') == FW_AUTO_ERR_BOOT_TYPE
assert c.auto_update_firmware('', 'cold') == FW_AUTO_SCHEDULED
@mock.patch('sonic_platform.component.subprocess.check_output')
def test_bios_component(self, mock_check_output):
c = ComponentBIOS()
mock_check_output.return_value = '1.0'
assert c.get_firmware_version() == '1.0'
mock_check_output.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
c.get_firmware_version()
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
assert c.get_firmware_update_notification('') == \
'Immediate cold reboot is required to complete BIOS firmware update'
c.onie_updater.is_non_onie_firmware_update_supported = mock.MagicMock(return_value=False)
assert not c.install_firmware('')
c.update_firmware('')
c.onie_updater.is_non_onie_firmware_update_supported = mock.MagicMock(return_value=True)
c._check_file_validity = mock.MagicMock(return_value=False)
assert not c.install_firmware('')
c._check_file_validity = mock.MagicMock(return_value=True)
c.onie_updater.update_firmware = mock.MagicMock()
assert c.install_firmware('')
c.onie_updater.update_firmware = mock.MagicMock(side_effect=RuntimeError(''))
assert not c.install_firmware('')
@mock.patch('sonic_platform.component.subprocess.check_output')
def test_bios_2201_component(self, mock_check_output):
c = ComponentBIOSSN2201()
mock_check_output.return_value = 'Version: 1.0'
assert c.get_firmware_version() == '1.0'
mock_check_output.return_value = ''
assert c.get_firmware_version() == 'Unknown version'
mock_check_output.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
c.get_firmware_version()
@mock.patch('sonic_platform.component.MPFAManager.cleanup', mock.MagicMock())
@mock.patch('sonic_platform.component.MPFAManager.extract', mock.MagicMock())
@mock.patch('sonic_platform.component.subprocess.check_call')
@mock.patch('sonic_platform.component.MPFAManager.get_path')
@mock.patch('sonic_platform.component.MPFAManager.get_metadata')
@mock.patch('sonic_platform.component.os.path.exists')
def test_cpld_component(self, mock_exists, mock_get_meta_data, mock_get_path, mock_check_call):
c = ComponentCPLD(1)
c._read_generic_file = mock.MagicMock(side_effect=[None, '1', None])
assert c.get_firmware_version() == 'CPLD000000_REV0100'
assert c.get_firmware_update_notification('a.txt') == \
'Immediate power cycle is required to complete CPLD1 firmware update'
assert c.get_firmware_update_notification('a.vme') == \
'Power cycle (with 30 sec delay) or refresh image is required to complete CPLD1 firmware update'
mock_meta_data = mock.MagicMock()
mock_meta_data.has_option = mock.MagicMock(return_value=False)
mock_get_meta_data.return_value = mock_meta_data
with pytest.raises(RuntimeError):
c.get_available_firmware_version('')
mock_meta_data.has_option = mock.MagicMock(return_value=True)
mock_meta_data.get = mock.MagicMock(return_value='1.1')
assert c.get_available_firmware_version('') == '1.1'
c._check_file_validity = mock.MagicMock(return_value=False)
assert not c._install_firmware('')
c._check_file_validity = mock.MagicMock(return_value=True)
c._ComponentCPLD__get_mst_device = mock.MagicMock(return_value=None)
assert not c._install_firmware('')
c._ComponentCPLD__get_mst_device = mock.MagicMock(return_value='some dev')
assert c._install_firmware('')
mock_check_call.side_effect = subprocess.CalledProcessError(1, None)
assert not c._install_firmware('')
c._install_firmware = mock.MagicMock()
mock_meta_data.has_option = mock.MagicMock(return_value=False)
with pytest.raises(RuntimeError):
c.install_firmware('a.mpfa')
mock_get_path.return_value = '/tmp'
mock_meta_data.has_option = mock.MagicMock(return_value=True)
mock_meta_data.get = mock.MagicMock(return_value='some_firmware')
c.install_firmware('a.mpfa')
c._install_firmware.assert_called_with('/tmp/some_firmware')
c._install_firmware.reset_mock()
c.install_firmware('a.txt')
c._install_firmware.assert_called_with('a.txt')
mock_meta_data.has_option = mock.MagicMock(return_value=False)
with pytest.raises(RuntimeError):
c.update_firmware('a.mpfa')
mock_meta_data.has_option = mock.MagicMock(side_effect=[True, False])
with pytest.raises(RuntimeError):
c.update_firmware('a.mpfa')
mock_meta_data.has_option = mock.MagicMock(return_value=True)
mock_meta_data.get = mock.MagicMock(side_effect=['burn', 'refresh'])
c._install_firmware.reset_mock()
c.update_firmware('a.mpfa')
assert c._install_firmware.call_count == 2
c._install_firmware.reset_mock()
c._install_firmware.return_value = False
mock_meta_data.get = mock.MagicMock(side_effect=['burn', 'refresh'])
c.update_firmware('a.mpfa')
assert c._install_firmware.call_count == 1
mock_exists.return_value = False
assert c.auto_update_firmware('', '') == FW_AUTO_ERR_IMAGE
mock_exists.return_value = True
assert c.auto_update_firmware('', '') == FW_AUTO_ERR_BOOT_TYPE
c.install_firmware = mock.MagicMock(return_value=False)
assert c.auto_update_firmware('', 'cold') == FW_AUTO_ERR_UNKNOWN
c.install_firmware = mock.MagicMock(return_value=True)
assert c.auto_update_firmware('', 'cold') == FW_AUTO_SCHEDULED
@mock.patch('sonic_platform.component.ComponentCPLD._read_generic_file', mock.MagicMock(return_value='3'))
def test_cpld_get_component_list(self):
component_list = ComponentCPLD.get_component_list()
assert len(component_list) == 3
for index, item in enumerate(component_list):
assert item.name == 'CPLD{}'.format(index + 1)
def test_cpld_get_mst_device(self):
ComponentCPLD.MST_DEVICE_PATH = '/tmp/mst'
os.system('rm -rf /tmp/mst')
c = ComponentCPLD(1)
assert c._ComponentCPLD__get_mst_device() is None
os.makedirs(ComponentCPLD.MST_DEVICE_PATH)
assert c._ComponentCPLD__get_mst_device() is None
with open('/tmp/mst/mt0_pci_cr0', 'w+') as f:
f.write('dummy')
assert c._ComponentCPLD__get_mst_device() == '/tmp/mst/mt0_pci_cr0'
@mock.patch('sonic_platform.component.subprocess.check_call')
def test_cpld_2201_component(self, mock_check_call):
c = ComponentCPLDSN2201(1)
assert c._install_firmware('')
mock_check_call.side_effect = subprocess.CalledProcessError(1, None)
assert not c._install_firmware('')
@mock.patch('sonic_platform.component.MPFAManager.cleanup')
@mock.patch('sonic_platform.component.MPFAManager.extract')
def test_mpfa_manager_context(self, mock_extract, mock_cleanup):
with MPFAManager('some_path') as mpfa:
assert isinstance(mpfa, MPFAManager)
mock_extract.assert_called_once()
mock_cleanup.assert_not_called()
mock_cleanup.assert_called_once()
@mock.patch('sonic_platform.component.tempfile.mkdtemp', mock.MagicMock(return_value='/tmp/mpfa'))
@mock.patch('sonic_platform.component.subprocess.check_call', mock.MagicMock())
@mock.patch('sonic_platform.component.os.path.isfile')
def test_mpfa_manager_extract_cleanup(self, mock_isfile):
m = MPFAManager('some_path')
mock_isfile.return_value = False
with pytest.raises(RuntimeError):
m.extract()
mock_isfile.return_value = True
with pytest.raises(RuntimeError):
m.extract()
m = MPFAManager('some_path.mpfa')
mock_isfile.side_effect = [True, False]
with pytest.raises(RuntimeError):
m.extract()
mock_isfile.side_effect = None
os.makedirs('/tmp/mpfa', exist_ok=True)
with open('/tmp/mpfa/metadata.ini', 'w+') as f:
f.write('[section1]\n')
f.write('a=b')
m = MPFAManager('some_path.mpfa')
m.extract()
assert m.get_path() == '/tmp/mpfa'
assert m.is_extracted()
assert m.get_metadata() is not None
# extract twice and verify no issue
m.extract()
m.cleanup()
assert m.get_path() is None
assert not m.is_extracted()
assert m.get_metadata() is None
def test_onie_updater_parse_onie_version(self):
o = ONIEUpdater()
onie_year, onie_month, onie_major, onie_minor, onie_release, onie_baudrate = \
o.parse_onie_version('2022.08-5.3.0010-9600')
assert onie_year == '2022'
assert onie_month == '08'
assert onie_major == '5'
assert onie_minor == '3'
assert onie_release == '0010'
assert onie_baudrate == '9600'
with pytest.raises(RuntimeError):
o.parse_onie_version('invalid', is_base=True)
with pytest.raises(RuntimeError):
o.parse_onie_version('invalid', is_base=False)
onie_year, onie_month, onie_major, onie_minor, onie_release, onie_baudrate = \
o.parse_onie_version('2022.08-5.3.0010-9600', is_base=True)
assert onie_year is None
assert onie_month is None
assert onie_major == '5'
assert onie_minor == '3'
assert onie_release == '0010'
assert onie_baudrate is None
assert o.get_onie_required_version() == o.ONIE_VERSION_REQUIRED
@mock.patch('sonic_platform.component.ONIEUpdater.get_onie_version')
@mock.patch('sonic_platform.component.device_info.get_platform')
def test_onie_updater_is_non_onie_firmware_update_supported(self, mock_platform, mock_version):
mock_platform.return_value = 'x86_64-nvidia_sn5600-r0'
o = ONIEUpdater()
mock_version.return_value = '2022.08-5.3.0010-9600'
assert o.is_non_onie_firmware_update_supported()
mock_version.return_value = '2022.08-5.1.0010-9600'
assert not o.is_non_onie_firmware_update_supported()
mock_platform.return_value = 'x86_64-nvidia_sn2201-r0'
o = ONIEUpdater()
assert o.is_non_onie_firmware_update_supported()
def test_onie_updater_get_onie_version(self):
o = ONIEUpdater()
o._ONIEUpdater__mount_onie_fs = mock.MagicMock()
o._ONIEUpdater__umount_onie_fs = mock.MagicMock()
mock_os_open = mock.mock_open(read_data='')
with mock.patch('sonic_platform.component.open', mock_os_open):
with pytest.raises(RuntimeError):
o.get_onie_version()
o._ONIEUpdater__umount_onie_fs.assert_called_once()
mock_os_open = mock.mock_open(read_data='onie_version')
with mock.patch('sonic_platform.component.open', mock_os_open):
with pytest.raises(RuntimeError):
o.get_onie_version()
mock_os_open = mock.mock_open(read_data='onie_version=2022.08-5.1.0010-9600')
with mock.patch('sonic_platform.component.open', mock_os_open):
assert o.get_onie_version() == '2022.08-5.1.0010-9600'
@mock.patch('sonic_platform.component.subprocess.check_output')
def test_onie_updater_get_onie_firmware_info(self, mock_check_output):
o = ONIEUpdater()
o._ONIEUpdater__mount_onie_fs = mock.MagicMock()
o._ONIEUpdater__umount_onie_fs = mock.MagicMock()
mock_check_output.return_value = 'a'
with pytest.raises(RuntimeError):
o.get_onie_firmware_info('')
o._ONIEUpdater__umount_onie_fs.assert_called_once()
mock_check_output.return_value = 'a=b'
fi = o.get_onie_firmware_info('')
assert fi == {'a':'b'}
mock_check_output.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
o.get_onie_firmware_info('')
def test_onie_updater_update_firmware(self):
o = ONIEUpdater()
o._ONIEUpdater__stage_update = mock.MagicMock()
o._ONIEUpdater__trigger_update = mock.MagicMock()
o._ONIEUpdater__is_update_staged = mock.MagicMock()
o._ONIEUpdater__unstage_update = mock.MagicMock()
o.update_firmware('')
o._ONIEUpdater__stage_update.assert_called_once()
o._ONIEUpdater__trigger_update.assert_called_once()
o._ONIEUpdater__is_update_staged.assert_not_called()
o._ONIEUpdater__unstage_update.assert_not_called()
o._ONIEUpdater__is_update_staged.return_value = False
o._ONIEUpdater__stage_update.side_effect = RuntimeError('')
with pytest.raises(RuntimeError):
o.update_firmware('')
o._ONIEUpdater__unstage_update.assert_not_called()
o._ONIEUpdater__is_update_staged.return_value = True
with pytest.raises(RuntimeError):
o.update_firmware('')
o._ONIEUpdater__unstage_update.assert_called_once()
@mock.patch('sonic_platform.component.os.path.lexists')
@mock.patch('sonic_platform.component.os.path.exists', mock.MagicMock(return_value=False))
@mock.patch('sonic_platform.component.os.mkdir', mock.MagicMock())
@mock.patch('sonic_platform.component.subprocess.check_call', mock.MagicMock())
@mock.patch('sonic_platform.component.os.symlink', mock.MagicMock())
@mock.patch('sonic_platform.component.check_output_pipe', mock.MagicMock())
def test_onie_updater_mount_onie_fs(self, mock_lexists):
o = ONIEUpdater()
o._ONIEUpdater__umount_onie_fs = mock.MagicMock()
mock_lexists.return_value = False
mp = o._ONIEUpdater__mount_onie_fs()
assert mp == '/mnt/onie-fs'
o._ONIEUpdater__umount_onie_fs.assert_not_called()
mock_lexists.return_value = True
o._ONIEUpdater__mount_onie_fs()
o._ONIEUpdater__umount_onie_fs.assert_called_once()
@mock.patch('sonic_platform.component.os.rmdir')
@mock.patch('sonic_platform.component.os.path.exists', mock.MagicMock(return_value=True))
@mock.patch('sonic_platform.component.subprocess.check_call')
@mock.patch('sonic_platform.component.os.path.ismount', mock.MagicMock(return_value=True))
@mock.patch('sonic_platform.component.os.unlink')
@mock.patch('sonic_platform.component.os.path.islink', mock.MagicMock(return_value=True))
def test_onie_updater_umount_onie_fs(self, mock_unlink, mock_check_call, mock_rmdir):
o = ONIEUpdater()
o._ONIEUpdater__umount_onie_fs()
mock_unlink.assert_called_once()
mock_check_call.assert_called_once()
mock_rmdir.assert_called_once()
@mock.patch('sonic_platform.component.subprocess.check_output')
@mock.patch('sonic_platform.component.subprocess.check_call')
@mock.patch('sonic_platform.component.copyfile', mock.MagicMock())
def test_onie_updater_stage(self, mock_check_call, mock_check_output):
o = ONIEUpdater()
o._ONIEUpdater__stage_update('')
mock_check_call.assert_called_once()
mock_check_call.reset_mock()
o._ONIEUpdater__unstage_update('a.rom')
mock_check_call.assert_called_once()
mock_check_call.reset_mock()
o._ONIEUpdater__trigger_update(True)
mock_check_call.assert_called_with(o.ONIE_FW_UPDATE_CMD_UPDATE, universal_newlines=True)
mock_check_call.reset_mock()
o._ONIEUpdater__trigger_update(False)
mock_check_call.assert_called_with(o.ONIE_FW_UPDATE_CMD_INSTALL, universal_newlines=True)
mock_check_output.return_value = 'invalid/'
assert not o._ONIEUpdater__is_update_staged('')
mock_check_output.return_value = '00-'
assert o._ONIEUpdater__is_update_staged('')
mock_check_call.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
o._ONIEUpdater__stage_update('')
with pytest.raises(RuntimeError):
o._ONIEUpdater__unstage_update('')
with pytest.raises(RuntimeError):
o._ONIEUpdater__trigger_update(True)
mock_check_output.side_effect = subprocess.CalledProcessError(1, None)
with pytest.raises(RuntimeError):
o._ONIEUpdater__is_update_staged('')
def test_read_generic_file(self):
with pytest.raises(RuntimeError):
Component._read_generic_file('invalid', 1)
content = Component._read_generic_file(os.path.abspath(__file__), 1)
assert content == '#'
def test_check_file_validity(self):
c = ComponentONIE()
assert not c._check_file_validity('invalid')
assert c._check_file_validity(os.path.abspath(__file__))
c.image_ext_name = '.py'
assert c._check_file_validity(os.path.abspath(__file__))
c.image_ext_name = '.txt'
assert not c._check_file_validity(os.path.abspath(__file__))

View File

@ -0,0 +1,56 @@
#
# Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# 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 os
import pytest
import subprocess
import sys
if sys.version_info.major == 3:
from unittest import mock
else:
import mock
test_path = os.path.dirname(os.path.abspath(__file__))
modules_path = os.path.dirname(test_path)
sys.path.insert(0, modules_path)
from sonic_platform.device_data import DeviceDataManager
class TestDeviceData:
@mock.patch('sonic_platform.device_data.utils.read_int_from_file', mock.MagicMock(return_value=1))
def test_is_fan_hotswapable(self):
assert DeviceDataManager.is_fan_hotswapable()
@mock.patch('sonic_platform.device_data.utils.read_int_from_file', mock.MagicMock(return_value=1))
def test_get_linecard_sfp_count(self):
assert DeviceDataManager.get_linecard_sfp_count(1) == 1
@mock.patch('sonic_platform.device_data.utils.read_int_from_file', mock.MagicMock(return_value=1))
def test_get_gearbox_count(self):
assert DeviceDataManager.get_gearbox_count('') == 1
@mock.patch('sonic_platform.device_data.DeviceDataManager.get_platform_name', mock.MagicMock(return_value='x86_64-mlnx_msn3420-r0'))
def test_get_linecard_max_port_count(self):
assert DeviceDataManager.get_linecard_max_port_count() == 0
@mock.patch('sonic_platform.device_data.DeviceDataManager.get_platform_name', mock.MagicMock(return_value='x86_64-nvidia_sn2201-r0'))
def test_get_bios_component(self):
assert DeviceDataManager.get_bios_component() is not None

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2021-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -179,3 +179,40 @@ class TestLed:
psu = FixedPsu(0)
physical_led = psu.led
self._verify_non_shared_led(physical_led, psu)
def test_get_actual_color(self):
led = Led()
assert led._get_actual_color('red') is None
led.supported_colors.add('orange')
assert led._get_actual_color('red') is 'orange'
@mock.patch('os.path.exists')
@mock.patch('time.sleep', mock.MagicMock())
def test_wait_files_ready(self, mock_exists):
mock_exists.side_effect = [True, True]
led = Led()
assert led._wait_files_ready(['a', 'b'])
mock_exists.side_effect = [False, False, True, True]
assert led._wait_files_ready(['a', 'b'])
mock_exists.side_effect = None
mock_exists.return_value = False
assert not led._wait_files_ready(['a', 'b'])
@mock.patch('sonic_platform.utils.write_file')
@mock.patch('sonic_platform.led.Led.get_led_path', mock.MagicMock())
@mock.patch('sonic_platform.led.Led._stop_blink', mock.MagicMock())
@mock.patch('sonic_platform.led.Led.get_capability', mock.MagicMock())
@mock.patch('sonic_platform.device_data.DeviceDataManager.is_simx_platform', mock.MagicMock(return_value=False))
def test_get_set_led_status(self, mock_write):
led = Led()
led._led_id = 'fan'
led.supported_colors.add('red')
led.supported_colors.add('green')
assert not led.set_status('black')
assert led.set_status(led.STATUS_LED_COLOR_OFF)
assert mock_write.call_count == 2
mock_write.side_effect = ValueError('')
assert not led.set_status(led.STATUS_LED_COLOR_OFF)
led.supported_colors.clear()
assert led.get_status() == led.STATUS_LED_COLOR_OFF

View File

@ -0,0 +1,67 @@
#
# Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# 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 os
import sys
if sys.version_info.major == 3:
from unittest import mock
else:
import mock
test_path = os.path.dirname(os.path.abspath(__file__))
modules_path = os.path.dirname(test_path)
sys.path.insert(0, modules_path)
from sonic_platform.pcie import Pcie
class TestPcie:
@mock.patch('sonic_platform.pcie.Pcie._create_device_id_to_bus_map', mock.MagicMock())
@mock.patch('sonic_platform.pcie.Pcie.load_config_file', mock.MagicMock())
def test_get_pcie_check(self):
p = Pcie('')
p._device_id_to_bus_map = {}
p.confInfo = [
{
'id': '1f0b',
'dev': '00',
'fn': '00'
}
]
info = p.get_pcie_check()
assert info[0]['result'] == 'Failed'
p.check_pcie_sysfs = mock.MagicMock(return_value=False)
p._device_id_to_bus_map = {'1f0b': '00'}
info = p.get_pcie_check()
assert info[0]['result'] == 'Failed'
p.check_pcie_sysfs = mock.MagicMock(return_value=True)
info = p.get_pcie_check()
assert info[0]['result'] == 'Passed'
@mock.patch('sonic_platform.pcie.os.listdir')
@mock.patch('sonic_platform.pcie.Pcie.load_config_file', mock.MagicMock())
def test_create_device_id_to_bus_map(self, mock_dir):
p = Pcie('')
assert not p._device_id_to_bus_map
mock_dir.return_value = ['0000:01:00.0']
mock_os_open = mock.mock_open(read_data='0x23')
with mock.patch('sonic_platform.pcie.open', mock_os_open):
p._create_device_id_to_bus_map()
assert p._device_id_to_bus_map == {'23':'01'}

View File

@ -37,8 +37,14 @@ class TestPsu:
assert psu.get_model() == 'N/A'
assert psu.get_serial() == 'N/A'
assert psu.get_revision() == 'N/A'
avail, msg = psu.get_power_available_status()
assert not avail
assert msg == 'absence of power'
utils.read_int_from_file = mock.MagicMock(return_value=1)
assert psu.get_powergood_status()
avail, msg = psu.get_power_available_status()
assert avail
assert msg == ''
utils.read_int_from_file = mock.MagicMock(return_value=0)
assert not psu.get_powergood_status()
assert psu.get_presence()
@ -69,6 +75,7 @@ class TestPsu:
psu.psu_temp_threshold: 50678,
psu.psu_voltage_in: 102345,
psu.psu_current_in: 676,
psu.psu_power_max: 1234567
}
def mock_read_int_from_file(file_path, **kwargs):
@ -77,8 +84,12 @@ class TestPsu:
utils.read_int_from_file = mock_read_int_from_file
utils.read_str_from_file = mock.MagicMock(return_value='min max')
assert psu.get_presence() is True
assert psu.get_maximum_supplied_power() == 1.234567
mock_sysfs_content[psu.psu_presence] = 0
assert psu.get_presence() is False
avail, msg = psu.get_power_available_status()
assert not avail
assert msg == 'absence of PSU'
assert psu.get_powergood_status() is True
mock_sysfs_content[psu.psu_oper_status] = 0
@ -91,6 +102,7 @@ class TestPsu:
assert psu.get_temperature_high_threshold() is None
assert psu.get_input_voltage() is None
assert psu.get_input_current() is None
assert psu.get_maximum_supplied_power() is None
mock_sysfs_content[psu.psu_oper_status] = 1
assert psu.get_voltage() == 10.234

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2021-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -37,6 +37,7 @@ class TestSfp:
@mock.patch('sonic_platform.device_data.DeviceDataManager.get_linecard_max_port_count')
def test_sfp_index(self, mock_max_port):
sfp = SFP(0)
assert sfp.is_replaceable()
assert sfp.sdk_index == 0
assert sfp.index == 1
@ -267,3 +268,25 @@ class TestSfp:
assert sfp.set_lpmode(True)
mock_write.assert_called_with('/sys/module/sx_core/asic0/module0/power_mode_policy', '2')
@mock.patch('sonic_platform.sfp.SFP.read_eeprom')
def test_get_xcvr_api(self, mock_read):
sfp = SFP(0)
api = sfp.get_xcvr_api()
assert api is None
mock_read.return_value = bytearray([0x18])
api = sfp.get_xcvr_api()
assert api is not None
def test_rj45_basic(self):
sfp = RJ45Port(0)
assert not sfp.get_lpmode()
assert not sfp.reset()
assert not sfp.set_lpmode(True)
assert not sfp.get_error_description()
assert not sfp.get_reset_status()
assert sfp.read_eeprom(0, 0) is None
assert sfp.get_transceiver_info()
assert sfp.get_transceiver_bulk_status()
assert sfp.get_transceiver_threshold_info()
sfp.reinit()

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2021-2022 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2021-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -42,7 +42,6 @@ class TestThermal:
@mock.patch('sonic_platform.device_data.DeviceDataManager.get_platform_name', mock.MagicMock(return_value='x86_64-mlnx_msn2700-r0'))
def test_chassis_thermal(self):
from sonic_platform.thermal import THERMAL_NAMING_RULE
os.path.exists = mock.MagicMock(return_value=True)
chassis = Chassis()
thermal_list = chassis.get_all_thermals()
assert thermal_list
@ -123,9 +122,9 @@ class TestThermal:
thermal_name = rule['name']
assert thermal_name in thermal_dict
@mock.patch('os.path.exists', mock.MagicMock(return_value=True))
def test_psu_thermal(self):
from sonic_platform.thermal import initialize_psu_thermal, THERMAL_NAMING_RULE
os.path.exists = mock.MagicMock(return_value=True)
presence_cb = mock.MagicMock(return_value=(True, ''))
thermal_list = initialize_psu_thermal(0, presence_cb)
assert len(thermal_list) == 1
@ -147,9 +146,9 @@ class TestThermal:
assert thermal.get_high_threshold() is None
assert thermal.get_high_critical_threshold() is None
@mock.patch('os.path.exists', mock.MagicMock(return_value=True))
def test_sfp_thermal(self):
from sonic_platform.thermal import initialize_sfp_thermal, THERMAL_NAMING_RULE
os.path.exists = mock.MagicMock(return_value=True)
thermal_list = initialize_sfp_thermal(0)
assert len(thermal_list) == 1
thermal = thermal_list[0]
@ -162,66 +161,65 @@ class TestThermal:
assert thermal.get_position_in_parent() == 1
assert thermal.is_replaceable() == False
def test_get_temperature(self):
@mock.patch('sonic_platform.utils.read_float_from_file')
def test_get_temperature(self, mock_read):
from sonic_platform.thermal import Thermal
from sonic_platform import utils
thermal = Thermal('test', 'temp_file', None, None, 1)
utils.read_float_from_file = mock.MagicMock(return_value=35727)
mock_read.return_value = 35727
assert thermal.get_temperature() == 35.727
utils.read_float_from_file = mock.MagicMock(return_value=0.0)
mock_read.return_value = 0.0
assert thermal.get_temperature() is None
utils.read_float_from_file = mock.MagicMock(return_value=None)
mock_read.return_value = None
assert thermal.get_temperature() is None
def test_get_high_threshold(self):
@mock.patch('sonic_platform.utils.read_float_from_file')
def test_get_high_threshold(self, mock_read):
from sonic_platform.thermal import Thermal
from sonic_platform import utils
thermal = Thermal('test', None, None, None, 1)
assert thermal.get_high_threshold() is None
thermal.high_threshold = 'high_th_file'
utils.read_float_from_file = mock.MagicMock(return_value=25833)
mock_read.return_value = 25833
assert thermal.get_temperature() == 25.833
utils.read_float_from_file = mock.MagicMock(return_value=0.0)
mock_read.return_value = 0.0
assert thermal.get_temperature() is None
utils.read_float_from_file = mock.MagicMock(return_value=None)
mock_read.return_value = None
assert thermal.get_temperature() is None
def test_get_high_critical_threshold(self):
@mock.patch('sonic_platform.utils.read_float_from_file')
def test_get_high_critical_threshold(self, mock_read):
from sonic_platform.thermal import Thermal
from sonic_platform import utils
thermal = Thermal('test', None, None, None, 1)
assert thermal.get_high_critical_threshold() is None
thermal.high_critical_threshold = 'high_th_file'
utils.read_float_from_file = mock.MagicMock(return_value=120839)
mock_read.return_value = 120839
assert thermal.get_high_critical_threshold() == 120.839
utils.read_float_from_file = mock.MagicMock(return_value=0.0)
mock_read.return_value = 0.0
assert thermal.get_high_critical_threshold() is None
utils.read_float_from_file = mock.MagicMock(return_value=None)
mock_read.return_value = None
assert thermal.get_high_critical_threshold() is None
def test_set_thermal_algorithm_status(self):
@mock.patch('glob.iglob', mock.MagicMock(return_value=['thermal_zone1', 'thermal_zone2']))
@mock.patch('sonic_platform.utils.write_file')
def test_set_thermal_algorithm_status(self, mock_write):
from sonic_platform.thermal import Thermal, THERMAL_ZONE_FOLDER_WILDCARD, THERMAL_ZONE_POLICY_FILE, THERMAL_ZONE_MODE_FILE
from sonic_platform import utils
glob.iglob = mock.MagicMock(return_value=['thermal_zone1', 'thermal_zone2'])
utils.write_file = mock.MagicMock()
assert Thermal.set_thermal_algorithm_status(True, False)
for folder in glob.iglob(THERMAL_ZONE_FOLDER_WILDCARD):
utils.write_file.assert_any_call(os.path.join(folder, THERMAL_ZONE_POLICY_FILE), 'step_wise')
utils.write_file.assert_any_call(os.path.join(folder, THERMAL_ZONE_MODE_FILE), 'enabled')
mock_write.assert_any_call(os.path.join(folder, THERMAL_ZONE_POLICY_FILE), 'step_wise')
mock_write.assert_any_call(os.path.join(folder, THERMAL_ZONE_MODE_FILE), 'enabled')
assert Thermal.set_thermal_algorithm_status(False, False)
for folder in glob.iglob(THERMAL_ZONE_FOLDER_WILDCARD):
utils.write_file.assert_any_call(os.path.join(folder, THERMAL_ZONE_POLICY_FILE), 'user_space')
utils.write_file.assert_any_call(os.path.join(folder, THERMAL_ZONE_MODE_FILE), 'disabled')
mock_write.assert_any_call(os.path.join(folder, THERMAL_ZONE_POLICY_FILE), 'user_space')
mock_write.assert_any_call(os.path.join(folder, THERMAL_ZONE_MODE_FILE), 'disabled')
assert not Thermal.set_thermal_algorithm_status(False, False)
@ -283,10 +281,10 @@ class TestThermal:
mock_file_content[os.path.join('thermal_zone2', THERMAL_ZONE_TEMP_FILE)] = 81000
assert Thermal.get_min_allowed_cooling_level_by_thermal_zone() is None
@mock.patch('glob.iglob', mock.MagicMock(return_value=['thermal_zone1', 'thermal_zone2']))
def test_check_module_temperature_trustable(self):
from sonic_platform.thermal import Thermal
from sonic_platform import utils
glob.iglob = mock.MagicMock(return_value=['thermal_zone1', 'thermal_zone2'])
utils.read_int_from_file = mock.MagicMock(return_value=1)
assert Thermal.check_module_temperature_trustable() == 'untrust'

View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2021 NVIDIA CORPORATION & AFFILIATES.
# Copyright (c) 2021-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -122,12 +122,45 @@ class TestUtils:
def test_run_command(self):
output = utils.run_command(['ls'])
assert output
assert utils.run_command(['not_a_command']) is None
def test_run_command_exception(self):
output = utils.run_command(['ls'])
assert output
@mock.patch('sonic_platform.utils.load_json_file')
@mock.patch('os.path.exists')
@mock.patch('sonic_py_common.device_info.get_path_to_port_config_file', mock.MagicMock(return_value=''))
@mock.patch('sonic_py_common.device_info.get_path_to_hwsku_dir', mock.MagicMock(return_value='/tmp'))
def test_extract_RJ45_ports_index(self):
def test_extract_RJ45_ports_index(self, mock_exists, mock_load_json):
mock_exists.return_value = False
rj45_list = utils.extract_RJ45_ports_index()
assert rj45_list is None
mock_exists.return_value = True
platform_json = {
'interfaces': {
"Ethernet0": {
"index": "1",
"lanes": "0",
"breakout_modes": {
"1x1000[100,10]": ["etp1"]
}
}
}
}
hwsku_json = {
'interfaces': {
"Ethernet0": {
"default_brkout_mode": "1x1000[100,10]",
"port_type": "RJ45"
}
}
}
mock_load_json.side_effect = [platform_json, hwsku_json]
assert utils.extract_RJ45_ports_index() == [0]
def test_wait_until(self):
values = []
assert utils.wait_until(lambda: len(values) == 0, timeout=1)
@ -141,3 +174,20 @@ class TestUtils:
t.start()
assert utils.wait_until(lambda: len(values) > 0, timeout=5)
t.join()
def test_load_json_file(self):
assert utils.load_json_file('some_file') is None
mock_os_open = mock.mock_open(read_data='')
with mock.patch('sonic_platform.utils.open', mock_os_open):
assert utils.load_json_file('some_file') is None
mock_os_open = mock.mock_open(read_data='{"a": "b"}')
with mock.patch('sonic_platform.utils.open', mock_os_open):
data = utils.load_json_file('some_file')
assert data['a'] == 'b'
def test_read_key_value_file(self):
mock_os_open = mock.mock_open(read_data='a:b')
with mock.patch('sonic_platform.utils.open', mock_os_open):
assert utils.read_key_value_file('some_file') == {'a':'b'}

View File

@ -0,0 +1,129 @@
#
# Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# 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 os
import pytest
import sys
if sys.version_info.major == 3:
from unittest import mock
else:
import mock
test_path = os.path.dirname(os.path.abspath(__file__))
modules_path = os.path.dirname(test_path)
sys.path.insert(0, modules_path)
from sonic_platform.chassis import Chassis
from sonic_platform.watchdog import get_watchdog, \
WatchdogType2, \
WatchdogType1, \
is_mlnx_wd_main, \
is_wd_type2
class TestWatchdog:
@mock.patch('sonic_platform.watchdog.is_mlnx_wd_main')
@mock.patch('sonic_platform.watchdog.os.listdir')
def test_get_watchdog_no_device(self, mock_listdir, mock_is_main):
mock_listdir.return_value = []
assert get_watchdog() is None
mock_listdir.return_value = ['invalid']
mock_is_main.return_value = True
assert get_watchdog() is None
mock_listdir.return_value = ['watchdog1']
mock_is_main.return_value = False
assert get_watchdog() is None
@mock.patch('sonic_platform.watchdog.is_mlnx_wd_main')
@mock.patch('sonic_platform.watchdog.is_wd_type2')
@mock.patch('sonic_platform.watchdog.os.listdir', mock.MagicMock(return_value=['watchdog1', 'watchdog2']))
@mock.patch('sonic_platform.watchdog.WatchdogImplBase.open_handle', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.fcntl.ioctl', mock.MagicMock())
@pytest.mark.parametrize('test_para',
[(True, WatchdogType2), (False, WatchdogType1)])
def test_get_watchdog(self, mock_is_type2, mock_is_main, test_para):
mock_is_main.side_effect = lambda dev: dev == 'watchdog2'
mock_is_type2.return_value = test_para[0]
chassis = Chassis()
watchdog = chassis.get_watchdog()
assert isinstance(watchdog, test_para[1])
assert watchdog.watchdog_path == '/dev/watchdog2'
def test_is_mlnx_wd_main(self):
mock_os_open = mock.mock_open(read_data='mlx-wdt-main')
with mock.patch('sonic_platform.watchdog.open', mock_os_open):
assert is_mlnx_wd_main('')
mock_os_open = mock.mock_open(read_data='invalid')
with mock.patch('sonic_platform.watchdog.open', mock_os_open):
assert not is_mlnx_wd_main('')
mock_os_open.side_effect = IOError
with mock.patch('sonic_platform.watchdog.open', mock_os_open):
assert not is_mlnx_wd_main('')
@mock.patch('sonic_platform.watchdog.os.path.exists')
@pytest.mark.parametrize('test_para',
[True, False])
def test_is_wd_type2(self, mock_exists, test_para):
mock_exists.return_value = test_para
assert is_wd_type2('') is test_para
@mock.patch('sonic_platform.watchdog.WatchdogImplBase.open_handle', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.fcntl.ioctl', mock.MagicMock())
def test_arm_disarm_watchdog2(self):
watchdog = WatchdogType2('watchdog2')
assert watchdog.arm(-1) == -1
assert not watchdog.is_armed()
watchdog.arm(10)
assert watchdog.is_armed()
watchdog.arm(5)
assert watchdog.is_armed()
watchdog.disarm()
assert not watchdog.is_armed()
@mock.patch('sonic_platform.watchdog.WatchdogImplBase.open_handle', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.fcntl.ioctl', mock.MagicMock())
def test_arm_disarm_watchdog1(self):
watchdog = WatchdogType1('watchdog1')
assert watchdog.arm(-1) == -1
assert not watchdog.is_armed()
watchdog.arm(10)
assert watchdog.is_armed()
watchdog.arm(5)
assert watchdog.is_armed()
watchdog.disarm()
assert not watchdog.is_armed()
@mock.patch('sonic_platform.watchdog.WatchdogImplBase.open_handle', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.fcntl.ioctl', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.WatchdogImplBase._gettimeleft', mock.MagicMock(return_value=10))
def test_get_remaining_time_watchdog2(self):
watchdog = WatchdogType2('watchdog2')
assert watchdog.get_remaining_time() == -1
watchdog.arm(10)
assert watchdog.get_remaining_time() == 10
@mock.patch('sonic_platform.watchdog.WatchdogImplBase.open_handle', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.fcntl.ioctl', mock.MagicMock())
@mock.patch('sonic_platform.watchdog.WatchdogImplBase._gettimeleft', mock.MagicMock(return_value=10))
def test_get_remaining_time_watchdog1(self):
watchdog = WatchdogType1('watchdog2')
assert watchdog.get_remaining_time() == -1
watchdog.arm(10)
assert watchdog.get_remaining_time() > 0