mypy fixes

bombard/bombardier.py

@@ -109,7 +109,9 @@ def get_headers(request: Dict[str, Any], body_is_json: bool) -> Dict[str, Any]:
             result.update(predefined["json"])
         return result
 
-    def process_resp(self, ammo: Dict[str, Any], status: int, resp: str, elapsed: int, size: int):
+    def process_resp(
+        self, ammo: Dict[str, Any], status: int, resp: str, elapsed: int, size: int
+    ) -> None:
         request = ammo["request"]
         self.reporter.log(status, elapsed, request.get("name"), size)
         if status in self.ok:

bombard/pretty_ns.py

@@ -19,17 +19,14 @@
 True
 
 """
-from typing import Any, Callable, Optional
-
-time_ns: Optional[
-    Callable[[], int]
-] = None  # for Python3.7+ this is function from system library time
-
-
-# for earlier Python versions this is emulation of the Python3.7 time_ns
+import time
+from typing import Any, Optional
 
 
 def pretty_ns(elapsed_ns: int, fixed_units: Optional[str] = None) -> str:
+    """
+    for earlier Python versions this is emulation of the Python3.7 time_ns
+    """
     dividers = {
         "us": 1,
         "mks": 1000,
@@ -48,8 +45,15 @@ def pretty_ns(elapsed_ns: int, fixed_units: Optional[str] = None) -> str:
     return f"{result:.1f} {dividers['days']}"
 
 
-def emul_time_ns() -> int:
-    return int(perf_counter() * 10 ** 9)
+try:
+    time_ns = time.time_ns
+except AttributeError:
+    from time import perf_counter
+
+    def emul_time_ns() -> int:
+        return int(perf_counter() * 10 ** 9)
+
+    time_ns = emul_time_ns
 
 
 class Timer:
@@ -72,15 +76,6 @@ def pretty(self) -> str:
         return pretty_ns(self.ns)
 
 
-import time
-
-try:
-    time_ns = time.time_ns
-except AttributeError:
-    from time import perf_counter
-
-    time_ns = emul_time_ns
-
 if __name__ == "__main__":
     import doctest
 

bombard/report.py

@@ -10,7 +10,7 @@
 from copy import deepcopy
 from typing import Any, Dict, Optional, Set
 
-from bombard.pretty_ns import pretty_ns, time_ns
+from bombard import pretty_ns
 from bombard.pretty_sz import pretty_sz
 from bombard.terminal_colours import red
 
@@ -53,21 +53,21 @@ def __init__(
             name: array(params["type"]) for name, params in self.DIMENSIONS.items()
         }
 
-        self.start_ns = time_ns()
+        self.start_ns = pretty_ns.time_ns()
 
         self.time_units = time_units
         self.time_threshold_ns = time_threshold_ms * 10 ** 6
         self.ok = success_statuses
 
-        self.stat = {}  # stat[request type][status]
+        self.stat: Dict[Any, Any] = {}  # stat[request type][status]
         # todo implement cache
         # To select the best approach cache compare benchmarks for two versions:
         #  1) fill cache in log() and use in filter()/reduce()
         #  2) fill cache in filter() and use in reduce()
         # self.stat_by_group = {}  # cache
         # self.stat_by_request = {}  # cache
 
-    def group_name_by_status(self, status):
+    def group_name_by_status(self, status: int) -> str:
         """
         All this statuses should be in GROUPS
         """
@@ -76,7 +76,7 @@ def group_name_by_status(self, status):
         else:
             return FAIL_GROUP
 
-    def log(self, status, elapsed, request_name, response_size):
+    def log(self, status: int, elapsed: int, request_name: str, response_size: int) -> None:
         """
         Add result to the report
 
@@ -90,8 +90,8 @@ def log(self, status, elapsed, request_name, response_size):
         self.stat[request_name][status][SIZE].append(response_size)
 
     @property
-    def total_elapsed_ns(self):
-        return time_ns() - self.start_ns
+    def total_elapsed_ns(self) -> int:
+        return pretty_ns.time_ns() - self.start_ns
 
     def reduce(
         self,
@@ -173,8 +173,8 @@ def statuses_report(self, request_name_filter: str = None) -> str:
     def pretty_time(self, elapsed: int, paint: bool = True):
         return self.pretty_ns(elapsed * TIME_DENOMINATOR, paint)
 
-    def pretty_ns(self, elapsed_ns: int, paint: bool = True):
-        result = pretty_ns(elapsed_ns, self.time_units)
+    def pretty_ns(self, elapsed_ns: int, paint: bool = True) -> str:
+        result = pretty_ns.pretty_ns(elapsed_ns, self.time_units)
         if elapsed_ns > self.time_threshold_ns and paint:
             return red(result)
         else:

bombard/request_logging.py

@@ -13,9 +13,7 @@
 
 log = logging.getLogger()
 thread_data = threading.local()
-pretty_ns: Callable[
-    [int, Optional[Any]], str
-] = plain_ns  # we can redefine that to have customized formatting
+pretty_ns: Callable[[int], str] = plain_ns  # we can redefine that to have customized formatting
 
 
 def main_thread() -> None:
@@ -51,7 +49,7 @@ def receiving() -> None:
 
 
 class RequestFormatter(logging.Formatter):
-    def format(self, record):
+    def format(self, record: Any) -> str:
         record.threadid = thread_data.thread_id
         record.requestid = str(thread_data.request_id).rjust(4)
         record.colour = thread_data.colour
@@ -63,7 +61,7 @@ def format(self, record):
         return super().format(record)
 
 
-def setup_logging(level: int, log_file_name: Optional[str] = None):
+def setup_logging(level: int, log_file_name: Optional[str] = None) -> None:
     main_thread()
     handler = logging.StreamHandler()
     formatter = RequestFormatter(

tests/fake_jsonplaceholder.py

@@ -16,7 +16,7 @@ class FakeResp:
     def __init__(self, resp_body: str) -> None:
         self.body = resp_body
 
-    def read(self) -> str:
+    def read(self) -> Optional[str]:
         return self.body