path: root/src/arrow/dev/archery/archery/crossbow/core.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import os
import re
import fnmatch
import glob
import time
import logging
import mimetypes
import subprocess
import textwrap
from io import StringIO
from pathlib import Path
from datetime import date

import jinja2
from ruamel.yaml import YAML

try:
    import github3
    _have_github3 = True
except ImportError:
    github3 = object
    _have_github3 = False

try:
    import pygit2
except ImportError:
    PygitRemoteCallbacks = object
else:
    PygitRemoteCallbacks = pygit2.RemoteCallbacks

from ..utils.source import ArrowSources


for pkg in ["requests", "urllib3", "github3"]:
    logging.getLogger(pkg).setLevel(logging.WARNING)

logger = logging.getLogger("crossbow")


class CrossbowError(Exception):
    pass


def _flatten(mapping):
    """Converts a hierarchical mapping to a flat dictionary"""
    result = {}
    for k, v in mapping.items():
        if isinstance(v, dict):
            for ik, iv in _flatten(v).items():
                ik = ik if isinstance(ik, tuple) else (ik,)
                result[(k,) + ik] = iv
        elif isinstance(v, list):
            for ik, iv in enumerate(_flatten(v)):
                ik = ik if isinstance(ik, tuple) else (ik,)
                result[(k,) + ik] = iv
        else:
            result[(k,)] = v
    return result


def _unflatten(mapping):
    """Converts a flat tuple => object mapping to hierarchical one"""
    result = {}
    for path, value in mapping.items():
        parents, leaf = path[:-1], path[-1]
        # create the hierarchy until we reach the leaf value
        temp = result
        for parent in parents:
            temp.setdefault(parent, {})
            temp = temp[parent]
        # set the leaf value
        temp[leaf] = value

    return result


def _unflatten_tree(files):
    """Converts a flat path => object mapping to a hierarchical directories

    Input:
        {
            'path/to/file.a': a_content,
            'path/to/file.b': b_content,
            'path/file.c': c_content
        }
    Output:
        {
            'path': {
                'to': {
                    'file.a': a_content,
                    'file.b': b_content
                },
                'file.c': c_content
            }
        }
    """
    files = {tuple(k.split('/')): v for k, v in files.items()}
    return _unflatten(files)


def _render_jinja_template(searchpath, template, params):
    def format_all(items, pattern):
        return [pattern.format(item) for item in items]

    loader = jinja2.FileSystemLoader(searchpath)
    env = jinja2.Environment(loader=loader, trim_blocks=True,
                             lstrip_blocks=True,
                             undefined=jinja2.StrictUndefined)
    env.filters['format_all'] = format_all
    template = env.get_template(template)
    return template.render(**params)


# configurations for setting up branch skipping
# - appveyor has a feature to skip builds without an appveyor.yml
# - travis reads from the master branch and applies the rules
# - circle requires the configuration to be present on all branch, even ones
#   that are configured to be skipped
# - azure skips branches without azure-pipelines.yml by default
# - github skips branches without .github/workflows/ by default

_default_travis_yml = """
branches:
  only:
    - master
    - /.*-travis-.*/

os: linux
dist: trusty
language: generic
"""

_default_circle_yml = """
version: 2

jobs:
  build:
    machine: true

workflows:
  version: 2
  build:
    jobs:
      - build:
          filters:
            branches:
              only:
                - /.*-circle-.*/
"""

_default_tree = {
    '.travis.yml': _default_travis_yml,
    '.circleci/config.yml': _default_circle_yml
}


class GitRemoteCallbacks(PygitRemoteCallbacks):

    def __init__(self, token):
        self.token = token
        self.attempts = 0
        super().__init__()

    def push_update_reference(self, refname, message):
        pass

    def update_tips(self, refname, old, new):
        pass

    def credentials(self, url, username_from_url, allowed_types):
        # its a libgit2 bug, that it infinitely retries the authentication
        self.attempts += 1

        if self.attempts >= 5:
            # pygit2 doesn't propagate the exception properly
            msg = 'Wrong oauth personal access token'
            print(msg)
            raise CrossbowError(msg)

        if (allowed_types &
                pygit2.credentials.GIT_CREDENTIAL_USERPASS_PLAINTEXT):
            return pygit2.UserPass(self.token, 'x-oauth-basic')
        else:
            return None


