PUC-1410: fix - update BMC credentials code with Bmc class.

Move actions to Bmc class and update tests.

Update factory cred options.

fix test calls.

remediate test actions and cleanup methods in Bmc.

Author: RSabounds

python/understack-workflows/tests/test_bmc_credentials.py

@@ -1,37 +1,39 @@
+from unittest.mock import MagicMock
+
 import pytest
 
+from understack_workflows.bmc import RedfishRequestError
 from understack_workflows.bmc_credentials import set_bmc_password
 
 
 @pytest.fixture
-def mock_redfish(mocker):
-    mock = mocker.patch("understack_workflows.bmc_credentials._redfish_request")
-    mock.return_value = {"AccountService": {"@odata.id": "/testme"}}
+def mock_getsession(mocker):
+    mock = mocker.patch("understack_workflows.bmc_credentials.Bmc.get_session")
+    mock.return_value = "tOkEn", "/path/to/session/1234"
     return mock
 
 
 @pytest.fixture
-def mock_success_auth(mocker):
-    mock = mocker.patch("understack_workflows.bmc_credentials._verify_auth")
-    mock.return_value = "tOkEn", "/path/to/session/1234"
+def mock_close(mocker):
+    mock = mocker.patch("understack_workflows.bmc_credentials.Bmc.close_session")
     return mock
 
 
 @pytest.fixture
 def mock_fail_auth(mocker):
-    mock = mocker.patch("understack_workflows.bmc_credentials._verify_auth")
-    mock.return_value = None
+    mock_response = MagicMock()
+    mock_response.status_code = 402
+    mock_response.json.return_value = {"message": "Failure"}
+    mock = mocker.patch("requests.request", return_value=mock_response)
     return mock
 
 
-def test_set_bmc_password_noop(mock_success_auth, mock_redfish):
+def test_set_bmc_password_noop(mock_getsession, mock_close):
     set_bmc_password("1.2.3.4", "qwertyuiop")
-    assert mock_redfish.call_count == 1
-    mock_redfish.assert_called_with(
-        "1.2.3.4", "/path/to/session/1234", "tOkEn", "DELETE"
-    )
+    mock_getsession.assert_called_once()
+    mock_close.assert_called_with(session="/path/to/session/1234", token="tOkEn")
 
 
-def test_set_bmc_password_failed(mock_fail_auth, mock_redfish):
-    with pytest.raises(Exception):
+def test_set_bmc_password_failed(mock_fail_auth):
+    with pytest.raises(RedfishRequestError):
         set_bmc_password("1.2.3.4", "qwertyuiop")

python/understack-workflows/understack_workflows/bmc.py

@@ -1,5 +1,6 @@
 # pylint: disable=E1131,C0103
 
+import copy
 import logging
 import os
 from dataclasses import dataclass
@@ -10,10 +11,13 @@
 
 from understack_workflows.bmc_password_standard import standard_password
 from understack_workflows.helpers import credential
+from understack_workflows.helpers import setup_logger
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)  # type: ignore
 logging.getLogger("urllib3").setLevel(logging.WARNING)
 
