bodhi-ci

#!/usr/bin/python3
# Copyright © 2018 Red Hat, Inc.
#
# This file is part of Bodhi.
#
# 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.
"""Bodhi's CI command tool."""
import asyncio
import datetime
import functools
import multiprocessing
import os
import shutil
import signal
import subprocess
import sys
import typing
import uuid

import click


CONTAINER_NAME = 'bodhi-ci'
# We label the containers we run so it's easy to find them when we run _stop_all_jobs() at the end.
# UUID is used so that one bodhi-ci process does not stop jobs started by a different one.
CONTAINER_LABEL = 'purpose=bodhi-ci-{}'.format(uuid.uuid4())
# This template is used to generate the summary lines that are printed out at the end.
LABEL_TEMPLATE = '{:>8}-{:<27}'
PROJECT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
RELEASES = ('f28', 'f29', 'rawhide', 'pip')


def _set_concurrency(ctx, param, value):
    """
    Set up the concurrency_semaphore.

    Args:
        ctx (click.core.Context): The Click context, unused.
        param (click.core.Option): The option being handled. Unused.
        value (str): The value of the -j flag.
    Returns:
        str: The value of the -j flag.
    """
    Job.concurrency_semaphore = asyncio.Semaphore(value=value)
    # We don't want too many integration test jobs to run at once because they consume
    # many system resources and will fail due to timeouts. Therefore, let's set another semaphore
    # to limit how many of these can run at once that cannot be greater than an arbitrary value.
    # The arbitrary value was chosen because it seems to work in a Vagrant guest on a
    # Lenovo T480s.
    IntegrationJob.extra_concurrency_semaphore = asyncio.Semaphore(value=max(1, value / 4))
    return value


def _set_container_runtime(ctx, param, value):
    """
    Set up the container_runtime variable.

    Args:
        ctx (click.core.Context): The Click context, unused.
        param (click.core.Option): The option being handled. Unused.
        value (str): The value of the --container-runtime flag.
    Returns:
        str: The value of the --container-runtime flag.
    """
    global CONTAINER_NAME
    global container_runtime
    container_runtime = value
    if value == 'podman':
        # Workaround for https://github.com/containers/buildah/issues/1034
        CONTAINER_NAME = 'localhost/{}'.format(CONTAINER_NAME)
    return value


def _set_global(ctx, param, value):
    """
    Set up a global variable based on click input.

    Args:
        ctx (click.core.Context): The Click context, unused.
        param (click.core.Option): The option being handled. Used to find the global we are setting.
        value (str): The value of the flag.
    Returns:
        bool: The value of the flag.
    """
    globals()[param.name] = value
    return value


archive_option = click.option(
    '--archive/--no-archive', is_flag=True, default=True,
    help=("Collect *.xml from the tests and put them into test_results/."))
archive_path_option = click.option(
    '--archive-path', envvar='BODHI_CI_ARCHIVE_PATH',
    default=os.path.join(PROJECT_PATH, 'test_results'),
    help='Define where test results should be placed if -a is used.')
concurrency_option = click.option(
    '--concurrency', '-j', default=multiprocessing.cpu_count(), callback=_set_concurrency, type=int,
    help=('Number of concurrent processes to run. Integration test runs are separately limited due '
          'to resource contention. Defaults to the number of cores detected'))
container_runtime_option = click.option(
    '--container-runtime', '-c', default='docker', type=click.Choice(['docker', 'podman']),
    help='Select the container runtime to use. Defaults to docker.',
    callback=_set_container_runtime)
failfast_option = click.option('--failfast', '-x', is_flag=True,
                               help='Exit immediately upon error.', callback=_set_global)
init_option = click.option(
    '--init/--no-init', default=True, callback=_set_global,
    help=("Use docker's --init flag. This option only has an effect when docker is "
          "the container runtime being used."))
no_build_option = click.option(
    '--no-build', is_flag=True, callback=_set_global,
    help='Do not run docker build if the image already exists.')
pyver_option = click.option(
    '--pyver', '-p', default=[2, 3], multiple=True, type=int,
    help=("Limit to a particular Python major version. May be specified multiple times. "
          "Acceptable values: 2, 3"))
releases_option = click.option(
    '--release', '-r', default=list(RELEASES), multiple=True,
    help=("Limit to a particular release. May be specified multiple times. "
          "Acceptable values: {}".format(', '.join(RELEASES))))
tty_option = click.option('--tty/--no-tty', default=True, help='Allocate a pseudo-TTY.',
                          callback=_set_global)

container_runtime = 'docker'
failfast = False
init = True
# If True, we will try to skip running any builds if suitable builds already exist. Set by
# _set_global().
no_build = False
tty = False


@click.group()
def cli():
    """
    Bodhi's Continuous Integration helper script.
    """


@cli.command()
@archive_option
@archive_path_option
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@pyver_option
@releases_option
@tty_option
def all(archive, archive_path, concurrency, container_runtime, no_build, failfast, init, pyver,
        release, tty):
    """Run all the types of tests in parallel."""
    _run_jobs(_build_jobs_list('all', concurrency, release, pyvers=pyver, archive=archive,
                               archive_path=archive_path))


