#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Build and run BioLib GoogleTests.

Hardware suites are run in a separate process for every scanner. This keeps
all tests for one scanner together and allows them to reuse one enrolled
fingerprint. Per-scanner GoogleTest XML files are merged back into one report
per executable.
"""

from __future__ import print_function

import argparse
import os
from pathlib import Path
import platform
import re
import shutil
import subprocess
import sys
import threading
from typing import Dict, List, Optional, TextIO
import xml.etree.ElementTree as ET


TEST_EXECUTABLES = (
    "test_base",
    "test_scanner",
    "test_fingerprint",
    "test_authenticate",
    "test_multithread",
)
HARDWARE_EXECUTABLES = {
    "test_fingerprint",
    "test_authenticate",
    "test_multithread",
}
SCANNER_PATTERN = re.compile(r"/([A-Za-z][A-Za-z0-9_]*)\s+(?:#|$)")
INTEGER_XML_ATTRIBUTES = ("tests", "failures", "disabled", "errors", "skipped")

SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_BIN_DIR = SCRIPT_DIR / "out"
DEFAULT_BUILD_DIR = SCRIPT_DIR / "_build"
DEFAULT_JUNIT_DIR = SCRIPT_DIR


def get_os_string() -> str:
    system = platform.system().lower()
    machine = platform.machine().lower()
    architecture_names = {
        "amd64": "x86_64",
        "x64": "x86_64",
        "i386": "x86",
        "i686": "x86",
        "aarch64": "arm64",
    }
    machine = architecture_names.get(machine, machine)
    return "{}_{}".format(system, machine)


def _log(log_file: TextIO, text: str) -> None:
    sys.stdout.write(text)
    sys.stdout.flush()
    log_file.write(text)
    log_file.flush()


def _stream_process(
    command: List[str],
    cwd: Path,
    log_file: TextIO,
    environment: Optional[Dict[str, str]] = None,
) -> int:
    try:
        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=str(cwd),
            env=environment,
        )
    except OSError as error:
        _log(log_file, "[ERROR] {}\n".format(error))
        return 1

    def forward(stream) -> None:
        for line in iter(stream.readline, b""):
            _log(log_file, line.decode("utf-8", errors="replace"))

    stdout_thread = threading.Thread(
        target=forward, args=(process.stdout,), daemon=True
    )
    stderr_thread = threading.Thread(
        target=forward, args=(process.stderr,), daemon=True
    )
    stdout_thread.start()
    stderr_thread.start()
    stdout_thread.join()
    stderr_thread.join()
    return process.wait()


def _executable_path(bin_dir: Path, name: str) -> Path:
    suffix = ".exe" if sys.platform == "win32" else ""
    return bin_dir / (name + suffix)


def _build_project(
    build_dir: Path,
    bin_dir: Path,
    dont_clear: bool,
    log_file: TextIO,
) -> bool:
    if not dont_clear:
        _log(log_file, "Remove: {}\n".format(build_dir))
        shutil.rmtree(str(build_dir), ignore_errors=True)
        _log(log_file, "Remove: {}\n".format(bin_dir))
        shutil.rmtree(str(bin_dir), ignore_errors=True)

    configure = [
        "cmake",
        "-S",
        str(SCRIPT_DIR),
        "-B",
        str(build_dir),
        "-DCMAKE_BUILD_TYPE=Release",
        "-DCMAKE_CXX_STANDARD=17",
    ]
    if sys.platform != "win32":
        configure.extend([
            "-DCMAKE_C_COMPILER=gcc",
            "-DCMAKE_CXX_COMPILER=g++",
        ])
    build = [
        "cmake",
        "--build",
        str(build_dir),
        "--config",
        "Release",
        "-j",
    ]
    _log(log_file, "Configure: {}\n".format(" ".join(configure)))
    if _stream_process(configure, SCRIPT_DIR, log_file) != 0:
        return False
    _log(log_file, "Build: {}\n".format(" ".join(build)))
    return _stream_process(build, SCRIPT_DIR, log_file) == 0


def _discover_scanners(executable: Path) -> List[str]:
    environment = os.environ.copy()
    environment.pop("BIO_TEST_SCANNER", None)
    result = subprocess.run(
        [str(executable), "--gtest_list_tests"],
        cwd=str(executable.parent),
        env=environment,
        check=False,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError("failed to discover scanners:\n{}{}".format(
            result.stdout, result.stderr
        ))

    scanners: List[str] = []
    for line in result.stdout.splitlines():
        match = SCANNER_PATTERN.search(line)
        if match is not None and match.group(1) not in scanners:
            scanners.append(match.group(1))
    if not scanners:
        raise RuntimeError("no scanners found in GoogleTest parameter list")
    return scanners


def _merge_xml(inputs: List[Path], output: Path) -> None:
    merged = ET.Element("testsuites", {"name": "AllTests"})
    totals = {name: 0 for name in INTEGER_XML_ATTRIBUTES}
    total_time = 0.0
    timestamps: List[str] = []

    for input_path in inputs:
        root = ET.parse(str(input_path)).getroot()
        for name in INTEGER_XML_ATTRIBUTES:
            totals[name] += int(root.attrib.get(name, "0"))
        total_time += float(root.attrib.get("time", "0"))
        if root.attrib.get("timestamp"):
            timestamps.append(root.attrib["timestamp"])
        for suite in root.findall("testsuite"):
            merged.append(suite)

    for name, value in totals.items():
        if value != 0 or name in {"tests", "failures", "disabled", "errors"}:
            merged.set(name, str(value))
    merged.set("time", "{:.3f}".format(total_time))
    if timestamps:
        merged.set("timestamp", min(timestamps))

    output.parent.mkdir(parents=True, exist_ok=True)
    if hasattr(ET, "indent"):
        ET.indent(merged, space="  ")
    ET.ElementTree(merged).write(
        str(output), encoding="UTF-8", xml_declaration=True
    )


def _gtest_arguments(test_filter: Optional[str], xml_path: Optional[Path]) -> List[str]:
    arguments: List[str] = []
    if test_filter:
        arguments.append("--gtest_filter=" + test_filter)
    if xml_path is not None:
        arguments.append("--gtest_output=xml:" + str(xml_path))
    return arguments


def _run_regular_test(
    executable: Path,
    test_filter: Optional[str],
    xml_path: Optional[Path],
    log_file: TextIO,
) -> int:
    _log(log_file, "\n========== {} ==========\n".format(executable.stem))
    return _stream_process(
        [str(executable), *_gtest_arguments(test_filter, xml_path)],
        executable.parent,
        log_file,
    )


def _run_hardware_test(
    executable: Path,
    test_filter: Optional[str],
    xml_path: Optional[Path],
    excluded_scanners: List[str],
    log_file: TextIO,
) -> int:
    scanners = [
        scanner
        for scanner in _discover_scanners(executable)
        if scanner not in set(excluded_scanners)
    ]
    if not scanners:
        raise RuntimeError("all scanners were excluded for {}".format(executable.name))

    exit_code = 0
    scanner_xml_files: List[Path] = []
    for scanner in scanners:
        _log(
            log_file,
            "\n========== {} / {} ==========\n".format(executable.stem, scanner),
        )
        environment = os.environ.copy()
        environment["BIO_TEST_SCANNER"] = scanner

        scanner_xml = None
        if xml_path is not None:
            scanner_xml = xml_path.with_name(
                ".{}.{}.tmp.xml".format(xml_path.stem, scanner)
            )
        command = [
            str(executable),
            *_gtest_arguments(test_filter, scanner_xml),
        ]
        result = _stream_process(
            command, executable.parent, log_file, environment
        )
        if result != 0:
            exit_code = 1

        if scanner_xml is not None and scanner_xml.is_file():
            scanner_xml_files.append(scanner_xml)
        elif scanner_xml is not None:
            _log(log_file, "[ERROR] XML was not created for {}\n".format(scanner))
            exit_code = 1

    if xml_path is not None and scanner_xml_files:
        try:
            _merge_xml(scanner_xml_files, xml_path)
        finally:
            for scanner_xml in scanner_xml_files:
                try:
                    scanner_xml.unlink()
                except FileNotFoundError:
                    pass
        _log(log_file, "Combined XML: {}\n".format(xml_path))
    return exit_code


def run_tests(options) -> int:
    bin_dir = options.bin_dir.resolve()
    junit_dir = options.junit_dir.resolve() if options.junit_dir else None
    selected_tests = options.tests or list(TEST_EXECUTABLES)
    failed: List[str] = []

    log_path = options.log_file.resolve()
    with log_path.open("w", encoding="utf-8") as log_file:
        if options.build and not _build_project(
            options.build_dir.resolve(),
            bin_dir,
            options.dont_clear,
            log_file,
        ):
            return 1

        if not bin_dir.is_dir():
            _log(log_file, "[ERROR] Binary directory not found: {}\n".format(bin_dir))
            return 1
        if junit_dir is not None:
            junit_dir.mkdir(parents=True, exist_ok=True)

        for name in selected_tests:
            executable = _executable_path(bin_dir, name)
            if not executable.is_file():
                _log(log_file, "[ERROR] Test executable not found: {}\n".format(executable))
                failed.append(name)
                continue

            xml_path = (
                junit_dir / ("{}_{}.xml".format(name, get_os_string()))
                if junit_dir
                else None
            )
            try:
                if name in HARDWARE_EXECUTABLES:
                    result = _run_hardware_test(
                        executable,
                        options.test_filter,
                        xml_path,
                        options.exclude_scanner,
                        log_file,
                    )
                else:
                    result = _run_regular_test(
                        executable, options.test_filter, xml_path, log_file
                    )
            except (OSError, RuntimeError, ET.ParseError) as error:
                _log(log_file, "[ERROR] {}: {}\n".format(name, error))
                result = 1

            if result != 0:
                failed.append(name)

        summary = (
            "\n{}\nResult: {}/{} test executables passed.\n".format(
                "=" * 57,
                len(selected_tests) - len(failed),
                len(selected_tests),
            )
        )
        if failed:
            summary += "Failed: {}\n".format(", ".join(failed))
        summary += "{}\n".format("=" * 57)
        _log(log_file, summary)
    return 1 if failed else 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Build and run BioLib tests.")
    parser.add_argument(
        "-t",
        "--test",
        dest="tests",
        action="append",
        choices=TEST_EXECUTABLES,
        help="test executable to run; may be repeated, default: all",
    )
    parser.add_argument(
        "-f",
        "--filter",
        dest="test_filter",
        help="GoogleTest filter, for example '*ScanCheckJSON*:*ScanFingerprintInfo*'",
    )
    parser.add_argument(
        "-o",
        "--junit-dir",
        type=Path,
        default=DEFAULT_JUNIT_DIR,
        help="directory for JUnit XML reports, default: directory of run_tests.py",
    )
    parser.add_argument(
        "-b", "--build", action="store_true", help="build tests before running"
    )
    parser.add_argument(
        "-nc",
        "--dont-clear",
        action="store_true",
        help="do not remove build and output directories before building",
    )
    parser.add_argument(
        "--build-dir", type=Path, default=DEFAULT_BUILD_DIR
    )
    parser.add_argument("--bin-dir", type=Path, default=DEFAULT_BIN_DIR)
    parser.add_argument(
        "--exclude-scanner",
        action="append",
        default=[],
        help="scanner parameter to skip, for example ZKTeco1",
    )
    parser.add_argument(
        "--log-file",
        type=Path,
        default=SCRIPT_DIR / ("biolib_tests_" + get_os_string() + ".log"),
    )
    return run_tests(parser.parse_args())


if __name__ == "__main__":
    sys.exit(main())