linter fix 3

Author: KarolinaPomian

tests/validation/Engine/csv_report.py

@@ -111,7 +111,9 @@ def update_compliance_result(requirement_id: str, is_compliant: bool) -> None:
     global _compliance_results
 
     _compliance_results[requirement_id] = is_compliant
-    logger.info(f"Updated compliance for {requirement_id}: {'COMPLIANT' if is_compliant else 'NON-COMPLIANT'}")
+    logger.info(
+        f"Updated compliance for {requirement_id}: {'COMPLIANT' if is_compliant else 'NON-COMPLIANT'}"
+    )
 
 
 def csv_write_report(log_dir: str = ".") -> None:
@@ -140,7 +142,9 @@ def csv_write_report(log_dir: str = ".") -> None:
     if _compliance_results:
         try:
             with open(compliance_path, "w", newline="") as csvfile:
-                writer = csv.DictWriter(csvfile, fieldnames=["Requirement ID", "Status"])
+                writer = csv.DictWriter(
+                    csvfile, fieldnames=["Requirement ID", "Status"]
+                )
                 writer.writeheader()
                 for req_id, is_compliant in _compliance_results.items():
                     writer.writerow(
@@ -169,7 +173,9 @@ def get_test_summary() -> Dict[str, int | str]:
         "total": total_count,
         "passed": pass_count,
         "failed": fail_count,
-        "pass_rate": (f"{(pass_count / total_count * 100):.1f}%" if total_count > 0 else "N/A"),
+        "pass_rate": (
+            f"{(pass_count / total_count * 100):.1f}%" if total_count > 0 else "N/A"
+        ),
     }
 
 
@@ -197,7 +203,9 @@ def get_component_summary() -> Dict[str, Dict[str, int | str]]:
     for component in components:
         total = components[component]["total"]
         passed = components[component]["passed"]
-        components[component]["pass_rate"] = f"{(passed / total * 100):.1f}%" if total > 0 else "N/A"
+        components[component]["pass_rate"] = (
+            f"{(passed / total * 100):.1f}%" if total > 0 else "N/A"
+        )
 
     return components
 
@@ -217,5 +225,9 @@ def get_compliance_summary() -> Dict[str, int | str]:
         "total": total_count,
         "compliant": compliant_count,
         "non_compliant": non_compliant_count,
-        "compliance_rate": (f"{(compliant_count / total_count * 100):.1f}%" if total_count > 0 else "N/A"),
+        "compliance_rate": (
+            f"{(compliant_count / total_count * 100):.1f}%"
+            if total_count > 0
+            else "N/A"
+        ),
     }

tests/validation/Engine/logging_utils.py

@@ -77,7 +77,9 @@ def wrapper(*args, **kwargs) -> Any:
     return decorator
 
 
