Source code for thoth.adviser.prescription.v1.gh_release_notes

#!/usr/bin/env python3
# thoth-adviser
# Copyright(C) 2021 Fridolin Pokorny
# 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 3 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
# 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, see <>.

"""GitHub release notes pipeline unit."""

import attr

import logging
from typing import Any
from typing import Dict
from typing import Generator
from typing import List
from typing import Optional
from typing import Union
from typing import TYPE_CHECKING
from packaging.specifiers import SpecifierSet

from thoth.adviser.state import State
from voluptuous import Schema
from voluptuous import Required

from .unit import UnitPrescription

    from ...pipeline_builder import PipelineBuilderContext

_LOGGER = logging.getLogger(__name__)

[docs]@attr.s(slots=True) class GHReleaseNotesWrapPrescription(UnitPrescription): """GitHub release notes pipeline unit.""" CONFIGURATION_SCHEMA: Schema = Schema( { Required("package_name"): str, Required("release_notes"): PRESCRIPTION_GH_RELEASE_NOTES_WRAP_RUN_ENTRY_SCHEMA, Required("package_version"): PACKAGE_VERSION_REQUIRED_NAME_SCHEMA, Required("prescription"): Schema({"run": bool}), } ) CONFIGURATION_DEFAULT: Dict[str, Any] = {"package_name": None, "release_notes": None, "package_version": None} _configuration = attr.ib(type=Dict[str, Any], kw_only=True, factory=CONFIGURATION_DEFAULT.copy)
[docs] @staticmethod def is_wrap_unit_type() -> bool: """Check if this unit is of type wrap.""" return True
@staticmethod def _construct_release_notes_url( organization: str, repository: str, tag_version_prefix: Optional[str], locked_version: str ) -> str: prefix = tag_version_prefix if tag_version_prefix is not None else "" return f"{organization}/{repository}/releases/tag/{prefix}{locked_version}" @staticmethod def _yield_from_resolved_dependencies( run_prescription: Dict[str, Any], resolved_dependencies_prescription: Union[Dict[str, Any], List[Dict[str, Any]]], prescription_conf: Dict[str, Any], ) -> Generator[Dict[str, Any], None, None]: """Yield configuration based on resolved dependencies prescribed.""" if isinstance(resolved_dependencies_prescription, list): for item in resolved_dependencies_prescription: yield { "package_name": item["name"], "release_notes": run_prescription["release_notes"], "package_version": item, "prescription": prescription_conf, } else: yield { "package_name": resolved_dependencies_prescription["name"], "release_notes": run_prescription["release_notes"], "package_version": resolved_dependencies_prescription, "prescription": prescription_conf, }
[docs] @classmethod def should_include(cls, builder_context: "PipelineBuilderContext") -> Generator[Dict[str, Any], None, None]: """Include this pipeline unit.""" if not builder_context.is_included(cls): prescription: Dict[str, Any] = cls._PRESCRIPTION # type: ignore prescription_conf = {"run": False} match = prescription["match"] run = prescription["run"] if isinstance(match, list): for item in match: yield from cls._yield_from_resolved_dependencies( run, item["state"]["resolved_dependencies"], prescription_conf ) else: yield from cls._yield_from_resolved_dependencies( run, match["state"]["resolved_dependencies"], prescription_conf ) return None yield from () return None
[docs] def run(self, state: State) -> None: """Add release information to justification for selected packages.""" justification_addition = [] for resolved_package_tuple in state.resolved_dependencies.values(): conf_package_version = self.configuration["package_version"] if not conf_package_version or resolved_package_tuple[0] != conf_package_version["name"]: continue version = conf_package_version.get("version") if version is not None and resolved_package_tuple[1] not in SpecifierSet(version): continue develop = conf_package_version.get("develop") if develop is not None: package_version = self.context.get_package_version(resolved_package_tuple) if not package_version: # This is a programming error as the give dependency has to be registered in the context. _LOGGER.error( "No matching package version for %r registered in the context", resolved_package_tuple ) continue if package_version.develop != develop: continue index_url = conf_package_version.get("index_url") if not self._index_url_check(index_url, resolved_package_tuple[2]): continue if self._configuration["prescription"]["run"]: # Can happen if prescription states criteria that match multiple times. We add # justification only once in such cases. continue self._configuration["prescription"]["run"] = True release_notes_conf = self.configuration["release_notes"] justification_addition.append( { "type": "INFO", "message": f"Release notes for package {resolved_package_tuple[0]!r}", "link": self._construct_release_notes_url( organization=release_notes_conf["organization"], repository=release_notes_conf["repository"], tag_version_prefix=release_notes_conf.get("tag_version_prefix"), locked_version=resolved_package_tuple[1], ), "package_name": resolved_package_tuple[0], } ) state.add_justification(justification_addition)