import statistics
import time
import tempfile
-from typing import Callable, Dict, List
+from typing import Any, Callable, Dict, List, Optional
import unittest
import warnings
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
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:
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()
@functools.wraps(func)
def wrapper_perf_monitor(*args, **kwargs):
+ if config.config['unittests_ignore_perf']:
+ return func(*args, **kwargs)
+
if config.config['unittests_persistance_strategy'] == 'FILE':
filename = config.config['unittests_perfdb_filename']
helper = FileBasedPerfRegressionDataPersister(filename)
limit = slowest + stdev * 4
logger.debug(f'For {func_name}, max acceptable runtime is {limit:f}s')
logger.debug(f'For {func_name}, actual observed runtime was {run_time:f}s')
- if run_time > limit and not config.config['unittests_ignore_perf']:
+ if run_time > limit:
msg = f'''{func_id} performance has regressed unacceptably.
{slowest:f}s is the slowest runtime on record in {len(hist)} perf samples.
It just ran in {run_time:f}s which is 4+ stdevs slower than the slowest.
... print("This is a test!")
>>> print({record().readline()})
{'This is a test!\\n'}
+ >>> record().close()
"""
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
... print("This is a test!", file=sys.stderr)
>>> print({record().readline()})
{'This is a test!\\n'}
+ >>> record().close()
"""
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
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:
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__':