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 socket
import subprocess
import sys
import time

from lib.constants import BASE_DIR
from task import github

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")
    parser.add_argument('--revision', help="Revision of the PR head", required=True)
    opts = parser.parse_args()

    # keep track of the most recent pull.updated_at attribute
    last_updated_at = "0"
    api = github.GitHub(repo=opts.repo)

    def detect_collisions(opts):
        nonlocal last_updated_at

        api = github.GitHub(repo=opts.repo)

        try:
            pull = api.get(f"pulls/{opts.pull_number}")
        except (TimeoutError, socket.gaierror, github.GitHubError) as e:
            sys.stderr.write('Warning: collision detection failed, skipping: %s\n' % e)
            # GitHub API is sometimes flaky, don't break test just because it sometimes does not respond
            return None

        if pull:
            if pull["head"]["sha"] != opts.revision:
                # Ignore GitHub API servers with outdated information
                if pull["updated_at"] < last_updated_at:
                    sys.stderr.write(
                        f"tests-invoke: Ignoring pull_request response for updated_at {pull['updated_at']}"
                        f"; it is older than the previous {last_updated_at}\n")
                    return None

                return f"Newer revision {pull['head']['sha']} available on GitHub for this pull request"

            last_updated_at = pull["updated_at"]

            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]

        os.environ["TEST_REVISION"] = opts.revision
        os.environ["TEST_PULL"] = str(opts.pull_number)

        collision_detected = False
        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(f'Collision detected, cancelling this test run: {ret}\n')
                collision_detected = True
                ret = None
                p.terminate()
                p.wait(timeout=60)
            else:
                time.sleep(int(os.getenv("TEST_INVOKE_SLEEP", "60")))
        return_code = p.returncode
        sys.stderr.write(f"Test run finished, return code: {return_code}\n")
        if not api.has_open_prs(opts.revision) and not collision_detected and return_code != 0:
            logs_url = os.getenv("TEST_ATTACHMENTS_URL")
            test_scenario = os.getenv("TEST_SCENARIO", "")
            test_os = os.getenv("TEST_OS")
            title = f"Nightly tests did not succeed on {test_os}"
            body = f"Tests failed on {opts.revision}"

            if logs_url:
                body += f", [logs]({os.path.join(logs_url, 'log.html')})"
            if test_scenario != "":
                title += f"/{test_scenario}"

            if 'updates-testing' in test_scenario:
                assert test_os is not None
                if test_os.startswith("fedora-"):
                    rel = test_os.split("-")[-1]
                    base_url = "https://bodhi.fedoraproject.org"
                    url = f"{base_url}/updates/?search=&status=testing&status=stable&releases=F{rel}"
                    body += f"\n[bodhi updates]({url})"
            data = {
                "title": title,
                "body": body,
                "labels": ["nightly"]
            }
            api.post("issues", data)
        return return_code
    except RuntimeError as ex:
        ret = str(ex)

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


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