#!/usr/bin/python3
"""Test the operation of the wmanager executable."""

from __future__ import annotations

import contextlib
import dataclasses
import errno
import os
import pathlib
import re
import select
import signal
import subprocess
import sys
import tempfile
import time

from typing import (
    Callable,  # noqa: H301
    Dict,
    Iterator,
    List,
    Optional,
    Tuple,
    Union,
)


TITLE = "wmanager-test-wm"

RE_XEV = re.compile(
    r""" ^
    (?P<name> \w+ ) \s+
    event,
    (?P<rest> .* )
    $ """,
    re.X,
)

WIN_PROPS = {
    "Absolute upper-left X": "x0",
    "Absolute upper-left Y": "y0",
    "Width": "dx",
    "Height": "dy",
}

TEST_WMS = [
    ("first", "firstwm"),
    ("second", "/path/to/secondwm"),
    ("a third thing", "something with 'special' characters, maybe?"),
]


@dataclasses.dataclass
class WindowData:
    """Attributes of the window that wmanager created."""

    click_dy: int
    window_id: Optional[str]
    window_props: Dict[str, int]


class IncDecoder:
    """An incremental UTF-8 decoder."""

    buf: bytes
    decoded: str
    lines: List[str]

    def __init__(self) -> None:
        """Initialize an empty IncDecoder object."""
        self.buf = b""
        self.decoded = ""
        self.lines = []

    def add_bytes(self, data: bytes) -> None:
        """Add a couple of bytes, decode characters, split them into lines."""
        for value in data:
            self.buf += bytes([value])
            try:
                dec = self.buf.decode("UTF-8")
            except UnicodeDecodeError as err:
                sys.exit(f"Oof, err {err!r}")
                continue

            self.buf = b""
            self.decoded += dec
            if "\n" in self.decoded:
                lines = self.decoded.split("\n")
                self.decoded = lines.pop()
                self.lines.extend(line.rstrip("\r") for line in lines)

    def pop_lines(self) -> List[str]:
        """Return the decoded lines and reset them."""
        res, self.lines = self.lines, []
        return res

    def pop_decoded(self) -> str:
        """Return the incomplete decoded line."""
        res, self.decoded = self.decoded, ""
        return res


LinesHandler = Callable[[str, bool], None]  # pylint: disable=invalid-name


@dataclasses.dataclass(frozen=True)
class Child:
    """A child process's attributes."""

    proc: subprocess.Popen[bytes]
    fileno: int
    dec: IncDecoder
    handler: List[LinesHandler]


def spawn_child(cmd: List[str], poller: select.epoll, handler: LinesHandler) -> Child:
    """Spawn a child, create a decoder."""
    print(f"Spawning a child process: {' '.join(cmd)}")
    proc = subprocess.Popen(  # pylint: disable=consider-using-with
        cmd, shell=False, stdout=subprocess.PIPE, bufsize=0
    )
    print(f"- spawned child pid {proc.pid}")

    assert proc.stdout is not None
    fno = proc.stdout.fileno()
    os.set_blocking(fno, False)
    poller.register(fno)
    poller.modify(fno, select.EPOLLIN | select.EPOLLHUP)

    return Child(proc=proc, fileno=fno, dec=IncDecoder(), handler=[handler])


def ready_children(
    children: Dict[str, Child], poller: select.epoll
) -> Dict[str, Child]:
    """Poll the children for input."""
    by_fno = {child.fileno: name for name, child in children.items()}
    events = poller.poll()
    print(f"Events: {events!r}")
    return {name: children[name] for name in (by_fno[evt[0]] for evt in events)}


def read_available(child: Child, final: bool = False) -> bool:
    """Read and process data from the process's standard output,."""
    assert child.proc.stdout is not None
    res = False
    while True:
        try:
            data = os.read(child.proc.stdout.fileno(), 4096)
            if not data:
                break
            child.dec.add_bytes(data)
            res = True
        except BlockingIOError as err:
            if err.errno != errno.EAGAIN:
                raise
            break

    lines = child.dec.pop_lines()
    if final and child.dec.decoded:
        lines.append(child.dec.pop_decoded())

    if not final:
        for line in lines:
            child.handler[0](line, False)
    elif lines:
        for line in lines[:-1]:
            child.handler[0](line, False)
        child.handler[0](lines[-1], True)
    else:
        # Oof... mmkay.
        child.handler[0]("", True)

    return res


def cleanup(children: Dict[str, Child]) -> None:
    """Clean up any remaining child processes."""
    print("Checking for any remaining child processes")
    for name, child in children.items():
        print(f"- child '{name}' pid {child.proc.pid}")
        try:
            res = child.proc.poll()
            if res is not None:
                print(f"The {name} child exited unexpectedly with code {res}")
            else:
                print("  - trying to terminate the child")
                child.proc.terminate()
                res = child.proc.wait()
                if res not in (0, -signal.SIGTERM):
                    print(f"The {name} child exited with an unexpected code {res}")

            read_available(child, final=True)
        except BaseException as err:  # pylint: disable=broad-except
            print(
                f"An error occurred while trying to "
                f"terminate the {name} child: {err}",
                file=sys.stderr,
            )
            if child.proc.poll() is None:
                try:
                    child.proc.kill()
                except BaseException as k_err:  # pylint: disable=broad-except
                    print(
                        "An additional error occurred while trying to "
                        f"kill the {name} child: {k_err}",
                        file=sys.stderr,
                    )


