Source code for libka.excuses.excuses_yaml_parser

# -*- coding: utf-8 -*-
# pylint: disable=line-too-long
# kate: space-indent on; indent-width 4; replace-tabs on; indent-mode python; remove-trailing-space modified;
# vim: expandtab ts=4
# pylint: enable=line-too-long

############################################################################
#   Copyright © 2020 José Manuel Santamaría Lema <panfaust@gmail.com>      #
#                                                                          #
#   This program 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 2 of the License, or      #
#   (at your option) any later version.                                    #
############################################################################


"""This module just provides the ExcusesYAMLParser class"""

import sys
import urllib3
import yaml
from yaml import CLoader as Loader
import numcodecs

from libka.excuses.excuses_package_info import ExcusesPackageInfo
from libka.ka_print import ka_print_error




class ExcusesYAMLParser():
    """Parses an 'update-excuses' YAML url"""
    def __init__(self):
        self._excuses_entries = None



    def parse_excuses_url(self, excuses_url):
        """Extracts the data from a yaml url"""
        #Get the update excuses YAML file
        http = urllib3.PoolManager()
        try:
            response = http.request("GET", excuses_url)
        except Exception as exception: # pylint: disable=broad-except
            ka_print_error(str(exception))
            ka_print_error("Error downloading: %s" % excuses_url)
            sys.exit(1)
        #Uncompress the data
        lzma_codec = numcodecs.lzma.LZMA()
        try:
            decoded_data = lzma_codec.decode(response.data)
        except Exception as exception: # pylint: disable=broad-except
            ka_print_error(str(exception))
            ka_print_error("Couldn't uncompress with LZMA the url: %s" % excuses_url)
            sys.exit(1)
        #Load the YAML data
        yaml_object = yaml.load(decoded_data, Loader=Loader)
        self._excuses_entries = yaml_object["sources"]





    def get_package_info(self, filter_list=None):
        """
        Returns a dictionary indexed by source package name with all the excuses info
        for each source package.

        If `filter_list` is `None` it will return information for all the packages in
        the excuses file, if it's a list of source package names it will return the
        information only for the specified packages.
        """
        #Init the dictionary we are going to return
        package_info_map = {}
        #Browse the excuses entries to find out information
        #about the packages given in filter_list
        for excuses_entry in self._excuses_entries:
            src_pkg_name = excuses_entry["item-name"]
            if (filter_list is None) or (src_pkg_name in filter_list):
                #Build the object we are going to add to the dict
                current_package_info = ExcusesPackageInfo()
                #Set the source package name
                current_package_info.package_name = src_pkg_name
                #Set the old and new versions
                current_package_info.old_version = excuses_entry["old-version"]
                current_package_info.new_version = excuses_entry["new-version"]
                #Set the maintainer
                current_package_info.maintainer = excuses_entry["maintainer"]
                #Set the age
                try:
                    current_package_info.age = excuses_entry["policy_info"]["age"]["current-age"]
                    current_package_info.age = int(round(current_package_info.age))
                except KeyError:
                    current_package_info.age = None
                #Set the autopkgtests info
                try:
                    autopkgtest_info = excuses_entry["policy_info"]["autopkgtest"]
                except KeyError:
                    autopkgtest_info = []
                for i in autopkgtest_info:
                    if i == "verdict":
                        continue
                    current_package_info.autopkgtests[i] = {}
                    for arch in autopkgtest_info[i]:
                        current_package_info.autopkgtests[i][arch] = autopkgtest_info[i][arch][0]
                #Set the missing builds info
                try:
                    current_package_info.missing_builds = (
                        excuses_entry["missing-builds"]["on-architectures"])
                except KeyError:
                    current_package_info.missing_builds = []
                #Set the implicit depends info
                try:
                    current_package_info.implicit_depends = (
                        excuses_entry["dependencies"]["blocked-by"])
                except KeyError:
                    current_package_info.implicit_depends = []
                #Set the depends info
                try:
                    current_package_info.depends = excuses_entry["dependencies"]["migrate-after"]
                except KeyError:
                    current_package_info.depends = []
                #Set the reasons to not migrate
                current_package_info.reasons = excuses_entry["reason"]
                #Set if this package is a valid candidate for migration or not
                current_package_info.is_candidate = excuses_entry["is-candidate"]
                #Add the package info object to the dict
                package_info_map[src_pkg_name] = current_package_info
        #Return the final package info map
        return package_info_map