diff options
Diffstat (limited to 'unittest_utils.py')
| -rw-r--r-- | unittest_utils.py | 27 |
1 files changed, 16 insertions, 11 deletions
diff --git a/unittest_utils.py b/unittest_utils.py index b9746a8..81b339a 100644 --- a/unittest_utils.py +++ b/unittest_utils.py @@ -18,7 +18,7 @@ import random import statistics import time import tempfile -from typing import Callable, Dict, List +from typing import Any, Callable, Dict, List, Optional import unittest import warnings @@ -83,7 +83,7 @@ class PerfRegressionDataPersister(ABC): pass @abstractmethod - def load_performance_data(self) -> Dict[str, List[float]]: + def load_performance_data(self, method_id: str) -> Dict[str, List[float]]: pass @abstractmethod @@ -98,7 +98,7 @@ class PerfRegressionDataPersister(ABC): class FileBasedPerfRegressionDataPersister(PerfRegressionDataPersister): def __init__(self, filename: str): self.filename = filename - self.traces_to_delete = [] + self.traces_to_delete: List[str] = [] def load_performance_data(self, method_id: str) -> Dict[str, List[float]]: with open(self.filename, 'rb') as f: @@ -128,7 +128,7 @@ class DatabasePerfRegressionDataPersister(PerfRegressionDataPersister): f'SELECT * FROM runtimes_by_function WHERE function = "{method_id}";' ) ) - ret = {method_id: []} + ret: Dict[str, List[float]] = {method_id: []} for result in results.all(): ret[method_id].append(result['runtime']) results.close() @@ -283,14 +283,16 @@ class RecordStdout(object): def __init__(self) -> None: self.destination = tempfile.SpooledTemporaryFile(mode='r+') - self.recorder = None + self.recorder: Optional[contextlib.redirect_stdout] = None def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]: self.recorder = contextlib.redirect_stdout(self.destination) + assert self.recorder self.recorder.__enter__() return lambda: self.destination - def __exit__(self, *args) -> bool: + def __exit__(self, *args) -> Optional[bool]: + assert self.recorder self.recorder.__exit__(*args) self.destination.seek(0) return None @@ -310,14 +312,16 @@ class RecordStderr(object): def __init__(self) -> None: self.destination = tempfile.SpooledTemporaryFile(mode='r+') - self.recorder = None + self.recorder: Optional[contextlib.redirect_stdout[Any]] = None def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]: - self.recorder = contextlib.redirect_stderr(self.destination) + self.recorder = contextlib.redirect_stderr(self.destination) # type: ignore + assert self.recorder self.recorder.__enter__() return lambda: self.destination - def __exit__(self, *args) -> bool: + def __exit__(self, *args) -> Optional[bool]: + assert self.recorder self.recorder.__exit__(*args) self.destination.seek(0) return None @@ -331,7 +335,7 @@ class RecordMultipleStreams(object): def __init__(self, *files) -> None: self.files = [*files] self.destination = tempfile.SpooledTemporaryFile(mode='r+') - self.saved_writes = [] + self.saved_writes: List[Callable[..., Any]] = [] def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]: for f in self.files: @@ -339,10 +343,11 @@ class RecordMultipleStreams(object): f.write = self.destination.write return lambda: self.destination - def __exit__(self, *args) -> bool: + def __exit__(self, *args) -> Optional[bool]: for f in self.files: f.write = self.saved_writes.pop() self.destination.seek(0) + return None if __name__ == '__main__': |