def cnee_write(cmd: str) -> None:
    """Send a command via the cnee tool."""
    print(f"Sending the {cmd!r} command via cnee")
    res = subprocess.run(
        ["cnee", "--replay", "--file", "stdin"],
        bufsize=0,
        capture_output=True,
        check=False,
        encoding="us-ascii",
        input=cmd,
        shell=False,
    )
    if res.returncode != 0:
        sys.exit(
            f"The cnee tool could not handle {cmd!r}: "
            f"code {res.returncode}, error output: {res.stderr}"
        )


def xev_create_notify(name: str, lines: List[str], window_data: WindowData) -> None:
    """A window was created... I guess."""
    # pylint: disable=too-many-locals,too-many-statements
    if window_data.window_id is not None:
        print(f"we already have the {window_data.window_id} id, so yeah")
        return

    if len(lines) < 2:
        sys.exit(f"Expected at least two lines for {name}: {lines!r}")
    window: Optional[str] = None
    for param in (thing.strip() for thing in lines[1].split(",")):
        if not param:
            continue
        values = param.split(" ", 1)
        if values[0] == "window":
            if window is not None:
                sys.exit("Duplicate 'window' on the second {name} line: {lines!r}")
            window = values[1]
    if window is None:
        sys.exit(f"Expected 'window' on the second {name} line: {lines!r}")

    res = subprocess.run(
        ["env", "LC_ALL=C.UTF-8", "xwininfo", "-id", window, "-shape"],
        bufsize=0,
        check=False,
        capture_output=True,
        encoding="UTF-8",
        shell=False,
    )
    if res.returncode != 0:
        print("- xwininfo could not find it, guess it was not important")
        return
    output = res.stdout

    prefix_first = f"xwininfo: Window id: {window} "
    props: Dict[str, int] = {}

    InfoHandler = Callable[[str], Optional[str]]

    def gather_properties(line: str) -> Optional[str]:
        """Gather the key/value pairs."""
        print(f"- gather_properties({line!r})")
        if not line.startswith(" "):
            print("  - not a property line anymore")
            return None
        data = line.strip().split(":", 1)
        if len(data) != 2:
            print("  - not a key/value pair")
            return None
        key, value = data[0].strip(), data[1].strip()
        rkey = WIN_PROPS.get(key)
        print(f"  - key {key!r} rkey {rkey!r} value {value!r}")

        if rkey is None:
            print("  - we do not care about this one")
            return "gather_properties"
        if rkey in props:
            sys.exit(f"Duplicate key {key!r} in the xwininfo output for {window}")
        props[rkey] = int(value)
        if len(props) == len(WIN_PROPS):
            print("  - got them all!")
            return None

        return "gather_properties"

    def skip_empty_line(line: str) -> Optional[str]:
        """Expect a single empty line."""
        print(f"- skip_empty_lines({line!r})")
        if line:
            print("  - not empty, passing it to gather_properties")
            return gather_properties(line)

        return "skip_empty_line"

    def look_for_first(line: str) -> Optional[str]:
        """Look for the first line."""
        print(f"- look_for_first({line!r})")
        if not line:
            return "look_for_first"

        if not line.startswith(prefix_first):
            sys.exit(
                "Unexpected first line in the output of xwininfo "
                f"for {window}: {line!r}"
            )
        rest = line[len(prefix_first) :]
        if not (rest.startswith('"') and rest.endswith('"')):
            print("- no window title, not ours")
            return None

        title = rest[1:-1]
        print(f"- got window title {title!r}")
        if title != TITLE:
            print("  - apparently not ours")
            return None
        print("  - got the correct window title!")

        return "skip_empty_line"

    handlers: Dict[str, InfoHandler] = {
        "look_for_first": look_for_first,
        "gather_properties": gather_properties,
        "skip_empty_line": skip_empty_line,
    }

    handler: InfoHandler = look_for_first
    for line in output.splitlines():
        print(f"- shape line: {line!r}")
        new_handler = handler(line)
        if new_handler is None:
            break
        handler = handlers[new_handler]

    if handler != gather_properties:  # pylint: disable=comparison-with-callable
        print("Apparently not an interesting window at all")
        return

    if len(props) != len(WIN_PROPS):
        missing = sorted(key for key, value in WIN_PROPS.items() if value not in props)
        sys.exit(
            f"Did not find some properties in "
            f"the xwininfo output for {window!r}: {missing!r}"
        )

    window_data.window_id = window
    window_data.window_props = props

    click_x = props["x0"] + 20
    click_y = props["y0"] + window_data.click_dy
    cnee_write(f"fake-motion x={click_x} y={click_y} msec=2")
    cnee_write("fake-button button=1 msec=2")
    cnee_write("fake-key key=Return msec=2")
    print("Now wait for the wmanager process to exit")


