#!/usr/bin/env python3
# ex:ts=4:sw=4:sts=4:et
# -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*-
#
# patchtest: execute all unittest test cases discovered for a single patch
#
# Copyright (C) 2016 Intel Corporation
#
# SPDX-License-Identifier: GPL-2.0-only
#

import json
import logging
import os
import signal
import subprocess
import sys
import traceback
import unittest

# Include current path so test cases can see it
sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))

# Include patchtest library
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), '../meta/lib/patchtest'))

from patchtest_parser import PatchtestParser
from repo import PatchTestRepo

logger = logging.getLogger("patchtest")
loggerhandler = logging.StreamHandler()
loggerhandler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(loggerhandler)
logger.setLevel(logging.INFO)

def _format_test_description(test):
    return (test.id().split('.')[-1]
            .replace('_', ' ')
            .replace("cve", "CVE")
            .replace("signed off by", "Signed-off-by")
            .replace("upstream status", "Upstream-Status")
            .replace("non auh", "non-AUH")
            .replace("presence format", "presence"))


def _write_patchtest_result(line, logfile=None):
    print(line)
    if logfile:
        with open(logfile, "a") as f:
            f.write(line + "\n")

def getResult(patch, mergepatch, logfile=None):

    class PatchTestResult(unittest.TextTestResult):
        """ Patchtest TextTestResult """
        shouldStop  = True
        longMessage = False

        success = 'PASS'
        fail    = 'FAIL'
        skip    = 'SKIP'

        def startTestRun(self):
            self.repo_error   = False
            self.test_error   = False
            self.test_failure = False

            try:
                self.repo = PatchtestParser.repo = PatchTestRepo(
                    patch=patch,
                    repodir=PatchtestParser.repodir,
                    commit=PatchtestParser.basecommit,
                    branch=PatchtestParser.basebranch,
                )
            except:
                logger.error(traceback.print_exc())
                self.repo_error = True
                self.stop()
                return

            if mergepatch:
                self.repo.merge()

        def addError(self, test, err):
            self.test_error = True
            (ty, va, trace) = err
            traceback.print_exc()

        def addFailure(self, test, err):
            self.test_failure = True
            desc = _format_test_description(test)
            issue = json.loads(str(err[1]))["issue"]
            _write_patchtest_result('{}: {}: {} ({})'.format(self.fail, desc, issue, test.id()), logfile)

        def addSuccess(self, test):
            desc = _format_test_description(test)
            _write_patchtest_result('{}: {} ({})'.format(self.success, desc, test.id()), logfile)

        def addSkip(self, test, reason):
            desc = _format_test_description(test)
            issue = json.loads(str(reason))["issue"]
            _write_patchtest_result('{}: {}: {} ({})'.format(self.skip, desc, issue, test.id()), logfile)

        def stopTestRun(self):

            # in case there was an error on repo object creation, just return
            if self.repo_error:
                return

            self.repo.clean()

    return PatchTestResult

def _runner(resultklass, prefix=None):
    # load test with the corresponding prefix
    loader = unittest.TestLoader()
    if prefix:
        loader.testMethodPrefix = prefix

    # create the suite with discovered tests and the corresponding runner
    suite = loader.discover(
        start_dir=PatchtestParser.testdir,
        pattern=PatchtestParser.pattern,
        top_level_dir=PatchtestParser.topdir,
    )

    # if there are no test cases, just quit
    if not suite.countTestCases():
        return 2

    runner = unittest.TextTestRunner(resultclass=resultklass, verbosity=0)

    try:
        result = runner.run(suite)
    except Exception:
        traceback.print_exc()
        logger.error('patchtest: something went wrong')

    return 1 if (result.test_failure or result.test_error) else 0

def run(patch, logfile=None):
    """ Load, setup and run pre and post-merge tests """
    premerge_result = _runner(getResult(patch, False, logfile), 'pretest')
    postmerge_result = _runner(getResult(patch, True, logfile), 'test')

    print_result_message(premerge_result, postmerge_result)
    return premerge_result or postmerge_result

def print_result_message(preresult, postresult):
    print("----------------------------------------------------------------------\n")
    if preresult == 2 and postresult == 2:
        logger.error(
            "patchtest: No test cases found - did you specify the correct suite directory?"
        )
    if preresult == 1 or postresult == 1:
        logger.error(
            "WARNING: patchtest: At least one patchtest caused a failure or an error - please check https://wiki.yoctoproject.org/wiki/Patchtest for further guidance"
        )
    else:
        logger.info("OK: patchtest: All patchtests passed")
    print("----------------------------------------------------------------------\n")

def main():
    patch_path = PatchtestParser.patch_path

    git_status = subprocess.run(
        ['git', '-C', PatchtestParser.repodir, 'status'],
        capture_output=True, text=True,
    ).stdout
    status_matches = ["Changes not staged for commit", "Changes to be committed"]
    if any(match in git_status for match in status_matches):
        logger.error("patchtest: there are uncommitted changes in the target repo that would be overwritten. Please commit or restore them before running patchtest")
        return 1

    builddir = os.environ.get('BUILDDIR')
    if builddir:
        bblayers_conf = os.path.join(builddir, 'conf', 'bblayers.conf')
        if os.path.exists(bblayers_conf):
            with open(bblayers_conf) as f:
                if 'meta-selftest' not in f.read():
                    logger.error(
                        "patchtest: meta-selftest layer not found in %s - add it to BBLAYERS before running patchtest" % bblayers_conf
                    )
                    return 1

    if os.path.isdir(patch_path):
        patch_list = [os.path.join(patch_path, filename) for filename in sorted(os.listdir(patch_path))]
    else:
        patch_list = [patch_path]

    interrupted = False
    previous_sigint = signal.getsignal(signal.SIGINT)

    def _sigint_handler(signum, frame):
        nonlocal interrupted
        interrupted = True
        # Restore previous handler so a second CTRL+C exits immediately
        signal.signal(signal.SIGINT, previous_sigint)
        signal.default_int_handler(signum, frame)

    signal.signal(signal.SIGINT, _sigint_handler)

    ret = 0
    for patch in patch_list:
        if interrupted:
            break

        if os.path.getsize(patch) == 0:
            logger.error('patchtest: patch is empty')
            signal.signal(signal.SIGINT, previous_sigint)
            return 1

        logger.info('Testing patch %s' % patch)

        log_path = None
        if PatchtestParser.log_results:
            log_path = patch + ".testresult"
            with open(log_path, "a") as f:
                f.write("Patchtest results for patch '%s':\n\n" % patch)

        try:
            result = run(patch, log_path)
            ret = ret or result
        except KeyboardInterrupt:
            interrupted = True

        if interrupted:
            logger.error('\npatchtest: interrupted')
            signal.signal(signal.SIGINT, previous_sigint)
            return 1

    signal.signal(signal.SIGINT, previous_sigint)
    return ret

if __name__ == '__main__':
    ret = 1

    # Parse the command line arguments and store it on the PatchtestParser namespace
    PatchtestParser.set_namespace()

    # set debugging level
    if PatchtestParser.debug:
        logger.setLevel(logging.DEBUG)

    # if topdir not define, default it to testdir
    if not PatchtestParser.topdir:
        PatchtestParser.topdir = PatchtestParser.testdir

    try:
        ret = main()
    except Exception:
        traceback.print_exc(5)

    sys.exit(ret)