mypy

bombard/bombardier.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Dict, List, Mapping, Optional, Set
+from typing import Any, Dict, List, Optional, Set, Union
 from urllib.parse import urlparse
 
 from bombard import request_logging
@@ -110,7 +110,7 @@ def get_headers(request: Dict[str, Any], body_is_json: bool) -> Dict[str, Any]:
         return result
 
     def process_resp(
-        self, ammo: Dict[str, Any], status: int, resp: str, elapsed: int, size: int
+        self, ammo: Dict[str, Any], status: Union[int, str], resp: str, elapsed: int, size: int
     ) -> None:
         request = ammo["request"]
         self.reporter.log(status, elapsed, request.get("name"), size)
@@ -133,6 +133,7 @@ def process_resp(
                         f'Cannot extract {request["extract"]} from {resp}:\n{e}', exc_info=True
                     )
             if "script" in request:
+                supply = None
                 try:
                     # Supply immediately repeats all changes in the self.supply so if the script spawns new
                     # requests they already get new values
@@ -171,7 +172,7 @@ def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
         """
         try:
             # setup logging ASAP and as safe as possible
-            if isinstance(ammo, Mapping):
+            if isinstance(ammo, dict):
                 request = ammo.get("request", {})
                 ammo_id = ammo.get("id", "")
                 ammo_name = request.get("name", "")
@@ -190,15 +191,15 @@ def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
                 headers = self.get_headers(request, body is not None)
                 pretty_url = self.beautify_url(url, method, body)
 
-                log.debug(
+                log.debug(  # pylint: disable=logging-not-lazy
                     f"Bomb to drop:\n{pretty_url}" + ("\n{body}" if body is not None else "")
                 )
-                if self.args.quiet:
-                    if ammo_id in self.show_request:
-                        print(f"{self.show_request[ammo_id].format(id=ammo_id):>15}\r", end="")
+                if self.args.quiet and ammo_id in self.show_request:
+                    print(f"{self.show_request[ammo_id].format(id=ammo_id):>15}\r", end="")
                 log.info(pretty_url)
 
                 start_ns = time_ns()
+                status: Union[str, int]
                 if self.args.dry:
                     status, resp = list(self.ok)[0], json.dumps(request.get("dry"))
                 else:
@@ -209,21 +210,22 @@ def worker(self, thread_id: int, ammo: Dict[str, Any]) -> None:
                 self.process_resp(ammo, status, resp, time_ns() - start_ns, len(resp))
 
                 self.resp_count += 1
-                if self.args.quiet:
-                    if self.resp_count in self.show_response:
-                        print(
-                            f"{self.show_response[self.resp_count].format(id=self.resp_count):>15}\r",
-                            end="",
-                        )
+                if self.args.quiet and self.resp_count in self.show_response:
+                    print(
+                        f"{self.show_response[self.resp_count].format(id=self.resp_count):>15}\r",
+                        end="",
+                    )
                 log.info(
                     self.status_coloured(status)
                     + f" ({pretty_sz(len(resp))}) "
                     + pretty_url
                     + " "
-                    + (red(resp) if status == EXCEPTION_STATUS else "")  # type: ignore
+                    + (red(resp) if status == EXCEPTION_STATUS else "")
                 )
             except Exception as e:
-                log.info(pretty_url + " " + red(str(e)), exc_info=True)
+                log.info(  # pylint: disable=logging-not-lazy
+                    pretty_url + " " + red(str(e)), exc_info=True
+                )
         finally:
             request_logging.main_thread()
 
@@ -263,7 +265,7 @@ def reload(
                 self.put({"id": self.job_count, "request": request, "supply": kwargs})
 
     def report(self) -> None:
-        log.warning(
+        log.warning(  # pylint: disable=logging-not-lazy
             "\n"
             + "=" * 100
             + "\n"

bombard/campaign_yaml.py

@@ -15,18 +15,18 @@
 
 class Yaml:
     @staticmethod
-    def load(*args: Any, **kwargs: Any) -> Any:
+    def load(stream: Any, Loader: Any = None) -> Any:  # pylint: disable=unused-argument
         """
         Mimics yaml interface for seamless injection
         """
-        return original_yaml.load(*args, **kwargs, Loader=IncludesLoader)
+        return original_yaml.load(stream, Loader=IncludesLoader)
 
     @staticmethod
-    def full_load(*args: Any, **kwargs: Any) -> Any:
+    def full_load(stream: Any, Loader: Any = None) -> Any:  # pylint: disable=unused-argument
         """
         Mimics yaml interface for seamless injection
         """
-        return original_yaml.load(*args, **kwargs, Loader=IncludesLoader)
+        return original_yaml.load(stream, Loader=IncludesLoader)
 
 
 yaml = Yaml()

bombard/http_request.py

@@ -1,6 +1,6 @@
 import http.client
 import ssl
-from typing import Optional
+from typing import Any, Dict, Optional, Tuple, Union
 from urllib.parse import urlparse
 
 EXCEPTION_STATUS = "!!!"
@@ -9,31 +9,33 @@
 def http_request(
     url: str,
     method: str = "GET",
-    headers: Optional[dict] = None,
+    headers: Optional[Dict[str, Any]] = None,
     body: Optional[str] = None,
     timeout: Optional[int] = None,
-) -> (int, dict):
+) -> Tuple[Union[int, str], Any]:
     """
     Make HTTP request.
 
     Returns tuple:
         <HTTP response status>, <response body>
     """
     try:
-        url = urlparse(url)
+        url_parsed = urlparse(url)
         kwargs = {"timeout": timeout} if timeout is not None else {}
-        if url.scheme.lower() == "https":
+        if url_parsed.scheme.lower() == "https":
             conn = http.client.HTTPSConnection(
-                url.netloc,
-                context=ssl._create_unverified_context(),
-                **kwargs,
+                url_parsed.netloc,
+                context=ssl._create_unverified_context(),  # type:ignore
+                **kwargs,  # type:ignore
             )
         else:
-            conn = http.client.HTTPConnection(
-                url.netloc,
-                **kwargs,
+            conn = http.client.HTTPConnection(  # type:ignore
+                url_parsed.netloc,
+                **kwargs,  # type:ignore
             )
-        conn.request(method, url.path, body=body, headers=headers if headers is not None else {})
+        conn.request(
+            method, url_parsed.path, body=body, headers=headers if headers is not None else {}
+        )
         resp = conn.getresponse()
         resp_body = resp.read()
     except Exception as e:

bombard/main.py

@@ -84,7 +84,9 @@ def init(args: Any) -> None:
 
 
 def start_campaign(args: Any, campaign_book: Dict[str, Any]) -> None:
-    log.debug("Starting bombard campaign with args\n" + " " * 4 + f"{args.__dict__}")
+    log.debug(  # pylint: disable=logging-not-lazy
+        "Starting bombard campaign with args\n" + " " * 4 + f"{args.__dict__}"
+    )
     log.debug(
         f'Loaded bombard campaign from "{args.file_name}": {len(campaign_book.get("ammo", {}))} ammo.'
     )

bombard/report.py

@@ -8,7 +8,7 @@
 import statistics
 from array import array
 from copy import deepcopy
-from typing import Any, Callable, Dict, Optional, Set
+from typing import Any, Callable, Dict, Optional, Set, Union
 
 from bombard import pretty_ns
 from bombard.pretty_sz import pretty_sz
@@ -35,7 +35,7 @@ def __init__(
         self,
         time_units: Optional[str] = "ms",
         time_threshold_ms: int = 1,
-        success_statuses: Set[int] = {200},
+        success_statuses: Set[int] = None,
     ):
         """
         :param time_units: time representation fixed in the units (see names in bombard.pretty_ns)
