|
7 | 7 | import timeit
|
8 | 8 | import unittest
|
9 | 9 | from time import sleep, time
|
10 |
| -from typing import Callable, ParamSpec, TypeVar |
| 10 | +from typing import Any, Callable, ParamSpec, TypeVar |
11 | 11 |
|
12 | 12 | import six
|
13 | 13 |
|
|
44 | 44 | BASE_MEMORY_USAGE = "BASE_MEMORY_USAGE"
|
45 | 45 |
|
46 | 46 |
|
47 |
| -def parameterized_list(cases: list[str]) -> list[str]: |
| 47 | +def parameterized_list(cases: list[Any]) -> list[tuple[str, Any]]: |
48 | 48 | """
|
49 | 49 | Creates a list of cases for parameterized to use to run tests.
|
50 | 50 |
|
@@ -119,7 +119,7 @@ def _get_config_name() -> str:
|
119 | 119 | return get_config_details()["name"]
|
120 | 120 |
|
121 | 121 |
|
122 |
| -def get_config_details() -> dict(): |
| 122 | +def get_config_details() -> dict: |
123 | 123 | """
|
124 | 124 | Returns the current config name after waiting for up to WAIT_FOR_SERVER_TIMEOUT seconds
|
125 | 125 | for it to be readable
|
@@ -292,7 +292,7 @@ def _start_stop_ioc_is_a_start(is_a_start: bool, ioc_name: str) -> None:
|
292 | 292 | wait_for_ioc_start_stop(timeout=IOCS_START_STOP_TIMEOUT, is_start=is_a_start, ioc_name=ioc_name)
|
293 | 293 |
|
294 | 294 |
|
295 |
| -def bulk_start_ioc(ioc_list: list[str]) -> tuple[bool, bool]: |
| 295 | +def bulk_start_ioc(ioc_list: list[str]) -> tuple[list[str], list[str]]: |
296 | 296 | """
|
297 | 297 | start a list of IOCs in bulk
|
298 | 298 | :param ioc_list: a list of the names of the IOCs to start
|
@@ -323,7 +323,7 @@ def bulk_start_ioc(ioc_list: list[str]) -> tuple[bool, bool]:
|
323 | 323 | return failed_to_start, not_in_proc_serv
|
324 | 324 |
|
325 | 325 |
|
326 |
| -def bulk_stop_ioc(ioc_list: list[str]) -> bool: |
| 326 | +def bulk_stop_ioc(ioc_list: list[str]) -> list[str]: |
327 | 327 | """
|
328 | 328 | Stops a list of IOCs in bulk
|
329 | 329 | :param ioc_list: a list of the names of the IOCs to stop
|
@@ -482,16 +482,16 @@ def wait_for_string_pvs_to_not_be_empty(
|
482 | 482 | return pv_values
|
483 | 483 |
|
484 | 484 |
|
485 |
| -def retry_on_failure(max_times: int) -> Callable[[], None]: |
| 485 | +def retry_on_failure(max_times: int) -> Callable[[Callable[P, T]], Callable[P, None]]: |
486 | 486 | """
|
487 | 487 | Decorator that will retry running a test if it failed.
|
488 | 488 | :param max_times: Maximum number of times to retry running the test
|
489 | 489 | :return: the decorator
|
490 | 490 | """
|
491 | 491 |
|
492 |
| - def decorator(func: Callable[P, T]) -> Callable[P, T]: |
| 492 | + def decorator(func: Callable[P, T]) -> Callable[P, None]: |
493 | 493 | @six.wraps(func)
|
494 |
| - def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: |
| 494 | + def wrapper(*args: P.args, **kwargs: P.kwargs) -> None: |
495 | 495 | err = None
|
496 | 496 | for attempt in range(max_times):
|
497 | 497 | try:
|
|
0 commit comments