common.py

import os
import allure
import pytest
import subprocess
import signal
import pandas

g_cur_path = os.path.dirname(__file__)
g_main_path = os.path.abspath(os.path.join(g_cur_path, "../../"))
g_log_path = "{}/log/".format(g_cur_path)

g_timeout_factor = int(os.getenv("TIMEOUT_FACTOR", "1"))
g_dtk_version = os.getenv("DTK_VERSION", "dtk-24.04.1")
g_test_type = os.getenv("TEST_TYPE", "release")

def clean_log(log_path):
    if os.path.exists(log_path):
        try:
            result = subprocess.run(["rm", "-rf", log_path], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        except subprocess.CalledProcessError as e:
            print(f"An error occurred while trying to remove log file: {e}")

if not os.path.exists(g_log_path):
    try:
        result = subprocess.run(["mkdir", "-p", g_log_path], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except subprocess.CalledProcessError as e:
        print(f"An error occurred while trying to create directory: {e}")


def tear_down(process_handle, timeout):
    try:
        process_handle.communicate(timeout=timeout * g_timeout_factor)
        proc_status = process_handle.poll()
        if 0 != proc_status:
            print("process run error, ret code: {}".format(proc_status))
            return False
    except subprocess.TimeoutExpired:
        os.killpg(process_handle.pid, signal.SIGKILL)
        print("run timout!")
        return False
    return True

def analysis_result(log_file):
    b_pass = True
    with open(log_file, "r") as f:
        for eachline in f.readlines():
            if "FAIL" in eachline or "fail" in eachline:
                b_pass = False
                break
            if "ERROR" in eachline or "error" in eachline:
                b_pass = False
                break
            if "test skip" in eachline:
                allure.attach.file(log_file, name="run_log", attachment_type=allure.attachment_type.TEXT)
                pytest.skip(eachline)
                break
    return b_pass


def run_command(command, log_file, timeout=10):
    cmd_to_log = "echo '{}' 2>&1 >>{}; {} 2>&1 >>{}".format(command, log_file, command, log_file)
    try:
        process_handle = subprocess.Popen(
            cmd_to_log, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        result = tear_down(process_handle, timeout)
        status = analysis_result(log_file)
        if status == False:
            allure.attach.file(log_file, name="TEST LOG", attachment_type=allure.attachment_type.TEXT)
            assert False, "run failed!"
    except subprocess.TimeoutExpired:
        os.killpg(process_handle.pid, signal.SIGKILL)
        allure.attach.file(log_file, name="TEST LOG", attachment_type=allure.attachment_type.TEXT)
        assert False, "timeout!"


def gen_test_suites_from_csv(csv_file, test_suites):
    pd_file = pandas.read_csv(csv_file)
    pd_data = pandas.DataFrame(pd_file)
    for index in pd_data.index:
        eachline = pd_data.loc[index]
        if g_test_type is None or g_test_type in eachline["type"]:
            dc_tmp = {"case": eachline["case"],
                      "suite": eachline["suite"],
                      "run_dir": eachline["run_dir"],
                      "run_cmd": eachline["run_cmd"],
                      "timeout": eachline["timeout"]}

            if pandas.isnull(eachline["run_args"]):
                dc_tmp["run_args"] = ""
            else:
                dc_tmp["run_args"] = eachline["run_args"]

            if dc_tmp["suite"] in test_suites.keys(): 
                test_suites[dc_tmp["suite"]][dc_tmp["case"]] = dc_tmp
            else:
                test_suites[dc_tmp["suite"]] = {}
                test_suites[dc_tmp["suite"]][dc_tmp["case"]] = dc_tmp