Skip to content

Commit

Permalink
Merge pull request #103 from canonical/add-fw-scripts
Browse files Browse the repository at this point in the history 
Refinements for fw updates
  • Loading branch information
@nancyc12
nancyc12 authored Sep 27, 2023
2 parents db7ee05 + 8b9a29c commit edea55d
Show file tree
Hide file tree
Showing 5 changed files with 30 additions and 25 deletions.
0 extras/README → extras/README.md
View file
File renamed without changes.
31 changes: 17 additions & 14 deletions extras/devices/LVFS/LVFS.py
100755 → 100644
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,7 @@ class LVFSDevice(AbstractDevice):

fw_update_type = "LVFS"
vendor = ["HP", "Dell Inc.", "LENOVO"]
reboot_timeout = 600

def __init__(self, ipaddr: str, user: str, password: str):
super().__init__(ipaddr, user, password)
reboot_timeout = 900

def run_cmd(
self, cmd: str, raise_stderr: bool = True, timeout: int = 30
Expand All @@ -36,13 +33,19 @@ def run_cmd(
ssh_cmd = f'ssh -t {SSH_OPTS} {self.user}@{self.ipaddr} "{cmd}"'
else:
ssh_cmd = f'sshpass -p {self.password} ssh -t {SSH_OPTS} {self.user}@{self.ipaddr} "{cmd}"'
logger.debug("Run command: %s", ssh_cmd)
r = subprocess.run(
ssh_cmd,
shell=True,
capture_output=True,
timeout=timeout,
)
logger.debug("Run command on DUT: %s", ssh_cmd)
try:
r = subprocess.run(
ssh_cmd,
shell=True,
capture_output=True,
timeout=timeout,
)
except subprocess.TimeoutExpired as e:
if raise_stderr:
raise e
else:
return 124, f"command timeout after {timeout}s", ""
rc, stdout, stderr = (
r.returncode,
r.stdout.decode().strip(),
Expand All @@ -67,7 +70,7 @@ def get_fw_info(self):
self._parse_fwupd_raw(stdout)

def _install_fwupd(self):
self.run_cmd("sudo apt update", timeout=120)
self.run_cmd("sudo apt update", raise_stderr=False, timeout=120)
self.run_cmd("sudo apt install -y fwupd")
rc, stdout, stderr = self.run_cmd("sudo fwupdmgr --version")
logger.debug(stdout)
Expand Down Expand Up @@ -290,7 +293,7 @@ def check_connectable(self, timeout: int):
shell=True,
universal_newlines=True,
).strip()
if status != "0":
if status != "0" and status != "124":
err_msg = f"Failed to SSH to {self.ipaddr} after {timeout}s"
logger.error(err_msg)
raise RuntimeError(err_msg)
Expand All @@ -313,4 +316,4 @@ class LenovoNB(LVFSDevice):
"""

fw_update_type = "LVFS-ext"
vendor = "LENOVO"
vendor = ["LENOVO"]
12 changes: 6 additions & 6 deletions extras/devices/LVFS/tests/test_LVFS.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ def test_upgradable(self):
Test if upgrade function returns True. And test if downgrade function
returns False.
"""
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")
device._parse_fwupd_raw(fwupd_data.GET_DEVICES_RESPONSE_DATA)
Expand All @@ -36,7 +36,7 @@ def test_downgradable(self):
Test if upgrade function returns False. And test if downgrade function
returns True.
"""
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")

Expand All @@ -53,7 +53,7 @@ def test_downgradable(self):
def test_check_results_failed_state(self):
"""Validate UpdateState check in check_results"""
global device_results
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")
device._parse_fwupd_raw(fwupd_data.GET_DEVICES_RESPONSE_DATA)
Expand All @@ -65,7 +65,7 @@ def test_check_results_failed_state(self):
def test_check_results_mismatched_version(self):
"""Validate version check in check_results"""
global device_results
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")
device._parse_fwupd_raw(fwupd_data.GET_DEVICES_RESPONSE_DATA)
Expand All @@ -75,7 +75,7 @@ def test_check_results_mismatched_version(self):
def test_check_results_good(self):
"""Test if check_results works with a valid case"""
global device_results
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")
device._parse_fwupd_raw(fwupd_data.GET_DEVICES_RESPONSE_DATA)
Expand All @@ -91,7 +91,7 @@ def test_check_results_error(self):
get-results```
"""
global device_results
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd
device = LVFSDevice("", "", "")
device._parse_fwupd_raw(fwupd_data.GET_DEVICES_RESPONSE_DATA)
Expand Down
8 changes: 4 additions & 4 deletions extras/tests/test_upgrade_fw.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,15 +69,15 @@ def mock_run_cmd_nossh(*args, **kwargs):

def test_detect_device_supported(self):
"""Test if detects_device returns a correct device class"""
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd_supported
device = upgrade_fw.detect_device("", "", "")
self.assertTrue(isinstance(device, LVFSDevice))

def test_detect_device_unsupported(self):
"""Test if detects_device exits while given a unsupported device"""
with pytest.raises(SystemExit) as pytest_wrapped_e:
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd_unsupported
upgrade_fw.detect_device("", "", "")
self.assertEqual(pytest_wrapped_e.type, SystemExit)
Expand All @@ -89,7 +89,7 @@ def test_detect_device_unsupported(self):
def test_detect_device_fail(self):
"""Test if detects_device exits while given a device without dmi data"""
with pytest.raises(SystemExit) as pytest_wrapped_e:
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd_fail
upgrade_fw.detect_device("", "", "")
self.assertEqual(pytest_wrapped_e.type, SystemExit)
Expand All @@ -101,7 +101,7 @@ def test_detect_device_fail(self):
def test_detect_device_nossh(self):
"""Test if detects_device exits while given an unreachable device"""
with pytest.raises(SystemExit) as pytest_wrapped_e:
with patch("devices.LVFS.LVFS.LVFSDevice.run_cmd") as mock_path:
with patch("devices.LVFSDevice.run_cmd") as mock_path:
mock_path.side_effect = self.mock_run_cmd_nossh
upgrade_fw.detect_device("", "", "")
self.assertEqual(pytest_wrapped_e.type, SystemExit)
Expand Down
4 changes: 3 additions & 1 deletion extras/upgrade_fw.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,9 @@ def main():
target_device = detect_device(
args.device_ip,
target_device_username,
options=vars(args),
bmc_ip=args.bmc_ip,
bmc_user=args.bmc_user,
bmc_password=args.bmc_password,
)
else:
target_device = detect_device(
Expand Down

0 comments on commit edea55d

Please sign in to comment.