#!/usr/bin/env python3
# coding=utf-8

#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import os
import shutil
import time
from dataclasses import dataclass

from xdevice import DeviceTestType
from xdevice import IDriver
from xdevice import Plugin
from xdevice import platform_logger
from xdevice import DeviceLabelType
from xdevice import ShellHandler
from xdevice import ExecuteTerminate
from xdevice import get_plugin
from xdevice import JsonParser
from xdevice import get_config_value
from xdevice import get_kit_instances
from xdevice import check_result_report
from xdevice import get_device_log_file
from xdevice import get_test_component_version
from xdevice import ParamError
from ohos.constants import ParserType
from ohos.constants import ComType
from ohos.constants import CKit
from ohos.exception import LiteDeviceExecuteCommandError
from core.utils import get_filename_extension
from core.testkit.kit_lite import DeployKit
from core.config.resource_manager import ResourceManager
from core.config.config_manager import UserConfigManager

__all__ = ["LiteUnitTest", "CTestDriver", "JSUnitTestLiteDriver"]
LOG = platform_logger("LiteUnitTest")


def get_level_para_string(level_string):
    level_list = list(set(level_string.split(",")))
    level_para_string = ""
    for item in level_list:
        if not item.isdigit():
            continue
        item = item.strip(" ")
        level_para_string = f"{level_para_string}Level{item},"
    level_para_string = level_para_string.strip(",")
    return level_para_string


@dataclass
class GTestConst(object):
    exec_para_filter = "--gtest_filter"
    exec_para_level = "--gtest_testsize"