@cli.command()
@concurrency_option
@container_runtime_option
@failfast_option
@releases_option
@tty_option
def build(concurrency, container_runtime, failfast, release, tty):
    """Build the containers for testing."""
    _run_jobs(_build_jobs_list('build', concurrency, release))


@cli.command()
@concurrency_option
@container_runtime_option
@init_option
@releases_option
@tty_option
def clean(concurrency, container_runtime, init, release, tty):
    """Remove all builds pertaining to Bodhi CI."""
    buffer_output = concurrency != 1 or len(release) != 1
    clean_jobs = [
        CleanJob(r, buffer_output=buffer_output) for r in release
    ] + [
        IntegrationCleanJob(app_name="bodhi", release=r, buffer_output=buffer_output)
        for r in release
    ] + [
        IntegrationCleanJob(app_name=app_name, release="prod", buffer_output=buffer_output)
        for app_name in ("resultsdb", "waiverdb", "greenwave")
    ]
    _run_jobs(clean_jobs)


@cli.command()
@archive_option
@archive_path_option
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@pyver_option
@releases_option
@tty_option
def diff_cover(archive, archive_path, concurrency, container_runtime, no_build, failfast, init,
               pyver, release, tty):
    """Run the diff cover test."""
    _run_jobs(_build_jobs_list('diff_cover', concurrency, release, pyvers=pyver, archive=archive,
                               archive_path=archive_path))


@cli.command()
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@releases_option
@tty_option
def docs(concurrency, container_runtime, failfast, init, no_build, release, tty):
    """Build the docs."""
    _run_jobs(_build_jobs_list('docs', concurrency, release))


@cli.command()
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@releases_option
@tty_option
def flake8(concurrency, container_runtime, failfast, init, no_build, release, tty):
    """Run flake8 tests."""
    _run_jobs(_build_jobs_list('flake8', concurrency, release))


@cli.command()
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@releases_option
@tty_option
def mypy(concurrency, container_runtime, failfast, init, no_build, release, tty):
    """Run mypy tests."""
    _run_jobs(_build_jobs_list('mypy', concurrency, release))


@cli.command()
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@releases_option
@tty_option
def pydocstyle(concurrency, container_runtime, failfast, init, no_build, release, tty):
    """Run pydocstyle tests."""
    _run_jobs(_build_jobs_list('pydocstyle', concurrency, release))


@cli.command()
@archive_option
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@pyver_option
@releases_option
@archive_path_option
@tty_option
def unit(archive, concurrency, container_runtime, no_build, failfast, init, pyver, release,
         archive_path, tty):
    """Run the unit tests."""
    _run_jobs(_build_jobs_list('unit', concurrency, release, pyvers=pyver, archive=archive,
                               archive_path=archive_path))


@cli.command("integration-build")
@concurrency_option
@container_runtime_option
@failfast_option
@pyver_option
@releases_option
@tty_option
def integration_build(concurrency, container_runtime, failfast, pyver, release, tty):
    """Build the containers for integration testing."""
    _run_jobs(_build_jobs_list('integration-build', concurrency, release, pyvers=pyver))


@cli.command("integration-download")
@concurrency_option
@failfast_option
@tty_option
def integration_download(concurrency, failfast, tty):
    """Download the database dumps for integration testing."""
    _run_jobs(_build_jobs_list('integration-download', concurrency, releases=[], pyvers=[3]))


@cli.command()
@archive_option
@concurrency_option
@container_runtime_option
@failfast_option
@init_option
@no_build_option
@pyver_option
@releases_option
@archive_path_option
@tty_option
def integration(archive, concurrency, container_runtime, no_build, failfast, init, pyver, release,
                archive_path, tty):
    """Run the integration tests."""
    _run_jobs(_build_jobs_list('integration', concurrency, release, pyvers=pyver, archive=archive,
                               archive_path=archive_path))


class ProgressReporter(object):
    """Report progress on the provided jobs.

    Attributes:
        jobs: A list of the Jobs to report on.
    """

    def __init__(self, jobs: typing.List['Job']):
        """
        Args:
            jobs: A list of the Jobs to report on.
        """
        self._jobs = []  # type: typing.List[Job]
        for job in jobs:
            self.register_job(job)

    def register_job(self, job):
        """Register a job to be reported.

        Args:
            job (Job): A job to report the status of.
        """
        self._jobs.append(job)
        loop = asyncio.get_event_loop()
        loop.create_task(self._print_on_complete(job))

    async def _print_on_complete(self, job):
        """Print the status of a job when it's complete."""
        await job.complete.wait()
        self.print_status()

    def print_status(self):
        """Print a status report on all the jobs."""
        for job in self._jobs:
            click.echo(job.summary_line)
        click.echo('\n')


class EmptySemaphore(object):
    """
    Implements a no-op context manager.

    This is used as the initial value on Job for the concurrency_semaphore and
    extra_concurrency_semaphore attributes so that they can be assumed to work with the async with
    statement if they are used without being turned into Semaphores.
    """

    async def __aenter__(self, *args, **kwargs):
        """Calls pass."""
        pass

    async def __aexit__(self, *args, **kwargs):
        """Calls pass."""
        pass


