Make smart futures avoid polling.

[python_utils.git] / unittest_utils.py

diff --git a/unittest_utils.py b/unittest_utils.py
index 5987da6ba38194bf639b7382347ed599ff5bc0e9..bb1a9b432f49d7886b99f39418339e9f62e0cee1 100644 (file)
--- a/unittest_utils.py
+++ b/unittest_utils.py
@@ -1,9 +1,10 @@
 #!/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.
+   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.
 """
 
 import contextlib
@@ -55,13 +56,14 @@ _db = '/home/scott/.python_unittest_performance_db'
 
 
 def check_method_for_perf_regressions(func: Callable) -> Callable:
-    """This is meant to be used on a method in a class that subclasses
+    """
+    This is meant to be used on a method in a class that subclasses
     unittest.TestCase.  When thus decorated it will time the execution
     of the code in the method, compare it with a database of
     historical perfmance, and fail the test with a perf-related
     message if it has become too slow.
-    """
 
+    """
     def load_known_test_performance_characteristics():
         with open(_db, 'rb') as f:
             return pickle.load(f)
@@ -106,7 +108,7 @@ def check_method_for_perf_regressions(func: Callable) -> Callable:
             )
         else:
             stdev = statistics.stdev(hist)
-            limit = hist[-1] + stdev * 3
+            limit = hist[-1] + stdev * 5
             logger.debug(
                 f'Max acceptable performace for {func.__name__} is {limit:f}s'
             )
@@ -116,12 +118,14 @@ def check_method_for_perf_regressions(func: Callable) -> Callable:
             ):
                 msg = f'''{func_id} performance has regressed unacceptably.
 {hist[-1]:f}s is the slowest record in {len(hist)} db perf samples.
-It just ran in {run_time:f}s which is >3 stdevs slower than the slowest sample.
+It just ran in {run_time:f}s which is >5 stdevs slower than the slowest sample.
 Here is the current, full db perf timing distribution:
 
-{hist}'''
-                slf = args[0]
+'''
+                for x in hist:
+                    msg += f'{x:f}\n'
                 logger.error(msg)
+                slf = args[0]
                 slf.fail(msg)
             else:
                 hist.append(run_time)
@@ -147,15 +151,19 @@ def check_all_methods_for_perf_regressions(prefix='test_'):
 
 
 def breakpoint():
+    """Hard code a breakpoint somewhere; drop into pdb."""
     import pdb
     pdb.set_trace()
 
 
 class RecordStdout(object):
     """
-        with uu.RecordStdout() as record:
-            print("This is a test!")
-        print({record().readline()})
+    Record what is emitted to stdout.
+
+    >>> with RecordStdout() as record:
+    ...     print("This is a test!")
+    >>> print({record().readline()})
+    {'This is a test!\\n'}
     """
 
     def __init__(self) -> None:
@@ -170,14 +178,18 @@ class RecordStdout(object):
     def __exit__(self, *args) -> bool:
         self.recorder.__exit__(*args)
         self.destination.seek(0)
-        return True
+        return None
 
 
 class RecordStderr(object):
     """
-        with uu.RecordStderr() as record:
-            print("This is a test!", file=sys.stderr)
-        print({record().readline()})
+    Record what is emitted to stderr.
+
+    >>> import sys
+    >>> with RecordStderr() as record:
+    ...     print("This is a test!", file=sys.stderr)
+    >>> print({record().readline()})
+    {'This is a test!\\n'}
     """
 
     def __init__(self) -> None:
@@ -192,17 +204,14 @@ class RecordStderr(object):
     def __exit__(self, *args) -> bool:
         self.recorder.__exit__(*args)
         self.destination.seek(0)
-        return True
+        return None
 
 
 class RecordMultipleStreams(object):
     """
-        with uu.RecordStreams(sys.stderr, sys.stdout) as record:
-            print("This is a test!")
-            print("This is one too.", file=sys.stderr)
-
-        print(record().readlines())
+    Record the output to more than one stream.
     """
+
     def __init__(self, *files) -> None:
         self.files = [*files]
         self.destination = tempfile.SpooledTemporaryFile(mode='r+')
@@ -218,3 +227,8 @@ class RecordMultipleStreams(object):
         for f in self.files:
             f.write = self.saved_writes.pop()
         self.destination.seek(0)
+
+
+if __name__ == '__main__':
+    import doctest
+    doctest.testmod()