def _git_ssh_to_https(url):
    return url.replace('git@github.com:', 'https://github.com/')


class Repo:
    """
    Base class for interaction with local git repositories

    A high level wrapper used for both reading revision information from
    arrow's repository and pushing continuous integration tasks to the queue
    repository.

    Parameters
    ----------
    require_https : boolean, default False
        Raise exception for SSH origin URLs
    """

    def __init__(self, path, github_token=None, remote_url=None,
                 require_https=False):
        self.path = Path(path)
        self.github_token = github_token
        self.require_https = require_https
        self._remote_url = remote_url
        self._pygit_repo = None
        self._github_repo = None  # set by as_github_repo()
        self._updated_refs = []

    def __str__(self):
        tpl = textwrap.dedent('''
            Repo: {remote}@{branch}
            Commit: {head}
        ''')
        return tpl.format(
            remote=self.remote_url,
            branch=self.branch.branch_name,
            head=self.head
        )

    @property
    def repo(self):
        if self._pygit_repo is None:
            self._pygit_repo = pygit2.Repository(str(self.path))
        return self._pygit_repo

    @property
    def origin(self):
        remote = self.repo.remotes['origin']
        if self.require_https and remote.url.startswith('git@github.com'):
            raise CrossbowError("Change SSH origin URL to HTTPS to use "
                                "Crossbow: {}".format(remote.url))
        return remote

    def fetch(self):
        refspec = '+refs/heads/*:refs/remotes/origin/*'
        self.origin.fetch([refspec])

    def push(self, refs=None, github_token=None):
        github_token = github_token or self.github_token
        if github_token is None:
            raise RuntimeError(
                'Could not determine GitHub token. Please set the '
                'CROSSBOW_GITHUB_TOKEN environment variable to a '
                'valid GitHub access token or pass one to --github-token.'
            )
        callbacks = GitRemoteCallbacks(github_token)
        refs = refs or []
        try:
            self.origin.push(refs + self._updated_refs, callbacks=callbacks)
        except pygit2.GitError:
            raise RuntimeError('Failed to push updated references, '
                               'potentially because of credential issues: {}'
                               .format(self._updated_refs))
        else:
            self.updated_refs = []

    @property
    def head(self):
        """Currently checked out commit's sha"""
        return self.repo.head

    @property
    def branch(self):
        """Currently checked out branch"""
        try:
            return self.repo.branches[self.repo.head.shorthand]
        except KeyError:
            return None  # detached

    @property
    def remote(self):
        """Currently checked out branch's remote counterpart"""
        try:
            return self.repo.remotes[self.branch.upstream.remote_name]
        except (AttributeError, KeyError):
            return None  # cannot detect

    @property
    def remote_url(self):
        """Currently checked out branch's remote counterpart URL

        If an SSH github url is set, it will be replaced by the https
        equivalent usable with GitHub OAuth token.
        """
        try:
            return self._remote_url or _git_ssh_to_https(self.remote.url)
        except AttributeError:
            return None

    @property
    def user_name(self):
        try:
            return next(self.repo.config.get_multivar('user.name'))
        except StopIteration:
            return os.environ.get('GIT_COMMITTER_NAME', 'unknown')

    @property
    def user_email(self):
        try:
            return next(self.repo.config.get_multivar('user.email'))
        except StopIteration:
            return os.environ.get('GIT_COMMITTER_EMAIL', 'unknown')

    @property
    def signature(self):
        return pygit2.Signature(self.user_name, self.user_email,
                                int(time.time()))

    def create_tree(self, files):
        builder = self.repo.TreeBuilder()

        for filename, content in files.items():
            if isinstance(content, dict):
                # create a subtree
                tree_id = self.create_tree(content)
                builder.insert(filename, tree_id, pygit2.GIT_FILEMODE_TREE)
            else:
                # create a file
                blob_id = self.repo.create_blob(content)
                builder.insert(filename, blob_id, pygit2.GIT_FILEMODE_BLOB)

        tree_id = builder.write()
        return tree_id

    def create_commit(self, files, parents=None, message='',
                      reference_name=None):
        if parents is None:
            # by default use the main branch as the base of the new branch
            # required to reuse github actions cache across crossbow tasks
            commit, _ = self.repo.resolve_refish("master")
            parents = [commit.id]
        tree_id = self.create_tree(files)

        author = committer = self.signature
        commit_id = self.repo.create_commit(reference_name, author, committer,
                                            message, tree_id, parents)
        return self.repo[commit_id]

    def create_branch(self, branch_name, files, parents=None, message='',
                      signature=None):
        # create commit with the passed tree
        commit = self.create_commit(files, parents=parents, message=message)

        # create branch pointing to the previously created commit
        branch = self.repo.create_branch(branch_name, commit)

        # append to the pushable references
        self._updated_refs.append('refs/heads/{}'.format(branch_name))

        return branch

    def create_tag(self, tag_name, commit_id, message=''):
        tag_id = self.repo.create_tag(tag_name, commit_id,
                                      pygit2.GIT_OBJ_COMMIT, self.signature,
                                      message)

        # append to the pushable references
        self._updated_refs.append('refs/tags/{}'.format(tag_name))

        return self.repo[tag_id]

    def file_contents(self, commit_id, file):
        commit = self.repo[commit_id]
        entry = commit.tree[file]
        blob = self.repo[entry.id]
        return blob.data

    def _parse_github_user_repo(self):
        m = re.match(r'.*\/([^\/]+)\/([^\/\.]+)(\.git)?$', self.remote_url)
        if m is None:
            raise CrossbowError(
                "Unable to parse the github owner and repository from the "
                "repository's remote url '{}'".format(self.remote_url)
            )
        user, repo = m.group(1), m.group(2)
        return user, repo

    def as_github_repo(self, github_token=None):
        """Converts it to a repository object which wraps the GitHub API"""
        if self._github_repo is None:
            if not _have_github3:
                raise ImportError('Must install github3.py')
            github_token = github_token or self.github_token
            username, reponame = self._parse_github_user_repo()
            session = github3.session.GitHubSession(
                default_connect_timeout=10,
                default_read_timeout=30
            )
            github = github3.GitHub(session=session)
            github.login(token=github_token)
            self._github_repo = github.repository(username, reponame)
        return self._github_repo

    def github_commit(self, sha):
        repo = self.as_github_repo()
        return repo.commit(sha)

    def github_release(self, tag):
        repo = self.as_github_repo()
        try:
            return repo.release_from_tag(tag)
        except github3.exceptions.NotFoundError:
            return None

    def github_upload_asset_requests(self, release, path, name, mime,
                                     max_retries=None, retry_backoff=None):
        if max_retries is None:
            max_retries = int(os.environ.get('CROSSBOW_MAX_RETRIES', 8))
        if retry_backoff is None:
            retry_backoff = int(os.environ.get('CROSSBOW_RETRY_BACKOFF', 5))

        for i in range(max_retries):
            try:
                with open(path, 'rb') as fp:
                    result = release.upload_asset(name=name, asset=fp,
                                                  content_type=mime)
            except github3.exceptions.ResponseError as e:
                logger.error('Attempt {} has failed with message: {}.'
                             .format(i + 1, str(e)))
                logger.error('Error message {}'.format(e.msg))
                logger.error('List of errors provided by Github:')
                for err in e.errors:
                    logger.error(' - {}'.format(err))

                if e.code == 422:
                    # 422 Validation Failed, probably raised because
                    # ReleaseAsset already exists, so try to remove it before
                    # reattempting the asset upload
                    for asset in release.assets():
                        if asset.name == name:
                            logger.info('Release asset {} already exists, '
                                        'removing it...'.format(name))
                            asset.delete()
                            logger.info('Asset {} removed.'.format(name))
                            break
            except github3.exceptions.ConnectionError as e:
                logger.error('Attempt {} has failed with message: {}.'
                             .format(i + 1, str(e)))
            else:
                logger.info('Attempt {} has finished.'.format(i + 1))
                return result

            time.sleep(retry_backoff)

        raise RuntimeError('Github asset uploading has failed!')

    def github_upload_asset_curl(self, release, path, name, mime):
        upload_url, _ = release.upload_url.split('{?')
        upload_url += '?name={}'.format(name)

        command = [
            'curl',
            '--fail',
            '-H', "Authorization: token {}".format(self.github_token),
            '-H', "Content-Type: {}".format(mime),
            '--data-binary', '@{}'.format(path),
            upload_url
        ]
        return subprocess.run(command, shell=False, check=True)

    def github_overwrite_release_assets(self, tag_name, target_commitish,
                                        patterns, method='requests'):
        # Since github has changed something the asset uploading via requests
        # got instable, so prefer the cURL alternative.
        # Potential cause:
        #    sigmavirus24/github3.py/issues/779#issuecomment-379470626
        repo = self.as_github_repo()
        if not tag_name:
            raise CrossbowError('Empty tag name')
        if not target_commitish:
            raise CrossbowError('Empty target commit for the release tag')

        # remove the whole release if it already exists
        try:
            release = repo.release_from_tag(tag_name)
        except github3.exceptions.NotFoundError:
            pass
        else:
            release.delete()

        release = repo.create_release(tag_name, target_commitish)
        for pattern in patterns:
            for path in glob.glob(pattern, recursive=True):
                name = os.path.basename(path)
                size = os.path.getsize(path)
                mime = mimetypes.guess_type(name)[0] or 'application/zip'

                logger.info(
                    'Uploading asset `{}` with mimetype {} and size {}...'
                    .format(name, mime, size)
                )

                if method == 'requests':
                    self.github_upload_asset_requests(release, path, name=name,
                                                      mime=mime)
                elif method == 'curl':
                    self.github_upload_asset_curl(release, path, name=name,
                                                  mime=mime)
                else:
                    raise CrossbowError(
                        'Unsupported upload method {}'.format(method)
                    )


