"""
This module handles compatibility issues between testcase format v2 and v3.
解决httprunner2 和 3 之间测试用例兼容性问题
"""
import os
import sys
from typing import List, Dict, Text, Union, Any

from loguru import logger

from httprunner import exceptions
from httprunner.loader import load_project_meta, convert_relative_project_root_dir
from httprunner.parser import parse_data
from httprunner.utils import sort_dict_by_custom_order


def convert_variables(
    raw_variables: Union[Dict, List, Text], test_path: Text
) -> Dict[Text, Any]:

    if isinstance(raw_variables, Dict):
        return raw_variables

    if isinstance(raw_variables, List):
        # [{"var1": 1}, {"var2": 2}]
        variables: Dict[Text, Any] = {}
        for var_item in raw_variables:
            if not isinstance(var_item, Dict) or len(var_item) != 1:
                raise exceptions.TestCaseFormatError(
                    f"Invalid variables format: {raw_variables}"
                )

            variables.update(var_item)

        return variables

    elif isinstance(raw_variables, Text):
        # get variables by function, e.g. ${get_variables()}
        project_meta = load_project_meta(test_path)
        variables = parse_data(raw_variables, {}, project_meta.functions)

        return variables

    else:
        raise exceptions.TestCaseFormatError(
            f"Invalid variables format: {raw_variables}"
        )


def _convert_jmespath(raw: Text) -> Text:
    if not isinstance(raw, Text):
        raise exceptions.TestCaseFormatError(f"Invalid jmespath extractor: {raw}")

    # content.xx/json.xx => body.xx
    if raw.startswith("content"):
        raw = f"body{raw[len('content'):]}"
    elif raw.startswith("json"):
        raw = f"body{raw[len('json'):]}"

    raw_list = []
    for item in raw.split("."):
        if "-" in item:
            # add quotes for field with separator
            # e.g. headers.Content-Type => headers."Content-Type"
            item = item.strip('"')
            raw_list.append(f'"{item}"')
        elif item.isdigit():
            # convert lst.0.name to lst[0].name
            if len(raw_list) == 0:
                logger.error(f"Invalid jmespath: {raw}")
                sys.exit(1)

            last_item = raw_list.pop()
            item = f"{last_item}[{item}]"
            raw_list.append(item)
        else:
            raw_list.append(item)

    return ".".join(raw_list)


def _convert_extractors(extractors: Union[List, Dict]) -> Dict:
    """ convert extract list(v2) to dict(v3)

    Args:
        extractors: [{"varA": "content.varA"}, {"varB": "json.varB"}]

    Returns:
        {"varA": "body.varA", "varB": "body.varB"}

    """
    v3_extractors: Dict = {}

    if isinstance(extractors, List):
        # [{"varA": "content.varA"}, {"varB": "json.varB"}]
        for extractor in extractors:
            if not isinstance(extractor, Dict):
                logger.error(f"Invalid extractor: {extractors}")
                sys.exit(1)
            for k, v in extractor.items():
                v3_extractors[k] = v
    elif isinstance(extractors, Dict):
        # {"varA": "body.varA", "varB": "body.varB"}
        v3_extractors = extractors
    else:
        logger.error(f"Invalid extractor: {extractors}")
        sys.exit(1)

    for k, v in v3_extractors.items():
        v3_extractors[k] = _convert_jmespath(v)

    return v3_extractors


def _convert_validators(validators: List) -> List:
    for v in validators:
        if "check" in v and "expect" in v:
            # format1: {"check": "content.abc", "assert": "eq", "expect": 201}
            v["check"] = _convert_jmespath(v["check"])

        elif len(v) == 1:
            # format2: {'eq': ['status_code', 201]}
            comparator = list(v.keys())[0]
            v[comparator][0] = _convert_jmespath(v[comparator][0])

    return validators


def _sort_request_by_custom_order(request: Dict) -> Dict:
    custom_order = [
        "method",
        "url",
        "params",
        "headers",
        "cookies",
        "data",
        "json",
        "files",
        "timeout",
        "allow_redirects",
        "proxies",
        "verify",
        "stream",
        "auth",
        "cert",
    ]
    return sort_dict_by_custom_order(request, custom_order)


def _sort_step_by_custom_order(step: Dict) -> Dict:
    custom_order = [
        "name",
        "variables",
        "request",
        "testcase",
        "setup_hooks",
        "teardown_hooks",
        "extract",
        "validate",
        "validate_script",
    ]
    return sort_dict_by_custom_order(step, custom_order)