@Plugin(type=Plugin.DRIVER, id=DeviceTestType.lite_cpp_test)
class LiteUnitTest(IDriver):
    """
    lite gtest test driver for L1
    """
    config = None
    log = platform_logger("LiteUnitTest")
    nfs_dir = ""
    mnt_cmd = ""
    lite_device = None
    result = None

    def __check_failed__(self, msg):
        self.log.error("check failed {}".format(msg))
        return

    def __check_environment__(self, device_options):
        pass

    def __check_config__(self, config):
        """
        1. check serial protocol
        2. login device
        3. NFS is available
        :param config: serial device
        :return:
        """
        self.log.error("Lite driver check config:{}".format(config))

    def __execute__(self, request):
        """

        1. select test case by subsystem, module, suite
        2. open test dir
        3、execute single test case, eg. ./test_demo
        :param request: contains test condition, sub_system
            module_name，test_suit，
        test_case，test_level，test_case_dir
        :return:
        """
        self.log.debug("Test suite FilePath: %s" %
                      request.root.source.source_file)
        self.lite_device = request.config.environment.devices[0]
        self.lite_device.connect()
        if not self._before_execute_test():
            self.log.error("open test dir failed")
            return
        self.log.debug("open test dir success")
        if self._execute_test(request) == "":
            self.log.error("execute test command failed")
            return
        self.log.info("execute test command success")
        if not self._after_execute_test(request):
            self.log.error("after execute test failed")
            return
        self.log.info("lite device execute request success")

    def _mount_nfs_server(self):
        #before execute each suits bin, mount nfs
        self.mnt_cmd = "mount {}".format(UserConfigManager().get_user_config(
            "NFS").get("mnt_cmd"))
        if self.mnt_cmd == "mount ":
            self.log.error("no configure for mount command")
            return

        filter_result, status, _ = \
            self.lite_device.execute_command_with_timeout(
            self.mnt_cmd, case_type=DeviceTestType.lite_cpp_test, timeout=3)
        if "already mounted" in filter_result:
            self.log.info("nfs has been mounted")
            return

        for i in range(0, 2):
            if status:
                self.log.info("execute mount command success")
                return
            self.log.info("try mount %d" % (i + 2))
            _, status, _ = self.lite_device.execute_command_with_timeout(
                self.mnt_cmd, case_type=DeviceTestType.lite_cpp_test,
                timeout=3)

        self.log.error("execute mount command failed")

    def _before_execute_test(self):
        """
        need copy test case to nfs dir
        :param request: nfs dir, test case path
        :return:
        """
        self.nfs_dir = \
            UserConfigManager().get_user_config("NFS").get("host_dir")
        if self.nfs_dir == "":
            self.log.error("no configure for nfs directory")
            return False
        self._mount_nfs_server()
        _, status, _ = \
            self.lite_device.execute_command_with_timeout("cd /{}".format(
                UserConfigManager().get_user_config("NFS").get("board_dir")),
            case_type=DeviceTestType.lite_cpp_test)
        if not status:
            self.log.error("pre execute command failed")
            return False
        self.log.info("pre execute command success")
        return True

    def _execute_test(self, request):
        test_case = request.root.source.source_file
        self.config = request.config
        test_para = self._get_test_para(self.config.testcase,
                                       self.config.testlevel)
        case_name = os.path.basename(test_case)
        if os.path.exists(os.path.join(self.nfs_dir, case_name)):
            os.remove(os.path.join(self.nfs_dir, case_name))
        result_name = case_name + ".xml"
        result_file = os.path.join(self.nfs_dir, result_name)
        if os.path.exists(result_file):
            os.remove(result_file)
        shutil.copyfile(test_case, os.path.join(self.nfs_dir, case_name))
        # push resource files
        resource_manager = ResourceManager()
        resource_data_dic, resource_dir = \
            resource_manager.get_resource_data_dic(test_case)
        resource_manager.lite_process_preparer_data(resource_data_dic, resource_dir)
        self.lite_device.execute_command_with_timeout(
            "chmod 777 {}".format(case_name),
            case_type=DeviceTestType.lite_cpp_test)
        test_command = "./%s %s" % (case_name, test_para)
        case_result, status, _ = \
            self.lite_device.execute_command_with_timeout(
            test_command, case_type=DeviceTestType.lite_cpp_test)
        if status:
            self.log.info("test case result:\n %s" % case_result)
            return
        self.log.error("failed case: %s" % test_case)

    def _get_test_para(self, testcase, testlevel):
        if "" != testcase and "" == testlevel:
            test_para = "%s=%s" % (GTestConst.exec_para_filter, testcase)
        elif "" == testcase and "" != testlevel:
            level_para = get_level_para_string(testlevel)
            test_para = "%s=%s" % (GTestConst.exec_para_level, level_para)
        else:
            test_para = ""
        return test_para

    def _after_execute_test(self, request):
        """
        copy test result to result dir
        :param request:
        :return:
        """
        if request.config is None:
            self.log.error("test config is null")
            return False
        report_path = request.config.report_path
        test_result = os.path.join(report_path, "result")
        test_case = request.root.source.source_file
        case_name = os.path.basename(test_case)
        if not os.path.exists(test_result):
            os.mkdir(test_result)
        sub_system_module = test_case.split(
            "unittest" + os.sep)[1].split(os.sep + "bin")[0]
        if os.sep in sub_system_module:
            sub_system = sub_system_module.split(os.sep)[0]
            module_name = sub_system_module.split(os.sep)[1]
            subsystem_dir = os.path.join(test_result, sub_system)
            if not os.path.exists(subsystem_dir):
                os.mkdir(subsystem_dir)
            module_dir = os.path.join(subsystem_dir, module_name)
            if not os.path.exists(module_dir):
                os.mkdir(module_dir)
            test_result = module_dir
        else:
            if sub_system_module != "":
                test_result = os.path.join(test_result, sub_system_module)
                if not os.path.exists(test_result):
                    os.mkdir(test_result)
        result_name = case_name + ".xml"
        result_file = os.path.join(self.nfs_dir, result_name)
        if not self._check_xml_exist(result_name):
            self.log.error("result xml file %s not exist." % result_name)
        if not os.path.exists(result_file):
            self.log.error("file %s not exist." % result_file)
            self._clear_nfs_space()
            return False
        file_name = os.path.basename(result_file)
        final_result = os.path.join(test_result, file_name)
        shutil.copyfile(result_file,
                        final_result)
        self.log.info("after execute test")
        self._clear_nfs_space()
        self.lite_device.close()
        return True

    def _clear_nfs_space(self):
        _, status, _ = \
            self.lite_device.execute_command_with_timeout(
                "cd ..",
                case_type=DeviceTestType.lite_cpp_test)
        _, status, _ = \
            self.lite_device.execute_command_with_timeout(
                "umount %s" % UserConfigManager().get_user_config("NFS").get("board_dir"),
                case_type=DeviceTestType.lite_cpp_test)
        shutil.rmtree(self.nfs_dir)
        os.mkdir(self.nfs_dir)

    def _check_xml_exist(self, xml_file, timeout=10):
        ls_command = \
            "ls /%s" % \
            UserConfigManager().get_user_config("NFS").get("board_dir")
        start_time = time.time()
        while time.time() - start_time < timeout:
            result, _, _ = self.lite_device.execute_command_with_timeout(
                command=ls_command, case_type=DeviceTestType.cpp_test_lite,
                timeout=5, receiver=None)
            if xml_file in result:
                return True
            time.sleep(5)
        return False

    def show_help_info(self):
        """
        show help info.
        """
        self.log.info("this is test driver for cpp test")
        return

    def show_driver_info(self):
        """
        show driver info.
        """
        self.log.info("this is test driver for cpp test")
        return

    def __result__(self):
        pass


