gh #44 L3 test case development

Updated c test and python code to read the commands from yaml file

host/tests/classes/hdmiCEC.py

@@ -21,12 +21,9 @@
 # *
 #* ******************************************************************************
 
-import subprocess
+import re
 import os
 import sys
-from enum import Enum, auto
-import re
-import yaml
 
 # Add parent directory to the system path for module imports
 dir_path = os.path.dirname(os.path.realpath(__file__))
@@ -119,13 +116,13 @@ def terminate(self):
         """
         result = self.utMenu.select(self.testSuite, "Close HDMI CEC")
 
-    def addLogicalAddress(self, logicalAddress:int):
+    def addLogicalAddress(self, logicalAddress:str='0'):
         """
         Adding the logical address of a specific device.
         For now Sink to support only the logical address 0.
 
         Args:
-            logicalAddress (int): The Logical address of the DUT.  This will be fixed to zero for a sink device for now.
+            logicalAddress (str): The Logical address of the DUT. This will be fixed to zero for a sink device for now.
 
         Returns:
             None
@@ -134,7 +131,7 @@ def addLogicalAddress(self, logicalAddress:int):
                 {
                     "query_type": "direct",
                     "query": "Enter Logical Address:",
-                    "input": str(logicalAddress)
+                    "input": logicalAddress
                 }
         ]
         result = self.utMenu.select(self.testSuite, "Add Logical Address", promptWithAnswers)
@@ -162,7 +159,7 @@ def getLogicalAddress(self):
             int: Logical address of the device.
         """
         result = self.utMenu.select( self.testSuite, "Get Logical Address")
-        connectionStatusPattern = r"Result HdmiCecGetLogicalAddress\(IN:handle:[.*\], OUT:logicalAddress:[.*\]) HDMI_CEC_STATUS:[.*\])"
+        connectionStatusPattern = r"Result HdmiCecGetLogicalAddress\(IN:handle:\[0x[0-9A-F]+\], OUT:logicalAddress:\[([0-9A-Fa-f]+)\]\)"
         logicalAddress = self.searchPattern(result, connectionStatusPattern)
 
         return logicalAddress
@@ -183,7 +180,7 @@ def getPhysicalAddress(self):
 
         return physicalAddress
 
-    def cecTransmitCmd(self, destLogicalAddress:int, cecCommand:int, cecData:list=None):
+    def cecTransmitCmd(self, destLogicalAddress:str, cecCommand:str, cecData:list=None):
         """
         Transmit/Broadcast the CEC command and data to the respective destination.
 
@@ -198,12 +195,12 @@ def cecTransmitCmd(self, destLogicalAddress:int, cecCommand:int, cecData:list=No
                 {
                     "query_type": "direct",
                     "query": "Enter a valid Destination Logical Address:",
-                    "input": str(destLogicalAddress)
+                    "input": destLogicalAddress
                 },
                 {
                     "query_type": "direct",
                     "query": "Enter CEC Command (in hex):",
-                    "input": str(cecCommand)
+                    "input": cecCommand
                 },
         ]
 