def _ensure_step_attachment(step: Dict) -> Dict:
    test_dict = {
        "name": step["name"],
    }

    if "variables" in step:
        test_dict["variables"] = step["variables"]

    if "setup_hooks" in step:
        test_dict["setup_hooks"] = step["setup_hooks"]

    if "teardown_hooks" in step:
        test_dict["teardown_hooks"] = step["teardown_hooks"]

    if "extract" in step:
        test_dict["extract"] = _convert_extractors(step["extract"])

    if "export" in step:
        test_dict["export"] = step["export"]

    if "validate" in step:
        if not isinstance(step["validate"], List):
            raise exceptions.TestCaseFormatError(
                f'Invalid teststep validate: {step["validate"]}'
            )
        test_dict["validate"] = _convert_validators(step["validate"])

    if "validate_script" in step:
        test_dict["validate_script"] = step["validate_script"]

    return test_dict


def ensure_testcase_v3_api(api_content: Dict) -> Dict:
    logger.info("convert api in v2 to testcase format v3")

    teststep = {
        "request": _sort_request_by_custom_order(api_content["request"]),
    }
    teststep.update(_ensure_step_attachment(api_content))

    teststep = _sort_step_by_custom_order(teststep)

    config = {"name": api_content["name"]}
    extract_variable_names: List = list(teststep.get("extract", {}).keys())
    if extract_variable_names:
        config["export"] = extract_variable_names

    return {
        "config": config,
        "teststeps": [teststep],
    }


def ensure_testcase_v3(test_content: Dict) -> Dict:
    logger.info("ensure compatibility with testcase format v2")

    v3_content = {"config": test_content["config"], "teststeps": []}

    if "teststeps" not in test_content:
        logger.error(f"Miss teststeps: {test_content}")
        sys.exit(1)

    if not isinstance(test_content["teststeps"], list):
        logger.error(
            f'teststeps should be list type, got {type(test_content["teststeps"])}: {test_content["teststeps"]}'
        )
        sys.exit(1)

    for step in test_content["teststeps"]:
        teststep = {}

        if "request" in step:
            teststep["request"] = _sort_request_by_custom_order(step.pop("request"))
        elif "api" in step:
            teststep["testcase"] = step.pop("api")
        elif "testcase" in step:
            teststep["testcase"] = step.pop("testcase")
        else:
            raise exceptions.TestCaseFormatError(f"Invalid teststep: {step}")

        teststep.update(_ensure_step_attachment(step))

        teststep = _sort_step_by_custom_order(teststep)
        v3_content["teststeps"].append(teststep)

    return v3_content


def ensure_cli_args(args: List) -> List:
    """ ensure compatibility with deprecated cli args in v2
    """
    # remove deprecated --failfast
    if "--failfast" in args:
        logger.warning(f"remove deprecated argument: --failfast")
        args.pop(args.index("--failfast"))

    # convert --report-file to --html
    if "--report-file" in args:
        logger.warning(f"replace deprecated argument --report-file with --html")
        index = args.index("--report-file")
        args[index] = "--html"
        args.append("--self-contained-html")

    # keep compatibility with --save-tests in v2
    if "--save-tests" in args:
        logger.warning(
            f"generate conftest.py keep compatibility with --save-tests in v2"
        )
        args.pop(args.index("--save-tests"))
        _generate_conftest_for_summary(args)

    return args


