"""
This module provides helper functions that will add common information of
Testplan execution to test report.
They could be used directly in testcases or provided to
pre/pose_start/stop hooks.
Also provided is a predefined testsuite that can be included in user's
Multitest directly.
"""
__all__ = [
"DriverLogCollector",
"get_hardware_info",
"log_pwd",
"log_hardware",
"log_cmd",
"log_environment",
"attach_log",
"attach_driver_logs_if_failed",
"clean_runpath_if_passed",
"TestplanExecutionInfo",
]
import logging
import os
import psutil
import shutil
import socket
import sys
from functools import reduce
from typing import Any, Dict, List, Optional
from testplan.common.entity import Environment
from testplan.common.utils.logger import TESTPLAN_LOGGER
from testplan.common.utils.path import pwd
from testplan.testing.multitest import testsuite, testcase
from testplan.testing.result import Result
def _prev_all_green(env: Environment, result: Result) -> bool:
return env.parent.report.passed and reduce( # type: ignore[union-attr]
lambda x, y: x and y.get("passed", True),
result.serialized_entries,
True,
)
[docs]
class DriverLogCollector:
"""
Customizable file collector class used for collecting driver logs.
:param name: Name of the object shown in the report.
:param description: Text description for the assertion.
:param ignore: List of patterns of file name to ignore when
attaching a directory.
:param file_pattern: List of patterns of file name to include when
attaching a directory. (Defaults: "stdout*", "stderr*")
:param recursive: Recursively traverse sub-directories and attach
all files, default is to only attach files in top directory.
:param failure_only: Only collect files on failure.
"""
def __init__(
self,
name: str = "DriverLogCollector",
description: str = "logs",
ignore: Optional[List[str]] = None,
file_pattern: Optional[List[str]] = None,
recursive: bool = True,
failure_only: bool = True,
) -> None:
self.__name__ = name
self.description = description
self.ignore = ignore
self.file_pattern = file_pattern or ["stdout*", "stderr*"]
self.recursive = recursive
self.failure_only = failure_only
def __call__(
self,
env: Environment,
result: Result,
) -> None:
"""
Attaches log files to the report for each driver.
"""
if not _prev_all_green(env, result) or not self.failure_only:
for driver in env:
result.attach(
path=driver.runpath,
description=f"Driver: {driver.name} - {self.description}", # type: ignore[attr-defined]
only=self.file_pattern,
recursive=self.recursive,
ignore=self.ignore,
)
[docs]
def get_hardware_info() -> Dict:
"""
Return a variety of host hardware information.
:return: dictionary of hardware information
"""
data = {
"CPU count": psutil.cpu_count(),
"CPU frequence": str(psutil.cpu_freq()),
"CPU percent": psutil.cpu_percent(interval=1, percpu=True),
"Memory": str(psutil.virtual_memory()),
"Swap": str(psutil.swap_memory()),
"Disk usage": str(psutil.disk_usage(os.getcwd())),
"Net interface addresses": psutil.net_if_addrs(),
"PID": os.getpid(),
}
load_avg = ("N/A", "N/A", "N/A")
try:
load_avg = psutil.getloadavg()
except Exception as exc:
print(exc)
data["Average load"] = dict(
zip(["Over 1 min", "Over 5 min", "Over 15 min"], load_avg)
)
return data
[docs]
def log_hardware(result: Result) -> None:
"""
Saves host hardware information to the report.
:param result: testcase result
"""
result.log(socket.getfqdn(), description="Current Host")
hardware = get_hardware_info()
result.dict.log(hardware, description="Hardware info") # type: ignore[attr-defined]
[docs]
def log_environment(result: Result) -> None:
"""
Saves host environment variable to the report.
:param result: testcase result
"""
result.dict.log( # type: ignore[attr-defined]
dict(os.environ), description="Current environment variable"
)
[docs]
def log_pwd(result: Result) -> None:
"""
Saves current path to the report.
:param result: testcase result
"""
result.log(pwd(), description="PWD environment")
result.log(os.getcwd(), description="Current real path")
[docs]
def log_cmd(result: Result) -> None:
"""
Saves command line arguments to the report.
:param result: testcase result
"""
result.log(sys.argv, description="Command")
result.log(
os.path.abspath(os.path.realpath(sys.argv[0])),
description="Resolved path",
)
[docs]
def attach_log(result: Result) -> None:
"""
Attaches top-level testplan.log file to the report.
:param result: testcase result
"""
log_handlers = TESTPLAN_LOGGER.handlers
for handler in log_handlers:
if isinstance(handler, logging.FileHandler):
result.attach(handler.baseFilename, description="Testplan log")
return
[docs]
def attach_driver_logs_if_failed(
env: Environment,
result: Result,
) -> None:
"""
Attaches stdout and stderr files to the report for each driver.
:param env: environment
:param result: testcase result
"""
stdout_logger = DriverLogCollector(
file_pattern=["stdout*"], description="stdout"
)
stderr_logger = DriverLogCollector(
file_pattern=["stderr*"], description="stderr"
)
stdout_logger(env, result)
stderr_logger(env, result)
[docs]
def clean_runpath_if_passed(
env: Environment,
result: Result,
) -> None:
"""
Deletes multitest-level runpath if the multitest passed.
:param env: environment
:param result: result object
"""
multitest = env.parent
if multitest is None:
return
if _prev_all_green(env, result) and multitest.report.passed: # type: ignore[attr-defined]
for subfile in os.listdir(multitest.runpath):
# TODO: Define scratch as a constant
if subfile != "scratch":
path = os.path.join(
os.path.abspath(multitest.runpath),
subfile,
)
if os.path.isfile(path) or os.path.islink(path):
os.remove(path)
elif os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True)
[docs]
@testsuite
class TestplanExecutionInfo:
"""
Utility testsuite to log generic information of Testplan execution.
"""
[docs]
@testcase
def environment(self, env: Environment, result: Result) -> None:
"""
Environment
"""
log_environment(result)
[docs]
@testcase
def path(self, env: Environment, result: Result) -> None:
"""
Execution path
"""
log_pwd(result)
log_cmd(result)
[docs]
@testcase
def hardware(self, env: Environment, result: Result) -> None:
"""
Host hardware
"""
log_hardware(result)
[docs]
@testcase
def logging(self, env: Environment, result: Result) -> None:
"""
Testplan log
"""
attach_log(result)