NEXMANAGE-950 Add Power Control capability from INBS (#587)

inbc-program/README.md

@@ -1,4 +1,4 @@
-# Intel® In-band Manageability Command-line Utility (INBC) 
+# Intel® In-band Manageability Command-line Utility (INBC)
 
 <details>
 <summary>Table of Contents</summary>
@@ -26,7 +26,7 @@
    14. [Source OS Add](#source-os-add)
    15. [Source OS Remove](#source-os-remove)
    16. [Source OS Update](#source-os-update)
-   15. [Source OS List](#source-os-list)
+   17. [Source OS List](#source-os-list)
 6. [Status Codes](#status-codes)
 7. [Return and Exit Codes](#return-and-exit-codes)
 

inbm/Changelog.md

@@ -4,12 +4,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ## NEXT - YYYY-MM-DD
-
-
+### Added
+ - (NEXMANAGE-950) Add Set Power State Capability to cloudadapter from UDM
 
 ## 4.2.6.2 - 2024-10-25
 ### Added
- - (NEXMANAGE-900) Add UpdateFirmware to the common proto file
+ - (NEXMANAGE-900) Add UpdateFirmware to the common proto file and cloudadapter support from UDM
 
 ### Changed
  - (NEXMANAGE-906) Truncate the granular log instead of removing it

inbm/cloudadapter-agent/cloudadapter/cloud/adapters/inbs/operation.py

@@ -6,7 +6,8 @@
 import logging
 import xml.etree.ElementTree as ET
 from google.protobuf.timestamp_pb2 import Timestamp
-from cloudadapter.pb.common.v1.common_pb2 import UpdateSystemSoftwareOperation, UpdateFirmwareOperation, RpcActivateOperation, Operation, Schedule
+from cloudadapter.pb.common.v1.common_pb2 import UpdateSystemSoftwareOperation, \
+    UpdateFirmwareOperation, SetPowerStateOperation, RpcActivateOperation, Operation, Schedule
 from cloudadapter.pb.inbs.v1.inbs_sb_pb2 import UpdateScheduledOperations
 
 logger = logging.getLogger(__name__)
@@ -94,7 +95,8 @@ def convert_operation_to_xml_manifests(operation: Operation) -> ET.Element:
 
     if not (operation.HasField('update_system_software_operation') 
             or operation.HasField('rpc_activate_operation') 
-            or operation.HasField('update_firmware_operation')):
+            or operation.HasField('update_firmware_operation')
+            or operation.HasField('set_power_state_operation')):
         raise ValueError("Operation type not supported")
 
     if len(operation.pre_operations) > 0:
@@ -108,13 +110,13 @@ def convert_operation_to_xml_manifests(operation: Operation) -> ET.Element:
     manifest = None
 
     if operation.HasField('update_system_software_operation'):
-        logger.debug("Converting UpdateSystemSoftwareOperation to XML manifest")
-        manifest = convert_system_software_operation_to_xml_manifest(operation.update_system_software_operation)
+         manifest = convert_system_software_operation_to_xml_manifest(operation.update_system_software_operation)
     elif operation.HasField('rpc_activate_operation'):
         manifest = convert_rpc_activate_operation_to_xml_manifest(operation.rpc_activate_operation)
     elif operation.HasField('update_firmware_operation'):
-        logger.debug("Converting UpdateFirmwareOperation to XML manifest")
         manifest = convert_firmware_operation_to_xml_manifest(operation.update_firmware_operation)
+    elif operation.HasField('set_power_state_operation'):
+        manifest = convert_power_state_operation_to_xml_manifest(operation.set_power_state_operation)
     else:
         raise ValueError("No valid operation found")
 
@@ -161,28 +163,36 @@ def convert_firmware_operation_to_xml_manifest(operation: UpdateFirmwareOperatio
     fota = ET.SubElement(type, 'fota', name="")
 
     # Fetch URL
-    if operation.url != '':
-        ET.SubElement(fota, 'fetch').text = operation.url
+    if not operation.url:
+        raise ValueError("Fetch URL cannot be unspecified")
+    ET.SubElement(fota, 'fetch').text = operation.url
 
-    if operation.bios_version:
-        ET.SubElement(fota, 'biosversion').text = operation.bios_version
+    if not operation.bios_version:
+        raise ValueError("BIOS Version cannot be unspecified")
+    ET.SubElement(fota, 'biosversion').text = operation.bios_version
 
     if operation.signature_version:
         ET.SubElement(fota, 'signatureversion').text = str(operation.signature_version)
 
     if operation.signature:
         ET.SubElement(fota, 'signature').text = operation.signature
 
-    if operation.manufacturer:
-        ET.SubElement(fota, 'manufacturer').text = operation.manufacturer
+    if not operation.manufacturer:
+        raise ValueError("Manufacturer cannot be unspecified")
+    ET.SubElement(fota, 'manufacturer').text = operation.manufacturer
 
-    if operation.product_name:
-        ET.SubElement(fota, 'product').text = operation.product_name
+    if not operation.product_name:
+        raise ValueError("Product name cannot be unspecified")
+    ET.SubElement(fota, 'product').text = operation.product_name
 
-    if operation.vendor:
-        ET.SubElement(fota, 'vendor').text = operation.vendor       
+    if not operation.vendor:
+        raise ValueError("Vendor cannot be unspecified")
+    ET.SubElement(fota, 'vendor').text = operation.vendor       
 
     # Release date in the required format
+    if operation.release_date == Timestamp():
+        raise ValueError("Release date cannot be unspecified")
+
     if operation.release_date.ToSeconds() > 0:
         release_date = Timestamp()
         release_date.FromDatetime(operation.release_date.ToDatetime())
@@ -257,3 +267,33 @@ def convert_system_software_operation_to_xml_manifest(operation: UpdateSystemSof
     xml_declaration = '<?xml version="1.0" encoding="utf-8"?>'
     xml_str = ET.tostring(manifest, encoding='utf-8', method='xml').decode('utf-8')
     return xml_declaration + '\n' + xml_str
+
+def convert_power_state_operation_to_xml_manifest(operation: SetPowerStateOperation) -> str:    
+    """Converts a SetPowerStateOperation message to an XML manifest string for Dispatcher."""
+
+    if operation.opcode == SetPowerStateOperation.POWER_STATE_UNSPECIFIED:
+        raise ValueError("Power state cannot be unspecified")
+
+    if operation.opcode == SetPowerStateOperation.POWER_STATE_ON:
+        raise ValueError("Power state ON is not supported as an Inband operation")
+
+    if operation.opcode == SetPowerStateOperation.POWER_STATE_RESET:
+        raise ValueError("Power state RESET is not supported as an Inband operation")
+
+    power_state = ''
+    if operation.opcode == SetPowerStateOperation.POWER_STATE_OFF:
+        power_state = 'shutdown'    
+    elif operation.opcode == SetPowerStateOperation.POWER_STATE_CYCLE:
+        power_state = 'restart'
+    else:
+        raise ValueError("Invalid power state")
+
+    # Create the root element
+    manifest = ET.Element('manifest')
+    ET.SubElement(manifest, 'type').text = 'cmd'
+    cmd = ET.SubElement(manifest, 'cmd').text = power_state
+
+    # Generate the XML string with declaration
+    xml_declaration = '<?xml version="1.0" encoding="utf-8"?>'
+    xml_str = ET.tostring(manifest, encoding='utf-8', method='xml').decode('utf-8')
+    return xml_declaration + '\n' + xml_str

inbm/cloudadapter-agent/tests/unit/cloud/adapters/inbs/test_operation.py

@@ -13,6 +13,7 @@
     PreOperation,
     PostOperation,
     ScheduledOperation,
+    SetPowerStateOperation,
     Schedule,
     SingleSchedule,
     RepeatedSchedule,
@@ -31,6 +32,7 @@
     convert_rpc_activate_operation_to_xml_manifest,
     convert_operation_to_xml_manifests,
     convert_updated_scheduled_operations_to_dispatcher_xml,
+    convert_power_state_operation_to_xml_manifest
 )
 
 RPC_OPERATION_LARGE = RpcActivateOperation(
@@ -135,7 +137,20 @@
     "</fota></type>"
     "</ota></manifest>"
 )
-
+RESTART_OPERATION = SetPowerStateOperation(
+    opcode=SetPowerStateOperation.POWER_STATE_CYCLE
+)
+RESTART_XML = (
+    '<?xml version="1.0" encoding="utf-8"?>\n'
+    "<manifest><type>cmd</type><cmd>restart</cmd></manifest>"
+)
+SHUTDOWN_OPERATION = SetPowerStateOperation(
+    opcode=SetPowerStateOperation.POWER_STATE_OFF
+)
+SHUTDOWN_XML = (
+    '<?xml version="1.0" encoding="utf-8"?>\n'
+    "<manifest><type>cmd</type><cmd>shutdown</cmd></manifest>"
+)
 
 # Test cases to convert UpdateScheduledOperations -> dispatcher XML (success)
 @pytest.mark.parametrize(
@@ -263,6 +278,20 @@ def test_convert_update_scheduled_operations_to_xml_manifest_exception(
         convert_updated_scheduled_operations_to_dispatcher_xml(request_id, uso)
     assert expected_exception_message == str(exc_info.value)
 
+# Test cases for function that checks XML manifest creation from software update operations
+@pytest.mark.parametrize(
+    "operation, expected_xml",
+    [
+        (RESTART_OPERATION, RESTART_XML),
+        (SHUTDOWN_OPERATION, SHUTDOWN_XML),
+    ],
+)
+def test_convert_power_state_operation_to_xml_manifest_success(
+    operation, expected_xml
+):
+    xml_manifest = convert_power_state_operation_to_xml_manifest(operation)
+    assert xml_manifest == expected_xml
+
 # Test cases for function that checks XML manifest creation from software update operations
 @pytest.mark.parametrize(
     "operation, expected_xml",
@@ -305,6 +334,104 @@ def test_convert_rpc_activate_operation_to_xml_manifest_success(
     rpc_xml_manifest = convert_rpc_activate_operation_to_xml_manifest(operation)
     assert rpc_xml_manifest == rpc_expected_xml
 
+@pytest.mark.parametrize(
+    "operation, exception_message",
+    [
+        (
+            UpdateFirmwareOperation(
+                url="http://example.com/update",
+                manufacturer="Intel",
+                product_name="Intel NUC",
+                vendor="Intel",    
+                release_date=Timestamp(seconds=int(datetime(2023, 1, 1).timestamp())),    
+            ),
+            "BIOS Version cannot be unspecified",
+        ),
+        (
+            UpdateFirmwareOperation(
+                bios_version="1.0",
+                manufacturer="Intel",
+                product_name="Intel NUC",
+                vendor="Intel",    
+                release_date=Timestamp(seconds=int(datetime(2023, 1, 1).timestamp())),    
+            ),
+            "Fetch URL cannot be unspecified",
+        ),
+        (
+            UpdateFirmwareOperation(
+                url="http://example.com/update",
+                bios_version="1.0",
+                product_name="Intel NUC",
+                vendor="Intel",    
+                release_date=Timestamp(seconds=int(datetime(2023, 1, 1).timestamp())),    
+            ),
+            "Manufacturer cannot be unspecified",
+        ),
+        (
+            UpdateFirmwareOperation(
+                url="http://example.com/update",
+                bios_version="1.0",
+                manufacturer="Intel",
+                product_name="Intel NUC",  
+                release_date=Timestamp(seconds=int(datetime(2023, 1, 1).timestamp())),    
+            ),
+            "Vendor cannot be unspecified",
+        ),
+        (
+            UpdateFirmwareOperation(
+                url="http://example.com/update",
+                bios_version="1.0",
+                manufacturer="Intel",
+                vendor="Intel", 
+                release_date=Timestamp(seconds=int(datetime(2023, 1, 1).timestamp())),    
+            ),
+            "Product name cannot be unspecified",
+        ),
+        (
+            UpdateFirmwareOperation(
+                url="http://example.com/update",
+                bios_version="1.0",
+                manufacturer="Intel",
+                product_name="Intel NUC",
+                vendor="Intel", 
+            ),
+            "Release date cannot be unspecified",
+        ),
+    ],
+)
+def test_convert_firmware_operation_to_xml_manifest_unspecified_error(
+    operation, exception_message
+):
+    with pytest.raises(ValueError) as excinfo:
+        convert_firmware_operation_to_xml_manifest(operation)
+    assert exception_message in str(excinfo.value)
+
+@pytest.mark.parametrize(
+    "power_state, exception_message",
+    [
+        (
+            SetPowerStateOperation.POWER_STATE_UNSPECIFIED,
+            "Power state cannot be unspecified",
+        ),
+        (
+            SetPowerStateOperation.POWER_STATE_ON,
+            "Power state ON is not supported as an Inband operation",
+        ),
+        (
+            SetPowerStateOperation.POWER_STATE_RESET,
+            "Power state RESET is not supported as an Inband operation",
+        ),
+    ],
+)
+def test_convert_power_state_operation_to_xml_manifest_unspecified_error(
+    power_state, exception_message
+):
+    operation = SetPowerStateOperation()
+    operation.opcode = power_state
+    with pytest.raises(ValueError) as excinfo:
+        convert_power_state_operation_to_xml_manifest(operation)
+    assert exception_message in str(excinfo.value)
+
 
 @pytest.mark.parametrize(
     "operation, exception_message",