class Job(object):
    """
    Represent a CI job.

    This is intended to be a superclass for specific CI jobs, such as building container images, or
    running tests.

    Attributes:
        archive_dir (str): A path where this job's test results is on the host. Only set if an
            archive is used (not used on all jobs).
        release (str): The release this job is testing.
        depends_on (list of Job): This Job will wait for all the jobs' complete Event in the list
            to start.
        cancelled (bool): True if this Job has been cancelled, False otherwise.
        complete (asyncio.Event): An Event that allows other Jobs to wait for this Job to complete.
            complete.set() is called at the end of run().
        returncode (int or None): If the Job's process has finished, this will be set to its
            returncode. Otherwise it will be None.
        skipped (bool): If True, this Job was skipped.
    """

    # A template to name the image that is built. CONTAINER_NAME and the release gets substituted
    # into the {}'s.
    _container_image_template = '{}/{}'
    # Limit how many Jobs can run at once. This is set by _set_concurrency().
    concurrency_semaphore = EmptySemaphore()
    # Jobs can set this if they want to have some additional limitations on how many of them can
    # run at once. The IntegrationJob uses this, for example, so that we don't run too many of
    # them no matter what the -j flag is set to. This is set by _set_concurrency().
    extra_concurrency_semaphore = EmptySemaphore()

    def __init__(self, release: str,
                 depends_on: typing.Union[typing.List['Job'], 'Job'] = None,
                 buffer_output: bool = False):
        """
        Initialize the new Job.

        Args:
            release: The release this Job pertains to.
            depends_on: A list of Job that this Job should wait to complete before
                starting. It is also possible to provide a Job instance if there is a single
                dependency.
            buffer_output: If True, this Job will buffer its stdout and stderr into its
                output property. If False, child processes will send their output straign to stdout
                and stderr.
        """
        self.release = release
        self.depends_on = depends_on or []
        if not isinstance(self.depends_on, list):
            self.depends_on = [self.depends_on]

        self.cancelled = False
        # Used to block dependent processes until this Job is done.
        self.complete = asyncio.Event()
        self.started = False
        self.returncode = None
        self.skipped = False
        self._container_image = self._container_image_template.format(CONTAINER_NAME, self.release)
        self._popen_kwargs = {'shell': False}  # type: typing.Mapping[str, typing.Union[bool, int]]
        if buffer_output:
            # Let's buffer the output so the user doesn't see a jumbled mess.
            self._popen_kwargs['stdout'] = subprocess.PIPE
            self._popen_kwargs['stderr'] = subprocess.STDOUT
        self._stdout = b''

    def __repr__(self):
        return "<{} release={!r}>".format(self.__class__.__name__, self.release)

    @property
    def label(self):
        """
        Return a label that represents this job.

        This label is used for the status line at the end of the bodhi-ci script, and is also
        prepended to each line of output.

        Returns:
            str: A label to represent this Job.
        """
        return LABEL_TEMPLATE.format(self.release, self._label)

    @property
    def output(self):
        """
        Run decode on the output, and then prepend label in front of each line.

        Returns:
            str: The output from the process.
        """
        if not self._stdout:
            return ''
        output = self._stdout.decode()
        return '\n'.join(['{}\t{}'.format(self.label, line) for line in output.split('\n')])

    async def run(self):
        """
        Run the job, returning itself.

        Returns:
            Job: Returns self.
        """
        try:
            for job in self.depends_on:
                await job.complete.wait()
                if job.returncode != 0 and not job.skipped:
                    # If the Job we depend on failed, we should cancel.
                    raise asyncio.CancelledError()

            async with self.extra_concurrency_semaphore, self.concurrency_semaphore:
                # It's possible that we got cancelled while we were waiting on the Semaphore.
                if not self.cancelled:
                    self._pre_start_hook()
                    self._start_time = datetime.datetime.utcnow()
                    self.started = True
                    if not self._popen_kwargs["shell"]:
                        process = await asyncio.create_subprocess_exec(
                            *self._command, **self._popen_kwargs)
                    else:
                        process = await asyncio.create_subprocess_shell(
                            *self._command, **self._popen_kwargs)

                    try:
                        self._stdout, stderr = await process.communicate()
                        if process.returncode < 0:
                            # A negative exit code means that our child process was sent a signal,
                            # so let's mark this task as cancelled.
                            raise asyncio.CancelledError()
                    except asyncio.CancelledError:
                        try:
                            process.terminate()
                        except ProcessLookupError:
                            # The process is already stopped, nothing to see here.
                            pass
                        cancelled_stdout, stderr = await process.communicate()
                        if self._stdout:
                            self._stdout = self._stdout + cancelled_stdout
                        else:
                            self._stdout = cancelled_stdout
                        raise
                    finally:
                        self.returncode = process.returncode

            if self.returncode:
                # If there was a failure, we need to raise an Exception in case the failfast flag
                # was set, so that _run_jobs() can cancel the remaining tasks.
                error = RuntimeError()
                error.result = self
                raise error

        except asyncio.CancelledError:
            self.cancelled = True
        finally:
            # If the job's been cancelled or successful, let's go ahead and print its output now.
            # Failed jobs will have their output printed at the end.
            if self.output and (self.returncode == 0 or self.cancelled):
                click.echo(self.output)

            self._finish_time = datetime.datetime.utcnow()
            # Kick off any tasks that were waiting for us to finish.
            self.complete.set()

        return self

    @property
    def summary_line(self):
        """
        Create a summary line for the Job.

        If the exit_code indicates failure, it is printed to the console immediately. Failed jobs'
        stdout is not printed until the end of the job, so this gives the user a way to know that a
        job failed before its output is printed, and they can ctrl-c to see its output.

        Returns:
            str: A summary line suitable to print at the end of the process.
        """
        blue_start = '\033[0;34m' if tty else ''
        yellow_start = '\033[0;33m' if tty else ''
        color_end = '\033[0m' if tty else ''
        if not self.complete.is_set():
            if not self.started:
                return '{}:  {}WAITING{}'.format(self.label, blue_start, color_end)
            else:
                return '{}:  {}RUNNING{}'.format(self.label, yellow_start, color_end)
        if self.cancelled:
            if hasattr(self, '_start_time'):
                return '{}:  {}CANCELED{}  [{}]'.format(self.label, yellow_start, color_end,
                                                        self._finish_time - self._start_time)
            return '{}:  {}CANCELED{}'.format(self.label, yellow_start, color_end)
        if self.skipped:
            return '{}:  {}SKIPPED{}'.format(self.label, blue_start, color_end)
        if self.returncode == 0:
            color_start = '\033[0;32m' if tty else ''
            return '{}:  {}SUCCESS!{}  [{}]'.format(self.label, color_start, color_end,
                                                    self._finish_time - self._start_time)
        else:
            color_start = '\033[0;31m' if tty else ''
            color_end = '\033[0m' if tty else ''
            return '{}:  {}FAILED{}    [{}] (exited with code: {})'.format(
                self.label, color_start, color_end, self._finish_time - self._start_time,
                self.returncode)

    def _convert_command_for_container(self, archive=False, archive_path=''):
        """
        Use this to convert self._command to run in a container.

        This method is a convenience method that allows Jobs to define their self._command
        attribute in a simple fashion, without having to redefine all the machinery to run the
        command in a container. This method replaces self._command with a command that will run it
        in a container.

        Args:
            archive_path (str): A path on the host to share as a volume into the container for
                its /results path.
            archive (bool): Whether to mount a shared volume from the host into the container for
                archival purposes.
        """
        args = [container_runtime, 'run', '--network', 'none', '--rm',
                '--label', CONTAINER_LABEL]

        # podman does not support --init: https://github.com/containers/libpod/issues/1670
        if init and container_runtime == 'docker':
            args.append('--init')

        if tty:
            args.append('-t')

        if archive:
            self.archive_dir = '{}/{}'.format(archive_path,
                                              '{}-{}'.format(self.release, self._label))
            args.extend(['-v', '{}:/results:z'.format(self.archive_dir)])

        args.append(self._container_image)
        args.extend(self._command)
        self._command = args

    def _pre_start_hook(self):
        """Print a message announcing that we are running now."""
        click.echo('Running {}'.format(' '.join(self._command)))


