kci_test

#!/usr/bin/env python3
#
# Copyright (C) 2019 Collabora Limited
# Author: Guillaume Tucker <guillaume.tucker@collabora.com>
#
# This module 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.
#
# This library 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 this library; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

import argparse
import glob
import json
import os
import requests
import sys

from kernelci.cli import Args, Command, parse_opts
import kernelci
import kernelci.config.lab
import kernelci.config.test
import kernelci.build
import kernelci.lab
import kernelci.test


# -----------------------------------------------------------------------------
# Commands
#

class cmd_validate(Command):
    help = "Validate the YAML configuration"
    opt_args = [Args.verbose]

    def __call__(self, configs, args):
        # ToDo: Use jsonschema

        entries = [
            'device_types',
            'file_system_types',
            'test_plans',
            'labs',
        ]
        err = kernelci.config.validate_yaml(opts.yaml_config, entries)
        if err:
            print(err)
            return False

        test_config_devices = list(
            str(config.device_type) for config in configs['test_configs']
        )
        for device_type in configs['device_types'].keys():
            if device_type not in test_config_devices:
                print("Device type has no test config: {}".format(device_type))
                return False

        for config in configs['test_configs']:
            err = kernelci.sort_check(config.test_plans.keys())
            if err:
                print("Plans broken order for {}: '{}' before '{}'".format(
                    config.device_type, err[0], err[1]))
                return False

        return True


class cmd_validate_rootfs_urls(Command):
    help = "Check that all the rootfs URLs are valid"
    opt_args = [Args.verbose]

    def __call__(self, config_data, args, **kwargs):
        file_systems_config = config_data['file_systems']
        rootfs_configs = config_data['rootfs_configs']
        for fs, fs_config in file_systems_config.items():
            fs_parts = fs.split('_')
            if fs_parts[0] == 'debian':
                fs_name = fs_parts[1]
                root_type = fs_parts[2]
            elif fs_parts[0].startswith('buildroot'):
                fs_name = fs_parts[0]
                root_type = fs_parts[1]
            else:
                continue
            rootfs_config = rootfs_configs.get(fs_name)
            url_formats = list(
                url_fmt for url_fmt in (
                    fs_config.get_url_format(fmt_type)
                    for fmt_type in set({root_type}).union({'nfs', 'ramdisk'})
                ) if url_fmt
            )
            if args.verbose:
                print(fs)
            for arch in rootfs_config.arch_list:
                for url_format in url_formats:
                    url = url_format.format(arch=arch)
                    resp = requests.head(url)
                    if args.verbose:
                        print(f"  {resp.status_code} {url}")
                    resp.raise_for_status()
        return True


class cmd_list_jobs(Command):
    help = "List all the jobs that need to be run for a given build and lab"
    args = [Args.lab_config]
    opt_args = [Args.user, Args.lab_token, Args.lab_json,
                Args.build_output, Args.install_path]

    def __call__(self, configs, args):
        path_args = (args.build_output, args.install_path)
        if not any(path_args) or all(path_args):
            print("Either --build-output or --install-path is required")
            return False

        install = (
            args.install_path if args.install_path else
            kernelci.build.Step.get_install_path(None, args.build_output)
        )
        meta = kernelci.build.Metadata(install)

        if meta.get('bmeta', 'build', 'status') != "PASS":
            return True

        lab_config = configs['labs'][args.lab_config]
        lab_api = kernelci.lab.get_api(
            lab_config, args.user, args.lab_token, args.lab_json)

        configs = kernelci.test.match_configs(
            configs['test_configs'], meta, lab_config)
        jobs = list()
        for device_type, plan in configs:
            if not lab_api.device_type_online(device_type):
                continue
            jobs.append((device_type.name, plan.name))

        for job in sorted(jobs):
            print(' '.join(job))

        return True


class cmd_list_plans(Command):
    help = "List all the existing test plan names"

    def __call__(self, configs, args):
        plans = set(plan.name
                    for plan in list(configs['test_plans'].values()))
        for plan in sorted(plans):
            print(plan)
        return True


class cmd_list_labs(Command):
    help = "List all the existing lab names"

    def __call__(self, configs, args):
        for lab in sorted(configs['labs'].keys()):
            print(lab)
        return True


