1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2010-2014 Intel Corporation 3# Copyright(c) 2022-2023 PANTHEON.tech s.r.o. 4# Copyright(c) 2022-2023 University of New Hampshire 5 6import atexit 7import json 8import os 9import subprocess 10import sys 11from enum import Enum 12from pathlib import Path 13from subprocess import SubprocessError 14 15from scapy.packet import Packet # type: ignore[import] 16 17from .exception import ConfigurationError 18 19 20class StrEnum(Enum): 21 @staticmethod 22 def _generate_next_value_(name: str, start: int, count: int, last_values: object) -> str: 23 return name 24 25 def __str__(self) -> str: 26 return self.name 27 28 29REGEX_FOR_PCI_ADDRESS = "/[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}.[0-9]{1}/" 30 31 32def check_dts_python_version() -> None: 33 if sys.version_info.major < 3 or (sys.version_info.major == 3 and sys.version_info.minor < 10): 34 print( 35 RED( 36 ( 37 "WARNING: DTS execution node's python version is lower than" 38 "python 3.10, is deprecated and will not work in future releases." 39 ) 40 ), 41 file=sys.stderr, 42 ) 43 print(RED("Please use Python >= 3.10 instead"), file=sys.stderr) 44 45 46def expand_range(range_str: str) -> list[int]: 47 """ 48 Process range string into a list of integers. There are two possible formats: 49 n - a single integer 50 n-m - a range of integers 51 52 The returned range includes both n and m. Empty string returns an empty list. 53 """ 54 expanded_range: list[int] = [] 55 if range_str: 56 range_boundaries = range_str.split("-") 57 # will throw an exception when items in range_boundaries can't be converted, 58 # serving as type check 59 expanded_range.extend(range(int(range_boundaries[0]), int(range_boundaries[-1]) + 1)) 60 61 return expanded_range 62 63 64def get_packet_summaries(packets: list[Packet]): 65 if len(packets) == 1: 66 packet_summaries = packets[0].summary() 67 else: 68 packet_summaries = json.dumps(list(map(lambda pkt: pkt.summary(), packets)), indent=4) 69 return f"Packet contents: \n{packet_summaries}" 70 71 72def RED(text: str) -> str: 73 return f"\u001B[31;1m{str(text)}\u001B[0m" 74 75 76class MesonArgs(object): 77 """ 78 Aggregate the arguments needed to build DPDK: 79 default_library: Default library type, Meson allows "shared", "static" and "both". 80 Defaults to None, in which case the argument won't be used. 81 Keyword arguments: The arguments found in meson_options.txt in root DPDK directory. 82 Do not use -D with them, for example: 83 meson_args = MesonArgs(enable_kmods=True). 84 """ 85 86 _default_library: str 87 88 def __init__(self, default_library: str | None = None, **dpdk_args: str | bool): 89 self._default_library = f"--default-library={default_library}" if default_library else "" 90 self._dpdk_args = " ".join( 91 ( 92 f"-D{dpdk_arg_name}={dpdk_arg_value}" 93 for dpdk_arg_name, dpdk_arg_value in dpdk_args.items() 94 ) 95 ) 96 97 def __str__(self) -> str: 98 return " ".join(f"{self._default_library} {self._dpdk_args}".split()) 99 100 101class _TarCompressionFormat(StrEnum): 102 """Compression formats that tar can use. 103 104 Enum names are the shell compression commands 105 and Enum values are the associated file extensions. 106 """ 107 108 gzip = "gz" 109 compress = "Z" 110 bzip2 = "bz2" 111 lzip = "lz" 112 lzma = "lzma" 113 lzop = "lzo" 114 xz = "xz" 115 zstd = "zst" 116 117 118class DPDKGitTarball(object): 119 """Create a compressed tarball of DPDK from the repository. 120 121 The DPDK version is specified with git object git_ref. 122 The tarball will be compressed with _TarCompressionFormat, 123 which must be supported by the DTS execution environment. 124 The resulting tarball will be put into output_dir. 125 126 The class supports the os.PathLike protocol, 127 which is used to get the Path of the tarball:: 128 129 from pathlib import Path 130 tarball = DPDKGitTarball("HEAD", "output") 131 tarball_path = Path(tarball) 132 133 Arguments: 134 git_ref: A git commit ID, tag ID or tree ID. 135 output_dir: The directory where to put the resulting tarball. 136 tar_compression_format: The compression format to use. 137 """ 138 139 _git_ref: str 140 _tar_compression_format: _TarCompressionFormat 141 _tarball_dir: Path 142 _tarball_name: str 143 _tarball_path: Path | None 144 145 def __init__( 146 self, 147 git_ref: str, 148 output_dir: str, 149 tar_compression_format: _TarCompressionFormat = _TarCompressionFormat.xz, 150 ): 151 self._git_ref = git_ref 152 self._tar_compression_format = tar_compression_format 153 154 self._tarball_dir = Path(output_dir, "tarball") 155 156 self._get_commit_id() 157 self._create_tarball_dir() 158 159 self._tarball_name = ( 160 f"dpdk-tarball-{self._git_ref}.tar.{self._tar_compression_format.value}" 161 ) 162 self._tarball_path = self._check_tarball_path() 163 if not self._tarball_path: 164 self._create_tarball() 165 166 def _get_commit_id(self) -> None: 167 result = subprocess.run( 168 ["git", "rev-parse", "--verify", self._git_ref], 169 text=True, 170 capture_output=True, 171 ) 172 if result.returncode != 0: 173 raise ConfigurationError( 174 f"{self._git_ref} is neither a path to an existing DPDK " 175 "archive nor a valid git reference.\n" 176 f"Command: {result.args}\n" 177 f"Stdout: {result.stdout}\n" 178 f"Stderr: {result.stderr}" 179 ) 180 self._git_ref = result.stdout.strip() 181 182 def _create_tarball_dir(self) -> None: 183 os.makedirs(self._tarball_dir, exist_ok=True) 184 185 def _check_tarball_path(self) -> Path | None: 186 if self._tarball_name in os.listdir(self._tarball_dir): 187 return Path(self._tarball_dir, self._tarball_name) 188 return None 189 190 def _create_tarball(self) -> None: 191 self._tarball_path = Path(self._tarball_dir, self._tarball_name) 192 193 atexit.register(self._delete_tarball) 194 195 result = subprocess.run( 196 'git -C "$(git rev-parse --show-toplevel)" archive ' 197 f'{self._git_ref} --prefix="dpdk-tarball-{self._git_ref + os.sep}" | ' 198 f"{self._tar_compression_format} > {Path(self._tarball_path.absolute())}", 199 shell=True, 200 text=True, 201 capture_output=True, 202 ) 203 204 if result.returncode != 0: 205 raise SubprocessError( 206 f"Git archive creation failed with exit code {result.returncode}.\n" 207 f"Command: {result.args}\n" 208 f"Stdout: {result.stdout}\n" 209 f"Stderr: {result.stderr}" 210 ) 211 212 atexit.unregister(self._delete_tarball) 213 214 def _delete_tarball(self) -> None: 215 if self._tarball_path and os.path.exists(self._tarball_path): 216 os.remove(self._tarball_path) 217 218 def __fspath__(self): 219 return str(self._tarball_path) 220