class BuildJob(Job):
    """
    Define a Job for building container images.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'build'
    # A template for finding the name of the Dockerfile to be used for this BuildJob. The release
    # gets substituted into the {}'s.
    _dockerfile_template = 'Dockerfile-{}'
    # Extra arguments to be passed to the container build command.
    _build_args = ['--force-rm', '--pull']

    def __init__(self, *args, **kwargs):
        """
        Initialize the BuildJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(BuildJob, self).__init__(*args, **kwargs)

        dockerfile = os.path.join(PROJECT_PATH, 'devel', 'ci',
                                  self._dockerfile_template.format(self.release))
        self._command = [container_runtime, 'build', '-t', self._container_image,
                         '-f', dockerfile, '.']
        if self._build_args:
            for arg in reversed(self._build_args):
                self._command.insert(2, arg)

    async def run(self):
        """
        Run the BuildJob, unless --no-build has been requested and the needed build already exists.

        Returns:
            BuildJob: Returns self.
        """
        if no_build and self._build_exists:
            self.complete.set()
            self.skipped = True
        else:
            await super(BuildJob, self).run()
        return self

    @property
    def _build_exists(self):
        """
        Determine whether a container image exists for this build job.

        Returns:
            bool: True if a build exists, False otherwise.
        """
        args = [container_runtime, 'images', self._container_image]
        images = subprocess.check_output(args).decode()
        if self._container_image in images:
            return True
        return False


class CleanJob(Job):
    """
    Define a Job for removing all container images built by bodhi-ci.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'clean'

    def __init__(self, *args, **kwargs):
        """
        Initialize the CleanJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(CleanJob, self).__init__(*args, **kwargs)

        self._command = [container_runtime, 'rmi', self._container_image]