def _generate_conftest_for_summary(args: List):

    for arg in args:
        if os.path.exists(arg):
            test_path = arg
            # FIXME: several test paths maybe specified
            break
    else:
        logger.error(f"No valid test path specified! \nargs: {args}")
        sys.exit(1)

    conftest_content = '''# NOTICE: Generated By HttpRunner.
import json
import os
import time

import pytest
from loguru import logger

from httprunner.utils import get_platform, ExtendJSONEncoder


@pytest.fixture(scope="session", autouse=True)
def session_fixture(request):
    """setup and teardown each task"""
    logger.info(f"start running testcases ...")

    start_at = time.time()

    yield

    logger.info(f"task finished, generate task summary for --save-tests")

    summary = {
        "success": True,
        "stat": {
            "testcases": {"total": 0, "success": 0, "fail": 0},
            "teststeps": {"total": 0, "failures": 0, "successes": 0},
        },
        "time": {"start_at": start_at, "duration": time.time() - start_at},
        "platform": get_platform(),
        "details": [],
    }

    for item in request.node.items:
        testcase_summary = item.instance.get_summary()
        summary["success"] &= testcase_summary.success

        summary["stat"]["testcases"]["total"] += 1
        summary["stat"]["teststeps"]["total"] += len(testcase_summary.step_datas)
        if testcase_summary.success:
            summary["stat"]["testcases"]["success"] += 1
            summary["stat"]["teststeps"]["successes"] += len(
                testcase_summary.step_datas
            )
        else:
            summary["stat"]["testcases"]["fail"] += 1
            summary["stat"]["teststeps"]["successes"] += (
                len(testcase_summary.step_datas) - 1
            )
            summary["stat"]["teststeps"]["failures"] += 1

        testcase_summary_json = testcase_summary.dict()
        testcase_summary_json["records"] = testcase_summary_json.pop("step_datas")
        summary["details"].append(testcase_summary_json)

    summary_path = r"{{SUMMARY_PATH_PLACEHOLDER}}"
    summary_dir = os.path.dirname(summary_path)
    os.makedirs(summary_dir, exist_ok=True)

    with open(summary_path, "w", encoding="utf-8") as f:
        json.dump(summary, f, indent=4, ensure_ascii=False, cls=ExtendJSONEncoder)

    logger.info(f"generated task summary: {summary_path}")

'''

    project_meta = load_project_meta(test_path)
    project_root_dir = project_meta.RootDir
    conftest_path = os.path.join(project_root_dir, "conftest.py")

    test_path = os.path.abspath(test_path)
    logs_dir_path = os.path.join(project_root_dir, "logs")
    test_path_relative_path = convert_relative_project_root_dir(test_path)

    if os.path.isdir(test_path):
        file_foder_path = os.path.join(logs_dir_path, test_path_relative_path)
        dump_file_name = "all.summary.json"
    else:
        file_relative_folder_path, test_file = os.path.split(test_path_relative_path)
        file_foder_path = os.path.join(logs_dir_path, file_relative_folder_path)
        test_file_name, _ = os.path.splitext(test_file)
        dump_file_name = f"{test_file_name}.summary.json"

    summary_path = os.path.join(file_foder_path, dump_file_name)
    conftest_content = conftest_content.replace(
        "{{SUMMARY_PATH_PLACEHOLDER}}", summary_path
    )

    dir_path = os.path.dirname(conftest_path)
    if not os.path.exists(dir_path):
        os.makedirs(dir_path)

    with open(conftest_path, "w", encoding="utf-8") as f:
        f.write(conftest_content)

    logger.info("generated conftest.py to generate summary.json")


def ensure_path_sep(path: Text) -> Text:
    """ ensure compatibility with different path separators of Linux and Windows
    """
    if "/" in path:
        path = os.sep.join(path.split("/"))

    if "\\" in path:
        path = os.sep.join(path.split("\\"))

    return path