@Plugin(type=Plugin.DRIVER, id=DeviceTestType.ctest_lite)
class CTestDriver(IDriver):
    """
    CTest is a test that runs a native test package on given lite device.
    """
    config = None
    result = ""
    error_message = ""
    version_cmd = "AT+CSV"

    def __init__(self):
        self.file_name = ""

    def __check_environment__(self, device_options):
        if len(device_options) != 1 or \
                device_options[0].label != DeviceLabelType.wifiiot:
            self.error_message = "check environment failed"
            return False
        return True

    def __check_config__(self, config=None):
        del config
        self.config = None

    def __execute__(self, request):
        from xdevice import Variables
        try:
            self.config = request.config
            self.config.device = request.config.environment.devices[0]

            if request.config.resource_path:
                current_dir = request.config.resource_path
            else:
                current_dir = Variables.exec_dir

            config_file = request.root.source.config_file.strip()
            if config_file:
                source = os.path.join(current_dir, config_file)
                self.file_name = os.path.basename(config_file).split(".")[0]
            else:
                source = request.root.source.source_string.strip()

            self._run_ctest(source=source, request=request)

        except (LiteDeviceExecuteCommandError, Exception) as exception:
            LOG.error(exception, error_no=getattr(exception, "error_no",
                                                  "00000"))
            self.error_message = exception
        finally:
            if request.root.source.test_name.startswith("{"):
                report_name = "report"
            else:
                report_name = get_filename_extension(
                    request.root.source.test_name)[0]

            self.result = check_result_report(request.config.report_path,
                                              self.result,
                                              self.error_message,
                                              report_name)

    def _run_ctest(self, source=None, request=None):
        if not source:
            LOG.error("Error: %s don't exist." % source, error_no="00101")
            return

        try:
            parsers = get_plugin(Plugin.PARSER, ParserType.ctest_lite)
            version = get_test_component_version(self.config)
            parser_instances = []
            for parser in parsers:
                parser_instance = parser.__class__()
                parser_instance.suites_name = self.file_name
                parser_instance.product_info.setdefault("Version", version)
                parser_instance.listeners = request.listeners
                parser_instances.append(parser_instance)
            handler = ShellHandler(parser_instances)

            reset_cmd = self._reset_device(request, source)
            self.result = "%s.xml" % os.path.join(request.config.report_path,
                                                  "result", self.file_name)

            self.config.device.device.com_dict.get(
                ComType.deploy_com).connect()

            result, _, error = self.config.device.device. \
                execute_command_with_timeout(
                    command=reset_cmd,
                    case_type=DeviceTestType.ctest_lite,
                    key=ComType.deploy_com,
                    timeout=90,
                    receiver=handler)

            device_log_file = get_device_log_file(request.config.report_path,
                                                  request.config.device.
                                                  __get_serial__())

            device_log_file_open = os.open(device_log_file, os.O_WRONLY |
                os.O_CREAT | os.O_APPEND, 0o755)
            with os.fdopen(device_log_file_open, "a") as file_name:
                file_name.write("{}{}".format(
                    "\n".join(result.split("\n")[0:-1]), "\n"))
                file_name.flush()
        finally:
            self.config.device.device.com_dict.get(ComType.deploy_com).close()

    def _reset_device(self, request, source):
        json_config = JsonParser(source)
        reset_cmd = []
        kit_instances = get_kit_instances(json_config,
                                          request.config.resource_path,
                                          request.config.testcases_path)
        from xdevice import Scheduler

        for (kit_instance, kit_info) in zip(kit_instances,
                                            json_config.get_kits()):
            if not isinstance(kit_instance, DeployKit):
                continue
            if not self.file_name:
                self.file_name = get_config_value(
                    'burn_file', kit_info)[0].split("\\")[-1].split(".")[0]
            reset_cmd = kit_instance.burn_command
            if not Scheduler.is_execute:
                raise ExecuteTerminate("ExecuteTerminate", error_no="00300")

            kit_instance.__setup__(self.config.device,
                source_file=request.root.source.source_file.strip())

        reset_cmd = [int(item, 16) for item in reset_cmd]
        return reset_cmd

    def __result__(self):
        return self.result if os.path.exists(self.result) else ""



