Add results.csv preparation. Add new logger levels

Author: andremiszcz

tests/validation/Engine/const.py

@@ -7,6 +7,8 @@
 DEFAULT_INPUT_PATH = "/opt/intel/input_path/"
 DEFAULT_OUTPUT_PATH = "/opt/intel/output_path/"
 
+TESTCMD_LVL = 24  # Custom logging level for test commands
+
 # time for establishing connection for example between TX and RX in st2110
 MTL_ESTABLISH_TIMEOUT = 2
 MCM_ESTABLISH_TIMEOUT = 5

tests/validation/Engine/rx_tx_app_engine_mcm.py

@@ -11,7 +11,7 @@
 
 import Engine.rx_tx_app_client_json
 import Engine.rx_tx_app_connection_json
-from Engine.const import LOG_FOLDER, RX_TX_APP_ERROR_KEYWORDS, DEFAULT_OUTPUT_PATH
+from Engine.const import LOG_FOLDER, RX_TX_APP_ERROR_KEYWORDS, DEFAULT_OUTPUT_PATH, TESTCMD_LVL
 from Engine.mcm_apps import (
     get_media_proxy_port,
     output_validator,
@@ -279,6 +279,7 @@ def start(self):
                 f"Starting Tx app with payload: {self.payload.payload_type} on {self.host}"
             )
             cmd = self._get_app_cmd("Tx")
+            logger.log(TESTCMD_LVL, f"Tx command: {cmd}")
             self.process = self.host.connection.start_process(
                 cmd, shell=True, stderr_to_stdout=True, cwd=self.app_path
             )
@@ -334,6 +335,7 @@ def start(self):
                 f"Starting Rx app with payload: {self.payload.payload_type} on {self.host}"
             )
             cmd = self._get_app_cmd("Rx")
+            logger.log(TESTCMD_LVL, f"Rx command: {cmd}")
             self.process = self.host.connection.start_process(
                 cmd, shell=True, stderr_to_stdout=True, cwd=self.app_path
             )

tests/validation/common/ffmpeg_handler/ffmpeg.py

@@ -9,7 +9,7 @@
     SSHRemoteProcessEndException,
     RemoteProcessInvalidState,
 )
-from Engine.const import LOG_FOLDER
+from Engine.const import LOG_FOLDER, TESTCMD_LVL
 from Engine.mcm_apps import save_process_log
 
 from .ffmpeg_io import FFmpegIO
@@ -77,6 +77,7 @@ def __init__(self, host, ffmpeg_instance: FFmpeg, log_path=None):
     def start(self):
         """Starts the FFmpeg process on the host, waits for the process to start."""
         cmd = self.ff.get_command()
+        logger.log(TESTCMD_LVL, f"ffmpeg command: {cmd}")
         ffmpeg_process = self.host.connection.start_process(
             cmd, stderr_to_stdout=True, shell=True
         )

tests/validation/conftest.py

@@ -7,6 +7,7 @@
 import shutil
 from ipaddress import IPv4Interface
 from pathlib import Path
+from typing import Dict
 
 from mfd_host import Host
 from mfd_connect.exceptions import (
@@ -16,6 +17,7 @@
 import pytest
 
 from common.nicctl import Nicctl
+
 from Engine.const import (
     ALLOWED_FFMPEG_VERSIONS,
     DEFAULT_FFMPEG_PATH,
@@ -32,11 +34,35 @@
     MCM_BUILD_PATH,
     MTL_BUILD_PATH,
     OPENH264_VERSION_TAG,
+    TESTCMD_LVL
+)
+from Engine.csv_report import (
+    csv_add_test,
+    csv_write_report,
+    update_compliance_result,
+
 )
 from Engine.mcm_apps import MediaProxy, MeshAgent, get_mcm_path, get_mtl_path
 from datetime import datetime
+from mfd_common_libs.custom_logger import add_logging_level
+from mfd_common_libs.log_levels import TEST_FAIL, TEST_INFO, TEST_PASS
+from pytest_mfd_logging.amber_log_formatter import AmberLogFormatter
+
 
 logger = logging.getLogger(__name__)
+phase_report_key = pytest.StashKey[Dict[str, pytest.CollectReport]]()
+
+
+@pytest.hookimpl(wrapper=True, tryfirst=True)
+def pytest_runtest_makereport(item, call):
+    # execute all other hooks to obtain the report object
+    rep = yield
+
+    # store test results for each phase of a call, which can
+    # be "setup", "call", "teardown"
+    item.stash.setdefault(phase_report_key, {})[rep.when] = rep
+
+    return rep
 
 
 @pytest.fixture(scope="function")
@@ -770,3 +796,89 @@ def log_interface_driver_info(hosts: dict[str, Host]) -> None:
             logger.info(
                 f"Interface {interface.name} on host {host.name} uses driver: {driver_info.driver_name} ({driver_info.driver_version})"
             )
+
+
+@pytest.fixture(scope="session", autouse=True)
+def log_session():
+    add_logging_level("TESTCMD", TESTCMD_LVL)
+
+    today = 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="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)
+    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="n/a",
+        result_note="n/a",
+    )
+
+
+@pytest.fixture(scope="session")
+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")

tests/validation/pytest.ini

@@ -1,8 +1,4 @@
 [pytest]
 pythonpath = .
-log_cli = true
 log_cli_level = info
 log_file = pytest.log
-log_file_level = debug
-log_file_format = %(asctime)s,%(msecs)03d %(levelname)-8s %(filename)s:%(lineno)d %(message)s
-log_file_date_format = %Y-%m-%d %H:%M:%S