class Queue(Repo):

    def _latest_prefix_id(self, prefix):
        pattern = re.compile(r'[\w\/-]*{}-(\d+)'.format(prefix))
        matches = list(filter(None, map(pattern.match, self.repo.branches)))
        if matches:
            latest = max(int(m.group(1)) for m in matches)
        else:
            latest = -1
        return latest

    def _next_job_id(self, prefix):
        """Auto increments the branch's identifier based on the prefix"""
        latest_id = self._latest_prefix_id(prefix)
        return '{}-{}'.format(prefix, latest_id + 1)

    def latest_for_prefix(self, prefix):
        latest_id = self._latest_prefix_id(prefix)
        if latest_id < 0:
            raise RuntimeError(
                'No job has been submitted with prefix {} yet'.format(prefix)
            )
        job_name = '{}-{}'.format(prefix, latest_id)
        return self.get(job_name)

    def date_of(self, job):
        # it'd be better to bound to the queue repository on deserialization
        # and reorganize these methods to Job
        branch_name = 'origin/{}'.format(job.branch)
        branch = self.repo.branches[branch_name]
        commit = self.repo[branch.target]
        return date.fromtimestamp(commit.commit_time)

    def jobs(self, pattern):
        """Return jobs sorted by its identifier in reverse order"""
        job_names = []
        for name in self.repo.branches.remote:
            origin, name = name.split('/', 1)
            result = re.match(pattern, name)
            if result:
                job_names.append(name)

        for name in sorted(job_names, reverse=True):
            yield self.get(name)

    def get(self, job_name):
        branch_name = 'origin/{}'.format(job_name)
        branch = self.repo.branches[branch_name]
        try:
            content = self.file_contents(branch.target, 'job.yml')
        except KeyError:
            raise CrossbowError(
                'No job is found with name: {}'.format(job_name)
            )

        buffer = StringIO(content.decode('utf-8'))
        job = yaml.load(buffer)
        job.queue = self
        return job

    def put(self, job, prefix='build'):
        if not isinstance(job, Job):
            raise CrossbowError('`job` must be an instance of Job')
        if job.branch is not None:
            raise CrossbowError('`job.branch` is automatically generated, '
                                'thus it must be blank')

        if job.target.remote is None:
            raise CrossbowError(
                'Cannot determine git remote for the Arrow repository to '
                'clone or push to, try to push the `{}` branch first to have '
                'a remote tracking counterpart.'.format(job.target.branch)
            )
        if job.target.branch is None:
            raise CrossbowError(
                'Cannot determine the current branch of the Arrow repository '
                'to clone or push to, perhaps it is in detached HEAD state. '
                'Please checkout a branch.'
            )

        # auto increment and set next job id, e.g. build-85
        job._queue = self
        job.branch = self._next_job_id(prefix)

        # create tasks' branches
        for task_name, task in job.tasks.items():
            # adding CI's name to the end of the branch in order to use skip
            # patterns on travis and circleci
            task.branch = '{}-{}-{}'.format(job.branch, task.ci, task_name)
            params = {
                **job.params,
                "arrow": job.target,
                "queue_remote_url": self.remote_url
            }
            files = task.render_files(job.template_searchpath, params=params)
            branch = self.create_branch(task.branch, files=files)
            self.create_tag(task.tag, branch.target)
            task.commit = str(branch.target)

        # create job's branch with its description
        return self.create_branch(job.branch, files=job.render_files())


