fix: update logger library v2

conftest.py

@@ -9,7 +9,9 @@ import sys
 from datetime import datetime
 from enum import Enum
 from statistics import mean
-from typing import Any, Callable, Dict, Match, Optional, Tuple, List
+from typing import Any, Callable, Dict, Generator, Optional, Tuple, List
+from re import Match
+
 import numpy as np
 import matplotlib.pyplot as plt
 from dataclasses import dataclass
@@ -49,6 +51,7 @@ OBJ_ID = 2
 OBJ_ADDRESS = 1
 CID = 2
 PARAM_NAME = 3
+PARAM_VAL = 4
 PARAM_SUCCESS = "success"
 PARAM_FAIL = "fail"
 
@@ -84,6 +87,7 @@ class MbParameter:
         status,
         cid,
         response_time,
+        value
     ) -> None:
         self.name: str = name
         self.instance_address: bytes = instance_address
@@ -92,13 +96,45 @@ class MbParameter:
         self.status: str = status
         self.cid: bytes = cid
         self.response_time: Optional[int] = response_time
+        self.value: Optional[bytes] = value
 
     def __repr__(self) -> str:
         if self.obj_tag == MASTER_TAG:
-            return f"Parameter name:{self.name}, Obj type: {self.obj_tag}, Object ID:{self.instance_address!r}, Transaction time:{self.transaction_timestamp}, Status:{self.status}, Cid:{self.cid}, Master Response time:{self.response_time}"
+            return f"Parameter name:{self.name}, Obj tag: {self.obj_tag}, Object ID:{self.instance_address!r}, Transaction time:{self.transaction_timestamp}, Status:{self.status}, Cid:{self.cid}, Value: {self.value}, Response time:{self.response_time}"
         else:
-            return f"Parameter name:{self.name}, Obj type: {self.obj_tag}, Object ID:{self.instance_address}, Transaction time:{self.transaction_timestamp}, Status:{self.status}"
+            return f"Parameter name:{self.name}, Obj tag: {self.obj_tag}, Object ID:{self.instance_address}, Transaction time:{self.transaction_timestamp}, Status:{self.status}"
 
+    def get_name(self) -> str:
+        """Retrieve parameter name (string)."""
+        return self.name
+
+    def get_inst(self) -> bytes:
+        """Retrieve parameter instance address (bytes)."""
+        return self.instance_address
+
+    def get_timestamp(self) -> int:
+        """Retrieve parameter transaction timestamp (int)."""
+        return self.transaction_timestamp
+
+    def get_tag(self) -> str:
+        """Retrieve parameter tag (string)."""
+        return self.obj_tag
+
+    def get_status(self) -> str:
+        """Retrieve parameter status (string)."""
+        return self.status
+
+    def get_cid(self) -> bytes:
+        """Retrieve parameter cid (bytes)."""
+        return self.cid
+
+    def get_response_time(self) -> Optional[int]:
+        """Retrieve parameter response time (ms) or None."""
+        return self.response_time
+
+    def get_value(self) -> Optional[bytes]:
+        """Retrieve parameter value (bytes) or a sentinel b\"N/A\"."""
+        return self.value if self.value is not None else b"N/A"
 
 class MbObject:
     def __init__(self, tag, id, object_creation_timestamp) -> None:
@@ -108,7 +144,7 @@ class MbObject:
         self.parameters: List[MbParameter] = []
         self.parameter_count: int = 0
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"Obj Tag: {self.tag}, Obj ID: {self.id}, Creation Timestamp: {self.object_creation_timestamp}"
 
     def add_parameter(
@@ -119,9 +155,10 @@ class MbObject:
         tag: str,
         status: str,
         cid: bytes,
+        value: Optional[bytes]
     ) -> MbParameter:
         """The function add to list master or slave parameters"""
