File: test.py

package info (click to toggle)
dpdk 25.11-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 127,892 kB
  • sloc: ansic: 2,358,479; python: 16,426; sh: 4,474; makefile: 1,713; awk: 70
file content (158 lines) | stat: -rw-r--r-- 4,886 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2025 Arm Limited

"""Common test utilities.

This module provides utility functions for test cases, including logging, verification.
"""

import json
from datetime import datetime

from api.artifact import Artifact
from framework.context import get_ctx
from framework.exception import InternalError, SkippedTestException, TestCaseVerifyError
from framework.logger import DTSLogger


def get_current_test_case_name() -> str:
    """The name of the current test case.

    Raises:
        InternalError: If no test is running.
    """
    current_test_case = get_ctx().local.current_test_case
    if current_test_case is None:
        raise InternalError("No current test case")
    return current_test_case.name


def log(message: str) -> None:
    """Log the given message with the level 'INFO'.

    Args:
        message: String representing the message to log.

    Raises:
        InternalError: If no test is running.
    """
    get_logger().info(message)


def log_debug(message: str) -> None:
    """Log the given message with the level 'DEBUG'.

    Args:
        message: String representing the message to log.

    Raises:
        InternalError: If no test is running.
    """
    get_logger().debug(message)


def verify(condition: bool, failure_description: str) -> None:
    """Verify `condition` and handle failures.

    When `condition` is :data:`False`, raise an exception and log the last 10 commands
    executed on both the SUT and TG.

    Args:
        condition: The condition to check.
        failure_description: A short description of the failure
            that will be stored in the raised exception.

    Raises:
        TestCaseVerifyError: If `condition` is :data:`False`.
    """
    if not condition:
        fail(failure_description)


def verify_else_skip(condition: bool, skip_reason: str) -> None:
    """Verify `condition` and handle skips.

    When `condition` is :data:`False`, raise a skip exception.

    Args:
        condition: The condition to check.
        skip_reason: Description of the reason for skipping.

    Raises:
        SkippedTestException: If `condition` is :data:`False`.
    """
    if not condition:
        skip(skip_reason)


def skip(skip_description: str) -> None:
    """Skip the current test case or test suite with a given description.

    Args:
        skip_description: Description of the reason for skipping.

    Raises:
        SkippedTestException: Always raised to indicate the test was skipped.
    """
    get_logger().debug(f"Test skipped: {skip_description}")
    raise SkippedTestException(skip_description)


def fail(failure_description: str) -> None:
    """Fail the current test case with a given description.

    Logs the last 10 commands executed on both the SUT and TG before raising an exception.

    Args:
        failure_description: Description of the reason for failure.

    Raises:
        TestCaseVerifyError: Always raised to indicate the test case failed.
    """
    get_logger().debug("A test case failed, showing the last 10 commands executed on SUT:")
    for command_res in get_ctx().sut_node.main_session.remote_session.history[-10:]:
        get_logger().debug(command_res.command)
    get_logger().debug("A test case failed, showing the last 10 commands executed on TG:")
    for command_res in get_ctx().tg_node.main_session.remote_session.history[-10:]:
        get_logger().debug(command_res.command)
    raise TestCaseVerifyError(failure_description)


def get_logger() -> DTSLogger:
    """Get a logger instance for tests.

    Raises:
        InternalError: If no test is running.
    """
    current_test_suite = get_ctx().local.current_test_suite
    if current_test_suite is None:
        raise InternalError("No current test suite")
    return current_test_suite._logger


def write_performance_json(
    performance_data: dict, filename: str = "performance_metrics.json"
) -> None:
    """Write performance test results to a JSON file in the test suite's output directory.

    This method creates a JSON file containing performance metrics in the test suite's
    output directory. The data can be a dictionary of any structure. No specific format
    is required.

    Args:
        performance_data: Dictionary containing performance metrics and results.
        filename: Name of the JSON file to create.

    Raises:
        InternalError: If performance data is not provided.
    """
    if not performance_data:
        raise InternalError("No performance data to write")

    perf_data = {"timestamp": datetime.now().isoformat(), **performance_data}
    perf_json_artifact = Artifact("local", filename)

    with perf_json_artifact.open("w") as json_file:
        json.dump(perf_data, json_file, indent=2)

    get_logger().info(f"Performance results written to: {perf_json_artifact.local_path}")