Fix pyright typing errors

pyright is an alternative type checker for Python. It's main advantage
over mypy is that it's a lot faster than mypy which makes it a lot more
pleasant to use in VSCode for in-editor type checking.

This commit fixes the typing errors flagged by pyright.

diff --git a/.github/workflows/ci-unit-test.yml b/.github/workflows/ci-unit-test.yml
index c31523237490ef5674d92fc8aca668c8940cfcb2..04a2bd9c28689b93230121122d16e76446a0e60c 100644 (file)
--- a/.github/workflows/ci-unit-test.yml
+++ b/.github/workflows/ci-unit-test.yml
@@ -15,7 +15,9 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Install
-      run: python3 -m pip install pytest mypy types-dataclasses black isort
+      run: |
+        python3 -m pip install pytest mypy black isort
+        npm install -g pyright
 
     - name: Check formatting
       run: python3 -m black --check mkosi/ tests/
@@ -23,8 +25,11 @@ jobs:
     - name: Check that imports are sorted
       run: python3 -m isort --verbose --check-only mkosi/
 
-    - name: Type Checking
-      run: python3 -m mypy mkosi
+    - name: Type Checking (mypy)
+      run: python3 -m mypy --python-version 3.7 mkosi
+
+    - name: Type Checking (pyright)
+      run: pyright .
 
     - name: Unit Tests
       run: python3 -m pytest
@@ -67,8 +72,6 @@ jobs:
 
     - name: Test venv installation
       run: |
-        sudo apt-get update
-        sudo apt-get -y --no-install-recommends install python3-venv
         python3 -m venv testvenv
         testvenv/bin/python3 -m pip install .
         testvenv/bin/mkosi -h

diff --git a/mkosi/__init__.py b/mkosi/__init__.py
index a7df072de4715fa0e473379125b6b8bba9a5c156..8e22fce60f77dc816a69624ac7aa0f81fa2d0fad 100644 (file)
--- a/mkosi/__init__.py
+++ b/mkosi/__init__.py
@@ -996,27 +996,6 @@ def luks_format(dev: str, passphrase: Dict[str, str]) -> None:
         )
 
 
-def luks_open(dev: str, passphrase: Dict[str, str]) -> str:
-    name = str(uuid.uuid4())
-
-    if passphrase["type"] == "stdin":
-        passphrase_content = (passphrase["content"] + "\n").encode("utf-8")
-        run(["cryptsetup", "open", "--type", "luks", dev, name], input=passphrase_content)
-    else:
-        assert passphrase["type"] == "file"
-        run(["cryptsetup", "--key-file", passphrase["content"], "open", "--type", "luks", dev, name])
-
-    return os.path.join("/dev/mapper", name)
-
-
-def luks_close(dev: Optional[str], text: str) -> None:
-    if dev is None:
-        return
-
-    with complete_step(text):
-        run(["cryptsetup", "close", dev])
-
-
 def luks_format_root(
     args: CommandLineArguments,
     loopdev: str,
@@ -1100,73 +1079,97 @@ def luks_format_tmp(args: CommandLineArguments, loopdev: str, do_run_build_scrip
         luks_format(partition(loopdev, args.tmp_partno), args.passphrase)
 
 
+@contextlib.contextmanager
+def luks_open(dev: str, passphrase: Dict[str, str], partition: str) -> Generator[str, None, None]:
+    name = str(uuid.uuid4())
+
+    with complete_step(f"Setting up LUKS on {partition}…"):
+        if passphrase["type"] == "stdin":
+            passphrase_content = (passphrase["content"] + "\n").encode("utf-8")
+            run(["cryptsetup", "open", "--type", "luks", dev, name], input=passphrase_content)
+        else:
+            assert passphrase["type"] == "file"
+            run(["cryptsetup", "--key-file", passphrase["content"], "open", "--type", "luks", dev, name])
+
+    path = os.path.join("/dev/mapper", name)
+
+    try:
+        yield path
+    finally:
+        with complete_step(f"Closing LUKS {partition}"):
+            run(["cryptsetup", "close", path])
+
+
 def luks_setup_root(
     args: CommandLineArguments, loopdev: str, do_run_build_script: bool, inserting_generated_root: bool = False
-) -> Optional[str]:
+) -> ContextManager[Optional[str]]:
     if args.encrypt != "all":
-        return None
+        return contextlib.nullcontext()
     if args.root_partno is None:
-        return None
+        return contextlib.nullcontext()
     if is_generated_root(args) and not inserting_generated_root:
-        return None
+        return contextlib.nullcontext()
     if do_run_build_script:
-        return None
+        return contextlib.nullcontext()
     assert args.passphrase is not None
 
-    with complete_step("Opening LUKS root partition…"):
-        return luks_open(partition(loopdev, args.root_partno), args.passphrase)
+    return luks_open(partition(loopdev, args.root_partno), args.passphrase, "root partition")
 
 
-def luks_setup_home(args: CommandLineArguments, loopdev: str, do_run_build_script: bool) -> Optional[str]:
+def luks_setup_home(
+    args: CommandLineArguments, loopdev: str, do_run_build_script: bool
+) -> ContextManager[Optional[str]]:
     if args.encrypt is None:
-        return None
+        return contextlib.nullcontext()
     if args.home_partno is None:
-        return None
+        return contextlib.nullcontext()
     if do_run_build_script:
-        return None
+        return contextlib.nullcontext()
     assert args.passphrase is not None
 
-    with complete_step("Opening LUKS home partition…"):
-        return luks_open(partition(loopdev, args.home_partno), args.passphrase)
+    return luks_open(partition(loopdev, args.home_partno), args.passphrase, "home partition")
 
 
-def luks_setup_srv(args: CommandLineArguments, loopdev: str, do_run_build_script: bool) -> Optional[str]:
+def luks_setup_srv(
+    args: CommandLineArguments, loopdev: str, do_run_build_script: bool
+) -> ContextManager[Optional[str]]:
     if args.encrypt is None:
-        return None
+        return contextlib.nullcontext()
     if args.srv_partno is None:
-        return None
+        return contextlib.nullcontext()
     if do_run_build_script:
-        return None
+        return contextlib.nullcontext()
     assert args.passphrase is not None
 
-    with complete_step("Opening LUKS server data partition…"):
-        return luks_open(partition(loopdev, args.srv_partno), args.passphrase)
+    return luks_open(partition(loopdev, args.srv_partno), args.passphrase, "server data partition")
 
 
-def luks_setup_var(args: CommandLineArguments, loopdev: str, do_run_build_script: bool) -> Optional[str]:
+def luks_setup_var(
+    args: CommandLineArguments, loopdev: str, do_run_build_script: bool
+) -> ContextManager[Optional[str]]:
     if args.encrypt is None:
-        return None
+        return contextlib.nullcontext()
     if args.var_partno is None:
-        return None
+        return contextlib.nullcontext()
     if do_run_build_script:
-        return None
+        return contextlib.nullcontext()
     assert args.passphrase is not None
 
-    with complete_step("Opening LUKS variable data partition…"):
-        return luks_open(partition(loopdev, args.var_partno), args.passphrase)
+    return luks_open(partition(loopdev, args.var_partno), args.passphrase, "variable data partition")
 
 
-def luks_setup_tmp(args: CommandLineArguments, loopdev: str, do_run_build_script: bool) -> Optional[str]:
+def luks_setup_tmp(
+    args: CommandLineArguments, loopdev: str, do_run_build_script: bool
+) -> ContextManager[Optional[str]]:
     if args.encrypt is None:
-        return None
+        return contextlib.nullcontext()
     if args.tmp_partno is None:
-        return None
+        return contextlib.nullcontext()
     if do_run_build_script:
-        return None
+        return contextlib.nullcontext()
     assert args.passphrase is not None
 
-    with complete_step("Opening LUKS temporary data partition…"):
-        return luks_open(partition(loopdev, args.tmp_partno), args.passphrase)
+    return luks_open(partition(loopdev, args.tmp_partno), args.passphrase, "temporary data partition")
 
 
 class LuksSetupOutput(NamedTuple):
@@ -1197,34 +1200,20 @@ def luks_setup_all(
         return
     assert loopdev is not None
 
-    try:
-        root = luks_setup_root(args, loopdev, do_run_build_script)
-        try:
-            home = luks_setup_home(args, loopdev, do_run_build_script)
-            try:
-                srv = luks_setup_srv(args, loopdev, do_run_build_script)
-                try:
-                    var = luks_setup_var(args, loopdev, do_run_build_script)
-                    try:
-                        tmp = luks_setup_tmp(args, loopdev, do_run_build_script)
-
-                        yield LuksSetupOutput(
-                            optional_partition(loopdev, args.root_partno) if root is None else root,
-                            optional_partition(loopdev, args.home_partno) if home is None else home,
-                            optional_partition(loopdev, args.srv_partno) if srv is None else srv,
-                            optional_partition(loopdev, args.var_partno) if var is None else var,
-                            optional_partition(loopdev, args.tmp_partno) if tmp is None else tmp,
-                        )
-                    finally:
-                        luks_close(tmp, "Closing LUKS temporary data partition")
-                finally:
-                    luks_close(var, "Closing LUKS variable data partition")
-            finally:
-                luks_close(srv, "Closing LUKS server data partition")
-        finally:
-            luks_close(home, "Closing LUKS home partition")
-    finally:
-        luks_close(root, "Closing LUKS root partition")
+    with contextlib.ExitStack() as stack:
+        root = stack.enter_context(luks_setup_root(args, loopdev, do_run_build_script))
+        home = stack.enter_context(luks_setup_home(args, loopdev, do_run_build_script))
+        srv = stack.enter_context(luks_setup_srv(args, loopdev, do_run_build_script))
+        var = stack.enter_context(luks_setup_var(args, loopdev, do_run_build_script))
+        tmp = stack.enter_context(luks_setup_tmp(args, loopdev, do_run_build_script))
+
+        yield LuksSetupOutput(
+            optional_partition(loopdev, args.root_partno) if root is None else root,
+            optional_partition(loopdev, args.home_partno) if home is None else home,
+            optional_partition(loopdev, args.srv_partno) if srv is None else srv,
+            optional_partition(loopdev, args.var_partno) if var is None else var,
+            optional_partition(loopdev, args.tmp_partno) if tmp is None else tmp,
+        )
 
 
 def prepare_root(args: CommandLineArguments, dev: Optional[str], cached: bool) -> None:
@@ -2953,12 +2942,12 @@ def set_root_password(args: CommandLineArguments, root: str, do_run_build_script
     if args.password == "":
         with complete_step("Deleting root password"):
 
-            def jj(line: str) -> str:
+            def delete_root_pw(line: str) -> str:
                 if line.startswith("root:"):
                     return ":".join(["root", ""] + line.split(":")[2:])
                 return line
 
-            patch_file(os.path.join(root, "etc/passwd"), jj)
+            patch_file(os.path.join(root, "etc/passwd"), delete_root_pw)
     elif args.password:
         with complete_step("Setting root password"):
             if args.password_is_hashed:
@@ -2966,12 +2955,12 @@ def set_root_password(args: CommandLineArguments, root: str, do_run_build_script
             else:
                 password = crypt.crypt(args.password, crypt.mksalt(crypt.METHOD_SHA512))
 
-            def jj(line: str) -> str:
+            def set_root_pw(line: str) -> str:
                 if line.startswith("root:"):
                     return ":".join(["root", password] + line.split(":")[2:])
                 return line
 
-            patch_file(os.path.join(root, "etc/shadow"), jj)
+            patch_file(os.path.join(root, "etc/shadow"), set_root_pw)
 
 
 def invoke_fstrim(args: CommandLineArguments, root: str, do_run_build_script: bool, for_cache: bool) -> None:
@@ -3686,17 +3675,15 @@ def insert_partition(
 
     MkosiPrinter.print_step("Writing partition...")
 
-    if args.root_partno == partno:
-        luks_format_root(args, loopdev, False, False, True)
-        dev = luks_setup_root(args, loopdev, False, True)
-    else:
-        dev = None
+    with contextlib.ExitStack() as stack:
+        if args.root_partno == partno:
+            luks_format_root(args, loopdev, False, False, True)
+            dev = stack.enter_context(luks_setup_root(args, loopdev, False, True))
+        else:
+            dev = None
 
-    path = dev if dev is not None else partition(loopdev, partno)
-    try:
+        path = dev if dev is not None else partition(loopdev, partno)
         run(["dd", f"if={blob.name}", f"of={path}", "conv=nocreat,sparse"])
-    finally:
-        luks_close(dev, "Closing LUKS root partition")
 
     args.ran_sfdisk = True
 

diff --git a/mkosi/backend.py b/mkosi/backend.py
index face357bf8d614edc2119f7b9b9d36dfb9ed9f81..e0c24544d8450edfb42a1e7d4dd41594b4b10a7a 100644 (file)
--- a/mkosi/backend.py
+++ b/mkosi/backend.py
@@ -418,7 +418,7 @@ def do_delay_interrupt() -> Generator[None, None, None]:
     # user can always exit mkosi even if a subprocess hangs by pressing CTRL+C twice.
     interrupted = False
 
-    def handler(signal: int, frame: FrameType) -> None:
+    def handler(signal: int, frame: Optional[FrameType]) -> None:
         nonlocal interrupted
         if interrupted:
             raise KeyboardInterrupt()

diff --git a/tests/test_config_parser.py b/tests/test_config_parser.py
index 98546b0d72b5a2504356016262be63a390a1d296..34d27ce07df84331d57b6628386ff879219b5768 100644 (file)
--- a/tests/test_config_parser.py
+++ b/tests/test_config_parser.py
@@ -4,6 +4,7 @@ import configparser
 import copy
 import os
 
+from typing import Mapping, Any
 import pytest
 
 import mkosi
@@ -140,7 +141,7 @@ class MkosiConfig(object):
             "workspace_dir": None,
         }
 
-    def __eq__(self, other: [mkosi.CommandLineArguments]) -> bool:
+    def __eq__(self, other: Mapping[str, Any]) -> bool:
         """Compare the configuration returned by parse_args against self.reference_config"""
         if len(self.reference_config) != len(other):
             return False
@@ -179,7 +180,7 @@ class MkosiConfig(object):
         if prio < 1000:
             fname = "{:03d}_{}".format(prio, fname)
         config_parser = configparser.RawConfigParser()
-        config_parser.optionxform = str
+        config_parser.optionxform = lambda optionstr: str(optionstr)
 
         # Replace lists in dict before calling config_parser write file
         config_all_normalized = copy.deepcopy(config)
@@ -875,7 +876,7 @@ def test_builtin(tested_config, tmpdir):
     """Test if builtin config and reference config match"""
     with ChangeCwd(tmpdir.strpath):
         if "--all" in tested_config.cli_arguments:
-            with pytest.raises(mkosi.MkosiParseException):
+            with pytest.raises(mkosi.MkosiException):
                 args = mkosi.parse_args(tested_config.cli_arguments)
         else:
             args = mkosi.parse_args(tested_config.cli_arguments)