-        parameter = MbParameter(
+        parameter: MbParameter = MbParameter(
             name,
             instance_address,
             int(transaction_timestamp.decode("ascii")),
@@ -129,6 +166,7 @@ class MbObject:
             status,
             cid,
             None,
+            value
         )
         self.parameters.append(parameter)
         self.parameter_count += 1
@@ -150,27 +188,28 @@ class MbObject:
 
 
 class ModbusTestDut(IdfDut):
-    TEST_IP_PROMPT = r"Waiting IP([0-9]{1,2}) from stdin:"
+    TEST_IP_PROMPT = r"Waiting IP\(([0-9]{1,2})\) from stdin:"
     TEST_IP_ADDRESS_REGEXP = r"I \([0-9]+\) example_[a-z]+: [A-Za-z\-]* IPv4 [A-Za-z\"_:\s]*address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})"
     TEST_APP_NAME = r"I \([0-9]+\) [a-z_]+: Project name:\s+([_a-z]*)"
 
     TEST_EXPECT_STR_TIMEOUT = 120
+    TEST_IP_PROMPT_TOUT = 10
     TEST_ACK_TIMEOUT = 60
     TEST_MAX_CIDS = 8
 
     app: IdfApp
-    serial: IdfSerial
+    serial: IdfSerial  # type: ignore[override]
 
     def __init__(self, *args, **kwargs) -> None:  # type: ignore
         super().__init__(*args, **kwargs)
-        self.logger = logging.getLogger()
+        self.logger: logging.Logger = logging.getLogger()
         self.ip_address: Optional[str] = None
         self.app_name: Optional[str] = None
         self.dut_list: Optional[List[ModbusTestDut]] = None
         self.param_fail_count: int = 0
         self.param_ok_count: int = 0
-        self.test_stage = Stages.STACK_DEFAULT
-        self.dictionary = None
+        self.test_stage: Stages = Stages.STACK_DEFAULT
+        self.dictionary: Optional[Dict[Stages, bytes]] = None
         self.test_finish: bool = False
         self.mb_objects_count: int = (
             0  # number of objects (modbus instances in the test application)
@@ -206,9 +245,9 @@ class ModbusTestDut(IdfDut):
         # Workaround to get master/slave tag from DUT class
         # Checking if app_name contains the  field. Ex: modbus_tcp_master
         if MASTER_TAG in self.app_name.lower():
-            obj_tag = MASTER_TAG
+            obj_tag: str = MASTER_TAG
         elif SLAVE_TAG in self.app_name.lower():
-            obj_tag = SLAVE_TAG
+            obj_tag: str = SLAVE_TAG
         else:
             self.logger.error("Could not determine master/slave tag from app_name")
             raise RuntimeError from None
@@ -218,7 +257,7 @@ class ModbusTestDut(IdfDut):
 
     def add_object(self, tag: str, id: bytes, timestamp: bytes) -> MbObject:
         """The function add to list master or slave instances in the test"""
-        obj = MbObject(tag, id, timestamp)
+        obj: MbObject = MbObject(tag, id, timestamp)
         self.mb_objects.append(obj)
         self.mb_objects_count += 1
         return obj
@@ -228,9 +267,9 @@ class ModbusTestDut(IdfDut):
         objects: List[MbObject] = []
         self.check_mb_objects_list()
 
-        for object in self.mb_objects:
-            if tag == object.tag:
-                objects.append(object)
+        for obj in self.mb_objects:
+            if tag == obj.tag:
+                objects.append(obj)
 
         if not objects:
             self.logger.error("objects list from tag couldn't be retrieved")
@@ -241,22 +280,33 @@ class ModbusTestDut(IdfDut):
     def get_object_by_id(self, id: bytes) -> Optional[MbObject]:
         """The getter retrieves master or slave object by instance address"""
         self.check_mb_objects_list()
-        for object in self.mb_objects:
-            if id == object.id:
-                return object
+        for obj in self.mb_objects:
+            if id == obj.id:
+                return obj
 
         self.logger.error(f"couldn't find registered object with id: {id}")
         return None
 
-    def update_wrong_object_id(self, id: bytes) -> MbObject:
+    def update_wrong_object_id(self, id: bytes) -> Optional[MbObject]:
         """Scan registered object list checking for a wrongly parsed object ID which is
         not 10 characters long. Ex - 0x3ffbf7bc"""
         self.check_mb_objects_list()
-        for object in self.mb_objects:
-            if len(object.id) != 10:
-                self.logger.info(f"updating wrong object id: {object.id} to: {id}")
-                object.id = id
-                return object
+        for obj in self.mb_objects:
+            if len(obj.id) != 10:
+                self.logger.info(f"Updating wrong object id: {obj.id} to: {id}")
+                obj.id = id
+                return obj
+
+    def get_params_by_name(self, name: str) -> Optional[List[MbParameter]]:
+        """The getter retrieves parameters by name"""
+        # import sys, pdb; pdb.Pdb(stdout=sys.__stdout__).set_trace()
+        self.check_mb_objects_list()
+        params: List[MbParameter] = []
+        for obj in self.mb_objects:
+            for param in obj.parameters:
+                if name in str(param.name):
+                    params.append(param)
+        return params
 
     def get_params_by_object_id_and_status(
         self, instance_address: bytes, status: str
@@ -323,7 +373,7 @@ class ModbusTestDut(IdfDut):
     def add_dut_list(self, dut_instance: "ModbusTestDut") -> None:
         """The function keeps track of all DUTs involved in the test"""
         if self.dut_list is None:
-            self.dut_list: List[ModbusTestDut] = []
+            self.dut_list = []
 
         self.dut_list.append(dut_instance)
         return None
@@ -338,9 +388,9 @@ class ModbusTestDut(IdfDut):
 
     def get_avg_response_time_master(self) -> int:
         """The function iterates over master parameters to calculate mean response time"""
-        master_success_params = self.get_master_params_by_status(PARAM_SUCCESS)
+        master_success_params: List[MbParameter] = self.get_master_params_by_status(PARAM_SUCCESS)
         if master_success_params:
-            avg_response_time = [
+            avg_response_time: List[int] = [
                 param.response_time
                 for param in master_success_params
                 if param.response_time is not None
@@ -361,15 +411,17 @@ class ModbusTestDut(IdfDut):
         self, transaction_timestamp: bytes, master_address: bytes, cid: bytes
     ) -> None:
         """The function gets and saves master requests"""
-        request_response = MbRequestResponse(
-            int(transaction_timestamp.decode("ascii")), master_address, cid
+        request_response: MbRequestResponse = MbRequestResponse(
+            int(transaction_timestamp.decode("ascii")),
+            master_address,
+            cid
         )
         self.mb_request_response.append(request_response)
         return None
 
     def add_response_time_to_param(self, param: MbParameter) -> None:
         """The function add response time to parameter entry based on request list"""
-        last_request_response = self.mb_request_response[-1]
+        last_request_response: MbRequestResponse = self.mb_request_response[-1]
         if (
             param.instance_address == last_request_response.master_inst_address
             and param.cid == last_request_response.cid
@@ -409,15 +461,24 @@ class ModbusTestDut(IdfDut):
         self.logger.info(f"Project name registered: {self.app_name}")
         return self.app_name
 
-    def dut_send_ip(self, slave_ip: Optional[str]) -> Optional[int]:
+    def dut_send_ip(self, slave_ip: Optional[str] = None, port: Optional[str] = None) -> Optional[int]:
         """The function sends the slave IP address defined as a parameter to master"""
         addr_num: int = 0
-        self.expect(self.TEST_IP_PROMPT, timeout=self.TEST_EXPECT_STR_TIMEOUT)
+        try:
+            self.expect(self.TEST_IP_PROMPT, timeout=self.TEST_IP_PROMPT_TOUT)
+        except pexpect.TIMEOUT:
+            # Workaround for unreliable parsing of the master prompt.
+            # The expect() sometime does not catch it in spite it appears in the log.
+            # Send the IP address anyway after the timeout.
+            self.logger.error("Timeout waiting for IP prompt. Try to send address anyway.")
         if isinstance(slave_ip, str):
             for addr_num in range(0, self.TEST_MAX_CIDS):
-                message = r"IP{}={}".format(addr_num, slave_ip)
-                self.logger.info("{} sent to master".format(message))
+                message: str = r"IP{}={}".format(addr_num, slave_ip)
+                if isinstance(port, str) or isinstance(port, int):
+                    message += r";{}".format(str(port))
+                message += r"\r\n"
                 self.write(message)
+                self.logger.info("{} sent to master".format(message))
         return addr_num
 
     def dut_stats_info(self) -> ModbusDutStats:
@@ -483,7 +544,7 @@ class ModbusTestDut(IdfDut):
 
         label_axis = np.arange(len(names)) * label_space
 
-        text_box = (
+        text_box: str = (
             f"Master Average response time:{self.get_avg_response_time_master()} ms"
         )
 
@@ -503,7 +564,7 @@ class ModbusTestDut(IdfDut):
             label="Fail",
         )
         plt.xticks(label_axis, names)
-        props = dict(boxstyle="round", facecolor="wheat", alpha=alpha)
+        props: Dict[str, str | float] = dict(boxstyle="round", facecolor="wheat", alpha=alpha)
         ax.text(
             x_position_text,
             y_position_text,
@@ -520,8 +581,29 @@ class ModbusTestDut(IdfDut):
         plt.savefig(f"{file_name}.png")
         plt.show()
 
+    def get_item(self, data: Optional[List[Any]] = None, item: int = 0) -> bytes:
+        """
+        Safely return the requested item from a list as bytes.
+        On unexpected exceptions returns a bytes sentinel b"N/A".
+        """
+        try:
+            if not data or len(data) <= item:
+                # Return sentinel when data missing to keep callers' expectations of bytes
+                return b"N/A"
+            val = data[item]
+            if isinstance(val, bytes):
+                return val
+            if isinstance(val, str):
+                return val.encode("utf-8")
+            if isinstance(val, (int, float)):
+                return str(val).encode("utf-8")
+            # Fallback: stringify and encode any other object
+            return str(val).encode("utf-8")
+        except Exception:
+            return b"N/A"
+
     def expect_any(
-        self, *expect_items: Tuple[str, Callable], timeout: Optional[int]
+        self, *expect_items: Tuple[Optional[str], Callable], timeout: Optional[int]
     ) -> None:
         """
         expect_any(*expect_items, timeout=DEFAULT_TIMEOUT)
@@ -537,27 +619,26 @@ class ModbusTestDut(IdfDut):
         :keyword timeout: timeout for expect
         :return: matched item
         """
-
         def process_expected_item(
-            item_raw: Tuple[str, Callable[..., Any]],
+            item_raw: Tuple[Optional[str], Callable[..., Any]],
         ) -> Dict[str, Any]:
             # convert item raw data to standard dict
             item = {
-                "pattern": item_raw[0] if isinstance(item_raw, tuple) else item_raw,
+                "pattern": item_raw[0] if isinstance(item_raw, tuple) else item_raw or None,
                 "callback": item_raw[1] if isinstance(item_raw, tuple) else None,
                 "index": -1,
                 "ret": None,
             }
             return item
 
-        expect_items_list = [process_expected_item(item) for item in expect_items]
-        expect_patterns = [
+        expect_items_list: List[Dict[str, Any]] = [process_expected_item(item) for item in expect_items if isinstance(item, tuple) and item[0]]
+        expect_patterns: List[Any] = [
             item["pattern"] for item in expect_items_list if item["pattern"] is not None
         ]
-        match_item = None
+        match_item: Optional[Dict[str, Any]] = None
 
         if self.pexpect_proc is not None:
-            match_index = self.pexpect_proc.expect(expect_patterns, timeout)
+            match_index: int = self.pexpect_proc.expect(expect_patterns, timeout)
 
             if isinstance(match_index, int):
                 match_item = expect_items_list[match_index]  # type: ignore
@@ -579,83 +660,86 @@ class ModbusTestDut(IdfDut):
             raise RuntimeError from None
 
     def dut_test_start(
-        self, dictionary: Dict, timeout_value=TEST_EXPECT_STR_TIMEOUT
+        self, dictionary: Dict, timeout_value: Optional[int] = TEST_EXPECT_STR_TIMEOUT
     ) -> None:  # type: ignore
         """The method to initialize and handle test stages"""
 
-        def handle_get_ip4(data: Any) -> None:
+        def handle_get_ip4(data: Optional[Any]=None) -> None:
             """Handle get_ip v4"""
             # Synch for handling the case where the DUT dont stablish connection from the beginning
             # DUT reboot after failing reconnecting
             if self.test_stage.value >= Stages.STACK_IPV4.value:
                 self.logger.info(
-                    "%s: Reboot loop detected on stage [STACK_IPV4], ending test.",
-                    self.app_name,
+                    f"Handle: {self.app_name}: Reboot loop detected on stage [{Stages.STACK_IPV4.name}]: {str(data)}"
                 )
                 self.send_message_destroy_other_dut_instances()
                 self.send_message_destroy_dut()
                 self.test_finish = True
             else:
-                self.logger.info("%s[STACK_IPV4]: %s", self.app_name, str(data))
                 self.test_stage = Stages.STACK_IPV4
+                self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
 
-        def handle_get_ip6(data: Any) -> None:
+        def handle_get_ip6(data: Optional[Any]=None) -> None:
             """Handle get_ip v6"""
-            self.logger.info("%s[STACK_IPV6]: %s", self.app_name, str(data))
             self.test_stage = Stages.STACK_IPV6
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
 
-        def handle_init(data: Any) -> None:
+        def handle_init(data: Optional[Any]=None) -> None:
             """Handle init"""
-            self.logger.info("%s[STACK_INIT]: %s", self.app_name, str(data))
             self.test_stage = Stages.STACK_INIT
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
 
-        def handle_connect(data: Any) -> None:
+        def handle_connect(data: Optional[Any]=None) -> None:
             """Handle connect"""
-            self.logger.info("%s[STACK_CONNECT]: %s", self.app_name, str(data))
             self.test_stage = Stages.STACK_CONNECT
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
 
-        def handle_test_start(data: Any) -> None:
+        def handle_test_start(data: Optional[Any]=None) -> None:
             """Handle connect"""
-            self.logger.info("%s[STACK_START]: %s", self.app_name, str(data))
             self.test_stage = Stages.STACK_START
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
 
-        def handle_cid_response_time(data: Any) -> None:
+        def handle_cid_response_time(data: Optional[Any]=None) -> None:
             """Handle Cid sent request for response time calculation"""
-            self.logger.info(
-                "%s[STACK_CID_RESPONSE_TIME]: %s", self.app_name, str(data)
-            )
             self.test_stage = Stages.STACK_CID_RESPONSE_TIME
+            self.logger.info(
+                f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}"
+            )
             self.add_request_response(
-                data[TRANSACTION_TIMESTAMP], data[OBJ_ADDRESS], data[CID]
+                self.get_item(data, TRANSACTION_TIMESTAMP),
+                self.get_item(data, OBJ_ADDRESS),
+                self.get_item(data, CID)
             )
 
-        def handle_bad_connection(data: Any) -> None:
+        def handle_bad_connection(data: Optional[Any]=None) -> None:
             """Handle bad connection"""
-            self.logger.info(
-                "%s Reached the stage [STACK_BAD_CONNECTION]. Ending the test.",
-                self.app_name,
-            )
             self.test_stage = Stages.STACK_BAD_CONNECTION
+            self.logger.info(
+                f"Handle: {self.app_name}: Reached the stage [{self.test_stage.name}]. Ending the test."
+            )
             self.send_message_destroy_other_dut_instances()
             self.send_message_destroy_dut()
             self.test_finish = True
 
-        def handle_par_ok(data: Any) -> None:
+        def handle_par_ok(data: Optional[Any]=None) -> None:
             """Handle parameter ok"""
-            self.logger.info("%s[READ_PAR_OK]: %s", self.app_name, str(data))
+            self.test_stage = Stages.STACK_PAR_OK
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
             # Checking if MBobject exist in the object list
 
-            object_handle = self.get_object_by_id(data[OBJ_ADDRESS])
+            object_handle: Optional[MbObject] = self.get_object_by_id(self.get_item(data, OBJ_ADDRESS))
             if object_handle is None:
-                object_handle = self.update_wrong_object_id(data[OBJ_ADDRESS])
+                object_handle = self.update_wrong_object_id(self.get_item(data, OBJ_ADDRESS))
+            assert object_handle is not None
 
-            last_sucess_parameter = object_handle.add_parameter(
-                data[PARAM_NAME],
-                data[OBJ_ADDRESS],
-                data[TRANSACTION_TIMESTAMP],
+            last_sucess_parameter: MbParameter = object_handle.add_parameter(
+                self.get_item(data, PARAM_NAME),
+                self.get_item(data, OBJ_ADDRESS),
+                self.get_item(data, TRANSACTION_TIMESTAMP),
                 object_handle.tag,
                 PARAM_SUCCESS,
-                data[CID],
+                self.get_item(data, CID),
+                self.get_item(data, PARAM_VAL)
             )
             if object_handle.is_master():
                 self.add_response_time_to_param(
@@ -666,51 +750,51 @@ class ModbusTestDut(IdfDut):
                 self.logger.info(last_sucess_parameter)
 
             self.param_ok_count += 1
-            self.test_stage = Stages.STACK_PAR_OK
 
-        def handle_par_fail(data: Any) -> None:
+        def handle_par_fail(data: Optional[Any]) -> None:
             """Handle parameter fail"""
-            self.logger.info("%s[READ_PAR_FAIL]: %s", self.app_name, str(data))
+            self.test_stage = Stages.STACK_PAR_FAIL
+            self.logger.info(f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}")
             # Checking if MBobject exist in the object list
-            object_handle = self.get_object_by_id(data[OBJ_ADDRESS])
+            object_handle: Optional[MbObject] = self.get_object_by_id(self.get_item(data, OBJ_ADDRESS))
             if object_handle is None:
-                object_handle = self.update_wrong_object_id(data[OBJ_ADDRESS])
+                object_handle = self.update_wrong_object_id(self.get_item(data, OBJ_ADDRESS))
+            assert object_handle is not None
 
-            last_fail_parameter = object_handle.add_parameter(
-                data[PARAM_NAME],
-                data[OBJ_ADDRESS],
-                data[TRANSACTION_TIMESTAMP],
+            last_fail_parameter: MbParameter = object_handle.add_parameter(
+                self.get_item(data, PARAM_NAME),
+                self.get_item(data, OBJ_ADDRESS),
+                self.get_item(data, TRANSACTION_TIMESTAMP),
                 object_handle.tag,
                 PARAM_FAIL,
-                data[CID],
+                self.get_item(data, CID),
+                None
             )
             self.logger.info(last_fail_parameter)
 
             self.param_fail_count += 1
-            self.test_stage = Stages.STACK_PAR_FAIL
 
-        def handle_obj_create(data: Any) -> None:
+        def handle_obj_create(data: Optional[Any]) -> None:
             """Handle creation"""
-            self.logger.info(
-                "Object creation handled: %s[%s]: %s",
-                self.app_name,
-                Stages.STACK_OBJECT_CREATE.name,
-                str(data),
-            )
-            obj_tag = self.validate_object_creation_tag(data[OBJ_TAG].decode("ascii"))
-
-            last_add_object = self.add_object(
-                obj_tag, data[OBJ_ID], data[TRANSACTION_TIMESTAMP]
-            )
             self.test_stage = Stages.STACK_OBJECT_CREATE
+            self.logger.info(
+                f"Object creation handled: {self.app_name}[{self.test_stage.name}]: {str(data)}",
+            )
+            obj_tag: str = self.validate_object_creation_tag(self.get_item(data, OBJ_TAG).decode("ascii"))
+
+            last_add_object: MbObject = self.add_object(
+                obj_tag,
+                self.get_item(data, OBJ_ID),
+                self.get_item(data, TRANSACTION_TIMESTAMP)
+            )
             self.logger.info("New added object: %s", last_add_object)
 
-        def handle_destroy(data: Any) -> None:
+        def handle_destroy(data: Optional[Any]) -> None:
             """Handle destroy"""
-            self.logger.info(
-                "%s[%s]: %s", self.app_name, Stages.STACK_DESTROY.name, str(data)
-            )
             self.test_stage = Stages.STACK_DESTROY
+            self.logger.info(
+                f"Handle: {self.app_name}[{self.test_stage.name}]: {str(data)}"
+            )
             self.send_message_destroy_other_dut_instances()
             self.test_finish = True
 
@@ -735,10 +819,7 @@ class ModbusTestDut(IdfDut):
                 )
             except pexpect.TIMEOUT:
                 self.logger.info(
-                    "%s, expect timeout on stage %s (%s seconds)",
-                    self.app_name,
-                    self.test_stage.name,
-                    timeout_value,
+                    f"{self.app_name}, expect timeout on stage {self.test_stage.name} ({timeout_value} seconds)"
                 )
                 self.send_message_destroy_other_dut_instances()
                 self.send_message_destroy_dut()
@@ -746,12 +827,12 @@ class ModbusTestDut(IdfDut):
 
     def dut_check_errors(self) -> None:
         """Verify allowed percentage of errors for the dut"""
-        allowed_ok_percentage = (
-            self.param_ok_count / (self.param_ok_count + self.param_fail_count + 1)
-        ) * 100
-        if self.param_ok_count and (
-            allowed_ok_percentage > (100 - ALLOWED_PERCENT_OF_FAILS)
-        ):
+        allowed_ok_percentage : float = 0
+        if self.param_ok_count or self.param_fail_count:
+            allowed_ok_percentage = (
+                self.param_ok_count / (self.param_ok_count + self.param_fail_count) * 100
+            )
+        if allowed_ok_percentage > (100 - ALLOWED_PERCENT_OF_FAILS):
             self.logger.info(
                 "%s: ok_count: %d, fail count: %d",
                 self.app_name,
@@ -775,13 +856,13 @@ class ModbusTestDut(IdfDut):
 
 
 @pytest.fixture
-def case_tester(dut: IdfDut, **kwargs):  # type: ignore
+def case_tester(dut: IdfDut, **kwargs) -> Generator[CaseTester, Any, None]:  # type: ignore
     yield CaseTester(dut, **kwargs)
 
 
 @pytest.fixture(scope="session", autouse=True)
 def session_tempdir() -> str:
-    _tmpdir = os.path.join(
+    _tmpdir: str = os.path.join(
         os.path.dirname(__file__),
         "pytest_embedded_log",
         datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
@@ -850,8 +931,10 @@ def build_dir(
         check_dirs.append(f"build_{config}")
     check_dirs.append("build")
 
+    binary_path = ""
+
     for check_dir in check_dirs:
-        binary_path = os.path.join(app_path, check_dir)
+        binary_path: str = os.path.join(app_path, check_dir)
         if os.path.isdir(binary_path):
             logging.info(f"find valid binary path: {binary_path}")
             return check_dir