@@ -57,17 +57,11 @@ def __init__(
 
         self.time_units = time_units
         self.time_threshold_ns = time_threshold_ms * 10 ** 6
-        self.ok = success_statuses
+        self.ok = success_statuses or {200}
 
         self.stat: Dict[
             str, Dict[int, Dict[str, array[Any]]]
         ] = {}  # stat[request type][status]['time'|'size']
-        # 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: int) -> str:
         """
@@ -77,7 +71,9 @@ def group_name_by_status(self, status: int) -> str:
             return SUCCESS_GROUP
         return FAIL_GROUP
 
-    def log(self, status: int, elapsed: int, request_name: str, response_size: int) -> None:
+    def log(
+        self, status: Union[int, str], elapsed: int, request_name: str, response_size: int
+    ) -> None:
         """
         Add result to the report
 
@@ -86,9 +82,10 @@ def log(self, status: int, elapsed: int, request_name: str, response_size: int)
         :param request_name: Request name or None
         :param response_size: Response body size
         """
-        self.stat.setdefault(request_name, {}).setdefault(status, deepcopy(self.STAT_DEFAULT))
-        self.stat[request_name][status][TIME].append(elapsed // TIME_DENOMINATOR)
-        self.stat[request_name][status][SIZE].append(response_size)
+        if isinstance(status, int):  # do not log exceptions
+            self.stat.setdefault(request_name, {}).setdefault(status, deepcopy(self.STAT_DEFAULT))
+            self.stat[request_name][status][TIME].append(elapsed // TIME_DENOMINATOR)
+            self.stat[request_name][status][SIZE].append(response_size)
 
     @property
     def total_elapsed_ns(self) -> int:
@@ -170,10 +167,10 @@ def filtered_report(
 
     def statuses_report(self, request_name_filter: str = None) -> str:
         return ", ".join(
-            [f"{group} {self.reduce(len, TIME, group, request_name_filter)}" for group in GROUPS]
+            f"{group} {self.reduce(len, TIME, group, request_name_filter)}" for group in GROUPS
         )
 
-    def pretty_time(self, elapsed: int, paint: bool = True):
+    def pretty_time(self, elapsed: int, paint: bool = True) -> str:
         return self.pretty_ns(elapsed * TIME_DENOMINATOR, paint)
 
     def pretty_ns(self, elapsed_ns: int, paint: bool = True) -> str:

tests/stdout_capture.py

@@ -46,9 +46,9 @@ def stderr(self) -> Optional[str]:
     @property
     def output(self) -> Optional[str]:
         """ stdout and strerr separated by new line """
-        if self.stdout is not None and self.stderr is not None:
-            return "\n".join([self.stdout, self.stderr])
-        elif self.stdout is not None:
+        if self.stdout is not None:
+            if self.stderr is not None:
+                return "\n".join([self.stdout, self.stderr])
             return self.stdout
         return self.stderr