1# SPDX-License-Identifier: GPL-2.0-only 2# This file is part of Scapy 3# See https://scapy.net/ for more information 4# Copyright (C) Nils Weiss <nils@we155.de> 5 6# scapy.contrib.description = AutomotiveTestCaseExecutorConfiguration 7# scapy.contrib.status = library 8 9import inspect 10from threading import Event 11 12from scapy.contrib.automotive import log_automotive 13from scapy.contrib.automotive.scanner.graph import Graph 14from scapy.contrib.automotive.scanner.test_case import AutomotiveTestCaseABC 15from scapy.contrib.automotive.scanner.staged_test_case import StagedAutomotiveTestCase # noqa: E501 16 17# Typing imports 18from typing import ( 19 Any, 20 Union, 21 List, 22 Type, 23 Set, 24 cast, 25) 26 27 28class AutomotiveTestCaseExecutorConfiguration(object): 29 """ 30 Configuration storage for AutomotiveTestCaseExecutor. 31 32 The following keywords are used in the AutomotiveTestCaseExecutor: 33 verbose: Enables verbose output and logging 34 debug: Will raise Exceptions on internal errors 35 36 :param test_cases: List of AutomotiveTestCase classes or instances. 37 Classes will get instantiated in this initializer. 38 :param kwargs: Configuration for every AutomotiveTestCase in test_cases 39 and for the AutomotiveTestCaseExecutor. TestCase local 40 configuration and global configuration for all TestCase 41 objects are possible. All keyword arguments given will 42 be stored for every TestCase. To define a local 43 configuration for one TestCase only, the keyword 44 arguments need to be provided in a dictionary. 45 To assign a configuration dictionary to a TestCase, the 46 keyword need to identify the TestCase by the following 47 pattern. 48 ``MyTestCase_kwargs={"someConfig": 42}`` 49 The keyword is composed from the TestCase class name and 50 the postfix '_kwargs'. 51 52 Example: 53 >>> config = AutomotiveTestCaseExecutorConfiguration([MyTestCase], global_config=42, MyTestCase_kwargs={"localConfig": 1337}) # noqa: E501 54 """ 55 def __setitem__(self, key, value): 56 # type: (Any, Any) -> None 57 self.__dict__[key] = value 58 59 def __getitem__(self, key): 60 # type: (Any) -> Any 61 return self.__dict__[key] 62 63 def _generate_test_case_config(self, test_case_cls): 64 # type: (Type[AutomotiveTestCaseABC]) -> None 65 # try to get config from kwargs 66 if test_case_cls in self.test_case_clss: 67 return 68 69 self.test_case_clss.add(test_case_cls) 70 71 kwargs_name = test_case_cls.__name__ + "_kwargs" 72 self.__setattr__(test_case_cls.__name__, self.global_kwargs.pop( 73 kwargs_name, dict())) 74 75 # apply global config 76 val = self.__getattribute__(test_case_cls.__name__) 77 for kwargs_key, kwargs_val in self.global_kwargs.items(): 78 if "_kwargs" in kwargs_key: 79 continue 80 if kwargs_key not in val.keys(): 81 val[kwargs_key] = kwargs_val 82 self.__setattr__(test_case_cls.__name__, val) 83 84 def add_test_case(self, test_case): 85 # type: (Union[AutomotiveTestCaseABC, Type[AutomotiveTestCaseABC], StagedAutomotiveTestCase, Type[StagedAutomotiveTestCase]]) -> None # noqa: E501 86 if inspect.isclass(test_case): 87 test_case_class = cast(Union[Type[AutomotiveTestCaseABC], 88 Type[StagedAutomotiveTestCase]], 89 test_case) 90 if issubclass(test_case_class, StagedAutomotiveTestCase): 91 self.add_test_case(test_case_class()) # type: ignore 92 elif issubclass(test_case_class, AutomotiveTestCaseABC): 93 self.add_test_case(test_case_class()) 94 else: 95 raise TypeError( 96 "Provided class is not in " 97 "Union[Type[AutomotiveTestCaseABC], " 98 "Type[StagedAutomotiveTestCase]]") 99 100 elif isinstance(test_case, AutomotiveTestCaseABC): 101 self.test_cases.append(test_case) 102 self._generate_test_case_config(test_case.__class__) 103 if isinstance(test_case, StagedAutomotiveTestCase): 104 self.stages.append(test_case) 105 for tc in test_case.test_cases: 106 self.staged_test_cases.append(tc) 107 self._generate_test_case_config(tc.__class__) 108 else: 109 raise TypeError( 110 "Provided instance or class of " 111 "StagedAutomotiveTestCase or AutomotiveTestCaseABC") 112 113 def __init__(self, test_cases, **kwargs): 114 # type: (Union[List[Union[AutomotiveTestCaseABC, Type[AutomotiveTestCaseABC]]], List[Type[AutomotiveTestCaseABC]]], Any) -> None # noqa: E501 115 self.verbose = kwargs.get("verbose", False) 116 self.debug = kwargs.get("debug", False) 117 self.unittest = kwargs.pop("unittest", False) 118 self.delay_enter_state = kwargs.pop("delay_enter_state", 0) 119 self.state_graph = Graph() 120 self.test_cases = list() # type: List[AutomotiveTestCaseABC] 121 self.stages = list() # type: List[StagedAutomotiveTestCase] 122 self.staged_test_cases = list() # type: List[AutomotiveTestCaseABC] 123 self.test_case_clss = set() # type: Set[Type[AutomotiveTestCaseABC]] 124 self.stop_event = Event() 125 self.global_kwargs = kwargs 126 self.global_kwargs["stop_event"] = self.stop_event 127 128 for tc in test_cases: 129 self.add_test_case(tc) 130 131 log_automotive.debug("The following configuration was created") 132 log_automotive.debug(self.__dict__) 133 134 def __reduce__(self): # type: ignore 135 f, t, d = super(AutomotiveTestCaseExecutorConfiguration, self).__reduce__() # type: ignore # noqa: E501 136 137 try: 138 del d["tps"] 139 except KeyError: 140 pass 141 142 try: 143 del d["stop_event"] 144 except KeyError: 145 pass 146 147 try: 148 del d["global_kwargs"]["stop_event"] 149 except KeyError: 150 pass 151 152 for tc in d["test_cases"]: 153 try: 154 del d[tc.__class__.__name__]["stop_event"] 155 except KeyError: 156 pass 157 158 for tc in d["staged_test_cases"]: 159 try: 160 del d[tc.__class__.__name__]["stop_event"] 161 except KeyError: 162 pass 163 164 try: 165 del d["global_kwargs"]["stop_event"] 166 except KeyError: 167 pass 168 169 return f, t, d 170