class DiffCoverJob(Job):
    """
    Define a Job for running diff-cover on the test results.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'diff-cover'

    def __init__(self, archive, archive_path, pyver, *args, **kwargs):
        """
        Initialize the DiffCoverJob.

        See the superclass's docblock for details about additional accepted parameters.

        Args:
            archive (bool): If True, set up the volume mount so we can retrieve the test results
                from the container.
            archive_path (str): A path on the host to share as a volume into the container for
                its /results path.
            pyver (int): Which major version of Python we are testing with.
        """
        super(DiffCoverJob, self).__init__(*args, **kwargs)

        self.archive_path = archive_path

        self._command = ['/usr/bin/diff-cover', '/results/coverage.xml',
                         '--compare-branch=origin/develop', '--fail-under=100']
        self._label = '{}-py{}'.format(self._label, pyver)
        self._convert_command_for_container(archive=archive, archive_path=archive_path)

    def _pre_start_hook(self):
        """
        Copy the coverage.xml from the unit test job to the diff_cover container, then announce.
        """
        if not os.path.exists(self.archive_dir):
            os.makedirs(self.archive_dir)
        shutil.copy(os.path.join(self.depends_on[0].archive_dir, 'coverage.xml'),
                    os.path.join(self.archive_dir, 'coverage.xml'))

        super(DiffCoverJob, self)._pre_start_hook()


class DocsJob(Job):
    """
    Define a Job for building docs.

    See the Job superclass's docblock for details about its attributes.
    """

    _command = [
        '/usr/bin/bash', '-c',
        ('/usr/bin/python3 setup.py develop && make -C docs clean && make -C docs html && make '
         '-C docs man')]
    _label = 'docs'

    def __init__(self, *args, **kwargs):
        """
        Initialize the DocsJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(DocsJob, self).__init__(*args, **kwargs)

        self._convert_command_for_container()


class Flake8Job(Job):
    """
    Define a Job for running flake8.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'flake8'

    def __init__(self, *args, **kwargs):
        """
        Initialize the Flake8Job.

        See the superclass's docblock for details about accepted parameters.
        """
        super(Flake8Job, self).__init__(*args, **kwargs)

        if self.release == 'pip':
            self._command = ['/usr/local/bin/flake8']
        else:
            self._command = ['/usr/bin/flake8']
        self._convert_command_for_container()


class MyPyBuildJob(BuildJob):
    """
    Define a Job for building mypy container images.

    See the Job superclass's docblock for details about its attributes.
    """

    # We don't want the --pull build arg because we are based on a container built by bodhi-ci.
    _build_args = []  # type: typing.List[str]
    _container_image_template = '{}/{}/mypy'
    _dockerfile_template = 'Dockerfile-mypy-{}'
    _label = 'mypy-build'


class MyPyJob(Job):
    """
    Define a Job for running mypy.

    See the Job superclass's docblock for details about its attributes.
    """

    _container_image_template = '{}/{}/mypy'
    _label = 'mypy'

    def __init__(self, *args, **kwargs):
        """
        Initialize the MyPyJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(MyPyJob, self).__init__(*args, **kwargs)

        self._command = ['/usr/local/bin/mypy', '--follow-imports', 'silent',
                         'bodhi/server/bugs.py', 'devel/ci/bodhi-ci']
        self._convert_command_for_container()


class PydocstyleJob(Job):
    """
    Define a Job for running pydocstyle.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'pydocstyle'

    def __init__(self, *args, **kwargs):
        """
        Initialize the PydocstyleJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(PydocstyleJob, self).__init__(*args, **kwargs)

        if self.release == 'pip':
            self._command = ['/usr/local/bin/pydocstyle', 'bodhi']
        else:
            self._command = ['/usr/bin/pydocstyle', 'bodhi']
        self._convert_command_for_container()


class StopJob(Job):
    """
    Define a Job for stopping all containers started by this process.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'stop'

    def __init__(self, *args, **kwargs):
        """
        Initialize the StopJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(StopJob, self).__init__(*args, **kwargs)

        self._command = [container_runtime, 'stop', self.release]
        self._popen_kwargs['stdout'] = subprocess.DEVNULL

    def _pre_start_hook(self):
        """Do not announce this Job; it is noisy."""
        pass


