Add type annotations to download.py and fetcher.py

Run mypy with --install-types and --non-interactive
options to install all suggested stub packages.
In our case this is the stub for the requests package.

Include urllib3 to the ignored imports.

Signed-off-by: Teodora Sechkova <tsechkova@vmware.com>

Author: sechkova

setup.cfg

@@ -22,3 +22,6 @@ files =
 
 [mypy-securesystemslib.*]
 ignore_missing_imports = True
+
+[mypy-urllib3.*]
+ignore_missing_imports = True

tox.ini

@@ -51,6 +51,6 @@ commands =
     # work, unfortunately each subdirectory has to be ignored explicitly.
     pylint -j 0 tuf --ignore=tuf/api,tuf/api/serialization,tuf/ngclient,tuf/ngclient/_internal
 
-    mypy
+    mypy --install-types --non-interactive
 
     bandit -r tuf

tuf/ngclient/_internal/download.py

@@ -26,18 +26,24 @@
 import logging
 import tempfile
 import timeit
+from typing import IO
 from urllib import parse
 
 from securesystemslib import formats as sslib_formats
 
-import tuf
-from tuf import exceptions, formats
+from tuf import exceptions, formats, settings
+from tuf.ngclient import FetcherInterface
 
 # See 'log.py' to learn how logging is handled in TUF.
 logger = logging.getLogger(__name__)
 
 
-def download_file(url, required_length, fetcher, strict_required_length=True):
+def download_file(
+    url: str,
+    required_length: int,
+    fetcher: FetcherInterface,
+    strict_required_length: bool = True,
+) -> IO[bytes]:
     """
     <Purpose>
       Given the url and length of the desired file, this function opens a
@@ -92,7 +98,7 @@ def download_file(url, required_length, fetcher, strict_required_length=True):
     # the downloaded file.
     temp_file = tempfile.TemporaryFile()  # pylint: disable=consider-using-with
 
-    average_download_speed = 0
+    average_download_speed = 0.0
     number_of_bytes_received = 0
 
     try:
@@ -110,7 +116,7 @@ def download_file(url, required_length, fetcher, strict_required_length=True):
                 number_of_bytes_received / seconds_spent_receiving
             )
 
-            if average_download_speed < tuf.settings.MIN_AVERAGE_DOWNLOAD_SPEED:
+            if average_download_speed < settings.MIN_AVERAGE_DOWNLOAD_SPEED:
                 logger.debug(
                     "The average download speed dropped below the minimum"
                     " average download speed set in tuf.settings.py."
@@ -142,7 +148,12 @@ def download_file(url, required_length, fetcher, strict_required_length=True):
         return temp_file
 
 
-def download_bytes(url, required_length, fetcher, strict_required_length=True):
+def download_bytes(
+    url: str,
+    required_length: int,
+    fetcher: FetcherInterface,
+    strict_required_length: bool = True,
+) -> bytes:
     """Download bytes from given url
 
     Returns the downloaded bytes, otherwise like download_file()
@@ -154,11 +165,11 @@ def download_bytes(url, required_length, fetcher, strict_required_length=True):
 
 
 def _check_downloaded_length(
-    total_downloaded,
-    required_length,
-    strict_required_length=True,
-    average_download_speed=None,
-):
+    total_downloaded: int,
+    required_length: int,
+    strict_required_length: bool,
+    average_download_speed: float,
+) -> None:
     """
     <Purpose>
       A helper function which checks whether the total number of downloaded
@@ -216,7 +227,7 @@ def _check_downloaded_length(
 
             # If the average download speed is below a certain threshold, we
             # flag this as a possible slow-retrieval attack.
-            if average_download_speed < tuf.settings.MIN_AVERAGE_DOWNLOAD_SPEED:
+            if average_download_speed < settings.MIN_AVERAGE_DOWNLOAD_SPEED:
                 raise exceptions.SlowRetrievalError
 
             raise exceptions.DownloadLengthMismatchError(
@@ -228,7 +239,7 @@ def _check_downloaded_length(
         # download the Timestamp or Root metadata, for which we have no
         # signed metadata; so, we must guess a reasonable required_length
         # for it.
-        if average_download_speed < tuf.settings.MIN_AVERAGE_DOWNLOAD_SPEED:
+        if average_download_speed < settings.MIN_AVERAGE_DOWNLOAD_SPEED:
             raise exceptions.SlowRetrievalError
 
         logger.debug(

tuf/ngclient/_internal/requests_fetcher.py

@@ -7,7 +7,7 @@
 
 import logging
 import time
-from typing import Optional
+from typing import Dict, Iterator, Optional
 from urllib import parse
 
 # Imports
@@ -31,7 +31,7 @@ class RequestsFetcher(FetcherInterface):
             session per scheme+hostname combination.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         # http://docs.python-requests.org/en/master/user/advanced/#session-objects:
         #
         # "The Session object allows you to persist certain parameters across
@@ -46,14 +46,14 @@ def __init__(self):
         # improve efficiency, but avoiding sharing state between different
         # hosts-scheme combinations to minimize subtle security issues.
         # Some cookies may not be HTTP-safe.
-        self._sessions = {}
+        self._sessions: Dict[str, requests.Session] = {}
 
         # Default settings
         self.socket_timeout: int = 4  # seconds
         self.chunk_size: int = 400000  # bytes
         self.sleep_before_round: Optional[int] = None
 
-    def fetch(self, url, required_length):
+    def fetch(self, url: str, required_length: int) -> Iterator[bytes]:
         """Fetches the contents of HTTP/HTTPS url from a remote server.
 
         Ensures the length of the downloaded data is up to 'required_length'.
@@ -93,7 +93,7 @@ def fetch(self, url, required_length):
         # Define a generator function to be returned by fetch. This way the
         # caller of fetch can differentiate between connection and actual data
         # download and measure download times accordingly.
-        def chunks():
+        def chunks() -> Iterator[bytes]:
             try:
                 bytes_received = 0
                 while True:
@@ -143,7 +143,7 @@ def chunks():
 
         return chunks()
 
-    def _get_session(self, url):
+    def _get_session(self, url: str) -> requests.Session:
         """Returns a different customized requests.Session per schema+hostname
         combination.
         """

tuf/ngclient/fetcher.py

@@ -6,6 +6,7 @@
 
 # Imports
 import abc
+from typing import Iterator
 
 
 # Classes
@@ -20,7 +21,7 @@ class FetcherInterface:
     __metaclass__ = abc.ABCMeta
 
     @abc.abstractmethod
-    def fetch(self, url, required_length):
+    def fetch(self, url: str, required_length: int) -> Iterator[bytes]:
         """Fetches the contents of HTTP/HTTPS url from a remote server.
 
         Ensures the length of the downloaded data is up to 'required_length'.