A Zippy test, consisting of a sequence of actions.
| 176 | |
| 177 | |
| 178 | class Test: |
| 179 | """A Zippy test, consisting of a sequence of actions.""" |
| 180 | |
| 181 | def __init__( |
| 182 | self, scenario: "Scenario", actions: int, max_execution_time: timedelta |
| 183 | ) -> None: |
| 184 | """Generate a new Zippy test. |
| 185 | |
| 186 | Args: |
| 187 | scenario: The Scenario to pick actions from. |
| 188 | actions: The number of actions to generate. |
| 189 | """ |
| 190 | self._scenario = scenario |
| 191 | self._actions: list[Action] = [] |
| 192 | self._final_actions: list[Action] = [] |
| 193 | self._capabilities = Capabilities() |
| 194 | self._actions_with_weight: dict[ActionOrFactory, float] = ( |
| 195 | self._scenario.actions_with_weight() |
| 196 | ) |
| 197 | self._state = State() |
| 198 | self._max_execution_time: timedelta = max_execution_time |
| 199 | |
| 200 | for action_or_factory in self._scenario.bootstrap(): |
| 201 | self._actions.extend(self.generate_actions(action_or_factory)) |
| 202 | |
| 203 | while len(self._actions) < actions: |
| 204 | action_or_factory = self._pick_action_or_factory() |
| 205 | self._actions.extend(self.generate_actions(action_or_factory)) |
| 206 | |
| 207 | def generate_actions(self, action_def: ActionOrFactory) -> list[Action]: |
| 208 | if isinstance(action_def, ActionFactory): |
| 209 | actions = action_def.new(capabilities=self._capabilities) |
| 210 | elif issubclass(action_def, Action): |
| 211 | actions = [action_def(capabilities=self._capabilities)] |
| 212 | else: |
| 213 | raise RuntimeError( |
| 214 | f"{type(action_def)} is not a subclass of {ActionFactory} or {Action}" |
| 215 | ) |
| 216 | |
| 217 | for action in actions: |
| 218 | print("test:", action) |
| 219 | self._capabilities._extend(action.provides()) |
| 220 | print(" - ", self._capabilities, action.provides()) |
| 221 | self._capabilities._remove(action.withholds()) |
| 222 | print(" - ", self._capabilities, action.withholds()) |
| 223 | |
| 224 | return actions |
| 225 | |
| 226 | def run(self, c: Composition) -> None: |
| 227 | """Run the Zippy test.""" |
| 228 | max_time = datetime.now() + self._max_execution_time |
| 229 | executed_count = len(self._actions) |
| 230 | for i, action in enumerate(self._actions): |
| 231 | print(action) |
| 232 | action.run(c, self._state) |
| 233 | if datetime.now() > max_time: |
| 234 | print( |
| 235 | f"--- Desired execution time of {self._max_execution_time} has been reached." |