ingest_mdir.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
#
# Copyright (C) 2019 Netronome Systems, Inc.

"""Command line interface for reading from a mdir

Script providing an ability to test local patch series.
On single patch series is expected as generated by git format-patch.
"""

import argparse
import configparser
import os
import re
import queue
import shutil
import sys
import tempfile
import time

from core import cmd
from core import log_open_sec, log_end_sec, log_init
from core import Patch
from core import Series
from core import Tree
from core import Tester

CONSOLE_WIDTH = None
NONINTERACTIVE = False
BOLD   = '\033[1m'
RED    = '\033[31m'
GREEN  = '\033[32m'
YELLOW = '\033[33m'
RESET  = '\033[0m'

config = configparser.ConfigParser()

config.add_section('dirs')
config.add_section('log')
config.add_section('tests')

parser = argparse.ArgumentParser()

patch_arg = parser.add_mutually_exclusive_group()
patch_arg.add_argument('--patch', help='path to the patch file')
patch_arg.add_argument('--mdir', help='path to the directory with the patches')

parser.add_argument('--tree', help='path to the tree to test on')
parser.add_argument('--tree-name', help='the tree name to expect')
parser.add_argument('--result-dir',
                    help='the directory where results will be generated')
parser.add_argument('--list-tests', action='store_true',
                    help='print all available tests and exit')
parser.add_argument('-d', '--disable-test', nargs='+',
                    help='disable test, can be specified multiple times')
parser.add_argument('-t', '--test', nargs='+',
                    help='run only specified tests. Note: full test name is needed, e.g. "patch/pylint" or "series/ynl" not just "pylint" or "ynl"')
parser.add_argument('--noninteractive', action='store_true',
                    help='Avoid printing terminal control characters to output which is not a terminal')
parser.add_argument('--dbg-print-run', help='print results of previous run')


def get_console_width():
    """ Get console width to avoid line wraps where we can. """

    global CONSOLE_WIDTH

    if CONSOLE_WIDTH is None:
        try:
            terminal_size = shutil.get_terminal_size()
            CONSOLE_WIDTH = terminal_size.columns
        except OSError:
            CONSOLE_WIDTH = 80
    return CONSOLE_WIDTH


def get_series_id(result_dir):
    """ Find an unused series ID. """

    i = 1
    while os.path.exists(os.path.join(result_dir, str(i))):
        i += 1
    return i


def __print_summary_result(offset, files, full_path):
    with open(os.path.join(full_path, "retcode"), "r", encoding="utf-8") as fp:
        retcode = int(fp.read())
    desc = None
    if "desc" in files:
        with open(os.path.join(full_path, "desc"), "r", encoding="utf-8") as fp:
            desc = fp.read().strip().replace('\n', ' ')

    failed = False

    if retcode == 0:
        print(GREEN  + "OKAY   " + RESET, end='')
    elif retcode == 250:
        print(YELLOW + "WARNING" + RESET, end='')
    else:
        print(RED    + "FAIL   " + RESET + f"({retcode})", end='')
        failed = True

    wrap_on_width = not NONINTERACTIVE and \
                    (desc and len(desc) + offset > get_console_width())
    if failed or wrap_on_width:
        print("\n", end=" ")
    if desc:
        print("", desc, end='')
        if failed:
            print("\n", end=" ")
    if failed:
        print(" Outputs:", full_path, end='')
    print('', flush=True)


def print_summary_singleton(print_state, files, full_path, patch_id):
    """
    Print summaries, single patch mode.
    Output differs if we have one patch vs many because tester will
    run the same test on all the patches in sequence.
    """

    if len(print_state['seen']) == 1:
        print()
        print(BOLD + "Series level tests:")

    if patch_id != print_state['last_patch']:
        print_state['last_patch'] = patch_id
        print(BOLD + "Patch level tests:")

    test_name = os.path.basename(full_path)

    print(BOLD + f" {test_name:32}", end='')
    __print_summary_result(41, files, full_path)


def print_summary_series(print_state, files, full_path, patch_id):
    """ Print summaries, series mode (more than one patch). """

    test_name = os.path.basename(full_path)
    if test_name != print_state.get('last_test'):
        print_state['last_test'] = test_name
        print()
        print(BOLD + test_name)

    if patch_id >= 0:
        patch_str = f"Patch {patch_id + 1:<6}"
    else:
        patch_str = "Full series "

    print(BOLD + " " + patch_str, end='')
    __print_summary_result(21, files, full_path)


def print_test_summary(args, series, print_state, tests=None):
    """
    Report results based on files created by the tester in the filesystem.
    Track which files we have already as this function should be called
    periodically to check for new results, as the tester runs.
    """

    seen = print_state.get('seen', set())
    print_state['seen'] = seen
    print_state['last_patch'] = print_state.get('last_patch', -1)

    for full_path, _, files in os.walk(os.path.join(args.result_dir,
                                                    str(series.id))):
        if full_path in seen:
            continue
        if "summary" not in files:
            continue
        seen.add(full_path)

        rel_path = full_path[len(args.result_dir) + 1:].split('/')
        test_name = os.path.basename(full_path)

        if tests and test_name not in tests:
            continue

        patch_id = -1
        if len(rel_path) == 3:
            patch_id = int(rel_path[-2]) - 1

        if len(series.patches) == 1:
            print_summary_singleton(print_state, files, full_path, patch_id)
        else:
            print_summary_series(print_state, files, full_path, patch_id)