class UnitJob(Job):
    """
    Define a Job for running the unit tests.

    See the Job superclass's docblock for details about its attributes.

    Attributes:
        pyver (int): The Python version we are testing.
    """

    _label = 'unit'

    def __init__(self, archive, archive_path, pyver, *args, **kwargs):
        """
        Initialize the UnitJob.

        See the superclass's docblock for details about additional accepted parameters.

        Args:
            archive (bool): If True, set up the volume mount so we can retrieve the test results
                from the container.
            archive_path (str): A path on the host to share as a volume into the container for
                its /results path.
            pyver (int): Which major version of Python we are testing with.
        """
        super(UnitJob, self).__init__(*args, **kwargs)

        self.pyver = pyver
        self._label = '{}-py{}'.format(self._label, pyver)
        pytest_flags = '--junit-xml=nosetests.xml bodhi/tests'
        if failfast:
            pytest_flags += ' -x'

        test_command = ('({} setup.py develop && {} {} || (cp *.xml /results && exit 1)) '
                        '&& cp *.xml /results')

        if pyver == 2:
            # We do not support the server on Python 2:
            test_command = (
                'rm requirements.txt && touch requirements.txt && rm .coveragerc && '
                'rm -rf bodhi/tests/server && {}'.format(test_command))

        if self.release == 'pip':
            # pip installs some of the executables in different places than Fedora does.
            if pyver == 2:
                self._command = [
                    '/usr/bin/bash', '-c',
                    test_command.format('/usr/bin/python2', '/usr/bin/py.test', pytest_flags)]
            else:
                self._command = [
                    '/usr/bin/bash', '-c',
                    test_command.format('/usr/bin/python3', '/usr/local/bin/py.test', pytest_flags)]
        else:
            if pyver == 2:
                self._command = [
                    '/usr/bin/bash', '-c',
                    test_command.format('/usr/bin/python2', '/usr/bin/py.test-2', pytest_flags)]
            else:
                self._command = [
                    '/usr/bin/bash', '-c',
                    test_command.format('/usr/bin/python3', '/usr/bin/py.test-3', pytest_flags)]

        self._convert_command_for_container(archive=archive, archive_path=archive_path)

    async def run(self):
        """
        Run the UnitJob, unless --no-build has been requested and the needed coverage data exists.

        Returns:
            UnitJob: Returns self.
        """
        if no_build and os.path.exists(os.path.join(self.archive_dir, 'coverage.xml')):
            self.complete.set()
            self.skipped = True
        else:
            await super(UnitJob, self).run()
        return self


class IntegrationBuildJob(BuildJob):
    """
    Define a Job for building container images for integration testing.

    See the Job superclass's docblock for details about its attributes.
    """

    def __init__(self, app_name, *args, **kwargs):
        """
        Initialize the IntegrationBuildJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(IntegrationBuildJob, self).__init__(*args, **kwargs)

        self._app_name = app_name
        self._label = 'integration-build-{}'.format(app_name)
        self._container_image = '{}-integration-{}'.format(CONTAINER_NAME, app_name)
        dockerfile = os.path.join(
            PROJECT_PATH, 'devel', 'ci', 'integration', app_name, 'Dockerfile'
        )
        self._command = [container_runtime, 'build', '--force-rm', '--pull',
                         '-t', self._container_image, '-f', dockerfile, '.']

    def __repr__(self):
        return "<{} app={!r}>".format(self.__class__.__name__, self._app_name)


class IntegrationBodhiBuildJob(IntegrationBuildJob):
    """
    Build Bodhi in a container image for integration testing.

    See the Job superclass's docblock for details about its attributes.
    """

    def __init__(self, *args, **kwargs):
        """
        Initialize the IntegrationBodhiBuildJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(IntegrationBodhiBuildJob, self).__init__(*args, app_name="bodhi", **kwargs)

        self._container_image = '{}-integration-bodhi/{}'.format(CONTAINER_NAME, self.release)
        dockerfile = os.path.join(
            PROJECT_PATH, 'devel', 'ci', 'integration', 'bodhi',
            'Dockerfile-{}'.format(self.release),
        )
        self._command = [container_runtime, 'build', '--force-rm', '-t', self._container_image,
                         '-f', dockerfile, '.']


