Source code for rebasehelper.plugins.checkers

# -*- coding: utf-8 -*-
#
# This tool helps you rebase your package to the latest version
# Copyright (C) 2013-2019 Red Hat, Inc.
#
# 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 program 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 this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Authors: Petr Hráček <phracek@redhat.com>
#          Tomáš Hozza <thozza@redhat.com>
#          Nikola Forró <nforro@redhat.com>
#          František Nečas <fifinecas@seznam.cz>

import enum
import logging
import os
import shutil
from typing import Any, Dict, List, Optional, Type, Union, cast

from rebasehelper.logger import CustomLogger
from rebasehelper.plugins.plugin import Plugin
from rebasehelper.plugins.plugin_collection import PluginCollection
from rebasehelper.constants import RESULTS_DIR


logger: CustomLogger = cast(CustomLogger, logging.getLogger(__name__))


[docs]class CheckerCategory(enum.Enum): SOURCE: int = 1 SRPM: int = 2 RPM: int = 3
[docs]class BaseChecker(Plugin): """Base class of package checkers. Attributes: DEFAULT(bool): If True, the checker is run by default. CATEGORY(CheckerCategory): Category which determines when the checker is run. results_dir(str): Path where the results are stored. """ DEFAULT: bool = False CATEGORY: Optional[CheckerCategory] = None results_dir: Optional[str] = None
[docs] @classmethod def prepare_results_dir(cls): """Creates checker's results dir. Removes the existing results dir if it exists to avoid collisions. """ if os.path.exists(cls.results_dir): shutil.rmtree(cls.results_dir) os.makedirs(cls.results_dir)
[docs] @classmethod def get_checker_output_dir_short(cls): """Return short version of checker output directory""" return os.path.join(RESULTS_DIR, 'checkers', cls.name)
[docs] @classmethod def is_available(cls): raise NotImplementedError()
[docs] @classmethod def run_check(cls, results_dir, **kwargs): """Perform the check itself and return results.""" raise NotImplementedError()
[docs] @classmethod def format(cls, data): """Formats checker output to a readable text form.""" raise NotImplementedError()
[docs] @classmethod def get_category(cls): return cls.CATEGORY
[docs] @classmethod def get_underlined_title(cls, text, separator='='): return "\n{}\n{}".format(text, separator * len(text))
[docs] @classmethod def get_important_changes(cls, checker_output): # pylint: disable=unused-argument """ Each checker has an opportunity to highlight some of its output. This function is optional, as not all checkers provide output with critical information. Args: checker_output (dict): Dictionary with output from the given checker. Returns: list: List of strings to be output to CLI as warning messages. """ return []
[docs]class CheckerCollection(PluginCollection): """ Class representing the process of running various checkers on final packages. """
[docs] def run(self, results_dir: str, checker_name: str, **kwargs: Any) -> Optional[Dict[str, Any]]: """Runs a particular checker and returns the results. Args: results_dir: Path to a directory in which the checker should store the results. checker_name: Name of the checker to be run. Raises: NotImplementedError: If a checker with the given name doesn't exist. Returns: Results of the checker. """ try: checker = cast(Type[BaseChecker], self.plugins[checker_name]) except KeyError: return None if checker.CATEGORY != kwargs.get('category'): return None if not checker.is_available(): return None if checker.CATEGORY == CheckerCategory.SOURCE: logger.info("Running checks on source files using '%s'", checker_name) else: logger.info("Running checks on packages using '%s'", checker_name) return checker.run_check(results_dir, **kwargs)
[docs] def get_supported_plugins(self) -> List[str]: return [k for k, v in self.plugins.items() if v and cast(Type[BaseChecker], v).is_available()]
[docs] def get_default_plugins(self, return_one: bool = False) -> Union[str, List[str], None]: default = [k for k, v in self.plugins.items() if v and getattr(v, 'DEFAULT', False)] return default if not return_one else default[0] if default else None