def print_series_info(series):
    """ Print list of patches """

    if len(series.patches) > 2 and series.cover_letter is None:
        print(BOLD + "No cover letter" + RESET)
    elif series.cover_letter:
        print(BOLD + series.title + RESET)

    for p in series.patches:
        print("  " + f"[{p.id}] " + p.title)


def run_tester(args, tree, series):
    """ Run the tester, report results as they appear """

    summary_seen = {}

    try:
        done = queue.Queue()
        pending = queue.Queue()
        tester = Tester(args.result_dir, tree, pending, done,
                        config=config)
        tester.start()

        pending.put(series)
        pending.put(None)

        while done.empty():
            print_test_summary(args, series, summary_seen)
            time.sleep(0.2)
        # Finish, print the last test's result
        print_test_summary(args, series, summary_seen)
    except:
        print("Error / Interrupt detected, asking runner to stop")
        tester.should_die = True
        tester.join()
        raise
    finally:
        tester.join()


def load_patches(args):
    """ Load patches from specified location on disk """

    if args.dbg_print_run is None:
        series_id = get_series_id(args.result_dir)
    else:
        series_id = int(args.dbg_print_run)

    log_open_sec("Loading patches")
    try:
        if args.mdir:
            mdir = os.path.abspath(args.mdir)
            files = [os.path.join(mdir, f) for f in sorted(os.listdir(mdir))]
        else:
            files = [os.path.abspath(args.patch)]

        series = Series(ident=series_id)
        series.tree_selection_comment = "ingest_mdir"
        series.tree_mark_expected = False

        for f in files:
            with open(f, 'r', encoding="utf-8") as fp:
                data = fp.read()
                if re.search(r"\[.* 0+/\d.*\]", data) and \
                   not re.search(r"\n@@ -\d", data):
                    series.set_cover_letter(data)
                else:
                    series.add_patch(Patch(data))
    finally:
        log_end_sec()

    return series


def validate_test_list(test_list, all_test_names, parser_instance, error_description):
    """Check a list of test names against the set of all available tests."""
    if test_list:
        invalid_tests = [name for name in test_list if name not in all_test_names]

        if invalid_tests:
            invalid_str = ', '.join(invalid_tests)
            msg = f"the following {error_description} are invalid: {invalid_str}\n" \
                  f"Run with --list-tests to see available tests."
            parser_instance.error(msg)


def main():
    """ Main function """

    args = parser.parse_args()

    global NONINTERACTIVE
    if args.noninteractive:
        NONINTERACTIVE = True
        global BOLD, RED, GREEN, YELLOW, RESET
        BOLD = ''
        RED = ''
        GREEN = ''
        YELLOW = ''
        RESET = ''

    # Get all available test names for validation and --list-tests
    # We can instantiate a temporary Tester just for this purpose.
    tester_for_names = Tester(None, None, None, None, config=config)
    all_test_names = set(tester_for_names.get_test_names())

    # Handle --list-tests first (using the list we just fetched)
    if args.list_tests:
        print(' ', '\n  '.join(sorted(all_test_names)))
        return

    # Validate --test and --disable-test
    validate_test_list(args.test, all_test_names, parser, "tests")
    validate_test_list(args.disable_test, all_test_names, parser, "disabled tests")

    # Set configs after validation
    if args.test:
        config.set('tests', 'include', ','.join(args.test))

    if args.disable_test:
        config.set('tests', 'exclude', ','.join(args.disable_test))

    # If not listing tests, manually validate the other required arguments
    if not args.tree:
        parser.error("the following arguments are required: --tree")

    if not args.patch and not args.mdir:
        parser.error("one of the arguments --patch --mdir is required")

    args.tree = os.path.abspath(args.tree)

    if args.result_dir is None:
        args.result_dir = tempfile.mkdtemp()
    print("Saving output and logs to:", args.result_dir)

    config.set('log', 'type', 'org')
    config.set('log', 'dir', args.result_dir)
    config.set('log', 'path', "nipa.log")

    log_init(config.get('log', 'type'),
             os.path.join(args.result_dir, 'nipa.log'),
             force_single_thread=True)

    series = load_patches(args)

    tree_name = args.tree_name
    if tree_name is None:
        # Try to guess tree name from the patch subject, expecting subject
        # to be something like [PATCH tree-name 2/N].
        tags = re.search(
            r'Subject: \[(?:PATCH|RFC) (?:v\d+ )?([a-zA-Z-]+)(?: v\d+)?(?: \d*\/\d*)?\]',
            series.patches[0].raw_patch
        )
        if tags:
            tree_name = tags.group(1).strip()
            print("Tree name extracted from patches:", tree_name)
        else:
            tree_name = "unknown"
            print("Tree name unknown")

    # Default settings for networking trees:
    if tree_name.startswith('net'):
        if not args.disable_test:
            config.set('tests', 'exclude', 'patch/signed')

    print_series_info(series)

    if args.dbg_print_run:
        print_test_summary(args, series, {}, tests={'ynl', 'build_clang'})
        return

    try:
        tree = Tree(tree_name, tree_name, args.tree, current_branch=True)
    except cmd.CmdError:
        print("Can't assertain tree state, is a valid branch checked out?")
        raise
    head = tree.head_hash()
    tree.git_checkout(head)

    if not tree.check_applies(series):
        print("Patch series does not apply cleanly to the tree")
        os.sys.exit(1)

    tree.git_reset(head, hard=True)

    run_tester(args, tree, series)
    tree.git_checkout(tree.branch)
    tree.git_reset(head, hard=True)


if __name__ == "__main__":
    main()