def get_version(root, **kwargs):
    """
    Parse function for setuptools_scm that ignores tags for non-C++
    subprojects, e.g. apache-arrow-js-XXX tags.
    """
    from setuptools_scm.git import parse as parse_git_version

    # query the calculated version based on the git tags
    kwargs['describe_command'] = (
        'git describe --dirty --tags --long --match "apache-arrow-[0-9].*"'
    )
    version = parse_git_version(root, **kwargs)
    tag = str(version.tag)

    # We may get a development tag for the next version, such as "5.0.0.dev0",
    # or the tag of an already released version, such as "4.0.0".
    # In the latter case, we need to increment the version so that the computed
    # version comes after any patch release (the next feature version after
    # 4.0.0 is 5.0.0).
    pattern = r"^(\d+)\.(\d+)\.(\d+)"
    match = re.match(pattern, tag)
    major, minor, patch = map(int, match.groups())
    if 'dev' not in tag:
        major += 1

    return "{}.{}.{}.dev{}".format(major, minor, patch, version.distance)


class Serializable:

    @classmethod
    def to_yaml(cls, representer, data):
        tag = '!{}'.format(cls.__name__)
        dct = {k: v for k, v in data.__dict__.items() if not k.startswith('_')}
        return representer.represent_mapping(tag, dct)


