tests-invoke

#!/usr/bin/env python3

# This file is part of Cockpit.
#
# Copyright (C) 2016 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import argparse
import os
import subprocess
import sys
import time

from task import github
from machine.machine_core.directories import BASE_DIR

sys.dont_write_bytecode = True


def main():
    parser = argparse.ArgumentParser(description='Run integration tests')
    parser.add_argument('--repo', help="The repository in which the tested PR is opened", default=None)
    parser.add_argument('--pull-number', help="The number of the pull request to test",
                        required=True)
    parser.add_argument('--revision', help="Revision of the PR head", required=True)
    opts = parser.parse_args()

    def detect_collisions(opts):
        api = github.GitHub(repo=opts.repo)

        pull = api.get("pulls/{0}".format(opts.pull_number))
        if pull:
            if pull["head"]["sha"] != opts.revision:
                return "Newer revision available on GitHub for this pull request"
            if pull["state"] != "open":
                return "Pull request isn't open"

        return None

    try:
        entry_point = os.path.join(BASE_DIR, ".cockpit-ci", "run")
        alt_entry_point = os.path.join(BASE_DIR, "test", "run")
        if not os.path.exists(entry_point) and os.path.exists(alt_entry_point):
            entry_point = alt_entry_point
        cmd = ["timeout", "120m", entry_point]

        p = subprocess.Popen(cmd)
        while p.poll() is None:
            # Cancel test if PR gets updated while running
            ret = detect_collisions(opts)
            if ret:
                sys.stderr.write('Collision detected, cancelling this test run: {0}\n'.format(ret))
                ret = None
                p.terminate()
                p.wait(timeout=60)
            else:
                time.sleep(60)
        return_code = p.returncode
        sys.stderr.write("Test run finished, return code: {0}\n".format(return_code))
        return return_code
    except RuntimeError as ex:
        ret = str(ex)

    if ret:
        sys.stderr.write("tests-invoke: {0}\n".format(ret))
        return 1
    return 0


if __name__ == '__main__':
    sys.exit(main())