Add loging into file. Update results with compliance results fixture (#1214)

Update logging in tests with correct format. Fix some RC checking
problems.
Update fixture for results.csv preparation. 
Add template fixture for compliance results update"

Author: andremiszcz

tests/validation/conftest.py

@@ -1,15 +1,34 @@
 # # SPDX-License-Identifier: BSD-3-Clause
 # # Copyright 2024-2025 Intel Corporation
 # # Media Communications Mesh
+import datetime
 import logging
 import os
+import shutil
 import time
 from typing import Dict
 
 import pytest
 from common.nicctl import Nicctl
 from create_pcap_file.ramdisk import RamdiskPreparer
-from mtl_engine.stash import clear_result_media, remove_result_media
+from mfd_common_libs.custom_logger import add_logging_level
+from mfd_common_libs.log_levels import TEST_FAIL, TEST_INFO, TEST_PASS
+from mtl_engine.const import LOG_FOLDER, TESTCMD_LVL
+from mtl_engine.csv_report import (
+    csv_add_test,
+    csv_write_report,
+    update_compliance_result,
+)
+from mtl_engine.stash import (
+    clear_issue,
+    clear_result_log,
+    clear_result_media,
+    clear_result_note,
+    get_issue,
+    get_result_note,
+    remove_result_media,
+)
+from pytest_mfd_logging.amber_log_formatter import AmberLogFormatter
 
 logger = logging.getLogger(__name__)
 phase_report_key = pytest.StashKey[Dict[str, pytest.CollectReport]]()
@@ -88,32 +107,14 @@ def dma_port_list(request):
 
 
 @pytest.fixture(scope="session")
-def nic_port_list(hosts: dict, mtl_path) -> list:
+def nic_port_list(hosts: dict, mtl_path) -> None:
     for host in hosts.values():
-        # connection = host.connection
-        # connection.enable_sudo()
         nicctl = Nicctl(mtl_path, host)
         if int(host.network_interfaces[0].virtualization.get_current_vfs()) == 0:
             vfs = nicctl.create_vfs(host.network_interfaces[0].pci_address)
         vfs = nicctl.vfio_list()
         # Store VFs on the host object for later use
         host.vfs = vfs
-        # connection.disable_sudo()
-
-
-# def nic_port_list(request, media_config, hosts: dict = None) -> list:
-# vfs = []
-# if hosts:
-#     for host in hosts.values():
-#         if hasattr(host, "vfs"):
-#             vfs.extend(host.vfs)
-# if vfs:
-#     return vfs
-# # Fallback: use --nic parameter
-# nic_option = request.config.getoption("--nic")
-# if nic_option:
-#     return [nic.strip() for nic in nic_option.split(",") if nic.strip()]
-# raise RuntimeError("No VFs found and --nic parameter not provided!")
 
 
 @pytest.fixture(scope="session")
@@ -164,3 +165,94 @@ def pytest_addoption(parser):
     parser.addoption("--nic", help="list of PCI IDs of network devices")
     parser.addoption("--dma", help="list of PCI IDs of DMA devices")
     parser.addoption("--time", help="seconds to run every test (default=15)")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def log_session():
+    add_logging_level("TESTCMD", TESTCMD_LVL)
+
+    today = datetime.datetime.today()
+    folder = today.strftime("%Y-%m-%dT%H:%M:%S")
+    path = os.path.join(LOG_FOLDER, folder)
+    path_symlink = os.path.join(LOG_FOLDER, "latest")
+    try:
+        os.remove(path_symlink)
+    except FileNotFoundError:
+        pass
+    os.makedirs(path, exist_ok=True)
+    os.symlink(folder, path_symlink)
+    yield
+    shutil.copy("pytest.log", f"{LOG_FOLDER}/latest/pytest.log")
+    csv_write_report(f"{LOG_FOLDER}/latest/report.csv")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def compliance_report(request, log_session, test_config):
+    """
+    This function is used for compliance check and report.
+    """
+    # TODO: Implement compliance check logic. When tcpdump pcap is enabled, at the end of the test session all pcaps
+    # shall be send into EBU list.
+    # Pcaps shall be stored in the ramdisk, and then moved to the compliance
+    # folder or send into EBU list after each test finished and remove it from the ramdisk.
+    # Compliance report generation logic goes here after yield. Or in another class / function but triggered here.
+    # AFAIK names of pcaps contains test name so it can be matched with result of each test like in code below.
+    yield
+    if test_config.get("compliance", False):
+        logging.info("Compliance mode enabled, updating compliance results")
+        for item in request.session.items:
+            test_case = item.nodeid
+            update_compliance_result(test_case, "Fail")
+
+
+@pytest.fixture(scope="function", autouse=True)
+def log_case(request, caplog: pytest.LogCaptureFixture):
+    case_id = request.node.nodeid
+    case_folder = os.path.dirname(case_id)
+    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
+    format = AmberLogFormatter(formatter)
+    fh.setFormatter(format)
+    fh.setLevel(logging.DEBUG)
+    logger = logging.getLogger()
+    logger.addHandler(fh)
+    clear_result_log()
+    clear_issue()
+    yield
+    report = request.node.stash[phase_report_key]
+    if report["setup"].failed:
+        logging.log(level=TEST_FAIL, msg=f"Setup failed for {case_id}")
+        os.chmod(logfile, 0o4755)
+        result = "Fail"
+    elif ("call" not in report) or report["call"].failed:
+        logging.log(level=TEST_FAIL, msg=f"Test failed for {case_id}")
+        os.chmod(logfile, 0o4755)
+        result = "Fail"
+    elif report["call"].passed:
+        logging.log(level=TEST_PASS, msg=f"Test passed for {case_id}")
+        os.chmod(logfile, 0o755)
+        result = "Pass"
+    else:
+        logging.log(level=TEST_INFO, msg=f"Test skipped for {case_id}")
+        result = "Skip"
+
+    logger.removeHandler(fh)
+
+    commands = []
+    for record in caplog.get_records("call"):
+        if record.levelno == TESTCMD_LVL:
+            commands.append(record.message)
+
+    csv_add_test(
+        test_case=case_id,
+        commands="\n".join(commands),
+        result=result,
+        issue=get_issue(),
+        result_note=get_result_note(),
+    )
+
+    clear_result_note()

tests/validation/mtl_engine/GstreamerApp.py

@@ -2,12 +2,15 @@
 # Copyright(c) 2024-2025 Intel Corporation
 
 import hashlib
+import logging
 import os
 import time
 
 from mtl_engine.RxTxApp import prepare_tcpdump
 
-from .execute import log_fail, log_info, run
+from .execute import log_fail, run
+
+logger = logging.getLogger(__name__)
 
 
 def create_connection_params(
@@ -322,8 +325,8 @@ def execute_test(
 
     remote_host = host
 
-    log_info(f"TX Command: {' '.join(tx_command)}")
-    log_info(f"RX Command: {' '.join(rx_command)}")
+    logger.info(f"TX Command: {' '.join(tx_command)}")
+    logger.info(f"RX Command: {' '.join(rx_command)}")
 
     tx_process = None
     rx_process = None
@@ -332,7 +335,7 @@ def execute_test(
     try:
         if tx_first:
             # Start TX pipeline first
-            log_info("Starting TX pipeline...")
+            logger.info("Starting TX pipeline...")
             tx_process = run(
                 " ".join(tx_command),
                 cwd=build,
@@ -345,7 +348,7 @@ def execute_test(
             time.sleep(sleep_interval)
 
             # Start RX pipeline
-            log_info("Starting RX pipeline...")
+            logger.info("Starting RX pipeline...")
             rx_process = run(
                 " ".join(rx_command),
                 cwd=build,
@@ -357,7 +360,7 @@ def execute_test(
             )
         else:
             # Start RX pipeline first
-            log_info("Starting RX pipeline...")
+            logger.info("Starting RX pipeline...")
             rx_process = run(
                 " ".join(rx_command),
                 cwd=build,
@@ -370,7 +373,7 @@ def execute_test(
             time.sleep(sleep_interval)
 
             # Start TX pipeline
-            log_info("Starting TX pipeline...")
+            logger.info("Starting TX pipeline...")
             tx_process = run(
                 " ".join(tx_command),
                 cwd=build,
@@ -382,15 +385,15 @@ def execute_test(
             )
         # --- Start tcpdump after pipelines are running ---
         if tcpdump:
-            log_info("Starting tcpdump capture...")
+            logger.info("Starting tcpdump capture...")
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
 
         # Let the test run for the specified duration
-        log_info(f"Running test for {test_time} seconds...")
+        logger.info(f"Running test for {test_time} seconds...")
         time.sleep(test_time)
 
         # Terminate processes gracefully
-        log_info("Terminating processes...")
+        logger.info("Terminating processes...")
         if tx_process:
             try:
                 tx_process.terminate()
@@ -410,17 +413,17 @@ def execute_test(
             if rx_process and hasattr(rx_process, "stdout_text"):
                 output_rx = rx_process.stdout_text.splitlines()
                 for line in output_rx:
-                    log_info(f"RX Output: {line}")
+                    logger.info(f"RX Output: {line}")
         except Exception:
-            log_info("Could not retrieve RX output")
+            logger.info("Could not retrieve RX output")
 
         try:
             if tx_process and hasattr(tx_process, "stdout_text"):
                 output_tx = tx_process.stdout_text.splitlines()
                 for line in output_tx:
-                    log_info(f"TX Output: {line}")
+                    logger.info(f"TX Output: {line}")
         except Exception:
-            log_info("Could not retrieve TX output")
+            logger.info("Could not retrieve TX output")
 
     except Exception as e:
         log_fail(f"Error during test execution: {e}")
@@ -444,7 +447,7 @@ def execute_test(
 
     # Compare files for validation
     file_compare = compare_files(input_file, output_file)
-    log_info(f"File comparison: {file_compare}")
+    logger.info(f"File comparison: {file_compare}")
 
     return file_compare
 
@@ -473,17 +476,17 @@ def compare_files(input_file, output_file):
     if os.path.exists(input_file) and os.path.exists(output_file):
         input_file_size = os.path.getsize(input_file)
         output_file_size = os.path.getsize(output_file)
-        log_info(f"Input file size: {input_file_size}")
-        log_info(f"Output file size: {output_file_size}")
+        logger.info(f"Input file size: {input_file_size}")
+        logger.info(f"Output file size: {output_file_size}")
         if input_file_size != output_file_size:
             log_fail("File size is different")
             return False
 
         with open(input_file, "rb") as i_file, open(output_file, "rb") as o_file:
             i_hash = hashlib.md5(i_file.read()).hexdigest()
             o_hash = hashlib.md5(o_file.read()).hexdigest()
-            log_info(f"Input file hash: {i_hash}")
-            log_info(f"Output file hash: {o_hash}")
+            logger.info(f"Input file hash: {i_hash}")
+            logger.info(f"Output file hash: {o_hash}")
             if i_hash == o_hash:
                 return True