class Target(Serializable):
    """
    Describes target repository and revision the builds run against

    This serializable data container holding information about arrow's
    git remote, branch, sha and version number as well as some metadata
    (currently only an email address where the notification should be sent).
    """

    def __init__(self, head, branch, remote, version, email=None):
        self.head = head
        self.email = email
        self.branch = branch
        self.remote = remote
        self.version = version
        self.no_rc_version = re.sub(r'-rc\d+\Z', '', version)
        # Semantic Versioning 1.0.0: https://semver.org/spec/v1.0.0.html
        #
        # > A pre-release version number MAY be denoted by appending an
        # > arbitrary string immediately following the patch version and a
        # > dash. The string MUST be comprised of only alphanumerics plus
        # > dash [0-9A-Za-z-].
        #
        # Example:
        #
        #   '0.16.1.dev10' ->
        #   '0.16.1-dev10'
        self.no_rc_semver_version = \
            re.sub(r'\.(dev\d+)\Z', r'-\1', self.no_rc_version)

    @classmethod
    def from_repo(cls, repo, head=None, branch=None, remote=None, version=None,
                  email=None):
        """Initialize from a repository

        Optionally override detected remote, branch, head, and/or version.
        """
        assert isinstance(repo, Repo)

        if head is None:
            head = str(repo.head.target)
        if branch is None:
            branch = repo.branch.branch_name
        if remote is None:
            remote = repo.remote_url
        if version is None:
            version = get_version(repo.path)
        if email is None:
            email = repo.user_email

        return cls(head=head, email=email, branch=branch, remote=remote,
                   version=version)


