diff --git a/.virtualenv.requirements.txt b/.virtualenv.requirements.txt index 97b53c1a2a5..29f8ed6be69 100644 --- a/.virtualenv.requirements.txt +++ b/.virtualenv.requirements.txt @@ -23,3 +23,5 @@ xmltodict # json support that also knows about tuples simplejson + +typing_extensions; python_version < '3.8' diff --git a/kiwi/boot/image/builtin_kiwi.py b/kiwi/boot/image/builtin_kiwi.py index 2de8c0cec7e..b4ffb7da058 100644 --- a/kiwi/boot/image/builtin_kiwi.py +++ b/kiwi/boot/image/builtin_kiwi.py @@ -199,10 +199,9 @@ def create_initrd( compress = Compress( os.sep.join([self.target_dir, kiwi_initrd_basename]) ) - compress.xz( + self.initrd_filename = compress.xz( ['--check=crc32', '--lzma2=dict=1MiB', '--threads=0'] ) - self.initrd_filename = compress.compressed_filename def cleanup(self) -> None: for directory in self.temp_directories: diff --git a/kiwi/builder/container.py b/kiwi/builder/container.py index e03ad0acc3d..e860fb3b125 100644 --- a/kiwi/builder/container.py +++ b/kiwi/builder/container.py @@ -145,7 +145,7 @@ def create(self) -> Result: self.requested_container_type, self.root_dir, self.container_config ) self.filename = container_image.create( - self.filename, self.base_image, self.ensure_empty_tmpdirs, + self.filename, self.base_image or '', self.ensure_empty_tmpdirs, self.runtime_config.get_container_compression() # appx containers already contains a compressed root if self.requested_container_type != 'appx' else False diff --git a/kiwi/builder/kis.py b/kiwi/builder/kis.py index 8729d7f27f9..e02ae6555cb 100644 --- a/kiwi/builder/kis.py +++ b/kiwi/builder/kis.py @@ -124,8 +124,7 @@ def create(self) -> Result: if self.compressed: log.info('xz compressing root filesystem image') compress = Compress(self.image) - compress.xz(self.xz_options) - self.image = compress.compressed_filename + self.image = compress.xz(self.xz_options) log.info('Creating root filesystem MD5 checksum') checksum = Checksum(self.image) diff --git a/kiwi/container/__init__.py b/kiwi/container/__init__.py index 924f72aed0e..69b4726d6eb 100644 --- a/kiwi/container/__init__.py +++ b/kiwi/container/__init__.py @@ -22,6 +22,7 @@ ) from typing import Dict +from kiwi.container.base import ContainerImageBase from kiwi.exceptions import ( KiwiContainerImageSetupError ) @@ -40,7 +41,7 @@ def __init__(self) -> None: return None # pragma: no cover @staticmethod - def new(name: str, root_dir: str, custom_args: Dict=None): # noqa: E252 + def new(name: str, root_dir: str, custom_args: Dict=None) -> ContainerImageBase: # noqa: E252 name_map = { 'docker': 'OCI', 'oci': 'OCI', diff --git a/kiwi/container/appx.py b/kiwi/container/appx.py index 3f2a41e9916..a55f31578a3 100644 --- a/kiwi/container/appx.py +++ b/kiwi/container/appx.py @@ -26,13 +26,13 @@ from kiwi.utils.compress import Compress from kiwi.runtime_config import RuntimeConfig from kiwi.command import Command - +from kiwi.container.base import ContainerImageBase from kiwi.exceptions import KiwiContainerSetupError log = logging.getLogger('kiwi') -class ContainerImageAppx: +class ContainerImageAppx(ContainerImageBase): """ Create Appx container from a root directory for WSL(Windows Subsystem Linux) @@ -50,7 +50,7 @@ class ContainerImageAppx: 'metadata_path': 'directory' } """ - def __init__(self, root_dir: str, custom_args: Dict = {}): + def __init__(self, root_dir: str, custom_args: Dict[str, str] = {}): self.root_dir = root_dir self.wsl_config = custom_args or {} self.runtime_config = RuntimeConfig() @@ -72,7 +72,7 @@ def __init__(self, root_dir: str, custom_args: Dict = {}): def create( self, filename: str, base_image: str = '', ensure_empty_tmpdirs: bool = False, compress_archive: bool = False - ): + ) -> str: """ Create WSL/Appx archive @@ -126,7 +126,6 @@ def create( ) if compress_archive: compress = Compress(filename) - compress.xz(self.runtime_config.get_xz_options()) - filename = compress.compressed_filename + filename = compress.xz(self.runtime_config.get_xz_options()) return filename diff --git a/kiwi/container/base.py b/kiwi/container/base.py new file mode 100644 index 00000000000..9b1e8e2e1cd --- /dev/null +++ b/kiwi/container/base.py @@ -0,0 +1,27 @@ +# Copyright (c) 2023 SUSE LLC +# +# This file is part of kiwi. +# +# kiwi is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# kiwi is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with kiwi. If not, see +# +from abc import ABC, abstractmethod + + +class ContainerImageBase(ABC): + @abstractmethod + def create( + self, filename: str, base_image: str, + ensure_empty_tmpdirs: bool, compress_archive: bool = False + ) -> str: + pass # pragma: no cover diff --git a/kiwi/container/oci.py b/kiwi/container/oci.py index e04cf2fe179..37115edd662 100644 --- a/kiwi/container/oci.py +++ b/kiwi/container/oci.py @@ -17,20 +17,40 @@ # import os import logging -from typing import ( - Dict, List -) +import sys +from typing import Dict, List, Optional +if sys.version_info >= (3, 8): + from typing import TypedDict # pragma: no cover +else: # pragma: no cover + from typing_extensions import TypedDict # pragma: no cover # project from kiwi.utils.compress import Compress from kiwi.runtime_config import RuntimeConfig from kiwi.defaults import Defaults from kiwi.oci_tools import OCI +from kiwi.container.base import ContainerImageBase log = logging.getLogger('kiwi') -class ContainerImageOCI: +class OciConfig(TypedDict, total=False): + container_name: str + container_tag: str + additional_names: List[str] + entry_command: List[str] + entry_subcommand: List[str] + maintainer: str + user: str + workingdir: str + expose_ports: List[str] + volumes: List[str] + environment: Dict[str, str] + labels: Dict[str, str] + history: Dict[str, str] + + +class ContainerImageOCI(ContainerImageBase): """ Create oci container from a root directory @@ -63,13 +83,10 @@ class ContainerImageOCI: } } """ - def __init__(self, root_dir: str, transport: str, custom_args: Dict = {}): + def __init__(self, root_dir: str, transport: str, custom_args: Optional[OciConfig] = None) -> None: self.root_dir = root_dir self.archive_transport = transport - if custom_args: - self.oci_config = custom_args - else: - self.oci_config = {} + self.oci_config = custom_args or {} # for builds inside the buildservice we include a reference to the # specific build. Thus disturl label only exists inside the @@ -102,10 +119,11 @@ def __init__(self, root_dir: str, transport: str, custom_args: Dict = {}): self.oci_config['history']['created_by'] = \ Defaults.get_default_container_created_by() + def create( self, filename: str, base_image: str, ensure_empty_tmpdirs: bool = True, compress_archive: bool = False - ): + ) -> str: """ Create compressed oci system container tar archive @@ -174,8 +192,7 @@ def create( ) if compress_archive: compress = Compress(filename) - compress.xz(RuntimeConfig().get_xz_options()) - filename = compress.compressed_filename + filename = compress.xz(RuntimeConfig().get_xz_options()) return filename diff --git a/kiwi/iso_tools/xorriso.py b/kiwi/iso_tools/xorriso.py index 0ec7ae610e4..44a2c04b83d 100644 --- a/kiwi/iso_tools/xorriso.py +++ b/kiwi/iso_tools/xorriso.py @@ -17,14 +17,14 @@ # import os from typing import ( - Dict, List + Dict, List, Optional ) # project from kiwi.iso_tools.base import IsoToolsBase from kiwi.path import Path from kiwi.command import Command -from kiwi.exceptions import KiwiIsoToolError +from kiwi.exceptions import KiwiFileNotFound, KiwiIsoToolError from kiwi.defaults import Defaults @@ -63,7 +63,7 @@ def get_tool_name(self) -> str: raise KiwiIsoToolError('xorriso tool not found') def init_iso_creation_parameters( - self, custom_args: Dict[str, str] = None + self, custom_args: Optional[Dict[str, str]] = None ) -> None: """ Create a set of standard parameters @@ -113,9 +113,12 @@ def init_iso_creation_parameters( ] else: loader_file = self.boot_path + '/loader/isolinux.bin' - mbr_file = Path.which( + mbr_file_c = Path.which( 'isohdpfx.bin', Defaults.get_syslinux_search_paths() ) + if not mbr_file_c: + raise KiwiFileNotFound("isohdpfx.bin not found in the systlinux search paths") + mbr_file = mbr_file_c self.iso_loaders += [ '-boot_image', 'isolinux', 'bin_path={0}'.format( loader_file diff --git a/kiwi/package_manager/pacman.py b/kiwi/package_manager/pacman.py index 47527a5367d..95c68675f17 100644 --- a/kiwi/package_manager/pacman.py +++ b/kiwi/package_manager/pacman.py @@ -18,7 +18,7 @@ import re import logging from typing import ( - List, Dict + List, Dict, Optional ) # project @@ -42,7 +42,7 @@ class PackageManagerPacman(PackageManagerBase): pacman command environment from repository runtime configuration """ - def post_init(self, custom_args: List = None) -> None: + def post_init(self, custom_args: Optional[List[str]] = None) -> None: """ Post initialization method @@ -50,9 +50,7 @@ def post_init(self, custom_args: List = None) -> None: :param list custom_args: custom pacman arguments """ - self.custom_args = custom_args - if not custom_args: - self.custom_args = [] + self.custom_args: List[str] = custom_args or [] runtime_config = self.repository.runtime_config() self.pacman_args = runtime_config['pacman_args'] diff --git a/kiwi/path.py b/kiwi/path.py index 9d4c686510d..053bf841263 100644 --- a/kiwi/path.py +++ b/kiwi/path.py @@ -18,6 +18,7 @@ import os import logging import collections +from typing import Dict, List, Optional # project from kiwi.command import Command @@ -31,7 +32,7 @@ class Path: **Directory path helpers** """ @staticmethod - def sort_by_hierarchy(path_list): + def sort_by_hierarchy(path_list: List[str]) -> List[str]: """ Sort given list of path names by their hierachy in the tree @@ -47,7 +48,7 @@ def sort_by_hierarchy(path_list): :rtype: list """ - paths_at_depth = {} + paths_at_depth: Dict[int, List[str]] = {} for path in path_list: path_elements = path.split('/') path_depth = len(path_elements) @@ -64,7 +65,7 @@ def sort_by_hierarchy(path_list): return ordered_paths @staticmethod - def access(path, mode, **kwargs): + def access(path: str, mode: int, **kwargs) -> bool: """ Check whether path can be accessed with the given mode. @@ -98,7 +99,7 @@ def access(path, mode, **kwargs): return os.access(path, mode, **kwargs) @staticmethod - def create(path): + def create(path: str) -> None: """ Create path and all sub directories to target @@ -110,7 +111,7 @@ def create(path): ) @staticmethod - def wipe(path): + def wipe(path: str) -> None: """ Delete path and all contents @@ -122,7 +123,7 @@ def wipe(path): ) @staticmethod - def remove(path): + def remove(path: str) -> None: """ Delete empty path, causes an error if target is not empty @@ -133,7 +134,7 @@ def remove(path): ) @staticmethod - def rename(cur, new): + def rename(cur: str, new: str) -> None: """ Move path from cur name to new name @@ -145,7 +146,7 @@ def rename(cur, new): ) @staticmethod - def remove_hierarchy(root, path): + def remove_hierarchy(root: str, path: str) -> None: """ Recursively remove an empty path and its sub directories starting at a given root directory. Ignore non empty or @@ -182,7 +183,7 @@ def remove_hierarchy(root, path): ) @staticmethod - def move_to_root(root, elements): + def move_to_root(root: str, elements: List[str]) -> List[str]: """ Change the given path elements to a new root directory @@ -204,7 +205,7 @@ def move_to_root(root, elements): return result @staticmethod - def rebase_to_root(root, elements): + def rebase_to_root(root: str, elements: List[str]) -> List[str]: """ Include the root prefix for the given paths elements @@ -222,9 +223,10 @@ def rebase_to_root(root, elements): @staticmethod def which( - filename, alternative_lookup_paths=None, - custom_env=None, access_mode=None, root_dir=None - ): + filename: str, alternative_lookup_paths: Optional[List[str]]=None, + custom_env: Optional[Dict[str, str]]=None, access_mode: Optional[int]=None, + root_dir: Optional[str]=None + ) -> Optional[str]: """ Lookup file name in PATH @@ -267,3 +269,4 @@ def which( return location log.debug(' '.join(multipart_message)) + return None diff --git a/kiwi/utils/compress.py b/kiwi/utils/compress.py index 7876bd6a6da..d29b7dba9d0 100644 --- a/kiwi/utils/compress.py +++ b/kiwi/utils/compress.py @@ -17,6 +17,7 @@ # import os import logging +from typing import List, Optional # project from kiwi.utils.temporary import Temporary @@ -43,7 +44,7 @@ class Compress: :param str uncompressed_filename: Uncompressed file name path """ - def __init__(self, source_filename, keep_source_on_compress=False): + def __init__(self, source_filename: str, keep_source_on_compress: bool=False) -> None: if not os.path.exists(source_filename): raise KiwiFileNotFound( 'compression source file %s not found' % source_filename @@ -53,10 +54,10 @@ def __init__(self, source_filename, keep_source_on_compress=False): self.supported_zipper = [ 'xz', 'gzip' ] - self.compressed_filename = None - self.uncompressed_filename = None + self.compressed_filename: Optional[str] = None + self.uncompressed_filename: Optional[str] = None - def xz(self, options=None): + def xz(self, options: Optional[List[str]]=None) -> str: """ Create XZ compressed file @@ -64,6 +65,7 @@ def xz(self, options=None): """ if not options: options = Defaults.get_xz_compression_options() + assert options if self.keep_source: options.append('--keep') Command.run( @@ -72,7 +74,7 @@ def xz(self, options=None): self.compressed_filename = self.source_filename + '.xz' return self.compressed_filename - def gzip(self): + def gzip(self) -> str: """ Create gzip(max compression) compressed file """ @@ -87,7 +89,7 @@ def gzip(self): self.compressed_filename = self.source_filename + '.gz' return self.compressed_filename - def uncompress(self, temporary=False): + def uncompress(self, temporary: bool=False) -> str: """ Uncompress with format autodetection @@ -116,7 +118,7 @@ def uncompress(self, temporary=False): self.uncompressed_filename = self.temp_file.name return self.uncompressed_filename - def get_format(self): + def get_format(self) -> Optional[str]: """ Detect compression format @@ -138,3 +140,4 @@ def get_format(self): exc=str(exc) ) ) + return None diff --git a/setup.py b/setup.py index b997d437e39..22882ad812c 100755 --- a/setup.py +++ b/setup.py @@ -66,7 +66,8 @@ def run(self): 'lxml', 'requests', 'PyYAML', - 'simplejson' + 'simplejson', + 'typing_extensions; python_version < "3.8"', ], 'packages': ['kiwi'], 'cmdclass': { diff --git a/test/unit/builder/container_test.py b/test/unit/builder/container_test.py index bc1f9fba57d..cc31dfeb4db 100644 --- a/test/unit/builder/container_test.py +++ b/test/unit/builder/container_test.py @@ -145,7 +145,7 @@ def test_create(self, mock_image, mock_setup): 'docker', 'root_dir', self.container_config ) container_image.create.assert_called_once_with( - 'target_dir/image_name.x86_64-1.2.3.docker.tar', None, True, False + 'target_dir/image_name.x86_64-1.2.3.docker.tar', '', True, False ) assert self.container.result.add.call_args_list == [ call( diff --git a/test/unit/builder/kis_test.py b/test/unit/builder/kis_test.py index e6907fa80d3..f61541f7016 100644 --- a/test/unit/builder/kis_test.py +++ b/test/unit/builder/kis_test.py @@ -104,7 +104,7 @@ def test_create( mock_tar.return_value = tar compress = Mock() mock_compress.return_value = compress - compress.compressed_filename = 'compressed-file-name' + compress.xz.return_value = 'compressed-file-name' checksum = Mock() mock_checksum.return_value = checksum self.boot_image_task.required = Mock( diff --git a/test/unit/iso_tools/xorriso_test.py b/test/unit/iso_tools/xorriso_test.py index c2bdaa4654f..e62b9ae2cdf 100644 --- a/test/unit/iso_tools/xorriso_test.py +++ b/test/unit/iso_tools/xorriso_test.py @@ -4,7 +4,7 @@ from kiwi.defaults import Defaults from kiwi.iso_tools.xorriso import IsoToolsXorrIso -from kiwi.exceptions import KiwiIsoToolError +from kiwi.exceptions import KiwiFileNotFound, KiwiIsoToolError class TestIsoToolsXorrIso: @@ -26,6 +26,19 @@ def test_get_tool_name_raises(self, mock_exists): with raises(KiwiIsoToolError): self.iso_tool.get_tool_name() + @patch('kiwi.iso_tools.xorriso.Path.which') + def test_init_iso_creation_parameters_isolinux_fail_on_isohdpfx_not_found(self, mock_which): + mock_which.return_value = None + with raises(KiwiFileNotFound): + self.iso_tool.init_iso_creation_parameters( + { + 'mbr_id': 'app_id', + 'publisher': 'org', + 'preparer': 'preparer', + 'volume_id': 'vol_id' + } + ) + @patch('kiwi.iso_tools.xorriso.Path.which') def test_init_iso_creation_parameters_isolinux(self, mock_which): mock_which.return_value = '/usr/share/syslinux/isohdpfx.bin'