+logger = setup_logger(__name__)
+
 HEADERS = {
     "Accept": "application/json",
     "Content-Type": "application/json; charset=utf-8",
@@ -24,6 +28,14 @@ class RedfishRequestError(Exception):
     """Handle Exceptions from Redfish handler."""
 
 
+class AuthException(Exception):
+    """Authentication Exception."""
+
+
+class AccountServiceException(Exception):
+    """AccountService Query Exception."""
+
+
 @dataclass
 class Bmc:
     """Represent DRAC/iLo and know how to perform low-level query on it."""
@@ -35,9 +47,16 @@ def __init__(
         self.ip_address = ip_address
         self.username = username
         self.password = password if password else ""
+        self._base_path: str | None = None
         self._system_path: str | None = None
         self._manager_path: str | None = None
 
+    @property
+    def base_path(self) -> str:
+        """Read System path from BMC."""
+        self._system_path = self._base_path or self.get_base_path()
+        return self._system_path
+
     @property
     def system_path(self) -> str:
         """Read System path from BMC."""
@@ -58,6 +77,11 @@ def url(self):
         """Return base redfish URL."""
         return f"https://{self.ip_address}"
 
+    def get_base_path(self):
+        """Get Base Path."""
+        _result = "/redfish/v1/"
+        return _result
+
     def get_system_path(self):
         """Get System Path."""
         _result = self.redfish_request("/redfish/v1/Systems/")
@@ -68,24 +92,143 @@ def get_manager_path(self):
         _result = self.redfish_request("/redfish/v1/Managers/")
         return _result["Members"][0]["@odata.id"].rstrip("/")
 
+    def get_user_accounts(self, token: str | None = None) -> list[dict]:
+        """A vendor agnostic approach to crawling the API for BMC accounts."""
+        try:
+            # get account service
+            r = (
+                self.redfish_request(path=self.base_path, token=token)
+                if token
+                else self.redfish_request(path=self.base_path)
+            )
+            account_service_uri = r["AccountService"]["@odata.id"]
+            logger.debug("account_service_url: %s", account_service_uri)
+
+            # get account collection uri
+            r = (
+                self.redfish_request(path=account_service_uri, token=token)
+                if token
+                else self.redfish_request(path=account_service_uri)
+            )
+            accounts_uri = r["Accounts"]["@odata.id"]
+            logger.debug("accounts_url: %s", accounts_uri)
+
+            # get accounts
+            r = (
+                self.redfish_request(path=accounts_uri, token=token)
+                if token
+                else self.redfish_request(path=accounts_uri)
+            )
+            accounts = r["Members"]
+            logger.debug("accounts: %s", accounts)
+
+            return accounts
+        except AccountServiceException:
+            logger.exception("Can't fetch accounts from Redfish account service.")
+            raise
+
+    def set_bmc_creds(self, password: str, token: str | None = None):
+        """Change password for the account associated with the bmc."""
+        accounts = self.get_user_accounts(token)
+        matched_account = None
+        for account in accounts:
+            account_url = account["@odata.id"]
+            if token:
+                a = self.redfish_request(path=account_url, token=token)
+            else:
+                a = self.redfish_request(path=account_url)
+            if self.username == a["UserName"]:
+                logger.debug("found account: %s", a)
+                matched_account = a
+                break
+            if not matched_account:
+                raise AuthException(f"Unable to find BMC account for {self.username}")
+            account_uri = matched_account["@odata.id"]
+            _payload = {"Password": password}
+            if token:
+                self.redfish_request(
+                    method="PATCH", path=account_uri, token=token, payload=_payload
+                )
+            else:
+                self.redfish_request(method="PATCH", path=account_uri, payload=_payload)
+
+    def get_session(self, password: str) -> tuple[str | None, str | None]:
+        """Request a new session."""
+        _payload = {"UserName": self.username, "Password": password}
+        token, session = self.session_request(
+            method="POST",
+            path="/redfish/v1/SessionService/Sessions",
+            payload=_payload,
+        )
+        if token and session:
+            return token, session
+        else:
+            return None, None
+
+    def close_session(self, session: str, token: str | None = None) -> None:
+        """Close BMC token session."""
+        if token:
+            self.redfish_request(method="DELETE", path=session, token=token)
+        else:
+            self.redfish_request(method="DELETE", path=session)
+
+    def session_request(
+        self,
+        path: str,
+        method: str = "POST",
+        payload: dict | None = None,
+        verify: bool = False,
+        timeout: int = 30,
+    ) -> tuple[str | None, str | None]:
+        """Request a session via Redfish against the Bmc."""
+        _headers = copy.copy(HEADERS)
+        url = f"{self.url()}{path}"
+        r = requests.request(
+            method,
+            url,
+            verify=verify,
+            timeout=timeout,
+            json=payload,
+            headers=_headers,
+        )
+        if r.status_code >= 400:
+            raise RedfishRequestError(
+                f"BMC communications failure HTTP {r.status_code} "
+                + f"{r.reason} from {url} - {r.text}"
+            )
+        if r.text:
+            token = r.headers["X-Auth-Token"]
+            if "Location" in r.headers:
+                location = r.headers["Location"].split(self.ip_address)[1]
+            else:
+                location = r.json()["@odata.id"]
+
+            return (token, location)
+        else:
+            return (None, None)
+
     def redfish_request(
         self,
         path: str,
         method: str = "GET",
         payload: dict | None = None,
+        token: str | None = None,
         verify: bool = False,
         timeout: int = 30,
     ) -> dict:
         """Request a path via Redfish against the Bmc."""
+        _headers = copy.copy(HEADERS)
+        if token:
+            _headers.update({"X-Auth-Token": token})
         url = f"{self.url()}{path}"
         r = requests.request(
             method,
             url,
-            auth=(self.username, self.password),
+            auth=None if token else (self.username, self.password),
             verify=verify,
             timeout=timeout,
             json=payload,
-            headers=HEADERS,
+            headers=_headers,
         )
         if r.status_code >= 400:
             raise RedfishRequestError(