[elbe-devel] [PATCH v2 2/6] commands: check-build: Add cdrom checker

Torben Hohn torben.hohn at linutronix.de
Wed Aug 19 12:18:54 CEST 2020


On Mon, Aug 03, 2020 at 12:40:31PM -0400, Olivier Dion wrote:
> This checker will validate the bin-cdrom.iso and src-cdrom*.iso of a
> build.
> 
> It starts by generating a binary table using the source.xml.  This
> table is then incrementally validated by inspecting the bin-cdrom.iso.
> Every time that a binary is validated, sources are added to the source
> table.  The source table is not component aware yet.
> 
> Then, the source table is validated the same way, by incrementally
> validating sources found in the src-cdrom*.iso.
> 
> Signed-off-by: Olivier Dion <dion at linutronix.de>
> ---
>  elbepack/commands/check-build.py | 290 +++++++++++++++++++++++++++++++
>  elbepack/filesystem.py           |   7 +
>  2 files changed, 297 insertions(+)
> 
> diff --git a/elbepack/commands/check-build.py b/elbepack/commands/check-build.py
> index 2db8faf6..a39ac8ac 100644
> --- a/elbepack/commands/check-build.py
> +++ b/elbepack/commands/check-build.py
> @@ -3,6 +3,7 @@
>  #
>  # SPDX-License-Identifier: GPL-3.0-or-later
>  
> +import glob
>  import logging
>  import optparse
>  import os
> @@ -95,3 +96,292 @@ class CheckBase(object):
>  
>      def fail(self, reason):
>          raise CheckException(reason)
> +
> + at CheckBase.register("cdrom")
> +class CheckCdroms(CheckBase):
> +
> +    """Check for cdroms integrity"""
> +
> +    def extract_cdrom(self, tgt, cdrom):
> +        try:
> +            do('7z x -o"%s" "%s"' % (tgt, cdrom))
> +        except CommandError as E:
> +            self.fail("Failed to extract cdrom %s:\n%s" % (cdrom, E))
> +
> +    def dpkg_get_infos(self, path, fmt):
> +        """Get dpkg infos for .deb and .dsc file formats"""
> +        try:
> +            if path.endswith(".deb"):
> +                cmd = 'dpkg -f "%s" %s' % (path, " ".join(fmt))
> +            elif path.endswith(".dsc"):
> +                cmd = 'grep -E "^(%s):" %s' % ("|".join(fmt), path)
> +            return get_command_out(cmd).decode("utf-8")
> +        except CommandError as E:
> +            self.fail("Failed to get debian infos (%s) for %s:\n%s" %
> +                      ('|'.join(fmt), path, E))
> +
> +    @staticmethod
> +    def cmp_version(v1, v2):
> +        return command_out("dpkg --compare-versions %s eq %s" %
> +                           (v1, v2), output=DEVNULL)[0]
> +
> +    def do_src(self, sources, src_total):
> +        """Check for sources in src-cdrom*"""
> +
> +        # pylint: disable=too-many-locals
> +        # pylint: disable=too-many-branches
> +
> +        iso_it  = glob.iglob("src-cdrom*")
> +        src_cnt = 0
> +
> +        # For every src-cdrom*, extract it to a temporary directory
> +        # and find all *.dsc files
> +        for cdrom in iso_it:
> +            with TmpdirFilesystem() as tmp:
> +                self.extract_cdrom(tmp, cdrom)
> +                for _dir, _, files in os.walk(tmp):
> +                    for _file in files:

why dont you use Filesystem.walk_files() ?

> +                        if not _file.endswith('.dsc'):
> +                            continue
> +
> +                        infos = self.dpkg_get_infos(os.path.join(_dir, _file),
> +                                                    ["Source", "Version"])
> +                        src_name    = None
> +                        src_version = None
> +
> +                        for info in infos.split('\n'):
> +
> +                            if info.startswith("Source:"):
> +                                src_name = info.split('Source:')[1].strip(' ')
> +
> +                            # Same as for the binary version.  The
> +                            # PGP's signature contains a version field

PGP ? you mean pkg ?

> +                            elif info.startswith("Version:"):
> +                                if not src_version:
> +                                    src_version = info.split('Version:')[1].strip(' ')

i would have preferred that the parsing would be separated into a
separate class. 

But lets keep it like this. We are short on time.