class Task(Serializable):
    """
    Describes a build task and metadata required to render CI templates

    A task is represented as a single git commit and branch containing jinja2
    rendered files (currently appveyor.yml or .travis.yml configurations).

    A task can't be directly submitted to a queue, must belong to a job.
    Each task's unique identifier is its branch name, which is generated after
    submitting the job to a queue.
    """

    def __init__(self, ci, template, artifacts=None, params=None):
        assert ci in {
            'circle',
            'travis',
            'appveyor',
            'azure',
            'github',
            'drone',
        }
        self.ci = ci
        self.template = template
        self.artifacts = artifacts or []
        self.params = params or {}
        self.branch = None  # filled after adding to a queue
        self.commit = None  # filled after adding to a queue
        self._queue = None  # set by the queue object after put or get
        self._status = None  # status cache
        self._assets = None  # assets cache

    def render_files(self, searchpath, params=None):
        params = {**self.params, **(params or {}), "task": self}
        try:
            rendered = _render_jinja_template(searchpath, self.template,
                                              params=params)
        except jinja2.TemplateError as e:
            raise RuntimeError(
                'Failed to render template `{}` with {}: {}'.format(
                    self.template, e.__class__.__name__, str(e)
                )
            )

        tree = {**_default_tree, self.filename: rendered}
        return _unflatten_tree(tree)

    @property
    def tag(self):
        return self.branch

    @property
    def filename(self):
        config_files = {
            'circle': '.circleci/config.yml',
            'travis': '.travis.yml',
            'appveyor': 'appveyor.yml',
            'azure': 'azure-pipelines.yml',
            'github': '.github/workflows/crossbow.yml',
            'drone': '.drone.yml',
        }
        return config_files[self.ci]

    def status(self, force_query=False):
        _status = getattr(self, '_status', None)
        if force_query or _status is None:
            github_commit = self._queue.github_commit(self.commit)
            self._status = TaskStatus(github_commit)
        return self._status

    def assets(self, force_query=False, validate_patterns=True):
        _assets = getattr(self, '_assets', None)
        if force_query or _assets is None:
            github_release = self._queue.github_release(self.tag)
            self._assets = TaskAssets(github_release,
                                      artifact_patterns=self.artifacts,
                                      validate_patterns=validate_patterns)
        return self._assets


class TaskStatus:
    """
    Combine the results from status and checks API to a single state.

    Azure pipelines uses checks API which doesn't provide a combined
    interface like status API does, so we need to manually combine
    both the commit statuses and the commit checks coming from
    different API endpoint

    Status.state: error, failure, pending or success, default pending
    CheckRun.status: queued, in_progress or completed, default: queued
    CheckRun.conclusion: success, failure, neutral, cancelled, timed_out
                            or action_required, only set if
                            CheckRun.status == 'completed'

    1. Convert CheckRun's status and conclusion to one of Status.state
    2. Merge the states based on the following rules:
        - failure if any of the contexts report as error or failure
        - pending if there are no statuses or a context is pending
        - success if the latest status for all contexts is success
        error otherwise.

    Parameters
    ----------
    commit : github3.Commit
        Commit to query the combined status for.

    Returns
    -------
    TaskStatus(
        combined_state='error|failure|pending|success',
        github_status='original github status object',
        github_check_runs='github checks associated with the commit',
        total_count='number of statuses and checks'
    )
    """

    def __init__(self, commit):
        status = commit.status()
        check_runs = list(commit.check_runs())
        states = [s.state for s in status.statuses]

        for check in check_runs:
            if check.status == 'completed':
                if check.conclusion in {'success', 'failure'}:
                    states.append(check.conclusion)
                elif check.conclusion in {'cancelled', 'timed_out',
                                          'action_required'}:
                    states.append('error')
                # omit `neutral` conclusion
            else:
                states.append('pending')

        # it could be more effective, but the following is more descriptive
        combined_state = 'error'
        if len(states):
            if any(state in {'error', 'failure'} for state in states):
                combined_state = 'failure'
            elif any(state == 'pending' for state in states):
                combined_state = 'pending'
            elif all(state == 'success' for state in states):
                combined_state = 'success'

        # show link to the actual build, some of the CI providers implement
        # the statuses API others implement the checks API, so display both
        build_links = [s.target_url for s in status.statuses]
        build_links += [c.html_url for c in check_runs]

        self.combined_state = combined_state
        self.github_status = status
        self.github_check_runs = check_runs
        self.total_count = len(states)
        self.build_links = build_links


