From 8de08d4169bc1072be7fbd5a34abcda09a1f0f6f Mon Sep 17 00:00:00 2001 From: Daan De Meyer Date: Sun, 8 Aug 2021 13:58:25 +0100 Subject: [PATCH] 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. --- .github/workflows/ci-unit-test.yml | 13 ++- mkosi/__init__.py | 173 +++++++++++++---------------- mkosi/backend.py | 2 +- tests/test_config_parser.py | 7 +- 4 files changed, 93 insertions(+), 102 deletions(-) diff --git a/.github/workflows/ci-unit-test.yml b/.github/workflows/ci-unit-test.yml index c31523237..04a2bd9c2 100644 --- 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 a7df072de..8e22fce60 100644 --- 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 face357bf..e0c24544d 100644 --- 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 98546b0d7..34d27ce07 100644 --- 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) -- 2.47.2