> +
> +                        if src_name in sources:
> +
> +                            match = False
> +
> +                            for version in sources[src_name]:
> +
> +                                # Found a matching version; prune it
> +                                if self.cmp_version(version, src_version) == 0:
> +
> +                                    logging.info("Validating source %s_%s",
> +                                                 src_name, version)
> +
> +                                    sources[src_name].remove(version)
> +                                    src_cnt += 1
> +                                    match   = True
> +
> +                                    break
> +
> +                            # NOTE! - Because the way the source table is
> +                            # generated, it's not possible to have multiple time
> +                            # the same version of a source (you have different
> +                            # versions only).  However, this is totally possible
> +                            # for cdrom because of multiple components.  Thus,
> +                            # whenever the source table can handle per component
> +                            # sources, this case should emit an error instead of
> +                            # a warning
> +                            if not match:
> +                                logging.warning("Can't find matching version for source %s_%s.\n"
> +                                                "It might have already been validated",
> +                                                src_name, src_version)
> +                        else:
> +                            logging.error("Extra source %s_%s found",
> +                                          src_name, src_version)
> +                            self.ret = 1
> +
> +        # List missing sources
> +        for src_name in sources:
> +            for src_version in sources[src_name]:
> +                logging.error("Missing source %s_%s",
> +                              src_name, src_version)
> +
> +        logging.info("Succesfully validated %d source packages out of %d",
> +                     src_cnt, src_total)
> +
> +        if src_cnt != src_total:
> +            self.ret = 1
> +
> +    def do_bin(self):
> +        """Check for binaries in bin-cdrom*.
> +
> +        Return a tuple of the form ({ "source-name" : [versions ..] }, src_cnt).
> +
> +        Put in others, validating binary cdroms will give us back a
> +        dictionnary where the key are the source package name, and the
> +        values of these keys are lists of versions.  It will also
> +        return a total count of source that is in the dictionnary.
> +
> +        """
> +
> +        # NOTE! - The binary table should be okay with the current
> +        # bin-cdrom.iso.  However, the way that the source table is genrated is
> +        # not okay for now.  Indeed, it's not aware of components.  So for
> +        # example, if two components have a binary in common, they will share
> +        # the corresponding source in the source table.  The solution is not as
> +        # trivial as having a reference counter or appending to a list, because
> +        # a component could have multiple binaries that match to the same source
> +        # package.  Thus, the only way is to make the source table component
> +        # aware.
> +
> +        # pylint: disable=too-many-locals
> +        # pylint: disable=too-many-branches
> +        # pylint: disable=too-many-statements
> +
> +        # Every build has a source.xml where the list of binaries
> +        # installed can be found
> +        xml = etree("source.xml")
> +
> +        # Initial statistics fo the build
> +        bin_cnt   = 0
> +        src_cnt   = 0
> +        bin_total = 0
> +
> +        binaries = {}
> +        sources  = {}
> +
> +        # Create a dictionnary of the form {"bin-name": [versions ..]}
> +        # from the source.xml.  We do this by iterating over all <pkg>
> +        for tag in xml.all("./*/pkg"):
> +
> +            bin_pkg = tag.et.text
> +
> +            # Package already in the dictionnary? Add its version.
> +            # Otherwise, add a new entry into the dictionnary
> +            if bin_pkg in binaries:
> +                binaries[bin_pkg].append(tag.et.attrib["version"])
> +            else:
> +                binaries[bin_pkg] = [tag.et.attrib["version"]]
> +
> +            bin_total += 1

There is already XMLPackage but that would have to be isolated from
python-apt. 

and this is more performant. 


> +
> +        # For every bin-cdrom, create a temporary directory where to
> +        # extract it and find all *.deb files
> +        #
> +        # pylint: disable=too-many-nested-blocks
> +        for cdrom in glob.glob("bin-cdrom*"):
> +            with TmpdirFilesystem() as tmp:
> +                self.extract_cdrom(tmp, cdrom)
> +                for _dir, _, files in os.walk(tmp):
> +                    for _file in files:

tmp.walk_files()

