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