recipe_modules/release/api.py

# Copyright 2019 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import re

import attr

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
from recipe_engine import recipe_api

# Regex patterns for each part of a release version.
MAJOR_PATTERN = r"[0-9]{1,10}"
DATE_PATTERN = r"[0-9]{8}"
RELEASE_PATTERN = r"[0-9]{1,10}"
CANDIDATE_PATTERN = r"[1-9][0-9]{0,9}"


def int_match(pattern):
    """Constructs an attrs validator asserting that an int matches the pattern.

    Conforms to the attrs validator API, raising an exception if the
    given value does not match the validator's pattern.
    """
    regex = re.compile(r"^%s$" % pattern)

    def validator(_, attribute, value):
        match = regex.match(str(value))
        assert match, "%r value %r does not match regex %r" % (
            attribute.name,
            value,
            pattern,
        )

    return validator


@attr.s(eq=True, order=True)  # Set up __eq__() and __lt__() based on attribs.
class ReleaseVersion(object):
    """Representation of a release version, versioned by major number > date > release
    number > candidate number.

    ReleaseVersions are guaranteed to be valid release versions. An
    AssertionError will be raised if an invalid release version would be
    created upon initialization or modification.
    """

    # Note that the order of the attrib declarations is also the order in which
    # they are compared when doing less-than/greater-than comparisons. So it's
    # important that they be defined in order of "significance", most significant
    # first.
    major = attr.ib(converter=int, validator=int_match(MAJOR_PATTERN))
    date = attr.ib(converter=int, validator=int_match(DATE_PATTERN))
    release = attr.ib(converter=int, validator=int_match(RELEASE_PATTERN))
    candidate = attr.ib(converter=int, validator=int_match(CANDIDATE_PATTERN))

    _TEMPLATE = "releases/{major}.{date}.{release}.{candidate}"

    def __str__(self):
        return self._TEMPLATE.format(
            major=self.major,
            date=self.date,
            release=self.release,
            candidate=self.candidate,
        )

    @classmethod
    def get_regexp(
        cls,
        major=MAJOR_PATTERN,
        date=DATE_PATTERN,
        release=RELEASE_PATTERN,
        candidate=CANDIDATE_PATTERN,
    ):
        """Create a release version regexp.

        Default values can be overridden to restrict regexp matching.
        """
        return cls._TEMPLATE.replace(".", "\.").format(
            major="(?P<major>%s)" % major,
            date="(?P<date>%s)" % date,
            release="(?P<release>%s)" % release,
            candidate="(?P<candidate>%s)" % candidate,
        )

    @classmethod
    def from_string(cls, string):
        """Create a release version from a string."""
        match = re.search(cls.get_regexp(), string)
        assert match
        return cls(
            major=match.group("major"),
            date=match.group("date"),
            release=match.group("release"),
            candidate=match.group("candidate"),
        )


