#!/usr/bin/env python3
# vim: set syntax=python ts=4 :
#
# Copyright (c) 2018 Intel Corporation
# SPDX-License-Identifier: Apache-2.0

import os
import json
import logging
from colorama import Fore
import xml.etree.ElementTree as ET
import string
from datetime import datetime

logger = logging.getLogger('twister')
logger.setLevel(logging.DEBUG)

class Reporting:

    def __init__(self, plan, env) -> None:
        self.plan = plan #FIXME
        self.instances = plan.instances
        self.platforms = plan.platforms
        self.selected_platforms = plan.selected_platforms
        self.filtered_platforms = plan.filtered_platforms
        self.env = env
        self.timestamp = datetime.now().isoformat()
        self.outdir = os.path.abspath(env.options.outdir)

    @staticmethod
    def process_log(log_file):
        filtered_string = ""
        if os.path.exists(log_file):
            with open(log_file, "rb") as f:
                log = f.read().decode("utf-8")
                filtered_string = ''.join(filter(lambda x: x in string.printable, log))

        return filtered_string


    @staticmethod
    def xunit_testcase(eleTestsuite, name, classname, status, ts_status, reason, duration, runnable, stats, log, build_only_as_skip):
        fails, passes, errors, skips = stats

        if status in ['skipped', 'filtered']:
            duration = 0

        eleTestcase = ET.SubElement(
            eleTestsuite, "testcase",
            classname=classname,
            name=f"{name}",
            time=f"{duration}")

        if status in ['skipped', 'filtered']:
            skips += 1
            # temporarily add build_only_as_skip to restore existing CI report behaviour
            if ts_status == "passed" and not runnable:
                tc_type = "build"
            else:
                tc_type = status
            ET.SubElement(eleTestcase, 'skipped', type=f"{tc_type}", message=f"{reason}")
        elif status in ["failed", "blocked"]:
            fails += 1
            el = ET.SubElement(eleTestcase, 'failure', type="failure", message=f"{reason}")
            if log:
                el.text = log
        elif status == "error":
            errors += 1
            el = ET.SubElement(eleTestcase, 'error', type="failure", message=f"{reason}")
            if log:
                el.text = log
        elif status == 'passed':
            if not runnable and build_only_as_skip:
                ET.SubElement(eleTestcase, 'skipped', type="build", message="built only")
                skips += 1
            else:
                passes += 1
        else:
            if not status:
                logger.debug(f"{name}: No status")
                ET.SubElement(eleTestcase, 'skipped', type=f"untested", message="No results captured, testsuite misconfiguration?")
            else:
                logger.error(f"{name}: Unknown status '{status}'")

        return (fails, passes, errors, skips)

    # Generate a report with all testsuites instead of doing this per platform
    def xunit_report_suites(self, json_file, filename):

        json_data = {}
        with open(json_file, "r") as json_results:
            json_data = json.load(json_results)


        env = json_data.get('environment', {})
        version = env.get('zephyr_version', None)

        eleTestsuites = ET.Element('testsuites')
        all_suites = json_data.get("testsuites", [])

        suites_to_report = all_suites
            # do not create entry if everything is filtered out
        if not self.env.options.detailed_skipped_report:
            suites_to_report = list(filter(lambda d: d.get('status') != "filtered", all_suites))

        for suite in suites_to_report:
            duration = 0
            eleTestsuite = ET.SubElement(eleTestsuites, 'testsuite',
                                            name=suite.get("name"), time="0",
                                            timestamp = self.timestamp,
                                            tests="0",
                                            failures="0",
                                            errors="0", skipped="0")
            eleTSPropetries = ET.SubElement(eleTestsuite, 'properties')
            # Multiple 'property' can be added to 'properties'
            # differing by name and value
            ET.SubElement(eleTSPropetries, 'property', name="version", value=version)
            ET.SubElement(eleTSPropetries, 'property', name="platform", value=suite.get("platform"))
            ET.SubElement(eleTSPropetries, 'property', name="architecture", value=suite.get("arch"))

            total = 0
            fails = passes = errors = skips = 0
            handler_time = suite.get('execution_time', 0)
            runnable = suite.get('runnable', 0)
            duration += float(handler_time)
            ts_status = suite.get('status')
            for tc in suite.get("testcases", []):
                status = tc.get('status')
                reason = tc.get('reason', suite.get('reason', 'Unknown'))
                log = tc.get("log", suite.get("log"))

                tc_duration = tc.get('execution_time', handler_time)
                name = tc.get("identifier")
                classname = ".".join(name.split(".")[:2])
                fails, passes, errors, skips = self.xunit_testcase(eleTestsuite,
                    name, classname, status, ts_status, reason, tc_duration, runnable,
                    (fails, passes, errors, skips), log, True)

            total = errors + passes + fails + skips

            eleTestsuite.attrib['time'] = f"{duration}"
            eleTestsuite.attrib['failures'] = f"{fails}"
            eleTestsuite.attrib['errors'] = f"{errors}"
            eleTestsuite.attrib['skipped'] = f"{skips}"
            eleTestsuite.attrib['tests'] = f"{total}"

        result = ET.tostring(eleTestsuites)
        with open(filename, 'wb') as report:
            report.write(result)

    def xunit_report(self, json_file, filename, selected_platform=None, full_report=False):
        if selected_platform:
            selected = [selected_platform]
            logger.info(f"Writing target report for {selected_platform}...")
        else:
            logger.info(f"Writing xunit report {filename}...")
            selected = self.selected_platforms

        json_data = {}
        with open(json_file, "r") as json_results:
            json_data = json.load(json_results)


        env = json_data.get('environment', {})
        version = env.get('zephyr_version', None)

        eleTestsuites = ET.Element('testsuites')
        all_suites = json_data.get("testsuites", [])

        for platform in selected:
            suites = list(filter(lambda d: d['platform'] == platform, all_suites))
            # do not create entry if everything is filtered out
            if not self.env.options.detailed_skipped_report:
                non_filtered = list(filter(lambda d: d.get('status') != "filtered", suites))
                if not non_filtered:
                    continue

            duration = 0
            eleTestsuite = ET.SubElement(eleTestsuites, 'testsuite',
                                            name=platform,
                                            timestamp = self.timestamp,
                                            time="0",
                                            tests="0",
                                            failures="0",
                                            errors="0", skipped="0")
            eleTSPropetries = ET.SubElement(eleTestsuite, 'properties')
            # Multiple 'property' can be added to 'properties'
            # differing by name and value
            ET.SubElement(eleTSPropetries, 'property', name="version", value=version)

            total = 0
            fails = passes = errors = skips = 0
            for ts in suites:
                handler_time = ts.get('execution_time', 0)
                runnable = ts.get('runnable', 0)
                duration += float(handler_time)

                ts_status = ts.get('status')
                # Do not report filtered testcases
                if ts_status == 'filtered' and not self.env.options.detailed_skipped_report:
                    continue
                if full_report:
                    for tc in ts.get("testcases", []):
                        status = tc.get('status')
                        reason = tc.get('reason', ts.get('reason', 'Unknown'))
                        log = tc.get("log", ts.get("log"))

                        tc_duration = tc.get('execution_time', handler_time)
                        name = tc.get("identifier")
                        classname = ".".join(name.split(".")[:2])
                        fails, passes, errors, skips = self.xunit_testcase(eleTestsuite,
                            name, classname, status, ts_status, reason, tc_duration, runnable,
                            (fails, passes, errors, skips), log, True)
                else:
                    reason = ts.get('reason', 'Unknown')
                    name = ts.get("name")
                    classname = f"{platform}:{name}"
                    log = ts.get("log")
                    fails, passes, errors, skips = self.xunit_testcase(eleTestsuite,
                        name, classname, ts_status, ts_status, reason, duration, runnable,
                        (fails, passes, errors, skips), log, False)

            total = errors + passes + fails + skips

            eleTestsuite.attrib['time'] = f"{duration}"
            eleTestsuite.attrib['failures'] = f"{fails}"
            eleTestsuite.attrib['errors'] = f"{errors}"
            eleTestsuite.attrib['skipped'] = f"{skips}"
            eleTestsuite.attrib['tests'] = f"{total}"

        result = ET.tostring(eleTestsuites)
        with open(filename, 'wb') as report:
            report.write(result)

    def json_report(self, filename, version="NA", platform=None):
        logger.info(f"Writing JSON report {filename}")
        report = {}
        report["environment"] = {"os": os.name,
                                 "zephyr_version": version,
                                 "toolchain": self.env.toolchain,
                                 "commit_date": self.env.commit_date,
                                 "run_date": self.env.run_date
                                 }
        suites = []

        for instance in self.instances.values():
            if platform and platform != instance.platform.name:
                continue
            suite = {}
            handler_log = os.path.join(instance.build_dir, "handler.log")
            pytest_log = os.path.join(instance.build_dir, "twister_harness.log")
            build_log = os.path.join(instance.build_dir, "build.log")
            device_log = os.path.join(instance.build_dir, "device.log")

            handler_time = instance.metrics.get('handler_time', 0)
            used_ram = instance.metrics.get ("used_ram", 0)
            used_rom  = instance.metrics.get("used_rom",0)
            available_ram = instance.metrics.get("available_ram", 0)
            available_rom = instance.metrics.get("available_rom", 0)
            suite = {
                "name": instance.testsuite.name,
                "arch": instance.platform.arch,
                "platform": instance.platform.name,
                "path": instance.testsuite.source_dir_rel
            }
            if instance.run_id:
                suite['run_id'] = instance.run_id

            suite["runnable"] = False
            if instance.status != 'filtered':
                suite["runnable"] = instance.run

            if used_ram:
                suite["used_ram"] = used_ram
            if used_rom:
                suite["used_rom"] = used_rom

            suite['retries'] = instance.retries

            if instance.dut:
                suite["dut"] = instance.dut
            if available_ram:
                suite["available_ram"] = available_ram
            if available_rom:
                suite["available_rom"] = available_rom
            if instance.status in ["error", "failed"]:
                suite['status'] = instance.status
                suite["reason"] = instance.reason
                # FIXME
                if os.path.exists(pytest_log):
                    suite["log"] = self.process_log(pytest_log)
                elif os.path.exists(handler_log):
                    suite["log"] = self.process_log(handler_log)
                elif os.path.exists(device_log):
                    suite["log"] = self.process_log(device_log)
                else:
                    suite["log"] = self.process_log(build_log)
            elif instance.status == 'filtered':
                suite["status"] = "filtered"
                suite["reason"] = instance.reason
            elif instance.status == 'passed':
                suite["status"] = "passed"
            elif instance.status == 'skipped':
                suite["status"] = "skipped"
                suite["reason"] = instance.reason

            if instance.status is not None:
                suite["execution_time"] =  f"{float(handler_time):.2f}"
            suite["build_time"] =  f"{float(instance.build_time):.2f}"

            testcases = []

            if len(instance.testcases) == 1:
                single_case_duration = f"{float(handler_time):.2f}"
            else:
                single_case_duration = 0

            for case in instance.testcases:
                # freeform was set when no sub testcases were parsed, however,
                # if we discover those at runtime, the fallback testcase wont be
                # needed anymore and can be removed from the output, it does
                # not have a status and would otherwise be reported as skipped.
                if case.freeform and case.status is None and len(instance.testcases) > 1:
                    continue
                testcase = {}
                testcase['identifier'] = case.name
                if instance.status:
                    if single_case_duration:
                        testcase['execution_time'] = single_case_duration
                    else:
                        testcase['execution_time'] = f"{float(case.duration):.2f}"

                if case.output != "":
                    testcase['log'] = case.output

                if case.status == "skipped":
                    if instance.status == "filtered":
                        testcase["status"] = "filtered"
                    else:
                        testcase["status"] = "skipped"
                        testcase["reason"] = case.reason or instance.reason
                else:
                    testcase["status"] = case.status
                    if case.reason:
                        testcase["reason"] = case.reason

                testcases.append(testcase)

            suite['testcases'] = testcases

            if instance.recording is not None:
                suite['recording'] = instance.recording

            suites.append(suite)

        report["testsuites"] = suites
        with open(filename, "wt") as json_file:
            json.dump(report, json_file, indent=4, separators=(',',':'))


    def compare_metrics(self, filename):
        # name, datatype, lower results better
        interesting_metrics = [("used_ram", int, True),
                               ("used_rom", int, True)]

        if not os.path.exists(filename):
            logger.error("Cannot compare metrics, %s not found" % filename)
            return []

        results = []
        saved_metrics = {}
        with open(filename) as fp:
            jt = json.load(fp)
            for ts in jt.get("testsuites", []):
                d = {}
                for m, _, _ in interesting_metrics:
                    d[m] = ts.get(m, 0)
                ts_name = ts.get('name')
                ts_platform = ts.get('platform')
                saved_metrics[(ts_name, ts_platform)] = d

        for instance in self.instances.values():
            mkey = (instance.testsuite.name, instance.platform.name)
            if mkey not in saved_metrics:
                continue
            sm = saved_metrics[mkey]
            for metric, mtype, lower_better in interesting_metrics:
                if metric not in instance.metrics:
                    continue
                if sm[metric] == "":
                    continue
                delta = instance.metrics.get(metric, 0) - mtype(sm[metric])
                if delta == 0:
                    continue
                results.append((instance, metric, instance.metrics.get(metric, 0), delta,
                                lower_better))
        return results

    def footprint_reports(self, report, show_footprint, all_deltas,
                          footprint_threshold, last_metrics):
        if not report:
            return

        logger.debug("running footprint_reports")
        deltas = self.compare_metrics(report)
        warnings = 0
        if deltas:
            for i, metric, value, delta, lower_better in deltas:
                if not all_deltas and ((delta < 0 and lower_better) or
                                       (delta > 0 and not lower_better)):
                    continue

                percentage = 0
                if value > delta:
                    percentage = (float(delta) / float(value - delta))

                if not all_deltas and (percentage < (footprint_threshold / 100.0)):
                    continue

                if show_footprint:
                    logger.log(
                        logging.INFO if all_deltas else logging.WARNING,
                        "{:<25} {:<60} {} {:<+4}, is now {:6} {:+.2%}".format(
                        i.platform.name, i.testsuite.name,
                        metric, delta, value, percentage))

                warnings += 1

        if warnings:
            logger.warning("Found {} footprint deltas to {} as a baseline.".format(
                           warnings,
                           (report if not last_metrics else "the last twister run.")))

    def synopsis(self):
        cnt = 0
        example_instance = None
        detailed_test_id = self.env.options.detailed_test_id
        for instance in self.instances.values():
            if instance.status not in ["passed", "filtered", "skipped"]:
                cnt = cnt + 1
                if cnt == 1:
                    logger.info("-+" * 40)
                    logger.info("The following issues were found (showing the top 10 items):")

                logger.info(f"{cnt}) {instance.testsuite.name} on {instance.platform.name} {instance.status} ({instance.reason})")
                example_instance = instance
            if cnt == 10:
                break

        if cnt and example_instance:
            logger.info("")
            logger.info("To rerun the tests, call twister using the following commandline:")
            extra_parameters = '' if detailed_test_id else ' --no-detailed-test-id'
            logger.info(f"west twister -p <PLATFORM> -s <TEST ID>{extra_parameters}, for example:")
            logger.info("")
            logger.info(f"west twister -p {example_instance.platform.name} -s {example_instance.testsuite.name}"
                        f"{extra_parameters}")
            logger.info(f"or with west:")
            logger.info(f"west build -p -b {example_instance.platform.name} "
                        f"{example_instance.testsuite.source_dir_rel} -T {example_instance.testsuite.id}")
            logger.info("-+" * 40)

    def summary(self, results, ignore_unrecognized_sections, duration):
        failed = 0
        run = 0
        for instance in self.instances.values():
            if instance.status == "failed":
                failed += 1
            elif not ignore_unrecognized_sections and instance.metrics.get("unrecognized"):
                logger.error("%sFAILED%s: %s has unrecognized binary sections: %s" %
                             (Fore.RED, Fore.RESET, instance.name,
                              str(instance.metrics.get("unrecognized", []))))
                failed += 1

            # FIXME: need a better way to identify executed tests
            handler_time = instance.metrics.get('handler_time', 0)
            if float(handler_time) > 0:
                run += 1

        if results.total and results.total != results.skipped_configs:
            pass_rate = (float(results.passed) / float(results.total - results.skipped_configs))
        else:
            pass_rate = 0

        logger.info(
            "{}{} of {}{} test configurations passed ({:.2%}), {}{}{} failed, {}{}{} errored, {} skipped with {}{}{} warnings in {:.2f} seconds".format(
                Fore.RED if failed else Fore.GREEN,
                results.passed,
                results.total,
                Fore.RESET,
                pass_rate,
                Fore.RED if results.failed else Fore.RESET,
                results.failed,
                Fore.RESET,
                Fore.RED if results.error else Fore.RESET,
                results.error,
                Fore.RESET,
                results.skipped_configs,
                Fore.YELLOW if self.plan.warnings else Fore.RESET,
                self.plan.warnings,
                Fore.RESET,
                duration))

        total_platforms = len(self.platforms)
        # if we are only building, do not report about tests being executed.
        if self.platforms and not self.env.options.build_only:
            logger.info("In total {} test cases were executed, {} skipped on {} out of total {} platforms ({:02.2f}%)".format(
                results.cases - results.skipped_cases,
                results.skipped_cases,
                len(self.filtered_platforms),
                total_platforms,
                (100 * len(self.filtered_platforms) / len(self.platforms))
            ))

        built_only = results.total - run - results.skipped_configs
        logger.info(f"{Fore.GREEN}{run}{Fore.RESET} test configurations executed on platforms, \
{Fore.RED}{built_only}{Fore.RESET} test configurations were only built.")

    def save_reports(self, name, suffix, report_dir, no_update, platform_reports):
        if not self.instances:
            return

        logger.info("Saving reports...")
        if name:
            report_name = name
        else:
            report_name = "twister"

        if report_dir:
            os.makedirs(report_dir, exist_ok=True)
            filename = os.path.join(report_dir, report_name)
            outdir = report_dir
        else:
            outdir = self.outdir
            filename = os.path.join(outdir, report_name)

        if suffix:
            filename = "{}_{}".format(filename, suffix)

        if not no_update:
            json_file = filename + ".json"
            self.json_report(json_file, version=self.env.version)
            self.xunit_report(json_file, filename + ".xml", full_report=False)
            self.xunit_report(json_file, filename + "_report.xml", full_report=True)
            self.xunit_report_suites(json_file, filename + "_suite_report.xml")

            if platform_reports:
                self.target_report(json_file, outdir, suffix)


    def target_report(self, json_file, outdir, suffix):
        platforms = {inst.platform for _, inst in self.instances.items()}
        for platform in platforms:
            if suffix:
                filename = os.path.join(outdir,"{}_{}.xml".format(platform.normalized_name, suffix))
                json_platform_file = os.path.join(outdir,"{}_{}.json".format(platform.normalized_name, suffix))
            else:
                filename = os.path.join(outdir,"{}.xml".format(platform.normalized_name))
                json_platform_file = os.path.join(outdir,"{}.json".format(platform.normalized_name))
            self.xunit_report(json_file, filename, platform.name, full_report=True)
            self.json_report(json_platform_file, version=self.env.version, platform=platform.name)