publish-wrapper

#!/usr/bin/env python3

# This file is part of Cockpit.
#
# Copyright (C) 2020 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

from task import github
from task import sink

sys.dont_write_bytecode = True

HOSTNAME = socket.gethostname().split(".")[0]


def main():
    parser = argparse.ArgumentParser(description='Log to sink and publish into github')
    parser.add_argument('--publish', help='Publish results centrally to a sink',
                        default=os.environ.get("TEST_PUBLISH"))
    parser.add_argument('--test-name', help='Test name', required=True)
    parser.add_argument('--repo', help="The repository in which the tested PR is opened", default=None)
    parser.add_argument('--revision', help="Revision of the PR head", required=True)
    parser.add_argument('--github-context', help="The full context as written in github",
                        required=True)
    parser.add_argument('command', help="Command to wrap")
    opts = parser.parse_args()

    if not opts.publish:
        parser.error("Publishing address is required")

    try:
        task = PullTask(opts.repo)
        ret = task.run(opts)
    except RuntimeError as ex:
        ret = str(ex)

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


class PullTask():
    def __init__(self, repo):
        self.sink = None
        self.github_status_data = None
        self.api = github.GitHub(repo=repo)

    def start_publishing(self, opts):
        # build a unique file name for this test run
        id_context = opts.github_context.replace("@", "-").replace("#", "-")
        identifier = "-".join([
            opts.test_name.replace("/", "-"),
            opts.revision[0:8],
            id_context.replace("/", "-")
        ])

        description = "{0} [{1}]".format(github.TESTING, HOSTNAME)

        self.github_status_data = {
            "state": "pending",
            "context": opts.github_context,
            "description": description,
            "target_url": ":link"
        }

        status = {
            "github": {
                "token": self.api.token,
                "requests": [
                    # Set status to pending
                    {
                        "method": "POST",
                        "resource": self.api.qualify("commits/" + opts.revision + "/statuses"),
                        "data": self.github_status_data
                    }
                ],
                "watches": [{
                    "resource": self.api.qualify("commits/" + opts.revision + "/status?per_page=100"),
                    "result": {
                        "statuses": [
                            {
                                "context": opts.github_context,
                                "description": description,
                                "target_url": ":link"
                            }
                        ]
                    }
                }]
            },
            "revision": opts.revision,
            "onaborted": {
                "github": {
                    "token": self.api.token,
                    "requests": [
                        # Set status to error
                        {
                            "method": "POST",
                            "resource": self.api.qualify("statuses/" + opts.revision),
                            "data": {
                                "state": "error",
                                "context": opts.github_context,
                                "description": "Aborted without status",
                                "target_url": ":link"
                            }
                        }
                    ]
                },
            }
        }

        status["link"] = "log.html"
        status["extras"] = ["https://raw.githubusercontent.com/cockpit-project/bots/master/task/log.html"]

        self.sink = sink.Sink(opts.publish, identifier, status)

    def stop_publishing(self, ret):
        def mark_failed():
            if "github" in self.sink.status:
                self.github_status_data["state"] = "failure"

        def mark_passed():
            if "github" in self.sink.status:
                self.github_status_data["state"] = "success"

        if isinstance(ret, str):
            message = ret
            mark_failed()
            ret = 0
        elif ret == 0:
            message = "Tests passed"
            mark_passed()
        else:
            message = "Tests failed with code {0}".format(ret)
            mark_failed()
            ret = 0  # A failure, but not for this script
        self.sink.status["message"] = message
        if "github" in self.sink.status:
            self.github_status_data["description"] = message
        try:
            del self.sink.status["extras"]
        except KeyError:
            pass
        self.sink.flush()

        return ret

    def detect_collisions(self, opts):
        statuses = self.api.get("commits/{0}/statuses".format(opts.revision))
        for status in statuses:
            if status.get("context") == opts.github_context:
                if status.get("state") != "pending":
                    return "Status is not pending"
                if status.get("description") not in [github.NOT_TESTED, github.NOT_TESTED_DIRECT]:
                    return "Status description is not in [{0}, {1}]".format(
                        github.NOT_TESTED, github.NOT_TESTED_DIRECT)
                return None  # only check the newest status of the supplied context
        return None

    def run(self, opts):
        ret = self.detect_collisions(opts)
        if ret:
            sys.stderr.write("Collision detected: {0}".format(ret))
            sys.stderr.flush()
            return 0

        self.start_publishing(opts)
        os.environ["TEST_ATTACHMENTS"] = self.sink.attachments

        msg = "Running on: `{0}`\nCommand: `{1}`\n".format(HOSTNAME, opts.command)
        sys.stderr.write(msg)
        sys.stderr.flush()

        ret = subprocess.call(opts.command, shell=True)
        ret = self.stop_publishing(ret)

        return ret


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