sambacc: complete type annotations

pyproject.toml

@@ -20,3 +20,11 @@ quiet = true
 
 [tool.mypy]
 disallow_incomplete_defs = true
+
+[[tool.mypy.overrides]]
+module = "sambacc.*"
+disallow_untyped_defs = true
+
+[[tool.mypy.overrides]]
+module = "sambacc.commands.*"
+disallow_untyped_defs = false

sambacc/config.py

@@ -90,8 +90,8 @@ def shares(self) -> typing.Iterable[ShareConfig]:
 
 
 class GlobalConfig:
-    def __init__(self, source=None):
-        self.data = {}
+    def __init__(self, source: typing.Optional[typing.IO] = None) -> None:
+        self.data: dict[str, typing.Any] = {}
         if source is not None:
             self.load(source)
 

sambacc/container_dns.py

@@ -16,6 +16,8 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>
 #
 
+from __future__ import annotations
+
 import json
 import subprocess
 import typing
@@ -27,18 +29,22 @@
 
 
 class HostState:
-    def __init__(self, ref="", items=[]):
-        self.ref = ref
-        self.items = items
+    T = typing.TypeVar("T", bound="HostState")
+
+    def __init__(
+        self, ref: str = "", items: typing.Optional[list[HostInfo]] = None
+    ) -> None:
+        self.ref: str = ref
+        self.items: list[HostInfo] = items or []
 
     @classmethod
-    def from_dict(cls, d):
+    def from_dict(cls: typing.Type[T], d: dict[str, typing.Any]) -> T:
         return cls(
             ref=d["ref"],
             items=[HostInfo.from_dict(i) for i in d.get("items", [])],
         )
 