> +                        if not _file.endswith(".deb"):
> +                            continue
> +
> +                        # Extract informations from .deb
> +                        deb_path = os.path.join(_dir, _file)
> +                        infos    = self.dpkg_get_infos(deb_path, ["Package",
> +                                                                  "Source",
> +                                                                  "Version",
> +                                                                  "Built-Using"])
> +                        src_name    = None
> +                        src_version = None
> +                        bin_name    = None
> +                        bin_version = None
> +
> +                        for line in infos.split('\n'):
> +
> +                            # Package: <PACKAGE>
> +                            if line.startswith("Package:"):
> +                                bin_name = line.split('Package:')[1].strip(' \t')
> +
> +                            # Version: <VERSION>
> +                            #
> +                            # Skip PGP's version.  The package version is
> +                            # supposed to be before the PGP signature.  However,
> +                            # the PGP signature will put a 'Version' field.
> +                            # Thus, let's check if we already have found a
> +                            # binary version and don't overwrite it
> +                            elif line.startswith("Version:"):
> +                                if not bin_version:
> +                                    bin_version = line.split('Version:')[1].strip(' ')
> +
> +                            # Source: <SOURCE> [(VERSION)]
> +                            #
> +                            # This field is optional.  If it is not present, the
> +                            # source package default to the bin package
> +                            elif line.startswith("Source:"):
> +                                src_infos = line.split('Source:')[1].strip(' ').split(' ')
> +                                src_name = src_infos[0]
> +                                if len(src_infos) > 1:
> +                                    src_version = src_infos[1].strip("()")
> +
> +                            # Built-Using: <SRC (=VERSION)>...
> +                            #
> +                            # Sources list in the built-using field are
> +                            # seperated by a comma
> +                            elif line.startswith("Built-Using:"):
> +
> +                                built_using = line.split("Built-Using:")[1].strip(' ').split(',')
> +
> +                                for src in built_using:
> +
> +                                    name, version = src.strip(' ').split(' ', 1)
> +                                    version = version.strip("(= )")
> +
> +                                    # TODO - This is not component aware!
> +                                    if name in sources:
> +                                        if version not in sources[name]:
> +                                            sources[name].add(version)
> +                                            src_cnt += 1
> +                                    else:
> +                                        src_cnt += 1
> +                                        sources[name] = {version}
> +
> +
> +                        # No source was found
> +                        if src_name is None:
> +                            src_name    = bin_name
> +                            src_version = bin_version
> +
> +                        # No source version was found
> +                        elif src_version is None:
> +                            src_version = bin_version
> +
> +                        # TODO - This is not component aware!
> +                        #
> +                        # Let's build a dictionnary of sources of the form
> +                        # {"source-name" : [versions ..]}. Same as the binary
> +                        # dictionnary before
> +                        if src_name in sources:
> +                            if src_version not in sources[src_name]:
> +                                sources[src_name].add(src_version)
> +                                src_cnt += 1
> +                        else:
> +                            sources[src_name] = {src_version}
> +                            src_cnt += 1
> +
> +                        # Prune version of this binary
> +                        bin_cnt += 1
> +                        try:
> +                            binaries[bin_name].remove(bin_version)
> +                            logging.info("Validating binary %s_%s",
> +                                         bin_name, bin_version)
> +                            logging.info("Adding source %s_%s", src_name, src_version)
> +                        except KeyError:
> +                            logging.error("Foreign binary found %s_%s",
> +                                          bin_name, bin_version)
> +                            self.ret = 1
> +
> +        # List all missing binaries
> +        for bin_name in binaries:
> +            for bin_version in binaries[bin_name]:
> +                logging.error("Missing binary %s_%s", bin_name, bin_version)
> +
> +        logging.info("Succesfully validated %d binary packages out of %d",
> +                     bin_cnt, bin_total)
> +
> +        if bin_cnt != bin_total:
> +            self.ret = 1
> +
> +        return sources, src_cnt
> +
> +    def run(self):
> +        sources, src_cnt = self.do_bin()
> +        self.do_src(sources, src_cnt)
> +        return self.ret
> +
> diff --git a/elbepack/filesystem.py b/elbepack/filesystem.py
> index 51697907..714aadd4 100644
> --- a/elbepack/filesystem.py
> +++ b/elbepack/filesystem.py
> @@ -507,6 +507,13 @@ class TmpdirFilesystem (Filesystem):
>      def delete(self):
>          shutil.rmtree(self.path, True)
>  
> +    def __enter__(self):
> +        return self.path
> +
> +    def __exit__(self, exec_type, exec_value, tb):
> +        shutil.rmtree(self.path)
> +        return False
> +

errm... this seems to be misplaced.

>  
>  class ImgMountFilesystem(Filesystem):
>      def __init__(self, mntpoint, dev):
> -- 
> 2.28.0
> 
> _______________________________________________
> elbe-devel mailing list
> elbe-devel at linutronix.de
> https://lists.linutronix.de/mailman/listinfo/elbe-devel

-- 
Torben Hohn
Linutronix GmbH | Bahnhofstrasse 3 | D-88690 Uhldingen-Mühlhofen
Phone: +49 7556 25 999 18; Fax.: +49 7556 25 999 99

Hinweise zum Datenschutz finden Sie hier (Informations on data privacy 
can be found here): https://linutronix.de/kontakt/Datenschutz.php

Linutronix GmbH | Firmensitz (Registered Office): Uhldingen-Mühlhofen | 
Registergericht (Registration Court): Amtsgericht Freiburg i.Br., HRB700 
806 | Geschäftsführer (Managing Directors): Heinz Egger, Thomas Gleixner


More information about the elbe-devel mailing list