#!/usr/bin/env python3
-"""Helpers for unittests. Note that when you import this we
- automatically wrap unittest.main() with a call to
- bootstrap.initialize so that we getLogger config, commandline args,
- logging control, etc... this works fine but it's a little hacky so
- caveat emptor.
+# © Copyright 2021-2022, Scott Gasch
+
+"""Helpers for unittests.
+
+.. note::
+
+ When you import this we automatically wrap unittest.main()
+ with a call to bootstrap.initialize so that we getLogger
+ config, commandline args, logging control, etc... this works
+ fine but it's a little hacky so caveat emptor.
"""
-from abc import ABC, abstractmethod
import contextlib
import functools
import inspect
import pickle
import random
import statistics
-import time
import tempfile
-from typing import Any, Callable, Dict, List, Optional
+import time
import unittest
import warnings
+from abc import ABC, abstractmethod
+from typing import Any, Callable, Dict, List, Literal, Optional
+
+import sqlalchemy as sa
import bootstrap
import config
import function_utils
import scott_secrets
-import sqlalchemy as sa
-
-
logger = logging.getLogger(__name__)
-cfg = config.add_commandline_args(
- f'Logging ({__file__})', 'Args related to function decorators'
-)
+cfg = config.add_commandline_args(f'Logging ({__file__})', 'Args related to function decorators')
cfg.add_argument(
'--unittests_ignore_perf',
action='store_true',
class PerfRegressionDataPersister(ABC):
+ """A base class for a signature dealing with persisting perf
+ regression data."""
+
def __init__(self):
pass
class FileBasedPerfRegressionDataPersister(PerfRegressionDataPersister):
+ """A perf regression data persister that uses files."""
+
def __init__(self, filename: str):
+ super().__init__()
self.filename = filename
self.traces_to_delete: List[str] = []
class DatabasePerfRegressionDataPersister(PerfRegressionDataPersister):
+ """A perf regression data persister that uses a database backend."""
+
def __init__(self, dbspec: str):
+ super().__init__()
self.dbspec = dbspec
self.engine = sa.create_engine(self.dbspec)
self.conn = self.engine.connect()
def load_performance_data(self, method_id: str) -> Dict[str, List[float]]:
results = self.conn.execute(
- sa.text(
- f'SELECT * FROM runtimes_by_function WHERE function = "{method_id}";'
- )
+ sa.text(f'SELECT * FROM runtimes_by_function WHERE function = "{method_id}";')
)
ret: Dict[str, List[float]] = {method_id: []}
for result in results.all():
def save_performance_data(self, method_id: str, data: Dict[str, List[float]]):
self.delete_performance_data(method_id)
- for (method_id, perf_data) in data.items():
+ for (mid, perf_data) in data.items():
sql = 'INSERT INTO runtimes_by_function (function, runtime) VALUES '
for perf in perf_data:
- self.conn.execute(sql + f'("{method_id}", {perf});')
+ self.conn.execute(sql + f'("{mid}", {perf});')
def delete_performance_data(self, method_id: str):
sql = f'DELETE FROM runtimes_by_function WHERE function = "{method_id}"'
helper = FileBasedPerfRegressionDataPersister(filename)
elif config.config['unittests_persistance_strategy'] == 'DATABASE':
dbspec = config.config['unittests_perfdb_spec']
- dbspec = dbspec.replace(
- '<PASSWORD>', scott_secrets.MARIADB_UNITTEST_PERF_PASSWORD
- )
+ dbspec = dbspec.replace('<PASSWORD>', scott_secrets.MARIADB_UNITTEST_PERF_PASSWORD)
helper = DatabasePerfRegressionDataPersister(dbspec)
else:
raise Exception('Unknown/unexpected --unittests_persistance_strategy value')
func_id = function_utils.function_identifier(func)
func_name = func.__name__
- logger.debug(f'Watching {func_name}\'s performance...')
- logger.debug(f'Canonical function identifier = {func_id}')
+ logger.debug('Watching %s\'s performance...', func_name)
+ logger.debug('Canonical function identifier = "%s"', func_id)
try:
perfdb = helper.load_performance_data(func_id)
hist = perfdb.get(func_id, [])
if len(hist) < config.config['unittests_num_perf_samples']:
hist.append(run_time)
- logger.debug(f'Still establishing a perf baseline for {func_name}')
+ logger.debug('Still establishing a perf baseline for %s', func_name)
else:
stdev = statistics.stdev(hist)
- logger.debug(f'For {func_name}, performance stdev={stdev}')
+ logger.debug('For %s, performance stdev=%.2f', func_name, stdev)
slowest = hist[-1]
- logger.debug(f'For {func_name}, slowest perf on record is {slowest:f}s')
+ logger.debug('For %s, slowest perf on record is %.2fs', func_name, slowest)
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')
+ logger.debug('For %s, max acceptable runtime is %.2fs', func_name, limit)
+ logger.debug('For %s, actual observed runtime was %.2fs', func_name, run_time)
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.
for name, m in inspect.getmembers(cls, inspect.isfunction):
if name.startswith(prefix):
setattr(cls, name, check_method_for_perf_regressions(m))
- logger.debug(f'Wrapping {cls.__name__}:{name}.')
+ logger.debug('Wrapping %s:%s.', cls.__name__, name)
return cls
return decorate_the_testcase
-def breakpoint():
- """Hard code a breakpoint somewhere; drop into pdb."""
- import pdb
-
- pdb.set_trace()
-
-
-class RecordStdout(object):
+class RecordStdout(contextlib.AbstractContextManager):
"""
Record what is emitted to stdout.
"""
def __init__(self) -> None:
+ super().__init__()
self.destination = tempfile.SpooledTemporaryFile(mode='r+')
self.recorder: Optional[contextlib.redirect_stdout] = None
def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]:
self.recorder = contextlib.redirect_stdout(self.destination)
- assert self.recorder
+ assert self.recorder is not None
self.recorder.__enter__()
return lambda: self.destination
- def __exit__(self, *args) -> Optional[bool]:
- assert self.recorder
+ def __exit__(self, *args) -> Literal[False]:
+ assert self.recorder is not None
self.recorder.__exit__(*args)
self.destination.seek(0)
- return None
+ return False
-class RecordStderr(object):
+class RecordStderr(contextlib.AbstractContextManager):
"""
Record what is emitted to stderr.
"""
def __init__(self) -> None:
+ super().__init__()
self.destination = tempfile.SpooledTemporaryFile(mode='r+')
self.recorder: Optional[contextlib.redirect_stdout[Any]] = None
def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]:
self.recorder = contextlib.redirect_stderr(self.destination) # type: ignore
- assert self.recorder
+ assert self.recorder is not None
self.recorder.__enter__()
return lambda: self.destination
- def __exit__(self, *args) -> Optional[bool]:
- assert self.recorder
+ def __exit__(self, *args) -> Literal[False]:
+ assert self.recorder is not None
self.recorder.__exit__(*args)
self.destination.seek(0)
- return None
+ return False
-class RecordMultipleStreams(object):
+class RecordMultipleStreams(contextlib.AbstractContextManager):
"""
Record the output to more than one stream.
"""
def __init__(self, *files) -> None:
+ super().__init__()
self.files = [*files]
self.destination = tempfile.SpooledTemporaryFile(mode='r+')
self.saved_writes: List[Callable[..., Any]] = []
f.write = self.destination.write
return lambda: self.destination
- def __exit__(self, *args) -> Optional[bool]:
+ def __exit__(self, *args) -> Literal[False]:
for f in self.files:
f.write = self.saved_writes.pop()
self.destination.seek(0)
- return None
+ return False
if __name__ == '__main__':
projects / python_utils.git / blobdiff