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 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
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')

parser = argparse.ArgumentParser()

patch_arg = parser.add_mutually_exclusive_group(required=True)
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', required=True, 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('--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_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)
    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', ' ')

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

    if desc:
        if len(desc) > get_console_width() - 41:
            print()
        print("", desc, end='')
    print('', flush=True)


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)

    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', ' ')

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

    failed = False
    print(BOLD + " " + patch_str, end='')
    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

    if failed or (desc and len(desc) > get_console_width() - 21):
        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_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)
    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 main():
    """ Main function """

    args = parser.parse_args()

    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")

    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()