Add more type hints

src/nemu/environ.py

@@ -25,6 +25,7 @@ import socket
 import subprocess
 import sys
 import syslog
+import typing
 from syslog import LOG_ERR, LOG_WARNING, LOG_NOTICE, LOG_INFO, LOG_DEBUG
 from typing import TypeVar, Callable, Optional
 
@@ -164,14 +165,14 @@ def set_log_level(level: int):
     _log_level = level
 
 
-def set_log_output(stream):
+def set_log_output(stream: typing.TextIO):
     "Redirect console messages to the provided stream."
     global _log_stream
     assert hasattr(stream, "write") and hasattr(stream, "flush")
     _log_stream = stream
 
 
-def log_use_syslog(use=True, ident=None, logopt=0,
+def log_use_syslog(use=True, ident: str = None, logopt=0,
                    facility=syslog.LOG_USER):
     "Enable or disable the use of syslog for logging messages."
     global _log_use_syslog, _log_syslog_opts
@@ -207,23 +208,23 @@ def logger(priority: int, message: str):
     _log_stream.flush()
 
 
-def error(message):
+def error(message: str):
     logger(LOG_ERR, message)
 
 
-def warning(message):
+def warning(message: str):
     logger(LOG_WARNING, message)
 
 
-def notice(message):
+def notice(message: str):
     logger(LOG_NOTICE, message)
 
 
-def info(message):
+def info(message: str):
     logger(LOG_INFO, message)
 
 
-def debug(message):
+def debug(message: str):
     logger(LOG_DEBUG, message)
 
 

src/nemu/interface.py

@@ -20,7 +20,7 @@
 
 import os
 import weakref
-from typing import TypedDict
+from typing import TypedDict, Optional
 
 import nemu.iproute
 from nemu.environ import *
@@ -64,7 +64,7 @@ class Interface:
         return self._idx
 
     @property
-    def control(self):
+    def control(self) -> Optional["Interface"]:
         """Associated interface in the main name space (if it exists). Only
         control interfaces can be put into a Switch, for example."""
         return None
@@ -175,7 +175,7 @@ class NodeInterface(NSInterface):
         super(NodeInterface, self).__init__(node, ns.index)
 
     @property
-    def control(self):
+    def control(self) -> "Interface":
         return self._control
 
     def destroy(self):
@@ -334,7 +334,7 @@ class ExternalInterface(Interface):
     namespace."""
 
     @property
-    def control(self):
+    def control(self) -> "Interface":
         # This is *the* control interface
         return self
 
@@ -508,12 +508,12 @@ class Switch(ExternalInterface):
         self._apply_parameters({}, iface.control)
         del self._ports[iface.control.index]
 
-    def set_parameters(self, bandwidth=None,
-                       delay=None, delay_jitter=None,
-                       delay_correlation=None, delay_distribution=None,
-                       loss=None, loss_correlation=None,
-                       dup=None, dup_correlation=None,
-                       corrupt=None, corrupt_correlation=None):
+    def set_parameters(self, bandwidth: int = None,
+                       delay: float = None, delay_jitter: float = None,
+                       delay_correlation: float = None, delay_distribution: str = None,
+                       loss: float = None, loss_correlation: float = None,
+                       dup: float = None, dup_correlation: float = None,
+                       corrupt: float = None, corrupt_correlation: float = None):
         """Set the parameters that control the link characteristics. For the
         description of each, refer to netem documentation:
         http://www.linuxfoundation.org/collaborate/workgroups/networking/netem
@@ -542,6 +542,6 @@ class Switch(ExternalInterface):
             raise
         self._parameters = parameters
 
-    def _apply_parameters(self, parameters, port=None):
+    def _apply_parameters(self, parameters, port: Interface = None):
         for i in [port] if port else list(self._ports.values()):
             nemu.iproute.set_tc(i.index, **parameters)

src/nemu/iproute.py

@@ -878,11 +878,12 @@ def clear_tc(iface):
     execute([TC_PATH, "qdisc", "del", "dev", iface.name, "root"])
 
 
-def set_tc(iface, bandwidth=None, delay=None, delay_jitter=None,
-           delay_correlation=None, delay_distribution=None,
-           loss=None, loss_correlation=None,
-           dup=None, dup_correlation=None,
-           corrupt=None, corrupt_correlation=None):
+def set_tc(iface, bandwidth: int = None,
+           delay: float = None, delay_jitter: float = None,
+           delay_correlation: float = None, delay_distribution: str = None,
+           loss: float = None, loss_correlation: float = None,
+           dup: float = None, dup_correlation: float = None,
+           corrupt: float = None, corrupt_correlation: float = None):
     use_netem = bool(delay or delay_jitter or delay_correlation or
                      delay_distribution or loss or loss_correlation or dup or
                      dup_correlation or corrupt or corrupt_correlation)

src/nemu/protocol.py

@@ -179,7 +179,7 @@ class Server:
         debug("<Reply> %s" % s)
         return
 
-    def readline(self):
+    def readline(self) -> Optional[str]:
         "Read a line from the socket and detect connection break-up."
         # FIXME: should use the eintr_wrapper from environ; why was I using
         # readline instead of read?
@@ -312,7 +312,7 @@ class Server:
         self.reply(200, reply)
 
     def do_QUIT(self, cmdname):
-        self.reply(221, "Sayounara.");
+        self.reply(221, "Sayounara.")
         self._closed = True
 
     def do_PROC_CRTE(self, cmdname, executable, *argv):
@@ -824,23 +824,6 @@ class Client:
         self._forwarder = _spawn_x11_forwarder(server, sock, addr)
 
 
-def _b64_OLD(text: str | bytes) -> str:
-    if text is None:
-        # easier this way
-        text = ''
-    if type(text) is str:
-        btext = text.encode("utf-8")
-    elif type(text) is bytes:
-        btext = text
-    else:
-        btext = text
-    if len(btext) == 0 or any(x for x in btext if x <= ord(" ") or
-                                                  x > ord("z") or x == ord("=")):
-        return "=" + base64.b64encode(btext).decode("ascii")
-    else:
-        return text
-
-
 def _b64(text) -> str:
     if text is None:
         # easier this way