class ReleaseApi(recipe_api.RecipeApi):
    """The release module provides utilities for release commit and version generation
    on a git repository."""

    _COMMIT_MESSAGE_PREFIX = "[release]"
    _REVISION_LENGTH = 10
    _BUILDSET_TEMPLATE = "commit/gitiles/{stripped_https_remote}/+/{revision}"

    def get_major_number(self, repo_path):
        """Get the major number from the root of the input repository in the
        file named MILESTONE.

        Args:
          repo_path (Path): Local git repository to get major number from.

        Returns:
          int: Major number.
        """
        contents = self.m.file.read_text(
            "get major number",
            repo_path.join("MILESTONE"),
            test_data="0\n",
        )
        return int(contents.strip("\n"))

    def write_major_number(self, number, repo_path):
        """Write the major number in the root of the input repository in the
        file named MILESTONE.

        Args:
          repo_path (Path): Local git repository containing MILESTONE file at
            the root.
        """
        self.m.file.write_text(
            "write major number", repo_path.join("MILESTONE"), "%s\n" % str(number)
        )

    def get_initial_release_version(self, date, repo_path):
        """Get an initial release version.

        An initial release version depends on the input date and major number
        found at the root of the repository.

        Args:
          repo_path (Path): Local git repository to get major number from.
          date (str): Release date as YYYYMMDD.

        Returns:
          ReleaseVersion: Initial release version.
        """
        return ReleaseVersion(
            major=self.get_major_number(repo_path), date=date, release=0, candidate=1
        )

    def get_next_release_version(self, ref, date, repo_path):
        """Get next release version for a ref and date.

        If there is a release version reachable on ref for the input date, the
        next release version is the max existing release version reachable on ref
        with its release number incremented by 1, and candidate number reset to 1.
        Otherwise, the next release version is an initial release version for
        the input date.

        Args:
          ref (str): Target git ref.
          date (str): Release date as YYYYMMDD.
          repo_path (Path): Local git repository to query release versions.

        Returns:
          ReleaseVersion: Next release version.
        """
        release_versions = self.get_release_versions(
            ref=ref, date=date, repo_path=repo_path
        )
        release_version = self.get_initial_release_version(
            date=date, repo_path=repo_path
        )
        if release_versions:
            latest_release_version = max(release_versions)
            # If the latest release version on branch matches input date, increment
            # release number by 1.
            if latest_release_version.date == int(date):
                release_version = attr.evolve(
                    release_version,
                    release=latest_release_version.release + 1,
                )
        return release_version

    def get_next_candidate_version(self, ref, repo_path, add_track=False):
        """Get next candidate version for a ref.

        The next candidate version is ref's release version with its candidate
        version incremented by 1.

        Args:
          ref (str): Target git ref.
          repo_path (Path): Local git repository to resolve release version.
          add_track (bool): Whether to start a new track.

        Returns:
          ReleaseVersion: Next candidate version.

        Raises:
          StepFailure: input ref is not a git tag, or it is a tag but not a
            valid release version.
        """
        release_version = self.ref_to_release_version(ref=ref, repo_path=repo_path)
        if not release_version:
            return None
        if add_track:
            next_candidate = int(str(release_version.candidate) + "001")
        else:
            next_candidate = release_version.candidate + 1
        return attr.evolve(release_version, candidate=next_candidate)

    def get_release_versions(self, ref, repo_path, date=DATE_PATTERN):
        """Get release versions for a ref and date.

        Args:
          ref (str): Target git ref.
          date (str): Release date as YYYYMMDD. Accepts pattern.
          repo_path (Path): Local git repository to query release versions.

        Returns:
          list<ReleaseVersion>: Matching release versions.
        """
        with self.m.context(cwd=repo_path):
            tags = self.m.git(
                "get release versions on %s" % ref,
                "--no-pager",
                "tag",
                "--merged",
                ref,
                stdout=self.m.raw_io.output(),
            ).stdout
        matches = re.findall(ReleaseVersion.get_regexp(date=date), tags, re.MULTILINE)
        return [
            ReleaseVersion(major=major, date=date, release=release, candidate=candidate)
            for major, date, release, candidate in matches
        ]

    def lookup_builds(self, builders, revision, remote):
        """
        Lookup release builds for a revision.

        Args:
          builders (seq(str)): A sequence of fully-qualified builder names.
          revision (str): The desired revision of the builds.
          remote (str): Remote integration URL.
        """
        with self.m.step.nest("lookup builds") as presentation:
            buildset = self._BUILDSET_TEMPLATE.format(
                stripped_https_remote=remote.replace("https://", ""),
                revision=revision,
            )
            for builder in builders:
                build_id = self.m.buildsetlookup(
                    step_name="lookup %s" % builder,
                    builder=builder,
                    buildset=buildset,
                    # We want to show links to builds as soon as they exist. Any
                    # status is acceptable.
                    build_status=common_pb2.STATUS_UNSPECIFIED,
                    retries=10,
                    interval=60,
                )
                presentation.links[builder] = self.m.buildbucket.build_url(
                    build_id=build_id
                )

    def ref_to_release_version(self, ref, repo_path):
        """Convert a ref to a release version.

        Args:
          ref (str): Target git ref.
          repo_path (Path): Local git repository to describe ref in.

        Returns:
          ReleaseVersion|None: the ref's associated release version. None if
            the ref has no associated release version.

        Raises:
          StepFailure: input ref is not a valid release version.
        """
        with self.m.context(cwd=repo_path):
            tags = self.m.git.describe(step_name="ref to release version", commit=ref)
            if not tags:
                return None
            tag = tags[0]
        try:
            release_version = ReleaseVersion.from_string(tag)
        except AssertionError:
            raise self.m.step.StepFailure("Tag %s is not a release version" % tag)
        return release_version

    def get_branches(self, repo_path, remotes=False):
        """Get release branches.

        Args:
          repo_path (Path): Local git repository.
          remotes (bool): If set, return remote branches only. Otherwise, return
            local branches only.

        Returns:
          list(str): A list of release branches.
        """
        with self.m.context(cwd=repo_path):
            branches = self.m.git.branch(step_name="get branches", remotes=remotes)
        return [b for b in branches if self.validate_branch(b)]

    def shorten_revision(self, revision):
        """Shorten a revision.

        Args:
          revision (str): A git revision SHA1.

        Returns:
          str: Shortened revision.
        """
        return revision[: ReleaseApi._REVISION_LENGTH]

    def validate_branch(self, branch):
        """Validate branch name.

        Args:
          branch (str): Branch name to validate.

        Returns:
          bool: True if branch name is valid, otherwise False.
        """
        return bool(re.compile(r"^releases/[^/]+$").match(branch))

    def get_milestone_from_branch(self, branch):
        """Get the milestone number from a milestone branch, if it is a
        milestone branch.

        Args:
          branch (str): Branch name.

        Returns:
          int: Milestone number, or -1 if the branch is not a milestone branch.
        """
        m = re.match(r"^releases/f([0-9]+)$", branch)
        return int(m.group(1)) if m else -1

    def set_output_properties(self, presentation, release_revision, release_version):
        """
        Set output properties for a release. This may be used for UI purposes
        following a successful release operation, e.g. snap, cherry-pick.

        Args:
          presentation (Step): Step to attach properties and links to.
          release_revision (str): Git revision corresponding to a release version.
          release_version (ReleaseVersion): Release version.
        """
        presentation.properties["release_version"] = str(release_version)
        presentation.properties["release_revision"] = release_revision
        presentation.links[
            "release_version"
        ] = "http://go/fuchsia-release-version/%s" % str(release_version).replace(
            "releases/", ""
        )