XEV_HANDLERS: Dict[str, Callable[[str, List[str], WindowData], None]] = {
    "CreateNotify": xev_create_notify,
}


def handle_xev_line(
    lines: List[str], line: str, final: bool, window_data: WindowData
) -> None:
    """Handle a line of xev output."""
    if line:
        lines.append(line)
    if line and not final:
        return

    if not lines:
        return

    first, rest = lines[0], lines[1:]
    lines.clear()

    match = RE_XEV.match(first)
    if not match:
        sys.exit(f"Unexpected line from xev: {lines[0]}")
    name = match.group("name")
    handler = XEV_HANDLERS.get(name)
    if handler is not None:
        print(f"xev event {name}")
        handler(name, [match.group("rest")] + rest, window_data)
    else:
        print(f"xev event {name}, ignored")


def run_test(click_dy: int, expected: List[int]) -> Tuple[int, Optional[int]]:
    """Run a wmanager test with a click at the specified Y offset."""
    print(f"Running a wmanager test, click_dy {click_dy}")

    window_data = WindowData(click_dy=click_dy, window_id=None, window_props={})
    children = {}
    poller = select.epoll()
    wmoutput: List[str] = []
    try:
        print("about to spawn xev")
        xev_lines: List[str] = []
        children["xev"] = spawn_child(
            ["xev", "-root"],
            poller,
            lambda line, final: handle_xev_line(xev_lines, line, final, window_data),
        )
        print("sleeping for a little while")
        time.sleep(1)

        print("about to spawn wmanager")
        children["wmanager"] = spawn_child(
            [
                "wmanager",
                "-title",
                TITLE,
                "-r" + str(pathlib.Path(os.environ["HOME"]) / ".wmanagerrc"),
            ],
            poller,
            lambda line, _: wmoutput.append(line),
        )

        while "wmanager" in children:
            print("polling")
            ready = ready_children(children, poller)
            for name, child in ready.items():
                print(f"whee, got something for {name}")
                if not read_available(child):
                    print("- nothing more to read, unregistering")
                    poller.unregister(child.fileno)

                    print("- waiting for the child to exit")
                    res = child.proc.wait()

                    print("- any last words?")
                    read_available(child, final=True)
                    print(f"- {name} exited with code {res}\n")
                    del children[name]
    finally:
        cleanup(children)

    if wmoutput and not wmoutput[-1]:
        del wmoutput[-1]

    if len(wmoutput) != 1:
        sys.exit(f"Unexpected wmanager output: {wmoutput!r}")

    print(f"wmanager told us {wmoutput!r}")
    found = [idx for idx in expected if TEST_WMS[idx][1] == wmoutput[0]]
    if len(found) != 1:
        sys.exit(
            "Expected wmanager to tell us "
            + (
                repr(TEST_WMS[expected[0]][1])
                if len(expected) == 1
                else " or ".join(repr(TEST_WMS[idx][1]) for idx in expected)
            )
            + f", got {wmoutput[0]!r} instead"
        )

    return found[0], window_data.window_props.get("dy")


@contextlib.contextmanager
def tempdir(path: Optional[Union[pathlib.Path, str]] = None) -> Iterator[pathlib.Path]:
    """Create a temporary directory and eventually remove it."""
    temp = None
    try:
        temp = tempfile.mkdtemp(
            dir=str(path) if isinstance(path, pathlib.Path) else path
        )
        yield pathlib.Path(temp)
    finally:
        if temp is not None:
            subprocess.call(["rm", "-rf", "--", temp])


def main() -> None:
    """Main program: parse command-line options, run stuff."""
    with tempdir() as tempd:
        homedir = tempd / "home"
        print(f"Creating a temporary homedir at {homedir}")
        homedir.mkdir(0o755)
        os.environ["HOME"] = str(homedir)

        rcfile = homedir / ".wmanagerrc"
        rcfile.write_text(
            "".join(f"{first}={second}\n" for first, second in TEST_WMS),
            encoding="UTF-8",
        )

        click_dy = 0
        expected = [0]
        window_height = None
        while True:
            click_dy += 8
            if window_height is not None and click_dy >= window_height:
                sys.exit(
                    "Reached the bottom of the window "
                    f"({click_dy} >= {window_height}) too soon"
                )
                print(f"{click_dy} no less than {window_height}, we are done")
                break

            found, new_window_height = run_test(click_dy, expected)
            if window_height is None:
                window_height = new_window_height
            elif new_window_height != window_height:
                sys.exit(
                    f"Got both {window_height} and {new_window_height} "
                    "for the window height"
                )

            if found + 1 == len(TEST_WMS):
                print("Got the last one!")
                break
            expected = [found, found + 1]

        print(f"We are done! (window height {window_height})")


if __name__ == "__main__":
    main()