class IntegrationDumpDownloadJob(BuildJob):
    """
    Define a Job for downloading database dumps for integration testing.

    See the Job superclass's docblock for details about its attributes.
    """

    def __init__(self, app_name, *args, **kwargs):
        """
        Initialize the IntegrationDumpDownloadJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(IntegrationDumpDownloadJob, self).__init__(*args, **kwargs)

        self._app_name = app_name
        self._label = 'integration-dump-download-{}'.format(app_name)
        self.filepath = os.path.join(
            "devel", "ci", "integration", "dumps", "{}.dump".format(self._app_name))
        url = "https://infrastructure.fedoraproject.org/infra/db-dumps/{app}.dump.xz".format(app=self._app_name)
        self._popen_kwargs['shell'] = True
        self._command = [
            (f"curl -o {self.filepath}.xz {url} && xz -d --keep --force {self.filepath}.xz")]

    def __repr__(self):
        return "<{} app={!r}>".format(self.__class__.__name__, self._app_name)

    async def run(self) -> 'IntegrationDumpDownloadJob':
        """
        Run the download, unless we already have the file and it's recent enough.

        Returns:
            Returns self.
        """
        if os.path.exists(self.filepath):
            # st_mtime is going to use the filesystem's timestamp, not necessarily UTC. Thus, we
            # will use tz=None on fromttimestamp() so that the time is expressed in the system's
            # local time. Therefore, we also need to collect the current time in the system's local
            # time for comparison.
            modified_time = datetime.datetime.fromtimestamp(os.stat(self.filepath).st_mtime)
            if datetime.datetime.now() - modified_time < datetime.timedelta(days=1):
                # Our download is within a day and infrastructure only produces downloads once a
                # day, so let's skip this task.
                self.complete.set()
                self.skipped = True
        else:
            await super(IntegrationDumpDownloadJob, self).run()
        return self


class IntegrationJob(Job):
    """
    Define a Job for running the integration tests.

    See the Job superclass's docblock for details about its attributes.

    Attributes:
        pyver (int): The Python version we are testing.
    """

    _label = 'integration'

    def __init__(self, archive, archive_path, pyver, *args, **kwargs):
        """
        Initialize the IntegrationJob.

        See the superclass's docblock for details about additional accepted parameters.

        Args:
            archive (bool): If True, set up the volume mount so we can retrieve the test results
                from the container.
            archive_path (str): A path on the host to share as a volume into the container for
                its /results path.
            pyver (int): Which major version of Python we are testing with.
        """
        super(IntegrationJob, self).__init__(*args, **kwargs)

        self.pyver = pyver
        self._command = [sys.executable, '-m', 'pytest', '--no-cov', 'devel/ci/integration/tests/']
        bodhi_container_image = '{}-integration-bodhi/{}'.format(CONTAINER_NAME, self.release)
        self._popen_kwargs["env"] = os.environ.copy()
        self._popen_kwargs["env"]["BODHI_INTEGRATION_IMAGE"] = bodhi_container_image
        if failfast:
            self._command.append('-x')
        if archive:
            self._command.append(
                '--junit-xml={}/{}-{}/nosetests.xml'.format(
                    archive_path, self.release, self._label
                )
            )


class IntegrationCleanJob(Job):
    """
    Define a Job for removing all container images built by bodhi-ci.

    See the Job superclass's docblock for details about its attributes.
    """

    _label = 'integration-clean'

    def __init__(self, app_name, *args, **kwargs):
        """
        Initialize the IntegrationCleanJob.

        See the superclass's docblock for details about accepted parameters.
        """
        super(IntegrationCleanJob, self).__init__(*args, **kwargs)

        self._container_image = '{}-integration-{}'.format(CONTAINER_NAME, app_name)
        if app_name == "bodhi":
            self._container_image = "{}/{}".format(self._container_image, self.release)
        self._command = [container_runtime, 'rmi', self._container_image]


def _build_jobs_list(
        command: str, concurrency: int, releases: typing.Sequence[str],
        pyvers: typing.Sequence[int] = None, archive: bool = False,
        archive_path: str = '') -> typing.List[Job]:
    """
    Build and return a list of jobs to be run for the given command.

    Args:
        command: The name of the calling command.
        concurrency: The number of Jobs we will run in parallel.
        releases: The releases we are building Jobs for.
        pyvers: The Python versions we are testing.
        archive: Whether to set up an archive volume to mount in the container.
        archive_path: Which path on the host so share into the container when archive is True.
    Returns:
        A list of Jobs to be run.
    """
    if pyvers is None:
        pyvers = tuple()
    if command == 'all':
        buffer_output = concurrency != 1
    elif command == 'mypy':
        # mypy always only has one container and no releases.
        buffer_output = False
    elif command in ('diff_cover', 'unit'):
        buffer_output = concurrency != 1 and (len(releases) != 1 or len(pyvers) != 1)
    else:
        buffer_output = concurrency != 1 and len(releases) != 1
    jobs = []  # type: typing.List[Job]
    integration_deps_build_jobs = []  # type: typing.List[Job]
    if command in ('all', 'integration-build', 'integration', 'integration-download') \
            and 3 in pyvers:
        if command != 'integration-download':
            integration_deps_build_jobs.extend([
                IntegrationBuildJob(app_name=app_name, release="prod", buffer_output=buffer_output)
                for app_name in ("resultsdb", "waiverdb", "greenwave")
            ])
        # Download the DB dumps
        integration_deps_build_jobs.extend([
            IntegrationDumpDownloadJob(app_name="waiverdb", release="prod", buffer_output=buffer_output),
            IntegrationDumpDownloadJob(app_name="bodhi2", release="prod", buffer_output=buffer_output),
        ])
        jobs.extend(integration_deps_build_jobs)
    for release in releases:
        integration_build_jobs = []  # type: typing.List[Job]
        if command in ('all', 'build', 'docs', 'flake8', 'mypy', 'pydocstyle', 'diff_cover',
                       'unit', 'integration-build', 'integration'):
            # We don't want to build non-pip releases if the command is mypy; mypy only uses pip.
            if command != 'mypy' or release == 'pip':
                build_job = BuildJob(release, buffer_output=buffer_output)
                jobs.append(build_job)
        if command in ('all', 'docs'):
            docs_job = DocsJob(release, depends_on=build_job, buffer_output=buffer_output)
            jobs.append(docs_job)
        if command in ('all', 'flake8'):
            flake8_job = Flake8Job(release, depends_on=build_job, buffer_output=buffer_output)
            jobs.append(flake8_job)
        if command in ('all', 'mypy') and release == 'pip':
            mypy_build_job = MyPyBuildJob(release, depends_on=build_job, buffer_output=buffer_output)
            mypy_job = MyPyJob(release, depends_on=mypy_build_job, buffer_output=buffer_output)
            jobs.extend([mypy_build_job, mypy_job])
        if command in ('all', 'pydocstyle'):
            pydocstyle_job = PydocstyleJob(release, depends_on=build_job, buffer_output=buffer_output)
            jobs.append(pydocstyle_job)
        if command in ('all', 'diff_cover', 'unit'):
            for pyver in pyvers:
                unit_job = UnitJob(
                    archive=archive, archive_path=archive_path, pyver=pyver, release=release,
                    depends_on=build_job, buffer_output=buffer_output)
                jobs.append(unit_job)
                if command in ('all', 'diff_cover'):
                    if pyver == 2:
                        # We don't support Python 2 on the server, so don't enforce coverage there.
                        continue
                    diff_cover_job = DiffCoverJob(
                        archive=archive, archive_path=archive_path, pyver=pyver,
                        release=release, depends_on=unit_job, buffer_output=buffer_output)
                    jobs.append(diff_cover_job)
        if command in ('all', 'integration-build', 'integration') and 3 in pyvers:
            integration_build_jobs.extend([
                IntegrationBodhiBuildJob(
                    release=release, depends_on=build_job, buffer_output=buffer_output),
            ])
            jobs.extend(integration_build_jobs)
        if command in ('all', 'integration') and 3 in pyvers:
            integration_job = IntegrationJob(
                archive=archive, archive_path=archive_path, pyver=3, release=release,
                depends_on=integration_deps_build_jobs + integration_build_jobs,
                buffer_output=buffer_output)
            jobs.append(integration_job)

    return jobs


def _cancel_jobs(jobs: typing.List[Job]):
    """
    Mark the given jobs as cancelled.

    This is used as the SIGINT handler.

    Args:
        jobs: A list of Jobs which will have their cancelled attribute set to True.
    """
    for job in jobs:
        if job.returncode is None:
            job.cancelled = True


def _process_results(loop, done, pending):
    """
    Process the finished and pendings tasks and return error output and an exit code.

    This function is used by _run_processes() to generate the final stdout to be printed (which is
    going to be the output of the failed tasks since the cancelled tasks and successful tasks
    already had their output printed) and an exit code that bodhi-ci should use. Any pending tasks
    will be cancelled.

    Args:
        loop (asyncio.AbstractEventLoop): The event loop. This is used to cancel any pending tasks.
        done (set): A set of asyncio.Tasks that represent finished tasks.
        pending (set): A set of asyncio.Tasks that represent unfinshed tasks. These will be
            canceled.
    Returns:
        dict: A dictionary with three keys:
            'output': The output that should be printed.
            'returncode': The exit code that bodhi-ci should exit with.
    """
    returncode = 0
    error_output = ''

    if pending:
        for task in pending:
            task.cancel()
        future = asyncio.wait(pending)
        cancelled, pending = loop.run_until_complete(future)
        done = done | cancelled
        returncode = -signal.SIGINT

    for task in done:
        try:
            result = task.result()
        except RuntimeError as e:
            result = e.result
        if not result.cancelled and result.returncode:
            if result.output:
                error_output = '{}\n{}'.format(error_output, result.output)
            if not returncode:
                returncode = result.returncode

    return {'output': error_output, 'returncode': returncode}


def _run_jobs(jobs: typing.List[Job]):
    """
    Run the given jobs in parallel.

    Start a process for each Job. The stdout and stderr for each process is written to the
    terminal. Processes that exited with code 0 or were cancelled are output first, followed by any
    processes that failed. If any jobs failed, one of the failed jobs' exit code will be used to
    exit this process.

    Args:
        jobs: A list of Jobs to run.
    """
    if not jobs:
        click.echo("No jobs!")
        sys.exit(3)

    loop = asyncio.get_event_loop()

    progress_reporter = ProgressReporter(jobs)
    progress_reporter.print_status()

    processes = [j.run() for j in jobs]

    return_when = asyncio.ALL_COMPLETED
    if failfast:
        return_when = asyncio.FIRST_EXCEPTION
    future = asyncio.wait(processes, return_when=return_when)
    loop.add_signal_handler(signal.SIGINT, functools.partial(_cancel_jobs, jobs))

    try:
        done, pending = loop.run_until_complete(future)

        results = _process_results(loop, done, pending)
    finally:
        _stop_all_jobs(loop)

    # Now it's time to print any error output we collected, then exit or return.
    click.echo(results['output'])
    if results['returncode']:
        sys.exit(results['returncode'])


def _stop_all_jobs(loop: asyncio.AbstractEventLoop):
    """
    Stop all running docker jobs with the CONTAINER_LABEL.

    Even though we terminate() all of our child processes above, Docker does not always proxy
    signals through to the container, so we will do a final cleanup to make sure all the jobs we
    started in this process have been told to stop.

    Args:
        loop: The event loop.
    """
    args = [container_runtime, 'ps', '--filter=label={}'.format(CONTAINER_LABEL), '-q']
    processes = subprocess.check_output(args).decode()
    stop_jobs = [StopJob(process).run()
                 for process in processes.split('\n') if process]

    # If you give run_until_complete a future with no tasks, you will haz a sad (that's the
    # technical wording for a ValueError).
    if stop_jobs:
        stop_future = asyncio.wait(stop_jobs)
        loop.run_until_complete(stop_future)


cli()