@@ -218,6 +215,63 @@ def cecTransmitCmd(self, destLogicalAddress:int, cecCommand:int, cecData:list=No
 
         result = self.utMenu.select( self.testSuite, "Transmit CEC Command",promptWithAnswers)
 
+    def readCallbackDetails (self):
+        """
+        Parses the callback logs from the device.
+
+        Args:
+            None.
+
+        Returns:
+            dict: A dictionary with two keys:
+                - "Received": A list of dictionaries containing details about received opcodes.
+                - "Response": A list of dictionaries containing details about sent response opcodes.
+                Each dictionary contains the following keys:
+                    - "Opcode" (str): The opcode value in hexadecimal.
+                    - "Description" (str): A textual description of the opcode.
+                    - "Initiator" (str): The initiator address in hexadecimal.
+                    - "Destination" (str): The destination address in hexadecimal.
+                    - "Data" (list): The data associated with the opcode.
+        """
+        result = {
+            "Received": [],
+            "Response": []
+        }
+        callbackLogs = self.testSession.read_all()
+
+        received_pattern = re.compile(
+            r"Received Opcode: \[([^\]]+)\] \[([^\]]+)\] Initiator: \[([^\]]+)\], Destination: \[([^\]]+)\] Data: \[(.*?)\]"
+        )
+        sent_pattern = re.compile(
+            r"Sent Response Opcode: \[([^\]]+)\] \[([^\]]+)\] Initiator: \[([^\]]+)\], Destination: \[([^\]]+)\] Data: \[(.*?)\]"
+        )
+
+        def parse_data_field(data_field):
+            # Split the data field into an array of hex values
+            return ["0x" + value.strip() for value in data_field.split(":")]
+
+        for match in received_pattern.finditer(callbackLogs):
+            opcode, description, initiator, destination, data = match.groups()
+            result["Received"].append({
+                "Opcode": opcode,
+                "Description": description,
+                "Initiator": initiator,
+                "Destination": destination,
+                "Data": parse_data_field(data)
+            })
+
+        for match in sent_pattern.finditer(callbackLogs):
+            opcode, description, initiator, destination, data = match.groups()
+            result["Response"].append({
+                "Opcode": opcode,
+                "Description": description,
+                "Initiator": initiator,
+                "Destination": destination,
+                "Data": parse_data_field(data)
+            })
+
+        return result
+
     def __del__(self):
         """
         Cleans up and de-initializes the hdmi cec  helper by stopping the test menu.

host/tests/classes/hdmiCEC_testConfig.yml

@@ -26,7 +26,7 @@ hdmicec:
         #List of artifacts folders, test class copies the content of folder to the target device workspace
           - "../../../bin/"
         # exectute command, this will appended with the target device workspace path
-        execute: "run.sh"
+        execute: "run.sh -p sink_hdmiCEC.yml"
         type: UT-C # C (UT-C Cunit) / C++ (UT-G (g++ ut-core gtest backend))
         suites:
             0:

host/tests/hdmiCEC_L3_Tests/hdmiCECHelperClass.py

@@ -69,6 +69,11 @@ def __init__(self, testName:str, qcId:str, log:logModule=None ):
         self.targetWorkspace = self.cpe.get("target_directory")
         self.targetWorkspace = os.path.join(self.targetWorkspace, self.moduleName)
 
+        self.testCECCommands = os.path.join(dir_path, "hdmiCECTestCommands.yml")
+        hdmicec = ConfigRead(self.testCECCommands, self.moduleName)
+        self.cecCommands = hdmicec.fields.get(self.testName)
+        self.hdmiCECController = self.dut.hdmiCECController
+
 #    def testDownloadAssets(self):
 #        """
 #        Downloads the test artifacts listed in the test setup configuration.

host/tests/hdmiCEC_L3_Tests/hdmiCECTestCommands.yml

@@ -1,15 +1,72 @@
-hdmiCEC:
+hdmicec:
   test01_TransmitCECCommands:
-      - command: "0x04"  # Image View on
-        payload:
-        responses:
-      - command: "0x36"  # Standby
-        payload:
-        responses:
-      - command: "0x9F"  # Get CEC version
-        payload:
-        responses:
-          - type: "Get CEC Version"
-            loop_devices: false  # Apply this response to all devices in the device map
-            payload: ""  # Response pattern with placeholders
-            description: "Device request for CEC version"
+    - command: "0x82"  # Active Soruce
+      payload: ["0x00", "0x00"]
+      type: "Broadcast"
+    - command: "0x90"  # "Report power status
+      payload: ["0x00"]
+      type: "Direct"
+      response: null
+    - command: "0x47"  # Give OSD Name
+      payload: ['0x52', '0x44', '0x4b', '0x20', '0x56', '0x54', '0x53', '0x20', '0x44', '0x65', '0x76', '0x69', '0x63', '0x65']
+      type: "Direct"
+      response: null
+  test02_ReceiveCECCommands:
+    - command: "0x04"  # Image View on
+      payload:
+      type: "Direct"
+      response: null
+    - command: "0x85"  # Request Active source
+      payload:
+      type: "Broadcast"
+      response:
+        type: "Broadcast"
+        command: "0x82"
+        update_payload: true
+        payload: ["0x00", "0x00"]
+        description: "Active Soruce"
+    - command: "0x91"  # Get Menu Language
+      payload:
+      type: "Direct"
+      response:
+        type: "Broadcast"
+        command: "0x32"
+        update_payload: false
+        payload: ["0x65", "0x6E", "0x67"]
+        description: "Set Menu Language"
+    - command: "0x8C"  # Give Vendor ID
+      payload:
+      type: "Direct"
+      response:
+        type: "Broadcast"
+        command: "0x87"
+        update_payload: false
+        payload: ["0x00", "0x00","0x01"]
+        description: "Device Vendor Id"
+    - command: "0x8F"  # Give power status
+      payload:
+      type: "Direct"
+      response:
+        type: "Direct"
+        command: "0x90"
+        update_payload: false
+        payload: ["0x00"]
+        description: "Report power status"
+    - command: "0x9F"  # Get CEC version
+      payload:
+      type: "Direct"
+      response:
+        type: "Direct"
+        command: "0x9E"
+        update_payload: false
+        payload: ["0x05"]
+        description: "Device request for CEC version"
+    - command: "0x46"  # Give OSD Name
+      payload:
+      type: "Direct"
+      response:
+        type: "Direct"
+        command: "0x47"
+        update_payload: false
+        payload: ['0x52', '0x44', '0x4b', '0x20', '0x56', '0x54', '0x53', '0x20', '0x44', '0x65', '0x76', '0x69', '0x63', '0x65']
+        description: "Device request OSD name"

host/tests/hdmiCEC_L3_Tests/hdmiCEC_test01_TransmitCECCommands.py

@@ -50,11 +50,8 @@ def __init__(self, log:logModule=None):
         # Class variables
         self.testName  = "test01_TransmitCECCommands"
         self.qcID = '1'
-        self.sourceLogicalAddress = 0
-
-        self.testCECCommands = os.path.join(dir_path, "hdmiCECTestCommands.yml")
-        hdmicec = ConfigRead(self.testCECCommands, "hdmiCEC")
-        self.cecCommands = hdmicec.fields.get(self.testName)
+        self.tvLogicalAddress = '0'
+        self.broadcastAddress = 'f'
 
         super().__init__(self.testName, self.qcID, log)
 
@@ -74,37 +71,49 @@ def testFunction(self):
         self.testhdmiCEC.initialise()
 
         # Add the logical Address.
-        self.testhdmiCEC.addLogicalAddress(self.sourceLogicalAddress)
+        self.testhdmiCEC.addLogicalAddress(self.tvLogicalAddress)
+
+        # Get the logical Address.
+        deviceLogicalAddress = self.testhdmiCEC.getLogicalAddress()
 
-        self.cecDevices = self.cecAdapter.listDevices()
+        self.cecDevices = self.hdmiCECController.listDevices()
 
+        finalResult = True
         for device in self.cecDevices:
             logicalAddress = device["logical address"]
 
             # To bypass sending the message to TV
-            if logicalAddress == 0 or logicalAddress == 14:
+            if logicalAddress == '0' or logicalAddress == 'f':
                 continue
 
             for command in self.cecCommands:
+                result = False
                 cec = command.get("command")
                 payload = command.get("payload")
+                type = command.get("type")
+
+                destinationLogicalAddress = logicalAddress
+                if type == "Broadcast":
+                    destinationLogicalAddress = self.broadcastAddress
 
                 # Transmit Standby command to a specific destination address
-                self.testhdmiCEC.cecTransmitCmd(logicalAddress, cec, payload)
+                self.testhdmiCEC.cecTransmitCmd(destinationLogicalAddress, cec, payload)
+
+                self.log.stepStart(f'HdmiCecTx Source: {deviceLogicalAddress} Destination: {destinationLogicalAddress} CEC OPCode: {cec} Payload: {payload}')
 
-                self.log.stepStart(f'HdmiCecTx Source: {self.sourceLogicalAddress} Destination: {logicalAddress} CEC OPCode: {cec} Payload: {payload}')
+                result = self.hdmiCECController.checkTransmitStatus(deviceLogicalAddress, destinationLogicalAddress, cec, payload)
 
-                result = self.cecAdapter.checkTransmitStatus(self.sourceLogicalAddress, logicalAddress, cec, payload)
+                self.log.stepResult(result, f'HdmiCecTx Source: {deviceLogicalAddress} Destination: {destinationLogicalAddress} CEC OPCode: {cec} Payload: {payload}')
 
-                self.log.stepResult(result, f'HdmiCecTx Source: {self.sourceLogicalAddress} Destination: {logicalAddress} CEC OPCode: {cec} Payload: {payload}')
+                finalResult &= result
 
         # Remove the Logical Address
         self.testhdmiCEC.removeLogicalAddress()
 
         # Terminate dsAudio Module
         self.testhdmiCEC.terminate()
 
-        return result
+        return finalResult
 
 if __name__ == '__main__':
     summerLogName = os.path.splitext(os.path.basename(__file__))[0] + "_summery"