X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=unittest_utils.py;h=b259c6b4c98f687d8e3c8db4b41c280db7ac9d19;hb=ffed473528c4feb836253758e86f7839af98f57b;hp=b9746a80307ad512cee993aca29449e365e998b6;hpb=f77068f0dc494d9394cdacafc6f719730926839f;p=python_utils.git

diff --git a/unittest_utils.py b/unittest_utils.py
index b9746a8..b259c6b 100644
--- a/unittest_utils.py
+++ b/unittest_utils.py
@@ -7,7 +7,6 @@
    caveat emptor.
 """
 
-from abc import ABC, abstractmethod
 import contextlib
 import functools
 import inspect
@@ -16,20 +15,20 @@ import os
 import pickle
 import random
 import statistics
-import time
 import tempfile
-from typing import Callable, Dict, List
+import time
 import unittest
 import warnings
+from abc import ABC, abstractmethod
+from typing import Any, Callable, Dict, List, 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'
@@ -83,7 +82,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 +97,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 +127,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 +282,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 is not None
         self.recorder.__enter__()
         return lambda: self.destination
 
-    def __exit__(self, *args) -> bool:
+    def __exit__(self, *args) -> Optional[bool]:
+        assert self.recorder is not None
         self.recorder.__exit__(*args)
         self.destination.seek(0)
         return None
@@ -310,14 +311,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 is not None
         self.recorder.__enter__()
         return lambda: self.destination
 
-    def __exit__(self, *args) -> bool:
+    def __exit__(self, *args) -> Optional[bool]:
+        assert self.recorder is not None
         self.recorder.__exit__(*args)
         self.destination.seek(0)
         return None
@@ -331,7 +334,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 +342,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__':