-def log_ffmpeg_test(test_name: str, test_description: str = "", error_message: str = "") -> Callable:
+def log_ffmpeg_test(
+    test_name: str, test_description: str = "", error_message: str = ""
+) -> Callable:
     """Decorator to log FFmpeg test results"""
     return log_test_result(
         component=COMPONENT_FFMPEG,
@@ -87,7 +89,9 @@ def log_ffmpeg_test(test_name: str, test_description: str = "", error_message: s
     )
 
 
-def log_media_proxy_test(test_name: str, test_description: str = "", error_message: str = "") -> Callable:
+def log_media_proxy_test(
+    test_name: str, test_description: str = "", error_message: str = ""
+) -> Callable:
     """Decorator to log Media Proxy test results"""
     return log_test_result(
         component=COMPONENT_MEDIA_PROXY,
@@ -97,7 +101,9 @@ def log_media_proxy_test(test_name: str, test_description: str = "", error_messa
     )
 
 
-def log_mesh_agent_test(test_name: str, test_description: str = "", error_message: str = "") -> Callable:
+def log_mesh_agent_test(
+    test_name: str, test_description: str = "", error_message: str = ""
+) -> Callable:
     """Decorator to log Mesh Agent test results"""
     return log_test_result(
         component=COMPONENT_MESH_AGENT,
@@ -107,7 +113,9 @@ def log_mesh_agent_test(test_name: str, test_description: str = "", error_messag
     )
 
 
-def log_rxtx_app_test(test_name: str, test_description: str = "", error_message: str = "") -> Callable:
+def log_rxtx_app_test(
+    test_name: str, test_description: str = "", error_message: str = ""
+) -> Callable:
     """Decorator to log RxTx App test results"""
     return log_test_result(
         component=COMPONENT_RXTX_APP,
@@ -135,7 +143,9 @@ def parse_output_for_errors(
         logger.warning(f"Unknown component: {component}, using generic error keywords")
         error_keywords = ["error", "failed", "exception", "fatal"]
     else:
-        error_keywords = custom_error_keywords or COMPONENT_ERROR_KEYWORDS.get(component, [])
+        error_keywords = custom_error_keywords or COMPONENT_ERROR_KEYWORDS.get(
+            component, []
+        )
 
     errors = []
     for line in output.splitlines():

tests/validation/Engine/output_validator.py

@@ -54,7 +54,9 @@ def validate_ffmpeg(
         Returns:
             Dictionary with is_pass and errors fields
         """
-        result = parse_output_for_errors(COMPONENT_FFMPEG, output, custom_error_keywords)
+        result = parse_output_for_errors(
+            COMPONENT_FFMPEG, output, custom_error_keywords
+        )
 
         error_message = "\n".join(result["errors"]) if result["errors"] else ""
 
@@ -88,7 +90,9 @@ def validate_media_proxy(
         Returns:
             Dictionary with is_pass and errors fields
         """
-        result = parse_output_for_errors(COMPONENT_MEDIA_PROXY, output, custom_error_keywords)
+        result = parse_output_for_errors(
+            COMPONENT_MEDIA_PROXY, output, custom_error_keywords
+        )
 
         error_message = "\n".join(result["errors"]) if result["errors"] else ""
 
@@ -122,7 +126,9 @@ def validate_mesh_agent(
         Returns:
             Dictionary with is_pass and errors fields
         """
-        result = parse_output_for_errors(COMPONENT_MESH_AGENT, output, custom_error_keywords)
+        result = parse_output_for_errors(
+            COMPONENT_MESH_AGENT, output, custom_error_keywords
+        )
 
         error_message = "\n".join(result["errors"]) if result["errors"] else ""
 
@@ -156,7 +162,9 @@ def validate_rxtx_app(
         Returns:
             Dictionary with is_pass and errors fields
         """
-        result = parse_output_for_errors(COMPONENT_RXTX_APP, output, custom_error_keywords)
+        result = parse_output_for_errors(
+            COMPONENT_RXTX_APP, output, custom_error_keywords
+        )
 
         error_message = "\n".join(result["errors"]) if result["errors"] else ""
 
@@ -203,16 +211,26 @@ def validate_from_log_file(
 
             # Validate output based on component
             if component == COMPONENT_FFMPEG:
-                return self.validate_ffmpeg(output, test_name, duration, custom_error_keywords)
+                return self.validate_ffmpeg(
+                    output, test_name, duration, custom_error_keywords
+                )
             elif component == COMPONENT_MEDIA_PROXY:
-                return self.validate_media_proxy(output, test_name, duration, custom_error_keywords)
+                return self.validate_media_proxy(
+                    output, test_name, duration, custom_error_keywords
+                )
             elif component == COMPONENT_MESH_AGENT:
-                return self.validate_mesh_agent(output, test_name, duration, custom_error_keywords)
+                return self.validate_mesh_agent(
+                    output, test_name, duration, custom_error_keywords
+                )
             elif component == COMPONENT_RXTX_APP:
-                return self.validate_rxtx_app(output, test_name, duration, custom_error_keywords)
+                return self.validate_rxtx_app(
+                    output, test_name, duration, custom_error_keywords
+                )
             else:
                 logger.warning(f"Unknown component: {component}")
-                result = parse_output_for_errors(component, output, custom_error_keywords)
+                result = parse_output_for_errors(
+                    component, output, custom_error_keywords
+                )
 
                 error_message = "\n".join(result["errors"]) if result["errors"] else ""
 

tests/validation/conftest.py

@@ -145,12 +145,16 @@ def mesh_agent(hosts, test_config, log_path):
         mesh_ip = test_config.get("mesh_ip", None)
 
         if not mesh_ip:
-            logger.error(f"Host '{mesh_agent_name}' not found in topology.yaml and no mesh_ip provided.")
+            logger.error(
+                f"Host '{mesh_agent_name}' not found in topology.yaml and no mesh_ip provided."
+            )
             raise RuntimeError(
                 f"No mesh-agent name '{mesh_agent_name}' found in hosts and no mesh_ip provided in test_config."
             )
         else:
-            logger.info(f"Assumed that mesh agent is running, getting IP from topology config: {mesh_ip}")
+            logger.info(
+                f"Assumed that mesh agent is running, getting IP from topology config: {mesh_ip}"
+            )
             mesh_agent.external = True
             mesh_agent.mesh_ip = mesh_ip
             mesh_agent.p = test_config.get("mesh_port", mesh_agent.p)
@@ -223,7 +227,9 @@ def media_config(hosts: dict) -> None:
             if host.topology.extra_info.media_proxy.get("st2110", False):
                 pf_addr = host.network_interfaces[if_idx].pci_address.lspci
                 vfs = nicctl.vfio_list(pf_addr)
-                host.st2110_dev = host.topology.extra_info.media_proxy.get("st2110_dev", None)
+                host.st2110_dev = host.topology.extra_info.media_proxy.get(
+                    "st2110_dev", None
+                )
                 if not host.st2110_dev and not vfs:
                     nicctl.create_vfs(pf_addr)
                     vfs = nicctl.vfio_list(pf_addr)
@@ -238,25 +244,44 @@ def media_config(hosts: dict) -> None:
                         f"Still no VFs on interface {host.network_interfaces[if_idx].pci_address_lspci} even after creating VFs!"
                     )
                 host.vfs = vfs
-                host.st2110_ip = host.topology.extra_info.media_proxy.get("st2110_ip", f"192.168.0.{last_oct}")
+                host.st2110_ip = host.topology.extra_info.media_proxy.get(
+                    "st2110_ip", f"192.168.0.{last_oct}"
+                )
                 if_idx += 1
             if host.topology.extra_info.media_proxy.get("rdma", False):
-                if int(host.network_interfaces[if_idx].virtualization.get_current_vfs()) > 0:
-                    nicctl.disable_vf(str(host.network_interfaces[if_idx].pci_address.lspci))
+                if (
+                    int(
+                        host.network_interfaces[if_idx].virtualization.get_current_vfs()
+                    )
+                    > 0
+                ):
+                    nicctl.disable_vf(
+                        str(host.network_interfaces[if_idx].pci_address.lspci)
+                    )
                 net_adap_ips = host.network_interfaces[if_idx].ip.get_ips().v4
                 rdma_ip = host.topology.extra_info.media_proxy.get("rdma_ip", False)
                 if rdma_ip:
-                    rdma_ip = IPv4Interface(f"{rdma_ip}" if "/" in rdma_ip else f"{rdma_ip}/24")
+                    rdma_ip = IPv4Interface(
+                        f"{rdma_ip}" if "/" in rdma_ip else f"{rdma_ip}/24"
+                    )
                 elif net_adap_ips and not rdma_ip:
                     rdma_ip = net_adap_ips[0]
                 if not rdma_ip or (rdma_ip not in net_adap_ips):
-                    rdma_ip = IPv4Interface(f"192.168.1.{last_oct}/24") if not rdma_ip else rdma_ip
-                    logger.info(f"IP {rdma_ip} not found on RDMA network interface, setting: {rdma_ip}")
+                    rdma_ip = (
+                        IPv4Interface(f"192.168.1.{last_oct}/24")
+                        if not rdma_ip
+                        else rdma_ip
+                    )
+                    logger.info(
+                        f"IP {rdma_ip} not found on RDMA network interface, setting: {rdma_ip}"
+                    )
                     host.network_interfaces[if_idx].ip.add_ip(rdma_ip)
                 host.rdma_ip = str(rdma_ip.ip)
             logger.info(f"VFs on {host.name} are: {host.vfs}")
         except IndexError:
-            raise IndexError(f"Not enough network adapters available for tests! Expected: {if_idx+1}")
+            raise IndexError(
+                f"Not enough network adapters available for tests! Expected: {if_idx+1}"
+            )
         except AttributeError:
             logger.warning(
                 f"Extra info media proxy in topology config for {host.name} is not set, skipping media config setup for this host."
@@ -282,10 +307,16 @@ def cleanup_processes(hosts: dict) -> None:
             try:
                 connection = host.connection
                 # connection.enable_sudo()
-                connection.execute_command(f"pgrep -f '{pattern}'", stderr_to_stdout=True)
-                connection.execute_command(f"pkill -9 -f '{pattern}'", stderr_to_stdout=True)
+                connection.execute_command(
+                    f"pgrep -f '{pattern}'", stderr_to_stdout=True
+                )
+                connection.execute_command(
+                    f"pkill -9 -f '{pattern}'", stderr_to_stdout=True
+                )
             except Exception as e:
-                logger.warning(f"Failed to check/kill processes matching {pattern} on {host.name}: {e}")
+                logger.warning(
+                    f"Failed to check/kill processes matching {pattern} on {host.name}: {e}"
+                )
     logger.info("Cleanup of processes completed.")
 
 
@@ -311,15 +342,19 @@ def check_iommu(hosts: dict[str, Host]) -> None:
     iommu_not_enabled_hosts = []
     for host in hosts.values():
         try:
-            output = host.connection.execute_command("ls -1 /sys/kernel/iommu_groups | wc -l", shell=True, timeout=10)
+            output = host.connection.execute_command(
+                "ls -1 /sys/kernel/iommu_groups | wc -l", shell=True, timeout=10
+            )
             if int(output.stdout.strip()) == 0:
                 logger.error(f"IOMMU is not enabled on host {host.name}.")
                 iommu_not_enabled_hosts.append(host.name)
         except Exception as e:
             logger.exception(f"Failed to check IOMMU status on host {host.name}.")
             iommu_not_enabled_hosts.append(host.name)
     if iommu_not_enabled_hosts:
-        pytest.exit(f"IOMMU is not enabled on hosts: {', '.join(iommu_not_enabled_hosts)}. Aborting test session.")
+        pytest.exit(
+            f"IOMMU is not enabled on hosts: {', '.join(iommu_not_enabled_hosts)}. Aborting test session."
+        )
     else:
         logger.info("IOMMU is enabled on all hosts.")
 
@@ -332,14 +367,20 @@ def enable_hugepages(hosts: dict[str, Host]) -> None:
     for host in hosts.values():
         if not _check_hugepages(host):
             try:
-                host.connection.execute_command("sudo sysctl -w vm.nr_hugepages=2048", shell=True, timeout=10)
+                host.connection.execute_command(
+                    "sudo sysctl -w vm.nr_hugepages=2048", shell=True, timeout=10
+                )
                 logger.info(f"Hugepages enabled on host {host.name}.")
             except (RemoteProcessTimeoutExpired, ConnectionCalledProcessError):
                 logger.exception(f"Failed to enable hugepages on host {host.name}.")
-                pytest.exit(f"Failed to enable hugepages on host {host.name}. Aborting test session.")
+                pytest.exit(
+                    f"Failed to enable hugepages on host {host.name}. Aborting test session."
+                )
             if not _check_hugepages(host):
                 logger.error(f"Hugepages could not be enabled on host {host.name}.")
-                pytest.exit(f"Hugepages could not be enabled on host {host.name}. Aborting test session.")
+                pytest.exit(
+                    f"Hugepages could not be enabled on host {host.name}. Aborting test session."
+                )
         else:
             logger.info(f"Hugepages are already enabled on host {host.name}.")
 
@@ -349,7 +390,9 @@ def _check_hugepages(host: Host) -> bool:
     Check if hugepages are enabled on the host.
     """
     try:
-        output = host.connection.execute_command("cat /proc/sys/vm/nr_hugepages", shell=True, timeout=10)
+        output = host.connection.execute_command(
+            "cat /proc/sys/vm/nr_hugepages", shell=True, timeout=10
+        )
         return int(output.stdout.strip()) > 0
     except (RemoteProcessTimeoutExpired, ConnectionCalledProcessError):
         logger.exception(f"Failed to check hugepages status on host {host.name}.")
@@ -395,7 +438,9 @@ def log_case(request, caplog: pytest.LogCaptureFixture):
     os.makedirs(os.path.join(LOG_FOLDER, "latest", case_folder), exist_ok=True)
     logfile = os.path.join(LOG_FOLDER, "latest", f"{case_id}.log")
     fh = logging.FileHandler(logfile)
-    formatter = request.session.config.pluginmanager.get_plugin("logging-plugin").formatter
+    formatter = request.session.config.pluginmanager.get_plugin(
+        "logging-plugin"
+    ).formatter
     format = AmberLogFormatter(formatter)
     fh.setFormatter(format)
     fh.setLevel(logging.DEBUG)