@Plugin(type=Plugin.DRIVER, id=DeviceTestType.jsunit_test_lite)
class JSUnitTestLiteDriver(IDriver):
    """
    JSUnitTestDriver is a Test that runs a native test package on given device.
    """

    def __init__(self):
        self.result = ""
        self.error_message = ""
        self.kits = []
        self.config = None

    def __check_environment__(self, device_options):
        pass

    def __check_config__(self, config):
        pass

    def _get_driver_config(self, json_config):
        bundle_name = get_config_value('bundle-name',
                                       json_config.get_driver(), False)
        if not bundle_name:
            raise ParamError("Can't find bundle-name in config file.",
                             error_no="00108")
        else:
            self.config.bundle_name = bundle_name

        ability = get_config_value('ability',
                                   json_config.get_driver(), False)
        if not ability:
            self.config.ability = "default"
        else:
            self.config.ability = ability

    def __execute__(self, request):
        try:
            LOG.debug("Start execute xdevice extension JSUnit Test")

            self.config = request.config
            self.config.device = request.config.environment.devices[0]

            config_file = request.root.source.config_file
            suite_file = request.root.source.source_file

            if not suite_file:
                raise ParamError(
                    "test source '%s' not exists" %
                    request.root.source.source_string, error_no="00101")

            if not os.path.exists(config_file):
                LOG.error("Error: Test cases don't exist %s." % config_file,
                          error_no="00101")
                raise ParamError(
                    "Error: Test cases don't exist %s." % config_file,
                    error_no="00101")

            self.file_name = os.path.basename(
                request.root.source.source_file.strip()).split(".")[0]

            self.result = "%s.xml" % os.path.join(
                request.config.report_path, "result", self.file_name)

            json_config = JsonParser(config_file)
            self.kits = get_kit_instances(json_config,
                                          self.config.resource_path,
                                          self.config.testcases_path)

            self._get_driver_config(json_config)

            from xdevice import Scheduler
            for kit in self.kits:
                if not Scheduler.is_execute:
                    raise ExecuteTerminate("ExecuteTerminate",
                                           error_no="00300")
                if kit.__class__.__name__ == CKit.liteinstall:
                    kit.bundle_name = self.config.bundle_name
                kit.__setup__(self.config.device, request=request)

            self._run_jsunit(request)

        except Exception as exception:
            self.error_message = exception
        finally:
            report_name = "report" if request.root.source. \
                test_name.startswith("{") else get_filename_extension(
                request.root.source.test_name)[0]

            self.result = check_result_report(
                request.config.report_path, self.result, self.error_message,
                report_name)

            for kit in self.kits:
                kit.__teardown__(self.config.device)

            self.config.device.close()

    def _run_jsunit(self, request):
        parser_instances = []
        parsers = get_plugin(Plugin.PARSER, ParserType.jsuit_test_lite)
        for parser in parsers:
            parser_instance = parser.__class__()
            parser_instance.suites_name = self.file_name
            parser_instance.listeners = request.listeners
            parser_instances.append(parser_instance)
        handler = ShellHandler(parser_instances)

        command = "./bin/aa start -p %s -n %s" % \
                  (self.config.bundle_name, self.config.ability)
        result, _, error = self.config.device.execute_command_with_timeout(
            command=command,
            timeout=300,
            receiver=handler)

        device_log_file = get_device_log_file(request.config.report_path,
            request.config.device.
            __get_serial__())

        device_log_file_open = os.open(device_log_file, os.O_WRONLY |
            os.O_CREAT | os.O_APPEND, 0o755)
        with os.fdopen(device_log_file_open, "a") as file_name:
            file_name.write("{}{}".format(
                "\n".join(result.split("\n")[0:-1]), "\n"))
            file_name.flush()

    def __result__(self):
        return self.result if os.path.exists(self.result) else ""