diff --git a/src/hardware.py b/src/hardware.py index a1a19e23..1890520c 100644 --- a/src/hardware.py +++ b/src/hardware.py @@ -12,7 +12,7 @@ logger = logging.getLogger(__name__) -SUPPORTED_STORAGES = { +LSHW_SUPPORTED_STORAGES = { HWTool.SAS2IRCU: [ # Broadcom "SAS2004", @@ -33,6 +33,17 @@ ], } +HWINFO_SUPPORTED_STORAGES = { + HWTool.SSACLI: [ + [ + "Hardware Class: storage", + 'Vendor: pci 0x9005 "Adaptec"', + 'Device: pci 0x028f "Smart Storage PQI 12G SAS/PCIe 3"', + 'SubDevice: pci 0x1100 "Smart Array P816i-a SR Gen10"', + ] + ] +} + def lshw(class_filter: t.Optional[str] = None) -> t.Any: """Return lshw output as dict.""" @@ -66,3 +77,28 @@ def get_bmc_address() -> t.Optional[str]: except subprocess.CalledProcessError: logger.debug("IPMI is not available") return None + + +def hwinfo(*args: str) -> t.Dict[str, str]: + """Run hwinfo command and return output as dictionary. + + Args: + args: Probe for a particular hardware class. + Returns: + hw_info: hardware information dictionary + """ + apt.add_package("hwinfo", update_cache=False) + hw_classes = list(args) + for idx, hw_item in enumerate(args): + hw_classes[idx] = "--" + hw_item + hw_info_cmd = ["hwinfo"] + hw_classes + + output = subprocess.check_output(hw_info_cmd, text=True) + if "start debug info" in output.splitlines()[0]: + output = output.split("=========== end debug info ============")[1] + + hardwares: t.Dict[str, str] = {} + for item in output.split("\n\n"): + key = item.splitlines()[0].strip() + hardwares[key] = item + return hardwares diff --git a/src/hw_tools.py b/src/hw_tools.py index 7f0d1991..b5539af5 100644 --- a/src/hw_tools.py +++ b/src/hw_tools.py @@ -11,7 +11,7 @@ from abc import ABCMeta, abstractmethod from functools import lru_cache from pathlib import Path -from typing import Dict, List, Tuple +from typing import Dict, List, Set, Tuple from charms.operator_libs_linux.v0 import apt from ops.model import ModelError, Resources @@ -42,7 +42,13 @@ StorageVendor, SystemVendor, ) -from hardware import SUPPORTED_STORAGES, get_bmc_address, lshw +from hardware import ( + HWINFO_SUPPORTED_STORAGES, + LSHW_SUPPORTED_STORAGES, + get_bmc_address, + hwinfo, + lshw, +) from keys import HP_KEYS logger = logging.getLogger(__name__) @@ -354,18 +360,28 @@ def check(self) -> bool: return True -# Using cache here to avoid repeat call. -# The lru_cache should be clean everytime the hook been triggered. -@lru_cache -def raid_hw_verifier() -> List[HWTool]: - """Verify if the HWTool support RAID card exists on machine.""" - hw_info = lshw() - system_vendor = hw_info.get("vendor") - storage_info = lshw(class_filter="storage") +def _raid_hw_verifier_hwinfo() -> Set[HWTool]: + """Verify if a supported RAID card exists on the machine using the hwinfo command.""" + hwinfo_output = hwinfo("storage") + + tools = set() + for _, hwinfo_content in hwinfo_output.items(): + # ssacli + for support_storage in HWINFO_SUPPORTED_STORAGES[HWTool.SSACLI]: + if all(item in hwinfo_content for item in support_storage): + tools.add(HWTool.SSACLI) + return tools + + +def _raid_hw_verifier_lshw() -> Set[HWTool]: + """Verify if a supported RAID card exists on the machine using the lshw command.""" + lshw_output = lshw() + system_vendor = lshw_output.get("vendor") + lshw_storage = lshw(class_filter="storage") tools = set() - for info in storage_info: + for info in lshw_storage: _id = info.get("id") product = info.get("product") vendor = info.get("vendor") @@ -375,7 +391,7 @@ def raid_hw_verifier() -> List[HWTool]: if ( any( _product - for _product in SUPPORTED_STORAGES[HWTool.SAS3IRCU] + for _product in LSHW_SUPPORTED_STORAGES[HWTool.SAS3IRCU] if _product in product ) and vendor == StorageVendor.BROADCOM @@ -385,7 +401,7 @@ def raid_hw_verifier() -> List[HWTool]: if ( any( _product - for _product in SUPPORTED_STORAGES[HWTool.SAS2IRCU] + for _product in LSHW_SUPPORTED_STORAGES[HWTool.SAS2IRCU] if _product in product ) and vendor == StorageVendor.BROADCOM @@ -395,7 +411,9 @@ def raid_hw_verifier() -> List[HWTool]: if _id == "raid": # ssacli if system_vendor == SystemVendor.HP and any( - _product for _product in SUPPORTED_STORAGES[HWTool.SSACLI] if _product in product + _product + for _product in LSHW_SUPPORTED_STORAGES[HWTool.SSACLI] + if _product in product ): tools.add(HWTool.SSACLI) # perccli @@ -404,7 +422,17 @@ def raid_hw_verifier() -> List[HWTool]: # storcli elif driver == "megaraid_sas" and vendor == StorageVendor.BROADCOM: tools.add(HWTool.STORCLI) - return list(tools) + return tools + + +# Using cache here to avoid repeat call. +# The lru_cache should be clean everytime the hook been triggered. +@lru_cache +def raid_hw_verifier() -> List[HWTool]: + """Verify if the HWTool support RAID card exists on machine.""" + lshw_tools = _raid_hw_verifier_lshw() + hwinfo_tools = _raid_hw_verifier_hwinfo() + return list(lshw_tools | hwinfo_tools) # Using cache here to avoid repeat call. diff --git a/tests/unit/test_hardware.py b/tests/unit/test_hardware.py index c2de5e4d..127230b9 100644 --- a/tests/unit/test_hardware.py +++ b/tests/unit/test_hardware.py @@ -2,12 +2,98 @@ import unittest from unittest import mock -from hardware import get_bmc_address, lshw +import pytest + +from hardware import get_bmc_address, hwinfo, lshw + + +class TestHwinfo: + @pytest.mark.parametrize( + "hw_classes,expect_cmd,hwinfo_output,expect", + [ + ( + [], + ["hwinfo"], + ( + "" + "============ start debug info ============" + "random-string" + "random-string" + "random-string" + "random-string" + "=========== end debug info ============" + "10: key-a\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-a\n" + " Parent ID: parent-id-a\n" + "\n" + "11: key-b\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-b\n" + " Parent ID: parent-id-b\n" + ), + { + "10: key-a": ( + "10: key-a\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-a\n" + " Parent ID: parent-id-a" + ), + "11: key-b": ( + "11: key-b\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-b\n" + " Parent ID: parent-id-b\n" + ), + }, + ), + ( + ["storage"], + ["hwinfo", "--storage"], + ( + "" + "10: key-a\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-a\n" + " Parent ID: parent-id-a\n" + "\n" + "11: key-b\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-b\n" + " Parent ID: parent-id-b\n" + ), + { + "10: key-a": ( + "10: key-a\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-a\n" + " Parent ID: parent-id-a" + ), + "11: key-b": ( + "11: key-b\n" + " [Created at pci.386]\n" + " Unique ID: unique-id-b\n" + " Parent ID: parent-id-b\n" + ), + }, + ), + ], + ) + @mock.patch("hardware.apt") + @mock.patch("hardware.subprocess.check_output") + def test_hwinfo_output( + self, mock_subprocess, mock_apt, hw_classes, expect_cmd, hwinfo_output, expect + ): + mock_subprocess.return_value = hwinfo_output + output = hwinfo(*hw_classes) + mock_subprocess.assert_called_with(expect_cmd, text=True) + assert output == expect class TestLshw(unittest.TestCase): + @mock.patch("hardware.apt") @mock.patch("hardware.subprocess.check_output") - def test_lshw_list_output(self, mock_subprocess): + def test_lshw_list_output(self, mock_subprocess, mock_apt): mock_subprocess.return_value = """[{"expected_output": 1}]""" for class_filter in [None, "storage"]: output = lshw(class_filter) diff --git a/tests/unit/test_hw_tools.py b/tests/unit/test_hw_tools.py index ddf23bc2..667ecb2c 100644 --- a/tests/unit/test_hw_tools.py +++ b/tests/unit/test_hw_tools.py @@ -36,6 +36,8 @@ StorCLIStrategy, StrategyABC, TPRStrategyABC, + _raid_hw_verifier_hwinfo, + _raid_hw_verifier_lshw, bmc_hw_verifier, check_deb_pkg_installed, copy_to_snap_common_bin, @@ -744,6 +746,14 @@ def test_get_hw_tool_white_list(mock_raid_verifier, mock_bmc_hw_verifier): assert output == [4, 5, 6, 1, 2, 3] +@mock.patch("hw_tools._raid_hw_verifier_hwinfo", return_value=set([4, 5, 6])) +@mock.patch("hw_tools._raid_hw_verifier_lshw", return_value=set([1, 2, 3, 4])) +def test_raid_hw_verifier(mock_hw_verifier_lshw, mock_hw_verifier_hwinfo): + raid_hw_verifier.cache_clear() + output = raid_hw_verifier() + assert output == [1, 2, 3, 4, 5, 6] + + @pytest.mark.parametrize( "lshw_output, lshw_storage_output, expect", [ @@ -806,10 +816,47 @@ def test_get_hw_tool_white_list(mock_raid_verifier, mock_bmc_hw_verifier): ], ) @mock.patch("hw_tools.lshw") -def test_raid_hw_verifier(mock_lshw, lshw_output, lshw_storage_output, expect): +def test_raid_hw_verifier_lshw(mock_lshw, lshw_output, lshw_storage_output, expect): mock_lshw.side_effect = [lshw_output, lshw_storage_output] - raid_hw_verifier.cache_clear() - output = raid_hw_verifier() + output = _raid_hw_verifier_lshw() + case = unittest.TestCase() + case.assertCountEqual(output, expect) + + +@pytest.mark.parametrize( + "hwinfo_output, expect", + [ + ({}, []), + ( + { + "random-key-a": """ + [Created at pci.386] + Hardware Class: storage + Vendor: pci 0x9005 "Adaptec" + Device: pci 0x028f "Smart Storage PQI 12G SAS/PCIe 3" + SubDevice: pci 0x1100 "Smart Array P816i-a SR Gen10" + """ + }, + [HWTool.SSACLI], + ), + ( + { + "random-key-a": """ + [Created at pci.386] + Hardware Class: not-valid-class + Vendor: pci 0x9005 "Adaptec" + Device: pci 0x028f "Smart Storage PQI 12G SAS/PCIe 3" + SubDevice: pci 0x1100 "Smart Array P816i-a SR Gen10" + """ + }, + [], + ), + ], +) +@mock.patch("hw_tools.hwinfo") +def test_raid_hw_verifier_hwinfo(mock_hwinfo, hwinfo_output, expect): + mock_hwinfo.return_value = hwinfo_output + output = _raid_hw_verifier_hwinfo() case = unittest.TestCase() case.assertCountEqual(output, expect)