# Copyright (C) 2023 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
import asyncio
import decimal
import logging
import shutil
import subprocess
import xml.etree.ElementTree as ET
from typing import List, Optional, Union


class Error:
    """
    Base class for errors.

    It is used by several modules to define error types.
    """

    __match_args__ = ("message",)

    def __init__(self, message: str) -> None:
        self.message = message


class CommandTimedOut(Error):
    pass


class Command:
    @staticmethod
    async def run(
        arguments: List[str],
        output_file: Optional[str] = None,
        timeout: Optional[int] = None,
        cwd: Optional[str] = None,
    ) -> Optional[Error]:
        if output_file is None:
            process = await asyncio.create_subprocess_exec(
                *arguments,
                cwd=cwd,
            )
        else:
            with open(output_file, "a") as f:
                process = await asyncio.create_subprocess_exec(
                    *arguments,
                    stdout=f,
                    stderr=subprocess.STDOUT,
                    cwd=cwd,
                )

        try:
            await asyncio.wait_for(process.wait(), timeout=timeout)
        except asyncio.TimeoutError:
            command = " ".join(arguments)
            message = f'Command "{command}" timed out after {timeout} seconds'
            if output_file:
                message += f"; output was redirected to {output_file}"
            return CommandTimedOut(message)

        if process.returncode != 0:
            command = " ".join(arguments)
            message = f'Command "{command} returned non-zero code {process.returncode}'
            if output_file:
                message += f"; output was redirected to {output_file}"
            return Error(message)
        else:
            return None


class OperationTimedOut(Error):
    pass


class Operation:
    """
    Operation that can time out.
    """

    async def run(self) -> Optional[Error]:
        """
        Run the operation.

        This function is overridden by subclass implementations.
        """
        raise NotImplementedError()

    async def run_and_retry(
        self, retries: int, delay: int, logger: Optional[logging.Logger]
    ) -> Optional[Error]:
        """
        Run the operation. If it times out, retry.
        """
        error = await self.run()

        while isinstance(error, OperationTimedOut) and retries > 0:
            if logger is not None:
                logger.warning(f"Operation timed out, retrying after {delay} seconds")

            await asyncio.sleep(delay)
            error = await self.run()
            retries -= 1

        return error


class XmlParser:
    def __init__(self, element: ET.Element) -> None:
        self.element = element

    @staticmethod
    def load(file: str, tag: str) -> Union["XmlParser", Error]:
        try:
            tree = ET.parse(file)
        except ET.ParseError as error:
            return Error(str(error))
        else:
            element = tree.getroot()
            if element.tag != tag:
                return Error(f"Root element should be {tag} but was {element.tag}")
            else:
                return XmlParser(element)

    def children(self, tag: str) -> List["XmlParser"]:
        return [XmlParser(element) for element in self.element.findall(tag)]

    def child(self, tag: str) -> Union["XmlParser", Error]:
        children = self.children(tag)
        if len(children) == 0:
            return Error(f"{self.element.tag} has no {tag} children")
        elif len(children) == 1:
            return children[0]
        else:
            return Error(f"{self.element.tag} has multiple {tag} children")

    def string_attribute(self, name: str) -> Union[str, Error]:
        try:
            return self.element.attrib[name]
        except KeyError:
            return Error(f"{self.element.tag} has no attribute {name}")

    def integer_attribute(self, name: str) -> Union[int, Error]:
        string = self.string_attribute(name)
        match string:
            case Error() as error:
                return error

        try:
            return int(string)
        except ValueError:
            return Error(f"Could not parse {self.element.tag}.{name} as integer: {string}")

    def decimal_attribute(self, name: str) -> Union[decimal.Decimal, Error]:
        string = self.string_attribute(name)
        match string:
            case Error() as error:
                return error
        string = string.replace(",", ".")  # QtTest uses both, depending on the version.

        try:
            return decimal.Decimal(string)
        except decimal.InvalidOperation:
            return Error(f"Could not parse {self.element.tag}.{name} as decimal: {string}")


def check_for_executable(name: str) -> Optional[Error]:
    """
    Check that the executable is installed. Returns an error if not.
    """
    path = shutil.which(name)
    if path is None:
        return Error(f"Executable is not installed: {name}")

    return None