-    def __eq__(self, other):
+    def __eq__(self, other: typing.Any) -> bool:
         return (
             self.ref == other.ref
             and len(self.items) == len(other.items)
@@ -47,32 +53,36 @@ def __eq__(self, other):
 
 
 class HostInfo:
-    def __init__(self, name="", ipv4_addr="", target=""):
+    T = typing.TypeVar("T", bound="HostInfo")
+
+    def __init__(
+        self, name: str = "", ipv4_addr: str = "", target: str = ""
+    ) -> None:
         self.name = name
         self.ipv4_addr = ipv4_addr
         self.target = target
 
     @classmethod
-    def from_dict(cls, d):
+    def from_dict(cls: typing.Type[T], d: dict[str, typing.Any]) -> T:
         return cls(
             name=d["name"],
             ipv4_addr=d["ipv4"],
             target=d.get("target", ""),
         )
 
-    def __eq__(self, other):
+    def __eq__(self, other: typing.Any) -> bool:
         return (
             self.name == other.name
             and self.ipv4_addr == other.ipv4_addr
             and self.target == other.target
         )
 
 
-def parse(fh):
+def parse(fh: typing.IO) -> HostState:
     return HostState.from_dict(json.load(fh))
 
 
-def parse_file(path):
+def parse_file(path: str) -> HostState:
     with open(path) as fh:
         return parse(fh)
 
@@ -115,13 +125,20 @@ def parse_and_update(
 
 
 # TODO: replace this with the common version added to simple_waiter
-def watch(domain, source, update_func, pause_func, print_func=None):
+def watch(
+    domain: str,
+    source: str,
+    update_func: typing.Callable,
+    pause_func: typing.Callable,
+    print_func: typing.Optional[typing.Callable],
+) -> None:
     previous = None
     while True:
         try:
             previous, updated = update_func(domain, source, previous)
         except FileNotFoundError:
-            print_func(f"Source file [{source}] not found")
+            if print_func:
+                print_func(f"Source file [{source}] not found")
             updated = False
             previous = None
         if updated and print_func:

sambacc/ctdb.py

@@ -58,7 +58,11 @@ def next_state(state: NodeState) -> NodeState:
 
 
 class NodeNotPresent(KeyError):
-    def __init__(self, identity, pnn=None):
+    def __init__(
+        self,
+        identity: typing.Any,
+        pnn: typing.Optional[typing.Union[str, int]] = None,
+    ) -> None:
         super().__init__(identity)
         self.identity = identity
         self.pnn = pnn

sambacc/inotify_waiter.py

@@ -69,7 +69,7 @@ def acted(self) -> None:
     def wait(self) -> None:
         next(self._wait())
 
-    def _get_events(self):
+    def _get_events(self) -> list[typing.Any]:
         timeout = 1000 * self.timeout
         self._print("waiting {}ms for activity...".format(timeout))
         events = self._inotify.read(timeout=timeout)
@@ -85,7 +85,7 @@ def _get_events(self):
             and ((event.mask & _inotify.flags.CLOSE_WRITE) != 0)
         ]
 
-    def _wait(self):
+    def _wait(self) -> typing.Iterator[None]:
         while True:
             for event in self._get_events():
                 if event is None:

sambacc/jfile.py

@@ -21,20 +21,23 @@
 import fcntl
 import json
 import os
+import typing
 
 OPEN_RO = os.O_RDONLY
 OPEN_RW = os.O_CREAT | os.O_RDWR
 
 
-def open(path, flags, mode=0o644):
+def open(path: str, flags: int, mode: int = 0o644) -> typing.IO:
     """A wrapper around open to open JSON files for read or read/write.
     `flags` must be os.open type flags. Use `OPEN_RO` and `OPEN_RW` for
     convenience.
     """
     return os.fdopen(os.open(path, flags, mode), "r+")
 
 
-def load(fh, default=None):
+def load(
+    fh: typing.IO, default: typing.Optional[dict[str, typing.Any]] = None
+) -> typing.Any:
     """Similar to json.load, but returns the `default` value if fh refers to an
     empty file. fh must be seekable."""
     if fh.read(4) == "":
@@ -46,7 +49,7 @@ def load(fh, default=None):
     return data
 
 
-def dump(data, fh):
+def dump(data: typing.Any, fh: typing.IO) -> None:
     """Similar to json.dump, but truncates the file before writing in order
     to avoid appending data to the file. fh must be seekable.
     """
@@ -55,6 +58,6 @@ def dump(data, fh):
     json.dump(data, fh)
 
 
-def flock(fh):
+def flock(fh: typing.IO) -> None:
     """A simple wrapper around flock."""
     fcntl.flock(fh.fileno(), fcntl.LOCK_EX)

sambacc/join.py

@@ -29,12 +29,14 @@
 
 
 class JoinError(Exception):
-    def __init__(self, v):
+    def __init__(self, v: typing.Any) -> None:
         super().__init__(v)
-        self.errors = []
+        self.errors: list[typing.Any] = []
 
 
 _PROMPT = object()
+_PT = typing.TypeVar("_PT")
+_PW = typing.Union[str, _PT]
 
 
 class JoinBy(enum.Enum):
@@ -47,9 +49,13 @@ class UserPass:
     """Encapsulate a username/password pair."""
 
     username: str = "Administrator"
-    password: typing.Optional[str] = None
+    password: typing.Optional[_PW] = None
 
-    def __init__(self, username=None, password=None):
+    def __init__(
+        self,
+        username: typing.Optional[str] = None,
+        password: typing.Optional[_PW] = None,
+    ) -> None:
         if username is not None:
             self.username = username
         if password is not None:
@@ -65,8 +71,8 @@ class Joiner:
 
     _net_ads_join = samba_cmds.net["ads", "join"]
 
-    def __init__(self, marker=None):
-        self._sources = []
+    def __init__(self, marker: typing.Optional[str] = None) -> None:
+        self._sources: list[tuple[JoinBy, typing.Any]] = []
         self.marker = marker
 
     def add_source(

sambacc/netcmd_loader.py

@@ -43,7 +43,9 @@ def template_config(
 class NetCmdLoader:
     _net_conf = samba_cmds.net["conf"]
 
-    def _cmd(self, *args, **kwargs):
+    def _cmd(
+        self, *args: str, **kwargs: typing.Any
+    ) -> tuple[list[str], typing.Any]:
         cmd = list(self._net_conf[args])
         return cmd, subprocess.Popen(cmd, **kwargs)
 

sambacc/nsswitch_loader.py

@@ -22,10 +22,10 @@
 
 
 class NameServiceSwitchLoader(TextFileLoader):
-    def __init__(self, path):
+    def __init__(self, path: str) -> None:
         super().__init__(path)
-        self.lines = []
-        self.idx = {}
+        self.lines: list[str] = []
+        self.idx: dict[str, int] = {}
 
     def loadlines(self, lines: typing.Iterable[str]) -> None:
         """Load in the lines from the text source."""

sambacc/passdb_loader.py

@@ -16,6 +16,8 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>
 #
 
+import typing
+
 from sambacc import config
 
 # Do the samba python bindings not export any useful constants?
@@ -24,15 +26,15 @@
 ACB_PWNOEXP = 0x00000200
 
 
-def _samba_modules():
+def _samba_modules() -> tuple[typing.Any, typing.Any]:
     from samba.samba3 import param  # type: ignore
     from samba.samba3 import passdb  # type: ignore
 
     return param, passdb
 
 
 class PassDBLoader:
-    def __init__(self, smbconf=None):
+    def __init__(self, smbconf: typing.Any = None) -> None:
         param, passdb = _samba_modules()
         lp = param.get_context()
         if smbconf is None:

sambacc/passwd_loader.py

@@ -23,9 +23,9 @@
 
 
 class LineFileLoader(TextFileLoader):
-    def __init__(self, path):
+    def __init__(self, path: str) -> None:
         super().__init__(path)
-        self.lines = []
+        self.lines: list[str] = []
 
     def loadlines(self, lines: typing.Iterable[str]) -> None:
         """Load in the lines from the text source."""
@@ -43,9 +43,9 @@ def dumplines(self) -> typing.Iterable[str]:
 
 
 class PasswdFileLoader(LineFileLoader):
-    def __init__(self, path="/etc/passwd"):
+    def __init__(self, path: str = "/etc/passwd") -> None:
         super().__init__(path)
-        self._usernames = set()
+        self._usernames: set[str] = set()
 
     def readfp(self, fp: typing.IO) -> None:
         super().readfp(fp)
@@ -66,9 +66,9 @@ def add_user(self, user_entry: config.UserEntry) -> None:
 
 
 class GroupFileLoader(LineFileLoader):
-    def __init__(self, path="/etc/group"):
+    def __init__(self, path: str = "/etc/group") -> None:
         super().__init__(path)
-        self._groupnames = set()
+        self._groupnames: set[str] = set()
 
     def readfp(self, fp: typing.IO) -> None:
         super().readfp(fp)

sambacc/permissions.py

@@ -124,22 +124,22 @@ def path(self) -> str:
     def _full_path(self) -> str:
         return os.path.join(self._root, self._path.lstrip("/"))
 
-    def has_status(self):
+    def has_status(self) -> bool:
         try:
             self._get_status()
             return True
         except KeyError:
             return False
 
-    def status_ok(self):
+    def status_ok(self) -> bool:
         try:
             sval = self._get_status()
         except KeyError:
             return False
         curr_prefix = sval.split("/")[0]
         return curr_prefix == self._prefix
 
-    def update(self):
+    def update(self) -> None:
         if self.status_ok():
             return
         self._set_perms()
@@ -182,6 +182,6 @@ class AlwaysPosixPermsHandler(InitPosixPermsHandler):
     May be useful for testing and debugging.
     """
 
-    def update(self):
+    def update(self) -> None:
         self._set_perms()
         self._set_status()

sambacc/samba_cmds.py

@@ -159,19 +159,19 @@ def __repr__(self) -> str:
 samba_dc = SambaCommand("/usr/sbin/samba")
 
 
-def smbd_foreground():
+def smbd_foreground() -> SambaCommand:
     return smbd[
         "--foreground", _daemon_stdout_opt("smbd"), "--no-process-group"
     ]
 
 
-def winbindd_foreground():
+def winbindd_foreground() -> SambaCommand:
     return winbindd[
         "--foreground", _daemon_stdout_opt("winbindd"), "--no-process-group"
     ]
 
 
-def samba_dc_foreground():
+def samba_dc_foreground() -> SambaCommand:
     return samba_dc["--foreground"]