[((147, 11, 147, 59), 'httprunner.utils.sort_dict_by_custom_order', 'sort_dict_by_custom_order', ({(147, 37, 147, 44): 'request', (147, 46, 147, 58): 'custom_order'}, {}), '(request, custom_order)', False, 'from httprunner.utils import sort_dict_by_custom_order\n'), ((162, 11, 162, 56), 'httprunner.utils.sort_dict_by_custom_order', 'sort_dict_by_custom_order', ({(162, 37, 162, 41): 'step', (162, 43, 162, 55): 'custom_order'}, {}), '(step, custom_order)', False, 'from httprunner.utils import sort_dict_by_custom_order\n'), ((199, 4, 199, 58), 'loguru.logger.info', 'logger.info', ({(199, 16, 199, 57): '"""convert api in v2 to testcase format v3"""'}, {}), "('convert api in v2 to testcase format v3')", False, 'from loguru import logger\n'), ((220, 4, 220, 63), 'loguru.logger.info', 'logger.info', ({(220, 16, 220, 62): '"""ensure compatibility with testcase format v2"""'}, {}), "('ensure compatibility with testcase format v2')", False, 'from loguru import logger\n'), ((357, 19, 357, 47), 'httprunner.loader.load_project_meta', 'load_project_meta', ({(357, 37, 357, 46): 'test_path'}, {}), '(test_path)', False, 'from httprunner.loader import load_project_meta, convert_relative_project_root_dir\n'), ((359, 20, 359, 65), 'os.path.join', 'os.path.join', ({(359, 33, 359, 49): 'project_root_dir', (359, 51, 359, 64): '"""conftest.py"""'}, {}), "(project_root_dir, 'conftest.py')", False, 'import os\n'), ((361, 16, 361, 42), 'os.path.abspath', 'os.path.abspath', ({(361, 32, 361, 41): 'test_path'}, {}), '(test_path)', False, 'import os\n'), ((362, 20, 362, 58), 'os.path.join', 'os.path.join', ({(362, 33, 362, 49): 'project_root_dir', (362, 51, 362, 57): '"""logs"""'}, {}), "(project_root_dir, 'logs')", False, 'import os\n'), ((363, 30, 363, 74), 'httprunner.loader.convert_relative_project_root_dir', 'convert_relative_project_root_dir', ({(363, 64, 363, 73): 'test_path'}, {}), '(test_path)', False, 'from httprunner.loader import load_project_meta, convert_relative_project_root_dir\n'), ((365, 7, 365, 31), 'os.path.isdir', 'os.path.isdir', ({(365, 21, 365, 30): 'test_path'}, {}), '(test_path)', False, 'import os\n'), ((374, 19, 374, 64), 'os.path.join', 'os.path.join', ({(374, 32, 374, 47): 'file_foder_path', (374, 49, 374, 63): 'dump_file_name'}, {}), '(file_foder_path, dump_file_name)', False, 'import os\n'), ((379, 15, 379, 45), 'os.path.dirname', 'os.path.dirname', ({(379, 31, 379, 44): 'conftest_path'}, {}), '(conftest_path)', False, 'import os\n'), ((386, 4, 386, 65), 'loguru.logger.info', 'logger.info', ({(386, 16, 386, 64): '"""generated conftest.py to generate summary.json"""'}, {}), "('generated conftest.py to generate summary.json')", False, 'from loguru import logger\n'), ((52, 14, 52, 82), 'httprunner.exceptions.TestCaseFormatError', 'exceptions.TestCaseFormatError', ({(52, 45, 52, 81): 'f"""Invalid jmespath extractor: {raw}"""'}, {}), "(f'Invalid jmespath extractor: {raw}')", False, 'from httprunner import exceptions\n'), ((225, 8, 225, 55), 'loguru.logger.error', 'logger.error', ({(225, 21, 225, 54): 'f"""Miss teststeps: {test_content}"""'}, {}), "(f'Miss teststeps: {test_content}')", False, 'from loguru import logger\n'), ((226, 8, 226, 19), 'sys.exit', 'sys.exit', ({(226, 17, 226, 18): '(1)'}, {}), '(1)', False, 'import sys\n'), ((232, 8, 232, 19), 'sys.exit', 'sys.exit', ({(232, 17, 232, 18): '(1)'}, {}), '(1)', False, 'import sys\n'), ((259, 8, 259, 65), 'loguru.logger.warning', 'logger.warning', ({(259, 23, 259, 64): 'f"""remove deprecated argument: --failfast"""'}, {}), "(f'remove deprecated argument: --failfast')", False, 'from loguru import logger\n'), ((264, 8, 264, 80), 'loguru.logger.warning', 'logger.warning', ({(264, 23, 264, 79): 'f"""replace deprecated argument --report-file with --html"""'}, {}), "(f'replace deprecated argument --report-file with --html')", False, 'from loguru import logger\n'), ((271, 8, 273, 9), 'loguru.logger.warning', 'logger.warning', ({(272, 12, 272, 78): 'f"""generate conftest.py keep compatibility with --save-tests in v2"""'}, {}), "(\n    f'generate conftest.py keep compatibility with --save-tests in v2')", False, 'from loguru import logger\n'), ((283, 11, 283, 30), 'os.path.exists', 'os.path.exists', ({(283, 26, 283, 29): 'arg'}, {}), '(arg)', False, 'import os\n'), ((288, 8, 288, 69), 'loguru.logger.error', 'logger.error', ({(288, 21, 288, 68): 'f"""No valid test path specified! \nargs: {args}"""'}, {}), '(f"""No valid test path specified! \nargs: {args}""")', False, 'from loguru import logger\n'), ((289, 8, 289, 19), 'sys.exit', 'sys.exit', ({(289, 17, 289, 18): '(1)'}, {}), '(1)', False, 'import sys\n'), ((366, 26, 366, 78), 'os.path.join', 'os.path.join', ({(366, 39, 366, 52): 'logs_dir_path', (366, 54, 366, 77): 'test_path_relative_path'}, {}), '(logs_dir_path, test_path_relative_path)', False, 'import os\n'), ((369, 47, 369, 85), 'os.path.split', 'os.path.split', ({(369, 61, 369, 84): 'test_path_relative_path'}, {}), '(test_path_relative_path)', False, 'import os\n'), ((370, 26, 370, 80), 'os.path.join', 'os.path.join', ({(370, 39, 370, 52): 'logs_dir_path', (370, 54, 370, 79): 'file_relative_folder_path'}, {}), '(logs_dir_path, file_relative_folder_path)', False, 'import os\n'), ((371, 28, 371, 55), 'os.path.splitext', 'os.path.splitext', ({(371, 45, 371, 54): 'test_file'}, {}), '(test_file)', False, 'import os\n'), ((380, 11, 380, 35), 'os.path.exists', 'os.path.exists', ({(380, 26, 380, 34): 'dir_path'}, {}), '(dir_path)', False, 'import os\n'), ((381, 8, 381, 29), 'os.makedirs', 'os.makedirs', ({(381, 20, 381, 28): 'dir_path'}, {}), '(dir_path)', False, 'import os\n'), ((39, 23, 39, 51), 'httprunner.loader.load_project_meta', 'load_project_meta', ({(39, 41, 39, 50): 'test_path'}, {}), '(test_path)', False, 'from httprunner.loader import load_project_meta, convert_relative_project_root_dir\n'), ((40, 20, 40, 73), 'httprunner.parser.parse_data', 'parse_data', ({(40, 31, 40, 44): 'raw_variables', (40, 46, 40, 48): '{}', (40, 50, 40, 72): 'project_meta.functions'}, {}), '(raw_variables, {}, project_meta.functions)', False, 'from httprunner.parser import parse_data\n'), ((45, 14, 47, 9), 'httprunner.exceptions.TestCaseFormatError', 'exceptions.TestCaseFormatError', ({(46, 12, 46, 56): 'f"""Invalid variables format: {raw_variables}"""'}, {}), "(f'Invalid variables format: {raw_variables}')", False, 'from httprunner import exceptions\n'), ((106, 8, 106, 56), 'loguru.logger.error', 'logger.error', ({(106, 21, 106, 55): 'f"""Invalid extractor: {extractors}"""'}, {}), "(f'Invalid extractor: {extractors}')", False, 'from loguru import logger\n'), ((107, 8, 107, 19), 'sys.exit', 'sys.exit', ({(107, 17, 107, 18): '(1)'}, {}), '(1)', False, 'import sys\n'), ((187, 18, 189, 13), 'httprunner.exceptions.TestCaseFormatError', 'exceptions.TestCaseFormatError', ({(188, 16, 188, 64): 'f"""Invalid teststep validate: {step[\'validate\']}"""'}, {}), '(f"Invalid teststep validate: {step[\'validate\']}"\n    )', False, 'from httprunner import exceptions\n'), ((29, 22, 31, 17), 'httprunner.exceptions.TestCaseFormatError', 'exceptions.TestCaseFormatError', ({(30, 20, 30, 64): 'f"""Invalid variables format: {raw_variables}"""'}, {}), "(f'Invalid variables format: {raw_variables}')", False, 'from httprunner import exceptions\n'), ((98, 16, 98, 64), 'loguru.logger.error', 'logger.error', ({(98, 29, 98, 63): 'f"""Invalid extractor: {extractors}"""'}, {}), "(f'Invalid extractor: {extractors}')", False, 'from loguru import logger\n'), ((99, 16, 99, 27), 'sys.exit', 'sys.exit', ({(99, 25, 99, 26): '(1)'}, {}), '(1)', False, 'import sys\n'), ((70, 16, 70, 56), 'loguru.logger.error', 'logger.error', ({(70, 29, 70, 55): 'f"""Invalid jmespath: {raw}"""'}, {}), "(f'Invalid jmespath: {raw}')", False, 'from loguru import logger\n'), ((71, 16, 71, 27), 'sys.exit', 'sys.exit', ({(71, 25, 71, 26): '(1)'}, {}), '(1)', False, 'import sys\n'), ((244, 18, 244, 77), 'httprunner.exceptions.TestCaseFormatError', 'exceptions.TestCaseFormatError', ({(244, 49, 244, 76): 'f"""Invalid teststep: {step}"""'}, {}), "(f'Invalid teststep: {step}')", False, 'from httprunner import exceptions\n')]