class TaskAssets(dict):

    def __init__(self, github_release, artifact_patterns,
                 validate_patterns=True):
        # HACK(kszucs): don't expect uploaded assets of no atifacts were
        # defiened for the tasks in order to spare a bit of github rate limit
        if not artifact_patterns:
            return

        if github_release is None:
            github_assets = {}  # no assets have been uploaded for the task
        else:
            github_assets = {a.name: a for a in github_release.assets()}

        if not validate_patterns:
            # shortcut to avoid pattern validation and just set all artifacts
            return self.update(github_assets)

        for pattern in artifact_patterns:
            # artifact can be a regex pattern
            compiled = re.compile(f"^{pattern}$")
            matches = list(
                filter(None, map(compiled.match, github_assets.keys()))
            )
            num_matches = len(matches)

            # validate artifact pattern matches single asset
            if num_matches == 0:
                self[pattern] = None
            elif num_matches == 1:
                self[pattern] = github_assets[matches[0].group(0)]
            else:
                raise CrossbowError(
                    'Only a single asset should match pattern `{}`, there are '
                    'multiple ones: {}'.format(pattern, ', '.join(matches))
                )

    def missing_patterns(self):
        return [pattern for pattern, asset in self.items() if asset is None]

    def uploaded_assets(self):
        return [asset for asset in self.values() if asset is not None]


class Job(Serializable):
    """Describes multiple tasks against a single target repository"""

    def __init__(self, target, tasks, params=None, template_searchpath=None):
        if not tasks:
            raise ValueError('no tasks were provided for the job')
        if not all(isinstance(task, Task) for task in tasks.values()):
            raise ValueError('each `tasks` mus be an instance of Task')
        if not isinstance(target, Target):
            raise ValueError('`target` must be an instance of Target')
        if not isinstance(target, Target):
            raise ValueError('`target` must be an instance of Target')
        if not isinstance(params, dict):
            raise ValueError('`params` must be an instance of dict')

        self.target = target
        self.tasks = tasks
        self.params = params or {}  # additional parameters for the tasks
        self.branch = None  # filled after adding to a queue
        self._queue = None  # set by the queue object after put or get
        if template_searchpath is None:
            self._template_searchpath = ArrowSources.find().path
        else:
            self._template_searchpath = template_searchpath

    def render_files(self):
        with StringIO() as buf:
            yaml.dump(self, buf)
            content = buf.getvalue()
        tree = {**_default_tree, "job.yml": content}
        return _unflatten_tree(tree)

    def render_tasks(self, params=None):
        result = {}
        params = {
            **self.params,
            "arrow": self.target,
            **(params or {})
        }
        for task_name, task in self.tasks.items():
            files = task.render_files(self._template_searchpath, params)
            result[task_name] = files
        return result

    @property
    def template_searchpath(self):
        return self._template_searchpath

    @property
    def queue(self):
        assert isinstance(self._queue, Queue)
        return self._queue

    @queue.setter
    def queue(self, queue):
        assert isinstance(queue, Queue)
        self._queue = queue
        for task in self.tasks.values():
            task._queue = queue

    @property
    def email(self):
        return os.environ.get('CROSSBOW_EMAIL', self.target.email)

    @property
    def date(self):
        return self.queue.date_of(self)

    def show(self, stream=None):
        return yaml.dump(self, stream=stream)

    @classmethod
    def from_config(cls, config, target, tasks=None, groups=None, params=None):
        """
        Intantiate a job from based on a config.

        Parameters
        ----------
        config : dict
            Deserialized content of tasks.yml
        target : Target
            Describes target repository and revision the builds run against.
        tasks : Optional[List[str]], default None
            List of glob patterns for matching task names.
        groups : Optional[List[str]], default None
            List of exact group names matching predefined task sets in the
            config.
        params : Optional[Dict[str, str]], default None
            Additional rendering parameters for the task templates.

        Returns
        -------
        Job

        Raises
        ------
        Exception:
            If invalid groups or tasks has been passed.
        """
        task_definitions = config.select(tasks, groups=groups)

        # instantiate the tasks
        tasks = {}
        versions = {'version': target.version,
                    'no_rc_version': target.no_rc_version,
                    'no_rc_semver_version': target.no_rc_semver_version}
        for task_name, task in task_definitions.items():
            artifacts = task.pop('artifacts', None) or []  # because of yaml
            artifacts = [fn.format(**versions) for fn in artifacts]
            tasks[task_name] = Task(artifacts=artifacts, **task)

        return cls(target=target, tasks=tasks, params=params,
                   template_searchpath=config.template_searchpath)

    def is_finished(self):
        for task in self.tasks.values():
            status = task.status(force_query=True)
            if status.combined_state == 'pending':
                return False
        return True

    def wait_until_finished(self, poll_max_minutes=120,
                            poll_interval_minutes=10):
        started_at = time.time()
        while True:
            if self.is_finished():
                break

            waited_for_minutes = (time.time() - started_at) / 60
            if waited_for_minutes > poll_max_minutes:
                msg = ('Exceeded the maximum amount of time waiting for job '
                       'to finish, waited for {} minutes.')
                raise RuntimeError(msg.format(waited_for_minutes))

            logger.info('Waiting {} minutes and then checking again'
                        .format(poll_interval_minutes))
            time.sleep(poll_interval_minutes * 60)


