mypy finished

.github/workflows/ci.yml

@@ -35,6 +35,8 @@ jobs:
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+    - name: pre-commit
+      run: pre-commit run --all-files
     - name: Test with unittest
       run: |
         python -m unittest --verbose

.pre-commit-config.yaml

@@ -4,8 +4,6 @@ repos:
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer
-      - id: check-yaml
-        args: ["--unsafe"]
       - id: check-added-large-files
 
   - repo: https://github.com/myint/autoflake

bombard/bombardier.py

@@ -77,10 +77,15 @@ def __init__(
 
         super().__init__(args.threads)
 
-    def status_coloured(self, status: int) -> str:
+    def status_coloured(self, status: Union[str, int]) -> str:
+        """
+        Paint ok status as green and overload as red using terminal control codes.
+
+        If status has special string value (EXCEPTION_STATUS) paint it red.
+        """
         if status in self.ok:
             return green(str(status))
-        if status in self.overload:
+        if status in self.overload or status == EXCEPTION_STATUS:
             return dark_red(str(status))
         return red(str(status))
 
@@ -165,16 +170,16 @@ def beautify_url(
         query = query if len(query) < 15 else "?..." + query[-15:]
         return f"""{method} {urlparts.netloc}{path}{query}"""
 
-    def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
+    def worker(self, thread_id: int, job: Dict[str, Any]) -> None:
         """
         Thread callable.
         Strike ammo from queue.
         """
         try:
             # setup logging ASAP and as safe as possible
-            if isinstance(ammo, dict):
-                request = ammo.get("request", {})
-                ammo_id = ammo.get("id", "")
+            if isinstance(job, dict):
+                request = job.get("request", {})
+                ammo_id = job.get("id", "")
                 ammo_name = request.get("name", "")
             else:
                 request = {}
@@ -183,7 +188,7 @@ def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
             request_logging.sending(thread_id, ammo_id, ammo_name)
             pretty_url = ""  # we use it in `except`
             try:
-                ammo = apply_supply_dict(ammo, dict(self.supply, **ammo["supply"]))
+                job = apply_supply_dict(job, dict(self.supply, **job["supply"]))
 
                 url = request.get("url", "")
                 method = request["method"] if "method" in request else "GET"
@@ -207,7 +212,7 @@ def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
 
                 request_logging.receiving()
 
-                self.process_resp(ammo, status, resp, time_ns() - start_ns, len(resp))
+                self.process_resp(job, status, resp, time_ns() - start_ns, len(resp))
 
                 self.resp_count += 1
                 if self.args.quiet and self.resp_count in self.show_response:

bombard/report.py

@@ -35,7 +35,7 @@ def __init__(
         self,
         time_units: Optional[str] = "ms",
         time_threshold_ms: int = 1,
-        success_statuses: Set[int] = None,
+        success_statuses: Optional[Set[int]] = None,
     ):
         """
         :param time_units: time representation fixed in the units (see names in bombard.pretty_ns)
@@ -116,22 +116,25 @@ def reduce(
         return result
 
     def filter(
-        self, dimension_name: str, status_group_filter: str = None, request_name_filter: str = None
+        self,
+        dimension_name: str,
+        status_group_filter: Optional[str] = None,
+        request_name_filter: Optional[str] = None,
     ) -> MutableSequence[Any]:
         """
         Filter the dimension by group and/or request_name,
         Returns array with values from dimention_name
         """
-        dimension: MutableSequence[Any] = array(self.DIMENSIONS[dimension_name]["type"])
-        for request_name in self.stat:
+        dimension_values: MutableSequence[Any] = array(self.DIMENSIONS[dimension_name]["type"])
+        for request_name, statuses in self.stat.items():
             if request_name == request_name_filter or request_name_filter is None:
-                for status in self.stat[request_name]:
+                for status_name, status in statuses.items():
                     if (
-                        self.group_name_by_status(status) == status_group_filter
+                        self.group_name_by_status(status_name) == status_group_filter
                         or status_group_filter is None
                     ):
-                        dimension += self.stat[request_name][status][dimension_name]
-        return dimension
+                        dimension_values += status[dimension_name]
+        return dimension_values
 
     @staticmethod
     def dimension_stat_report(
@@ -149,26 +152,26 @@ def dimension_stat_report(
         )
 
     def filtered_report(
-        self, status_group_filter: str = None, request_name_filter: str = None
+        self, status_group_filter: Optional[str] = None, request_name_filter: Optional[str] = None
     ) -> str:
         """
         Filter by group and/or request_name.
         Returns report str with stats for all dimensions.
         """
         result = []
-        for dimension_name in self.DIMENSIONS:
-            dimension = self.filter(dimension_name, status_group_filter, request_name_filter)
-            if dimension:
+        for dimension_name, dimention_descr in self.DIMENSIONS.items():
+            dimension_values = self.filter(
+                dimension_name, status_group_filter, request_name_filter
+            )
+            if dimension_values:
                 result.append(
-                    self.dimension_stat_report(
-                        dimension, self.DIMENSIONS[dimension_name]["pretty_func"]
-                    )
+                    self.dimension_stat_report(dimension_values, dimention_descr["pretty_func"])
                 )
         if not result:
             return "No such requests"
         return "\n".join(result)
 
-    def statuses_report(self, request_name_filter: str = None) -> str:
+    def statuses_report(self, request_name_filter: Optional[str] = None) -> str:
         return ", ".join(
             f"{group} {self.reduce(len, TIME, group, request_name_filter)}" for group in GROUPS
         )
@@ -204,17 +207,17 @@ def report(self) -> str:
 {self.filtered_report(status_group)}
 """
             )
-        by_group = "\n".join(by_group)
+        by_group_str = "\n".join(by_group)  # we cannot use "\n" inside f-string
         by_request = []
         for request_name in self.stat:
             by_request.append(
                 f"""### {request_name}: {self.statuses_report(request_name)}
 {self.filtered_report(None, request_name)}
 """
             )
-        by_request = "\n".join(by_request)
+        by_request_str = "\n".join(by_request)  # we cannot use "\n" inside f-string
         return f"""{total_line}
 
-{by_group}
+{by_group_str}
 
-{by_request}"""
+{by_request_str}"""