From ca03fd5e08eca62164418a23ea9bb0c7d2d4cd80 Mon Sep 17 00:00:00 2001 From: Marek Mahut Date: Sun, 10 May 2026 17:13:20 +0200 Subject: [PATCH] feat: add multi-vendor detection --- pytest.ini | 5 ++- src/badfish/main.py | 18 ++++++--- tests/config.py | 12 ++++++ tests/test_reset_bmc.py | 16 +++++++- tests/test_reset_idrac.py | 2 +- tests/test_vendor_detection.py | 72 ++++++++++++++++++++++++++++++++++ tests/test_virtual_media.py | 21 ++++++++++ 7 files changed, 137 insertions(+), 9 deletions(-) create mode 100644 tests/test_vendor_detection.py diff --git a/pytest.ini b/pytest.ini index 3d3bcb98..17cdf948 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,2 +1,3 @@ -[tool.pytest.ini_options] -pythonpath = [".", "src"] +[pytest] +pythonpath = . src +asyncio_mode = auto diff --git a/src/badfish/main.py b/src/badfish/main.py index 6968f33e..c89fbd05 100755 --- a/src/badfish/main.py +++ b/src/badfish/main.py @@ -529,7 +529,15 @@ async def find_managers_resource(self): raw = await response.text("utf-8", "ignore") data = json.loads(raw.strip()) - self.vendor = "Dell" if data.get("Oem") and "Dell" in data["Oem"] else "Supermicro" + oem = data.get("Oem") or {} + if "Dell" in oem: + self.vendor = "Dell" + elif "Supermicro" in oem: + self.vendor = "Supermicro" + elif "Hpe" in oem: + self.vendor = "HPE" + else: + self.vendor = "Unknown" if "Managers" not in data: raise BadfishException("Managers resource not found") @@ -1117,7 +1125,7 @@ async def reboot_server(self, graceful=True): async def reset_idrac(self, wait=False): if self.vendor != "Dell": - self.logger.warning("Vendor isn't a Dell, if you are trying this on a Supermicro, use --bmc-reset instead.") + self.logger.warning("Vendor isn't a Dell, if you are trying this on a Supermicro or HPE, use --bmc-reset instead.") return False self.logger.debug("Running reset iDRAC.") _reset_types = await self.get_reset_types(manager=True) @@ -1154,8 +1162,8 @@ async def reset_idrac(self, wait=False): return True async def reset_bmc(self): - if self.vendor != "Supermicro": - self.logger.warning("Vendor isn't a Supermicro, if you are trying this on a Dell, use --racreset instead.") + if self.vendor not in ("Supermicro", "HPE"): + self.logger.warning("Vendor isn't a Supermicro or HPE, if you are trying this on a Dell, use --racreset instead.") return False self.logger.debug("Running reset BMC.") _reset_types = await self.get_reset_types(manager=True, bmc=True) @@ -1630,7 +1638,7 @@ async def boot_to_virtual_media(self): _uri = "%s%s" % (self.host_uri, self.system_resource) _headers = {"Content-Type": "application/json"} - if self.vendor == "Supermicro": + if self.vendor in ("Supermicro", "HPE"): _payload = {"Boot": {"BootSourceOverrideEnabled": "Once"}} _response = await self.get_request(_uri) diff --git a/tests/config.py b/tests/config.py index f9c72e68..fc0a2868 100644 --- a/tests/config.py +++ b/tests/config.py @@ -271,6 +271,18 @@ def render_device_dict(index, device): ROOT_RESP_SUPERMICRO, MAN_RESP, ] +ROOT_RESP_HPE = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},"Systems":{"@odata.id":"/redfish/v1/Systems"}, ' + '"RedfishVersion": "1.0.2","Oem":{"Hpe":{"Manager":[{}]}}}' +) +MAN_RESP_HPE = '{"Members":[{"@odata.id":"/redfish/v1/Managers/1"}]}' +INIT_RESP_HPE = [ + ROOT_RESP_HPE, + ROOT_RESP_HPE, + SYS_RESP, + ROOT_RESP_HPE, + MAN_RESP_HPE, +] RESPONSE_INIT_CREDENTIALS_UNAUTHORIZED = ( f"- ERROR - Failed to authenticate. Verify your credentials for {MOCK_HOST}\n" diff --git a/tests/test_reset_bmc.py b/tests/test_reset_bmc.py index 17c9b2c9..faeff0a7 100644 --- a/tests/test_reset_bmc.py +++ b/tests/test_reset_bmc.py @@ -3,6 +3,7 @@ from tests.config import ( BOOT_SEQ_RESPONSE_DIRECTOR, INIT_RESP, + INIT_RESP_HPE, INIT_RESP_SUPERMICRO, RESET_TYPE_RESP, RESET_TYPE_RESP_NO_ALLOWABLE_VALUES, @@ -66,4 +67,17 @@ def test_reset_bmc_wrong_vendor(self, mock_get, mock_post, mock_delete): self.boot_seq = BOOT_SEQ_RESPONSE_DIRECTOR self.args = [self.option_arg] _, err = self.badfish_call() - assert err == RESPONSE_RESET_WRONG_VENDOR % ("Supermicro", "Dell", "--racreset") + assert err == RESPONSE_RESET_WRONG_VENDOR % ("Supermicro or HPE", "Dell", "--racreset") + + @patch("aiohttp.ClientSession.delete") + @patch("aiohttp.ClientSession.post") + @patch("aiohttp.ClientSession.get") + def test_reset_bmc_hpe(self, mock_get, mock_post, mock_delete): + responses = INIT_RESP_HPE + [RESET_TYPE_RESP] + self.set_mock_response(mock_get, 200, responses) + self.set_mock_response(mock_post, [200, 200], "OK", True) + self.set_mock_response(mock_delete, 200, "OK") + self.boot_seq = BOOT_SEQ_RESPONSE_DIRECTOR + self.args = [self.option_arg] + _, err = self.badfish_call() + assert err == RESPONSE_RESET % ("200", "BMC", "BMC") diff --git a/tests/test_reset_idrac.py b/tests/test_reset_idrac.py index edc28c4b..927094aa 100644 --- a/tests/test_reset_idrac.py +++ b/tests/test_reset_idrac.py @@ -53,7 +53,7 @@ def test_reset_idrac_wrong_vendor(self, mock_get, mock_post, mock_delete): self.boot_seq = BOOT_SEQ_RESPONSE_DIRECTOR self.args = [self.option_arg] _, err = self.badfish_call() - assert err == RESPONSE_RESET_WRONG_VENDOR % ("Dell", "Supermicro", "--bmc-reset") + assert err == RESPONSE_RESET_WRONG_VENDOR % ("Dell", "Supermicro or HPE", "--bmc-reset") @patch("badfish.main.Badfish.wait_for_idrac_ready", new_callable=AsyncMock) @patch("aiohttp.ClientSession.delete") diff --git a/tests/test_vendor_detection.py b/tests/test_vendor_detection.py new file mode 100644 index 00000000..336c2f03 --- /dev/null +++ b/tests/test_vendor_detection.py @@ -0,0 +1,72 @@ +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from badfish.main import Badfish + + +ROOT_RESP_UNKNOWN_OEM = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},' + '"RedfishVersion":"1.0.2","Oem":{"Lenovo":{}}}' +) +MAN_RESP = '{"Members":[{"@odata.id":"/redfish/v1/Managers/1"}]}' + + +def _make_resp(body: str) -> MagicMock: + resp = MagicMock() + resp.status = 200 + resp.text = AsyncMock(return_value=body) + return resp + + +async def _init_vendor(root_body: str) -> str: + logger = MagicMock(spec=logging.Logger) + bf = Badfish("test_host", "user", "pass", logger, 1) + bf.http_client = MagicMock() + bf.http_client.get_request = AsyncMock( + side_effect=[_make_resp(root_body), _make_resp(MAN_RESP)] + ) + await bf.find_managers_resource() + return bf.vendor + + +@pytest.mark.asyncio +async def test_vendor_dell(): + root = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},' + '"RedfishVersion":"1.0.2","Oem":{"Dell":{"ServiceTag":"T35T7A6"}}}' + ) + assert await _init_vendor(root) == "Dell" + + +@pytest.mark.asyncio +async def test_vendor_supermicro(): + root = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},' + '"RedfishVersion":"1.0.2","Oem":{"Supermicro":{}}}' + ) + assert await _init_vendor(root) == "Supermicro" + + +@pytest.mark.asyncio +async def test_vendor_hpe(): + root = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},' + '"RedfishVersion":"1.0.2","Oem":{"Hpe":{"Manager":[{}]}}}' + ) + assert await _init_vendor(root) == "HPE" + + +@pytest.mark.asyncio +async def test_vendor_unknown_oem(): + assert await _init_vendor(ROOT_RESP_UNKNOWN_OEM) == "Unknown" + + +@pytest.mark.asyncio +async def test_vendor_no_oem(): + root = ( + '{"Managers":{"@odata.id":"/redfish/v1/Managers"},' + '"RedfishVersion":"1.0.2"}' + ) + assert await _init_vendor(root) == "Unknown" diff --git a/tests/test_virtual_media.py b/tests/test_virtual_media.py index 41d57139..f89726ce 100644 --- a/tests/test_virtual_media.py +++ b/tests/test_virtual_media.py @@ -9,6 +9,7 @@ BOOT_SOURCE_OVERRIDE_TARGET_CD, BOOT_SOURCE_OVERRIDE_TARGET_USBCD, INIT_RESP, + INIT_RESP_HPE, INIT_RESP_SUPERMICRO, JOB_OK_RESP, RESPONSE_BOOT_TO, @@ -431,6 +432,26 @@ def test_fail_sm(self, mock_get, mock_post, mock_delete, mock_patch): _, err = self.badfish_call() assert err == VMEDIA_BOOT_TO_SM_FAIL + @patch("aiohttp.ClientSession.patch") + @patch("aiohttp.ClientSession.delete") + @patch("aiohttp.ClientSession.post") + @patch("aiohttp.ClientSession.get") + def test_good_hpe(self, mock_get, mock_post, mock_delete, mock_patch): + responses_get = [ + VMEDIA_GET_VM_CONFIG_RESP_DELL, + VMEDIA_MEMBER_RM_DISK_RESP, + VMEDIA_MEMBER_CD_RESP, + BOOT_SOURCE_OVERRIDE_TARGET_USBCD, + ] + responses = INIT_RESP_HPE + responses_get + self.set_mock_response(mock_get, 200, responses) + self.set_mock_response(mock_post, 200, "OK") + self.set_mock_response(mock_delete, 200, "OK") + self.set_mock_response(mock_patch, 200, "OK") + self.args = [self.option_arg] + _, err = self.badfish_call() + assert err == VMEDIA_BOOT_TO_SM_PASS + class TestCheckRemoteImage(TestBase): option_arg = "--check-remote-image"