class Config(dict):

    def __init__(self, tasks, template_searchpath):
        super().__init__(tasks)
        self.template_searchpath = template_searchpath

    @classmethod
    def load_yaml(cls, path):
        path = Path(path)
        searchpath = path.parent
        rendered = _render_jinja_template(searchpath, template=path.name,
                                          params={})
        config = yaml.load(rendered)
        return cls(config, template_searchpath=searchpath)

    def show(self, stream=None):
        return yaml.dump(dict(self), stream=stream)

    def select(self, tasks=None, groups=None):
        config_groups = dict(self['groups'])
        config_tasks = dict(self['tasks'])
        valid_groups = set(config_groups.keys())
        valid_tasks = set(config_tasks.keys())
        group_whitelist = list(groups or [])
        task_whitelist = list(tasks or [])

        # validate that the passed groups are defined in the config
        requested_groups = set(group_whitelist)
        invalid_groups = requested_groups - valid_groups
        if invalid_groups:
            msg = 'Invalid group(s) {!r}. Must be one of {!r}'.format(
                invalid_groups, valid_groups
            )
            raise CrossbowError(msg)

        # merge the tasks defined in the selected groups
        task_patterns = [list(config_groups[name]) for name in group_whitelist]
        task_patterns = set(sum(task_patterns, task_whitelist))

        # treat the task names as glob patterns to select tasks more easily
        requested_tasks = set()
        for pattern in task_patterns:
            matches = fnmatch.filter(valid_tasks, pattern)
            if len(matches):
                requested_tasks.update(matches)
            else:
                raise CrossbowError(
                    "Unable to match any tasks for `{}`".format(pattern)
                )

        # validate that the passed and matched tasks are defined in the config
        invalid_tasks = requested_tasks - valid_tasks
        if invalid_tasks:
            msg = 'Invalid task(s) {!r}. Must be one of {!r}'.format(
                invalid_tasks, valid_tasks
            )
            raise CrossbowError(msg)

        return {
            task_name: config_tasks[task_name] for task_name in requested_tasks
        }

    def validate(self):
        # validate that the task groups are properly referening the tasks
        for group_name, group in self['groups'].items():
            for pattern in group:
                tasks = self.select(tasks=[pattern])
                if not tasks:
                    raise CrossbowError(
                        "The pattern `{}` defined for task group `{}` is not "
                        "matching any of the tasks defined in the "
                        "configuration file.".format(pattern, group_name)
                    )

        # validate that the tasks are constructible
        for task_name, task in self['tasks'].items():
            try:
                Task(**task)
            except Exception as e:
                raise CrossbowError(
                    'Unable to construct a task object from the '
                    'definition  of task `{}`. The original error message '
                    'is: `{}`'.format(task_name, str(e))
                )

        # validate that the defined tasks are renderable, in order to to that
        # define the required object with dummy data
        target = Target(
            head='e279a7e06e61c14868ca7d71dea795420aea6539',
            branch='master',
            remote='https://github.com/apache/arrow',
            version='1.0.0dev123',
            email='dummy@example.ltd'
        )

        for task_name, task in self['tasks'].items():
            task = Task(**task)
            files = task.render_files(
                self.template_searchpath,
                params=dict(
                    arrow=target,
                    queue_remote_url='https://github.com/org/crossbow'
                )
            )
            if not files:
                raise CrossbowError('No files have been rendered for task `{}`'
                                    .format(task_name))


# configure yaml serializer
yaml = YAML()
yaml.register_class(Job)
yaml.register_class(Task)
yaml.register_class(Target)