# mypy: allow-untyped-defs

import os
from multiprocessing import Queue, Event
from subprocess import PIPE
from threading import Thread
from mozprocess import ProcessHandlerMixin

from . import chrome_spki_certs
from .base import Browser, ExecutorBrowser
from .base import get_timeout_multiplier   # noqa: F401
from ..executors import executor_kwargs as base_executor_kwargs
from ..executors.executorcontentshell import (  # noqa: F401
    ContentShellCrashtestExecutor,
    ContentShellPrintRefTestExecutor,
    ContentShellRefTestExecutor,
    ContentShellTestharnessExecutor,
)


__wptrunner__ = {"product": "content_shell",
                 "check_args": "check_args",
                 "browser": "ContentShellBrowser",
                 "executor": {
                     "crashtest": "ContentShellCrashtestExecutor",
                     "print-reftest": "ContentShellPrintRefTestExecutor",
                     "reftest": "ContentShellRefTestExecutor",
                     "testharness": "ContentShellTestharnessExecutor",
                 },
                 "browser_kwargs": "browser_kwargs",
                 "executor_kwargs": "executor_kwargs",
                 "env_extras": "env_extras",
                 "env_options": "env_options",
                 "update_properties": "update_properties",
                 "timeout_multiplier": "get_timeout_multiplier",}


def check_args(**kwargs):
    pass


def browser_kwargs(logger, test_type, run_info_data, config, **kwargs):
    args = list(kwargs["binary_args"])

    args.append("--ignore-certificate-errors-spki-list=%s" %
        ','.join(chrome_spki_certs.IGNORE_CERTIFICATE_ERRORS_SPKI_LIST))

    webtranport_h3_port = config.ports.get('webtransport-h3')
    if webtranport_h3_port is not None:
        args.append(
            f"--origin-to-force-quic-on=web-platform.test:{webtranport_h3_port[0]}")

    # These flags are specific to content_shell - they activate web test protocol mode.
    args.append("--run-web-tests")
    args.append("-")

    return {"binary": kwargs["binary"],
            "binary_args": args}


def executor_kwargs(logger, test_type, test_environment, run_info_data,
                    **kwargs):
    executor_kwargs = base_executor_kwargs(test_type, test_environment, run_info_data,
                                           **kwargs)
    return executor_kwargs


def env_extras(**kwargs):
    return []


def env_options():
    return {"server_host": "127.0.0.1",
            "testharnessreport": "testharnessreport-content-shell.js"}


def update_properties():
    return (["debug", "os", "processor"], {"os": ["version"], "processor": ["bits"]})


class ContentShellBrowser(Browser):
    """Class that represents an instance of content_shell.

    Upon startup, the stdout, stderr, and stdin pipes of the underlying content_shell
    process are connected to multiprocessing Queues so that the runner process can
    interact with content_shell through its protocol mode.
    """

    def __init__(self, logger, binary="content_shell", binary_args=[], **kwargs):
        super().__init__(logger)

        self._args = [binary] + binary_args
        self._proc = None

    def start(self, group_metadata, **kwargs):
        self.logger.debug("Starting content shell: %s..." % self._args[0])

        # Unfortunately we need to use the Process class directly because we do not
        # want mozprocess to do any output handling at all.
        self._proc = ProcessHandlerMixin.Process(self._args, stdin=PIPE, stdout=PIPE, stderr=PIPE)
        if os.name == "posix":
            self._proc.pgid = ProcessHandlerMixin._getpgid(self._proc.pid)
            self._proc.detached_pid = None

        self._stdout_queue = Queue()
        self._stderr_queue = Queue()
        self._stdin_queue = Queue()
        self._io_stopped = Event()

        self._stdout_reader = self._create_reader_thread(self._proc.stdout, self._stdout_queue)
        self._stderr_reader = self._create_reader_thread(self._proc.stderr, self._stderr_queue)
        self._stdin_writer = self._create_writer_thread(self._proc.stdin, self._stdin_queue)

        # Content shell is likely still in the process of initializing. The actual waiting
        # for the startup to finish is done in the ContentShellProtocol.
        self.logger.debug("Content shell has been started.")

    def stop(self, force=False):
        self.logger.debug("Stopping content shell...")

        if self.is_alive():
            kill_result = self._proc.kill(timeout=5)
            # This makes sure any left-over child processes get killed.
            # See http://bugzilla.mozilla.org/show_bug.cgi?id=1760080
            if force and kill_result != 0:
                self._proc.kill(9, timeout=5)

        # We need to shut down these queues cleanly to avoid broken pipe error spam in the logs.
        self._stdout_reader.join(2)
        self._stderr_reader.join(2)

        self._stdin_queue.put(None)
        self._stdin_writer.join(2)

        for thread in [self._stdout_reader, self._stderr_reader, self._stdin_writer]:
            if thread.is_alive():
                self.logger.warning("Content shell IO threads did not shut down gracefully.")
                return False

        stopped = not self.is_alive()
        if stopped:
            self.logger.debug("Content shell has been stopped.")
        else:
            self.logger.warning("Content shell failed to stop.")

        return stopped

    def is_alive(self):
        return self._proc is not None and self._proc.poll() is None

    def pid(self):
        return self._proc.pid if self._proc else None

    def executor_browser(self):
        """This function returns the `ExecutorBrowser` object that is used by other
        processes to interact with content_shell. In our case, this consists of the three
        multiprocessing Queues as well as an `io_stopped` event to signal when the
        underlying pipes have reached EOF.
        """
        return ExecutorBrowser, {"stdout_queue": self._stdout_queue,
                                 "stderr_queue": self._stderr_queue,
                                 "stdin_queue": self._stdin_queue,
                                 "io_stopped": self._io_stopped}

    def check_crash(self, process, test):
        return not self.is_alive()

    def _create_reader_thread(self, stream, queue):
        """This creates (and starts) a background thread which reads lines from `stream` and
        puts them into `queue` until `stream` reports EOF.
        """
        def reader_thread(stream, queue, stop_event):
            while True:
                line = stream.readline()
                if not line:
                    break

                queue.put(line)

            stop_event.set()
            queue.close()
            queue.join_thread()

        result = Thread(target=reader_thread, args=(stream, queue, self._io_stopped), daemon=True)
        result.start()
        return result

    def _create_writer_thread(self, stream, queue):
        """This creates (and starts) a background thread which gets items from `queue` and
        writes them into `stream` until it encounters a None item in the queue.
        """
        def writer_thread(stream, queue):
            while True:
                line = queue.get()
                if not line:
                    break

                stream.write(line)
                stream.flush()

        result = Thread(target=writer_thread, args=(stream, queue), daemon=True)
        result.start()
        return result