class cmd_get_lab_info(Command):
    help = "Get the information about a lab into a JSON file"
    args = [Args.lab_config, Args.lab_json]
    opt_args = [Args.user, Args.lab_token]

    def __call__(self, configs, args):
        lab = configs['labs'][args.lab_config]
        lab_api = kernelci.lab.get_api(lab, args.user, args.lab_token)
        data = {
            'lab': lab.name,
            'lab_type': lab.lab_type,
            'url': lab.url,
            'devices': lab_api.devices,
        }
        dir = os.path.dirname(args.lab_json)
        if dir and not os.path.exists(dir):
            os.makedirs(dir)
        with open(args.lab_json, 'w') as json_file:
            json.dump(data, json_file, indent=4, sort_keys=True)
        return True


class cmd_generate(Command):
    help = "Generate the job definition for a given build"
    args = [Args.lab_config, Args.storage]
    opt_args = [Args.plan, Args.target, Args.output,
                Args.build_output, Args.install_path,
                Args.lab_json, Args.user, Args.lab_token, Args.db_config,
                Args.callback_id, Args.callback_dataset,
                Args.callback_type, Args.callback_url, Args.mach]

    def __call__(self, configs, args):
        if args.callback_id and not args.callback_url:
            print("--callback-url is required with --callback-id")
            return False

        path_args = (args.build_output, args.install_path)
        if not any(path_args) or all(path_args):
            print("Either --build-output or --install-path is required")
            return False

        install = (
            args.install_path
            or kernelci.build.Step.get_install_path(None, args.build_output)
        )
        meta = kernelci.build.Metadata(install)

        if meta.get('bmeta', 'build', 'status') != "PASS":
            return True

        lab_config = configs['labs'][args.lab_config]
        lab_api = kernelci.lab.get_api(
            lab_config, args.user, args.lab_token, args.lab_json)

        if args.target and args.plan:
            device_config = configs['device_types'][args.target]
            plan_config = configs['test_plans'][args.plan]
            jobs_list = [(device_config, plan_config)]
        else:
            jobs_list = []
            test_configs = kernelci.test.match_configs(
                configs['test_configs'], meta, lab_config)
            for device_config, plan_config in test_configs:
                if not lab_api.device_type_online(device_config):
                    continue
                if args.target and device_config.name != args.target:
                    continue
                if args.plan and plan_config.name != args.plan:
                    continue
                if args.mach and device_config.mach != args.mach:
                    continue
                jobs_list.append((device_config, plan_config))

        callback_opts = {
            'id': args.callback_id,
            'dataset': args.callback_dataset or 'all',
            'type': args.callback_type or 'kernelci',
            'url': args.callback_url,
        }
        db_config = (
            configs['db_configs'][args.db_config] if args.db_config else None
        )
        if args.output and not os.path.exists(args.output):
            os.makedirs(args.output)
        for device_config, plan_config in jobs_list:
            params = kernelci.test.get_params(
                meta, device_config, plan_config, args.storage)
            job = lab_api.generate(params, device_config, plan_config,
                                   callback_opts, db_config=db_config)
            if job is None:
                print("Failed to generate the job definition")
                return False
            if args.output:
                output_file = lab_api.save_file(job, args.output, params)
                print(output_file)
            else:
                print("# Job: {}".format(params['name']))
                print(job)
        return True


class cmd_submit(Command):
    help = "Submit job definitions to a lab"
    args = [Args.lab_config, Args.user, Args.lab_token, Args.jobs]

    def __call__(self, configs, args):
        lab = configs['labs'][args.lab_config]
        lab_api = kernelci.lab.get_api(lab, args.user, args.lab_token)
        job_paths = glob.glob(args.jobs)
        res = True
        for path in job_paths:
            if not os.path.isfile(path):
                continue
            try:
                job_id = lab_api.submit(path)
                print("{} {}".format(job_id, path))
            except Exception as e:
                print("ERROR {}: {}".format(path, e))
                res = False
        return res


# -----------------------------------------------------------------------------
# Main
#

if __name__ == '__main__':
    opts = parse_opts("kci_test", globals())
    configs = kernelci.config.load(opts.yaml_config)
    status = opts.command(configs, opts)
    sys.exit(0 if status is True else 1)