projects
/
python_utils.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Add a simple test for waitable presence.
[python_utils.git]
/
unittest_utils.py
diff --git
a/unittest_utils.py
b/unittest_utils.py
index ba9ca28f091bc70bd232cb6f059116cfc70d7fb9..f229df75e8b88825d66ca227d7e907d3dc725e1a 100644
(file)
--- a/
unittest_utils.py
+++ b/
unittest_utils.py
@@
-30,9
+30,7
@@
import function_utils
import scott_secrets
logger = logging.getLogger(__name__)
import scott_secrets
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',
cfg.add_argument(
'--unittests_ignore_perf',
action='store_true',
@@
-123,9
+121,7
@@
class DatabasePerfRegressionDataPersister(PerfRegressionDataPersister):
def load_performance_data(self, method_id: str) -> Dict[str, List[float]]:
results = self.conn.execute(
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():
)
ret: Dict[str, List[float]] = {method_id: []}
for result in results.all():
@@
-165,9
+161,7
@@
def check_method_for_perf_regressions(func: Callable) -> Callable:
helper = FileBasedPerfRegressionDataPersister(filename)
elif config.config['unittests_persistance_strategy'] == 'DATABASE':
dbspec = config.config['unittests_perfdb_spec']
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')
helper = DatabasePerfRegressionDataPersister(dbspec)
else:
raise Exception('Unknown/unexpected --unittests_persistance_strategy value')
@@
-286,12
+280,12
@@
class RecordStdout(object):
def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]:
self.recorder = contextlib.redirect_stdout(self.destination)
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]:
self.recorder.__enter__()
return lambda: self.destination
def __exit__(self, *args) -> Optional[bool]:
- assert self.recorder
+ assert self.recorder
is not None
self.recorder.__exit__(*args)
self.destination.seek(0)
return None
self.recorder.__exit__(*args)
self.destination.seek(0)
return None
@@
-315,12
+309,12
@@
class RecordStderr(object):
def __enter__(self) -> Callable[[], tempfile.SpooledTemporaryFile]:
self.recorder = contextlib.redirect_stderr(self.destination) # type: ignore
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]:
self.recorder.__enter__()
return lambda: self.destination
def __exit__(self, *args) -> Optional[bool]:
- assert self.recorder
+ assert self.recorder
is not None
self.recorder.__exit__(*args)
self.destination.seek(0)
return None
self.recorder.__exit__(*args)
self.destination.seek(0)
return None