tools/kci-maintainer

#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1-or-later
#
# Copyright (C) 2024 Collabora Limited
# Author: Denys Fedoryshchenko <denys.f@collabora.com>
'''
Tool to send checkout and jobretry requests to kernelci.org API server
Intended to be used by Linux kernel maintainers to trigger kernelci.org jobs
You need to have a valid Maestro API token to use this tool
'''

import requests
import json
import os
import argparse
import sys
import subprocess
import re


PIPELINE_ENDPOINTS = {
    'staging': 'https://staging.kernelci.org:9100/',
    'production': 'https://kernelci-pipeline.westus3.cloudapp.azure.com/',
    'local': 'http://localhost:8100/'
}

API_ENDPOINTS = {
    'staging': 'https://staging.kernelci.org:9000/',
    'production': 'https://kernelci-api.westus3.cloudapp.azure.com/',
    'local': 'http://localhost:9000/'
}

PIPELINE_URL = PIPELINE_ENDPOINTS['staging']
API_URL = API_ENDPOINTS['staging']


def get_token():
    # Two places to try get the token
    # environment variable KCI_TOKEN
    if 'KCI_TOKEN' in os.environ:
        return os.environ['KCI_TOKEN']
    # .token file
    if os.path.exists('.kci_token'):
        with open('.kci_token', 'r') as f:
            return f.read().strip()


def send_checkout_request1(token, nodeid, commit):
    '''
    1st way of sending checkout request
    nodeid + commit
    '''
    url = PIPELINE_URL + 'api/checkout'
    headers = {'Authorization': token, 'Content-Type': 'application/json'}
    data = {'nodeid': nodeid, 'commit': commit}
    response = requests.post(url, headers=headers, data=json.dumps(data))
    print(response.text)
    response.raise_for_status()
    return response.text


def send_checkout_request2(token, giturl, branch, commit, jobfilter):
    '''
    2nd way of sending checkout request
    repourl + branch + commit + jobfilter
    '''
    url = PIPELINE_URL + 'api/checkout'
    headers = {'Authorization': token, 'Content-Type': 'application/json'}
    data = {'url': giturl, 'branch': branch, 'commit': commit, 'jobfilter': jobfilter}
    response = requests.post(url, headers=headers, data=json.dumps(data))
    print(response.text)
    response.raise_for_status()
    return response.text


def send_jobretry_request(token, nodeid):
    url = PIPELINE_URL + 'api/jobretry'
    headers = {'Authorization': token, 'Content-Type': 'application/json'}
    data = {'nodeid': nodeid}
    response = requests.post(url, headers=headers, data=json.dumps(data))
    print(response.text)


def get_repo_info(repodir):
    if not os.path.exists(repodir):
        print('Repository directory does not exist')
        return

    os.chdir(repodir)
    repoinfo = {}
    try:
        repoinfo['url'] = os.popen('git config --get remote.origin.url').read().strip()
        repoinfo['commit'] = os.popen('git rev-parse HEAD').read().strip()
        repoinfo['branch'] = os.popen('git rev-parse --abbrev-ref HEAD').read().strip()
    except Exception as e:
        print('Failed to get repository info: %s' % e)
        return

    return repoinfo

def prepare_checkout(args):
    latest_sha = None
    token = get_token()
    if not token:
        token = args.token

    # Developer have several ways to provide checkout data
    # Always require commit ID
    # If node ID is not provided, we will try to get it:
    # 1. From repository developer pointed to
    #    (commit ID then can be also taken from there)
    # 2. From cli parameters

    if args.nodeid:
        if not args.commit:
            print('Commit ID is required')
            return
        return send_checkout_request1(token, args.nodeid, args.commit)

    if args.latest_commit:
        # retrieve latest commit from specific git tree and branch
        repo_url = args.repourl
        proc = subprocess.Popen(["git", "ls-remote", repo_url], stdout=subprocess.PIPE)
        stdout, stderr = proc.communicate()
        latest_sha = re.split(r'\t+', stdout.decode('ascii'))[0]
        print(f'Retrieved latest commit: {latest_sha}')

    if not args.commit and not latest_sha:
        print('Commit ID is required')
        return
    else:
        if not args.commit and latest_sha:
            args.commit = latest_sha

    if args.repodir:
        repoinfo = get_repo_info(args.repodir)

    if args.repourl:
        if not args.branch:
            print('Branch name is required')
            return
        repoinfo = {'url': args.repourl, 'branch': args.branch}

    if not repoinfo:
        print('Failed to get repository info')
        return

    if not args.jobfilter:
        # Just show warning, that lack of jobfilter will trigger all jobs
        print('Warning: jobfilter is not provided, all jobs will be triggered, this will'
              ' consume a lot of resources on kernelci.org.')
        if not args.nojobfilter:
            print('If you want to trigger all jobs, please provide --nojobfilter option')
            return

    if args.verbose:
        print(f'Repository info: {repoinfo}')
        print(f'Job filter: {args.jobfilter}')

    send_checkout_request2(token, repoinfo['url'], repoinfo['branch'], args.commit, args.jobfilter)


def send_patchset_request(token, args):
    if not args.patchurl:
        print('Patch URL is required')
        return

    if not args.nodeid:
        print('Node ID is required')
        return

    if not args.jobfilter and not args.nojobfilter:
        print('Job filter is required or --nojobfilter should be provided')
        return

    url = PIPELINE_URL + 'api/patchset'
    headers = {'Authorization': token, 'Content-Type': 'application/json'}
    data = {
        'nodeid': args.nodeid,
        'patchurl': args.patchurl
    }
    if args.jobfilter:
        data['jobfilter'] = args.jobfilter

    if args.verbose:
        print(json.dumps(data))
    response = requests.post(url, headers=headers, data=json.dumps(data))
    print(response.text)
    response.raise_for_status()


def main():
    global PIPELINE_URL, API_URL
    ap = argparse.ArgumentParser()
    token = get_token()
    if not token:
        ap.add_argument('-t', '--token', required=True, help='API token')

    ap.add_argument('--api', help='API server to use', choices=API_ENDPOINTS.keys())
    ap.add_argument('--checkout', action='store_true', help='Send checkout request')
    ap.add_argument('--jobretry', action='store_true', help='Retry job(test) request')
    ap.add_argument('--patchset', action='store_true', help='Send patchset request')
    ap.add_argument('-r', '--repodir', help='Local repository directory')
    ap.add_argument('-u', '--repourl', help='Repository URL')
    ap.add_argument('-b', '--branch', help='Branch name')
    ap.add_argument('-f', '--jobfilter', help='Job filter', nargs='+', action='extend', type=str)
    ap.add_argument('--nojobfilter', help='No job filter set', action='store_true')
    ap.add_argument('-n', '--nodeid', help='Node ID')
    ap.add_argument('-c', '--commit', help='Commit ID')
    ap.add_argument('--latest-commit', help='Retrieve latest commit from repository', action='store_true')
    ap.add_argument('-p', '--patchurl', help='Patch URL', nargs='+', action='extend', type=str)
    ap.add_argument('-v', '--verbose', action='store_true', help='Verbose output')

    args = ap.parse_args()

    if args.api:
        PIPELINE_URL = PIPELINE_ENDPOINTS[args.api]
        API_URL = API_ENDPOINTS[args.api]
        if not PIPELINE_URL or not API_URL:
            print('Invalid API server')
            return

    if not token and args.token:
        token = args.token

    if not token:
        print('API token is required')
        return

    if args.jobretry:
        if not args.nodeid:
            print('Node ID is required')
            return
        send_jobretry_request(token, args.nodeid)
        sys.exit(0)

    if args.checkout:
        prepare_checkout(args)
        sys.exit(0)

    if args.patchset:
        send_patchset_request(token, args)
        sys.exit(0)

    ap.print_help()


if __name__ == '__main__':
    main()