More type annotations

pyproject.toml

@@ -17,3 +17,6 @@ version = "{version}"
 [tool.black]
 line-length = 79
 quiet = true
+
+[tool.mypy]
+disallow_incomplete_defs = true

sambacc/addc.py

@@ -83,7 +83,7 @@ def create_group(name: str) -> None:
     subprocess.check_call(cmd)
 
 
-def add_group_members(group_name: str, members: typing.List[str]) -> None:
+def add_group_members(group_name: str, members: list[str]) -> None:
     cmd = _group_add_members_cmd(group_name, members)
     _logger.info("Adding group members: %r", cmd)
     subprocess.check_call(cmd)
@@ -95,7 +95,7 @@ def _provision_cmd(
     admin_password: str,
     dns_backend: typing.Optional[str] = None,
     domain: typing.Optional[str] = None,
-) -> typing.List[str]:
+) -> list[str]:
     if not dns_backend:
         dns_backend = "SAMBA_INTERNAL"
     if not domain:
@@ -120,7 +120,7 @@ def _join_cmd(
     admin_password: str,
     dns_backend: typing.Optional[str] = None,
     domain: typing.Optional[str] = None,
-) -> typing.List[str]:
+) -> list[str]:
     if not dns_backend:
         dns_backend = "SAMBA_INTERNAL"
     if not domain:
@@ -143,7 +143,7 @@ def _user_create_cmd(
     password: str,
     surname: typing.Optional[str],
     given_name: typing.Optional[str],
-) -> typing.List[str]:
+) -> list[str]:
     cmd = samba_cmds.sambatool[
         "user",
         "create",
@@ -157,7 +157,7 @@ def _user_create_cmd(
     return cmd
 
 
-def _group_add_cmd(name: str) -> typing.List[str]:
+def _group_add_cmd(name: str) -> list[str]:
     cmd = samba_cmds.sambatool[
         "group",
         "add",
@@ -166,9 +166,7 @@ def _group_add_cmd(name: str) -> typing.List[str]:
     return cmd
 
 
-def _group_add_members_cmd(
-    group_name: str, members: typing.List[str]
-) -> typing.List[str]:
+def _group_add_members_cmd(group_name: str, members: list[str]) -> list[str]:
     cmd = samba_cmds.sambatool[
         "group",
         "addmembers",

sambacc/commands/cli.py

@@ -39,12 +39,23 @@ class Fail(ValueError):
     pass
 
 
+class Parser(typing.Protocol):
+    def frank(self, x: str) -> str:
+        ...  # pragma: no cover
+
+    def set_defaults(self, **kwargs: typing.Any) -> None:
+        ...  # pragma: no cover
+
+    def add_argument(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> typing.Any:
+        ...  # pragma: no cover
+
+
 Command = namedtuple("Command", "name cmd_func arg_func cmd_help")
 
 
-def toggle_option(
-    parser: argparse.ArgumentParser, arg: str, dest: str, helpfmt: str
-) -> argparse.ArgumentParser:
+def toggle_option(parser: Parser, arg: str, dest: str, helpfmt: str) -> Parser:
     parser.add_argument(
         arg,
         action="store_true",
@@ -69,7 +80,7 @@ def get_help(cmd: Command) -> str:
     return ""
 
 
-def add_command(subparsers, cmd) -> None:
+def add_command(subparsers: typing.Any, cmd: Command) -> None:
     subparser = subparsers.add_parser(cmd.name, help=get_help(cmd))
     subparser.set_defaults(cfunc=cmd.cmd_func)
     if cmd.arg_func is not None:
@@ -96,7 +107,9 @@ def _wrapper(f):
 
         return _wrapper
 
-    def assemble(self, arg_func=None) -> argparse.ArgumentParser:
+    def assemble(
+        self, arg_func: typing.Callable = None
+    ) -> argparse.ArgumentParser:
         parser = argparse.ArgumentParser()
         if arg_func is not None:
             arg_func(parser)
@@ -105,7 +118,7 @@ def assemble(self, arg_func=None) -> argparse.ArgumentParser:
             add_command(subparsers, cmd)
         return parser
 
-    def dict(self) -> typing.Dict[str, Command]:
+    def dict(self) -> dict[str, Command]:
         """Return a dict mapping command names to Command object."""
         return {c.name: c for c in self._commands}
 

sambacc/commands/config.py

@@ -127,7 +127,8 @@ def _call_if_leader(
                 _logger.info("skipping config update. node not leader")
                 return previous, False
             _logger.info("checking for update. node is leader")
-            return cond_func(current, previous)
+            result = cond_func(current, previous)
+        return result
 
     return _call_if_leader
 

sambacc/commands/dcmain.py

@@ -16,6 +16,8 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>
 #
 
+import typing
+
 from . import addc
 from .cli import Fail
 from .main import (
@@ -31,7 +33,7 @@
 default_cfunc = addc.summary
 
 
-def main(args=None) -> None:
+def main(args: typing.Optional[typing.Sequence[str]] = None) -> None:
     cli = addc.dccommands.assemble(arg_func=global_args).parse_args(args)
     env_to_cli(cli)
     enable_logging(cli)

sambacc/commands/dns.py

@@ -91,12 +91,13 @@ def leader_update_func(
         domain: str,
         source: str,
         previous: typing.Optional[container_dns.HostState] = None,
-    ):
+    ) -> typing.Tuple[typing.Optional[container_dns.HostState], bool]:
         with best_leader_locator(iconfig) as ll:
             if not ll.is_leader():
                 _logger.info("skipping dns update. node not leader")
                 return previous, False
             _logger.info("checking for update. node is leader")
-            return update_func(domain, source, previous)
+            result = update_func(domain, source, previous)
+        return result
 
     return leader_update_func

sambacc/commands/initialize.py

@@ -17,6 +17,7 @@
 #
 
 import logging
+import typing
 
 from sambacc import ctdb
 from sambacc import paths
@@ -105,7 +106,9 @@ def setup_step_names():
 
 
 @commands.command(name="init")
-def init_container(ctx: Context, steps=None) -> None:
+def init_container(
+    ctx: Context, steps: typing.Optional[typing.Iterable[str]] = None
+) -> None:
     """Initialize the entire container environment."""
     steps = _default_setup_steps if steps is None else list(steps)
     cmds = setup_steps.dict()

sambacc/commands/join.py

@@ -17,19 +17,27 @@
 #
 
 import sys
+import typing
 
 import sambacc.join as joinutil
 
-from .cli import commands, Context, toggle_option, best_waiter, Fail
+from .cli import (
+    Context,
+    Fail,
+    Parser,
+    best_waiter,
+    commands,
+    toggle_option,
+)
 
 
-def _print_join_error(err) -> None:
+def _print_join_error(err: typing.Any) -> None:
     print(f"ERROR: {err}", file=sys.stderr)
     for suberr in getattr(err, "errors", []):
         print(f"  - {suberr}", file=sys.stderr)
 
 
-def _add_join_sources(joiner, cli) -> None:
+def _add_join_sources(joiner: joinutil.Joiner, cli: typing.Any) -> None:
     if cli.insecure or getattr(cli, "insecure_auto_join", False):
         upass = joinutil.UserPass(cli.username, cli.password)
         joiner.add_source(joinutil.JoinBy.PASSWORD, upass)
@@ -41,7 +49,7 @@ def _add_join_sources(joiner, cli) -> None:
         joiner.add_source(joinutil.JoinBy.INTERACTIVE, upass)
 
 
-def _join_args(parser) -> None:
+def _join_args(parser: Parser) -> None:
     parser.set_defaults(insecure=False, files=True, interactive=True)
     toggle_option(
         parser,
@@ -88,7 +96,7 @@ def join(ctx: Context) -> None:
         raise Fail("failed to join to a domain")
 
 
-def _must_join_args(parser) -> None:
+def _must_join_args(parser: Parser) -> None:
     parser.set_defaults(insecure=False, files=True, wait=True)
     toggle_option(
         parser,

sambacc/commands/main.py

@@ -33,15 +33,15 @@
 from . import join  # noqa: F401
 from . import run  # noqa: F401
 from . import users  # noqa: F401
-from .cli import commands, Fail
+from .cli import commands, Fail, Parser
 
 DEFAULT_CONFIG = "/etc/samba/container/config.json"
 DEFAULT_JOIN_MARKER = "/var/lib/samba/container-join-marker.json"
 
 default_cfunc = config_cmds.print_config
 
 
-def global_args(parser) -> None:
+def global_args(parser: Parser) -> None:
     parser.add_argument(
         "--config",
         action="append",
@@ -101,7 +101,13 @@ def global_args(parser) -> None:
     )
 
 
-def from_env(ns, var, ename, default=None, vtype=str) -> None:
+def from_env(
+    ns: typing.Any,
+    var: str,
+    ename: str,
+    default: typing.Any = None,
+    vtype: typing.Optional[typing.Callable] = str,
+) -> None:
     value = getattr(ns, var, None)
     if not value:
         value = os.environ.get(ename, "")
@@ -123,7 +129,7 @@ def split_paths(value):
     return out
 
 
-def env_to_cli(cli) -> None:
+def env_to_cli(cli: typing.Any) -> None:
     from_env(
         cli,
         "config",
@@ -165,7 +171,7 @@ def instance_config(self) -> config.InstanceConfig:
         return self._iconfig
 
 
-def pre_action(cli) -> None:
+def pre_action(cli: typing.Any) -> None:
     """Handle debugging/diagnostic related options before the target
     action of the command is performed.
     """
@@ -177,7 +183,7 @@ def pre_action(cli) -> None:
         samba_cmds.set_global_prefix([cli.samba_command_prefix])
 
 
-def enable_logging(cli) -> None:
+def enable_logging(cli: typing.Any) -> None:
     logger = logging.getLogger()
     logger.setLevel(logging.INFO)
     handler = logging.StreamHandler()
@@ -188,14 +194,14 @@ def enable_logging(cli) -> None:
     logger.addHandler(handler)
 
 
-def action_filter(cli) -> typing.Optional[str]:
+def action_filter(cli: typing.Any) -> typing.Optional[str]:
     for path in cli.skip_if_file or []:
         if os.path.exists(path):
             return f"skip-if-file: {path} exists"
     return None
 
 
-def main(args=None) -> None:
+def main(args: typing.Optional[typing.Sequence[str]] = None) -> None:
     cli = commands.assemble(arg_func=global_args).parse_args(args)
     env_to_cli(cli)
     enable_logging(cli)

sambacc/config.py

@@ -40,7 +40,7 @@
 FEATURES: typing.Final[str] = "instance_features"
 
 
-def read_config_files(fnames) -> GlobalConfig:
+def read_config_files(fnames: list[str]) -> GlobalConfig:
     """Read the global container config from the given filenames.
     At least one of the files from the fnames list must exist and contain
     a valid config. If none of the file names exist an error will be raised.
@@ -68,7 +68,7 @@ def read_config_files(fnames) -> GlobalConfig:
     return gconfig
 
 
-def check_config_data(data) -> dict:
+def check_config_data(data: dict[str, typing.Any]) -> dict[str, typing.Any]:
     """Return the config data or raise a ValueError if the config
     is invalid or incomplete.
     """
@@ -162,7 +162,7 @@ def ctdb_smb_config(self) -> CTDBSambaConfig:
             raise ValueError("ctdb not supported in configuration")
         return CTDBSambaConfig()
 
-    def ctdb_config(self) -> typing.Dict[str, str]:
+    def ctdb_config(self) -> dict[str, str]:
         """Common configuration of CTDB itself."""
         if not self.with_ctdb:
             return {}

sambacc/container_dns.py

@@ -77,11 +77,16 @@ def parse_file(path):
         return parse(fh)
 
 
-def match_target(state: HostState, target_name: str) -> typing.List[HostInfo]:
+def match_target(state: HostState, target_name: str) -> list[HostInfo]:
     return [h for h in state.items if h.target == target_name]
 
 
-def register(domain, hs, prefix=None, target_name: str = EXTERNAL) -> bool:
+def register(
+    domain: str,
+    hs: HostState,
+    prefix: typing.Optional[list[str]] = None,
+    target_name: str = EXTERNAL,
+) -> bool:
     updated = False
     for item in match_target(hs, target_name):
         ip = item.ipv4_addr
@@ -99,7 +104,7 @@ def parse_and_update(
     source: str,
     previous: typing.Optional[HostState] = None,
     target_name: str = EXTERNAL,
-    reg_func=register,
+    reg_func: typing.Callable = register,
 ) -> typing.Tuple[HostState, bool]:
     hs = parse_file(source)
     if previous is not None and hs == previous: