linters: set disallow_untyped_defs in mypy

craft_parts/actions.py

@@ -49,7 +49,7 @@ class ActionType(enum.IntEnum):
     UPDATE = 3
     REAPPLY = 4
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"{self.__class__.__name__}.{self.name}"
 
 

craft_parts/callbacks.py

@@ -126,13 +126,13 @@ def run_post_step(step_info: StepInfo) -> None:
     return _run_step(hook_list=_POST_STEP_HOOKS, step_info=step_info)
 
 
-def _run_step(*, hook_list: List[CallbackHook], step_info: StepInfo):
+def _run_step(*, hook_list: List[CallbackHook], step_info: StepInfo) -> None:
     for hook in hook_list:
         if not hook.step_list or step_info.step in hook.step_list:
             hook.function(step_info)
 
 
-def _ensure_not_defined(func: Callback, hook_list: List[CallbackHook]):
+def _ensure_not_defined(func: Callback, hook_list: List[CallbackHook]) -> None:
     for hook in hook_list:
         if func == hook.function:
             raise errors.CallbackRegistrationError(

craft_parts/ctl.py

@@ -52,7 +52,7 @@ def run(cls, cmd: str, args: List[str]) -> Optional[str]:
         raise RuntimeError(f"invalid command {cmd!r}")
 
 
-def _client(cmd: str, args: List[str]):
+def _client(cmd: str, args: List[str]) -> Optional[str]:
     """Execute a command in the running step processor.
 
     The control protocol client allows a user scriptlet to execute
@@ -101,7 +101,7 @@ def _client(cmd: str, args: List[str]):
     return retval
 
 
-def main():
+def main() -> None:
     """Run the ctl client cli."""
     if len(sys.argv) < 2:
         print(f"usage: {sys.argv[0]} <command> [arguments]")

craft_parts/errors.py

@@ -17,10 +17,11 @@
 """Craft parts errors."""
 
 import dataclasses
+import pathlib
 from typing import TYPE_CHECKING, List, Optional, Set
 
 if TYPE_CHECKING:
-    from pydantic.error_wrappers import ErrorDict
+    from pydantic.error_wrappers import ErrorDict, Loc
 
 
 @dataclasses.dataclass(repr=True)
@@ -120,7 +121,9 @@ def __init__(self, *, part_name: str, message: str):
         super().__init__(brief=brief, details=details, resolution=resolution)
 
     @classmethod
-    def from_validation_error(cls, *, part_name: str, error_list: List["ErrorDict"]):
+    def from_validation_error(
+        cls, *, part_name: str, error_list: List["ErrorDict"]
+    ) -> "PartSpecificationError":
         """Create a PartSpecificationError from a pydantic error list.
 
         :param part_name: The name of the part being processed.
@@ -146,7 +149,7 @@ def from_validation_error(cls, *, part_name: str, error_list: List["ErrorDict"])
         return cls(part_name=part_name, message="\n".join(formatted_errors))
 
     @classmethod
-    def _format_loc(cls, loc):
+    def _format_loc(cls, loc: "Loc") -> str:
         """Format location."""
         loc_parts = []
         for loc_part in loc:
@@ -160,11 +163,11 @@ def _format_loc(cls, loc):
             else:
                 raise RuntimeError(f"unhandled loc: {loc_part}")
 
-        loc = ".".join(loc_parts)
+        loc_str = ".".join(loc_parts)
 
         # Filter out internal __root__ detail.
-        loc = loc.replace(".__root__", "")
-        return loc
+        loc_str = loc_str.replace(".__root__", "")
+        return loc_str
 
 
 class CopyTreeError(PartsError):
@@ -266,7 +269,7 @@ def __init__(self, plugin_name: str, *, part_name: str):
 class OsReleaseIdError(PartsError):
     """Failed to determine the host operating system identification string."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         brief = "Unable to determine the host operating system ID."
 
         super().__init__(brief=brief)
@@ -275,7 +278,7 @@ def __init__(self):
 class OsReleaseNameError(PartsError):
     """Failed to determine the host operating system name."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         brief = "Unable to determine the host operating system name."
 
         super().__init__(brief=brief)
@@ -284,7 +287,7 @@ def __init__(self):
 class OsReleaseVersionIdError(PartsError):
     """Failed to determine the host operating system version."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         brief = "Unable to determine the host operating system version ID."
 
         super().__init__(brief=brief)
@@ -293,7 +296,7 @@ def __init__(self):
 class OsReleaseCodenameError(PartsError):
     """Failed to determine the host operating system version codename."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         brief = "Unable to determine the host operating system codename."
 
         super().__init__(brief=brief)
@@ -306,7 +309,7 @@ class FilesetError(PartsError):
     :param message: The error message.
     """
 
-    def __init__(self, *, name: str, message: str):
+    def __init__(self, *, name: str, message: str) -> None:
         self.name = name
         self.message = message
         brief = f"{name!r} fileset error: {message}."
@@ -321,7 +324,7 @@ class FilesetConflict(PartsError):
     :param conflicting_files: A set containing the conflicting file names.
     """
 
-    def __init__(self, conflicting_files: Set[str]):
+    def __init__(self, conflicting_files: Set[str]) -> None:
         self.conflicting_files = conflicting_files
         brief = "Failed to filter files: inconsistent 'stage' and 'prime' filesets."
         details = (
@@ -342,7 +345,7 @@ class FileOrganizeError(PartsError):
     :param message: The error message.
     """
 
-    def __init__(self, *, part_name, message):
+    def __init__(self, *, part_name: str, message: str) -> None:
         self.part_name = part_name
         self.message = message
         brief = f"Failed to organize part {part_name!r}: {message}."
@@ -360,7 +363,7 @@ class PartFilesConflict(PartsError):
 
     def __init__(
         self, *, part_name: str, other_part_name: str, conflicting_files: List[str]
-    ):
+    ) -> None:
         self.part_name = part_name
         self.other_part_name = other_part_name
         self.conflicting_files = conflicting_files
@@ -386,7 +389,7 @@ class StageFilesConflict(PartsError):
     :param conflicting_files: The list of confictling files.
     """
 
-    def __init__(self, *, part_name: str, conflicting_files: List[str]):
+    def __init__(self, *, part_name: str, conflicting_files: List[str]) -> None:
         self.part_name = part_name
         self.conflicting_files = conflicting_files
         indented_conflicting_files = (f"    {i}" for i in conflicting_files)
@@ -546,7 +549,9 @@ def __init__(self) -> None:
 class DebError(PartsError):
     """A "deb"-related command failed."""
 
-    def __init__(self, deb_path, command, exit_code):
+    def __init__(
+        self, deb_path: pathlib.Path, command: List[str], exit_code: int
+    ) -> None:
         brief = (
             f"Failed when handling {deb_path}: "
             f"command {command!r} exited with code {exit_code}."

craft_parts/executor/executor.py

@@ -19,7 +19,7 @@
 import logging
 import shutil
 from pathlib import Path
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from craft_parts import callbacks, overlays, packages, parts, plugins
 from craft_parts.actions import Action, ActionType
@@ -280,7 +280,7 @@ def __enter__(self) -> "ExecutionContext":
         self._executor.prologue()
         return self
 
-    def __exit__(self, *exc):
+    def __exit__(self, *exc: Any) -> None:
         self._executor.epilogue()
 
     def execute(

craft_parts/executor/filesets.py

@@ -30,7 +30,7 @@ def __init__(self, entries: List[str], *, name: str = ""):
         self._name = name
         self._list = entries
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"Fileset({self._list!r}, name={self._name!r})"
 
     @property

craft_parts/executor/migration.py

@@ -19,7 +19,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import Dict, List, Optional, Set, Tuple
+from typing import Callable, Dict, List, Optional, Set, Tuple
 
 from craft_parts import overlays
 from craft_parts.permissions import Permissions, filter_permissions
@@ -38,7 +38,7 @@ def migrate_files(
     missing_ok: bool = False,
     follow_symlinks: bool = False,
     oci_translation: bool = False,
-    fixup_func=lambda *args: None,
+    fixup_func: Callable[..., None] = lambda *args: None,
     permissions: Optional[List[Permissions]] = None,
 ) -> Tuple[Set[str], Set[str]]:
     """Copy or link files from a directory to another.

craft_parts/executor/part_handler.py

@@ -22,7 +22,7 @@
 import shutil
 from glob import iglob
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Set, cast
+from typing import Any, Callable, Dict, List, Optional, Sequence, Set, cast
 
 from typing_extensions import Protocol
 
@@ -269,7 +269,7 @@ def _run_build(
         *,
         stdout: Stream,
         stderr: Stream,
-        update=False,
+        update: bool = False,
     ) -> StepState:
         """Execute the build step for this part.
 
@@ -648,7 +648,9 @@ def _reapply_action(
                 f"cannot reapply step {step_name!r} of {self._part.name!r}"
             )
 
-    def _reapply_overlay(self, step_info, *, stdout: Stream, stderr: Stream) -> None:
+    def _reapply_overlay(
+        self, step_info: StepInfo, *, stdout: Stream, stderr: Stream
+    ) -> None:
         """Clean and repopulate the current part's layer, keeping its state."""
         shutil.rmtree(self._part.part_layer_dir)
         self._run_overlay(step_info, stdout=stdout, stderr=stderr)
@@ -847,7 +849,7 @@ def _clean_shared(self, step: Step, *, shared_dir: Path) -> None:
             )
             overlay_migration_state_path.unlink()
 
-    def _make_dirs(self):
+    def _make_dirs(self) -> None:
         dirs = [
             self._part.part_src_dir,
             self._part.part_build_dir,
@@ -861,7 +863,7 @@ def _make_dirs(self):
         for dir_name in dirs:
             os.makedirs(dir_name, exist_ok=True)
 
-    def _organize(self, *, overwrite=False):
+    def _organize(self, *, overwrite: bool = False) -> None:
         mapping = self._part.spec.organize_files
         organize_files(
             part_name=self._part.name,
@@ -894,7 +896,7 @@ def _fetch_stage_packages(self, *, step_info: StepInfo) -> Optional[List[str]]:
 
         return fetched_packages
 
-    def _fetch_stage_snaps(self):
+    def _fetch_stage_snaps(self) -> Optional[Sequence[str]]:
         """Download snap packages to the part's snap directory."""
         stage_snaps = self._part.spec.stage_snaps
         if not stage_snaps:
@@ -923,7 +925,7 @@ def _fetch_overlay_packages(self) -> None:
                 part_name=self._part.name, package_name=err.package_name
             )
 
-    def _unpack_stage_packages(self):
+    def _unpack_stage_packages(self) -> None:
         """Extract stage packages contents to the part's install directory."""
         pulled_packages = None
 
@@ -938,7 +940,7 @@ def _unpack_stage_packages(self):
             stage_packages=pulled_packages,
         )
 
-    def _unpack_stage_snaps(self):
+    def _unpack_stage_snaps(self) -> None:
         """Extract stage snap contents to the part's install directory."""
         stage_snaps = self._part.spec.stage_snaps
         if not stage_snaps:

craft_parts/executor/step_handler.py

@@ -158,14 +158,14 @@ def _builtin_stage(self) -> StepContents:
         srcdir = str(self._part.part_install_dir)
         files, dirs = filesets.migratable_filesets(stage_fileset, srcdir)
 
-        def pkgconfig_fixup(file_path):
+        def pkgconfig_fixup(file_path: str) -> None:
             if os.path.islink(file_path):
                 return
             if not file_path.endswith(".pc"):
                 return
             packages.fix_pkg_config(
                 prefix_prepend=self._part.stage_dir,
-                pkg_config_file=file_path,
+                pkg_config_file=Path(file_path),
                 prefix_trim=self._part.part_install_dir,
             )
 

craft_parts/infos.py

@@ -80,7 +80,7 @@ def __init__(
         project_name: Optional[str] = None,
         project_vars_part_name: Optional[str] = None,
         project_vars: Optional[Dict[str, str]] = None,
-        **custom_args,  # custom passthrough args
+        **custom_args: Any,  # custom passthrough args
     ):
         if not project_dirs:
             project_dirs = ProjectDirs()
@@ -101,7 +101,7 @@ def __init__(
 
         self.execution_finished = False
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> Any:
         if hasattr(self._dirs, name):
             return getattr(self._dirs, name)
 
@@ -255,7 +255,7 @@ def _ensure_valid_variable_name(self, name: str) -> None:
         if name not in self._project_vars:
             raise ValueError(f"{name!r} not in project variables")
 
-    def _set_machine(self, arch: Optional[str]):
+    def _set_machine(self, arch: Optional[str]) -> None:
         """Initialize host and target machine information based on the architecture.
 
         :param arch: The architecture to use. If empty, assume the
@@ -302,7 +302,7 @@ def __init__(
         self._part_state_dir = part.part_state_dir
         self.build_attributes = part.spec.build_attributes.copy()
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> Any:
         # Use composition and attribute cascading to avoid setting attributes
         # cumulatively in the init method.
         if hasattr(self._project_info, name):
@@ -404,7 +404,7 @@ def __init__(
         self.step_environment: Dict[str, str] = {}
         self.state: "Optional[states.StepState]" = None
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> Any:
         if hasattr(self._part_info, name):
             return getattr(self._part_info, name)
 

craft_parts/lifecycle_manager.py

@@ -94,7 +94,7 @@ def __init__(
         base_layer_hash: Optional[bytes] = None,
         project_vars_part_name: Optional[str] = None,
         project_vars: Optional[Dict[str, str]] = None,
-        **custom_args,  # custom passthrough args
+        **custom_args: Any,  # custom passthrough args
     ):
         # pylint: disable=too-many-locals
 

craft_parts/main.py

@@ -37,7 +37,7 @@
 from craft_parts import ActionType, Step
 
 
-def main():
+def main() -> None:
     """Run the command-line interface."""
     options = _parse_arguments()
 

craft_parts/overlays/chroot.py

@@ -32,7 +32,7 @@
 logger = logging.getLogger(__name__)
 
 
-def chroot(path: Path, target: Callable, *args, **kwargs) -> Any:
+def chroot(path: Path, target: Callable, *args: Any, **kwargs: Any) -> Any:
     """Execute a callable in a chroot environment.
 
     :param path: The new filesystem root.