"""End-to-end tests: run xidle-inhibit on a private D-Bus bus with fake lockers on PATH."""

import json
import os
import signal
import subprocess
import sys
import time
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import Any

import pytest
from gi.repository import Gio, GLib

# XIDLE_INHIBIT tests another copy, e.g. /usr/bin/xidle-inhibit.
SCRIPT = Path(os.environ.get("XIDLE_INHIBIT") or Path(__file__).resolve().parent.parent / "xidle-inhibit")
BUS_NAME = "org.freedesktop.ScreenSaver"
OBJECT_PATH = "/org/freedesktop/ScreenSaver"
STATE_FILE = "xidle-inhibit.json"
TIMEOUT = 5.0
DELAY = ["--release-delay", "1"]
SETTLE = 1.5  # longer than DELAY, to check that nothing happens

ENABLE, DISABLE = "xautolock -enable", "xautolock -disable"

# No <servicedir>: nothing is auto-started on this bus.
BUS_CONFIG = """<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
 "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
<busconfig>
  <type>session</type>
  <listen>unix:dir={dir}</listen>
  <auth>EXTERNAL</auth>
  <policy context="default">
    <allow send_destination="*" eavesdrop="true"/>
    <allow eavesdrop="true"/>
    <allow own="*"/>
  </policy>
</busconfig>
"""

FAKE_TOOL = '#!/bin/sh\necho "${0##*/} $*" >> "$FAKE_LOG"\n'


def wait_for(cond: Callable[[], bool], what: str, timeout: float = TIMEOUT) -> None:
    deadline = time.monotonic() + timeout
    while not cond():
        if time.monotonic() > deadline:
            raise AssertionError(f"timed out waiting for {what}")
        time.sleep(0.05)


class Bus:
    def __init__(self, root: Path) -> None:
        sock_dir = root / "sock"
        sock_dir.mkdir(parents=True)
        config = root / "bus.conf"
        config.write_text(BUS_CONFIG.format(dir=sock_dir))
        self.proc = subprocess.Popen(
            ["dbus-daemon", f"--config-file={config}", "--nofork", "--print-address=1"],
            stdout=subprocess.PIPE, text=True,
        )
        assert self.proc.stdout is not None
        self.address = self.proc.stdout.readline().strip()
        assert self.address, "dbus-daemon printed no address"

    def connect(self) -> Gio.DBusConnection:
        flags = Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION
        return Gio.DBusConnection.new_for_address_sync(self.address, flags, None, None)

    def stop(self) -> None:
        self.proc.terminate()
        self.proc.wait(TIMEOUT)


class Client:
    """One bus connection, so one D-Bus sender."""

    def __init__(self, bus: Bus) -> None:
        self.conn = bus.connect()
        name = self.conn.get_unique_name()
        assert name is not None
        self.name = name

    def call(
        self, method: str, args: GLib.Variant | None = None, reply: str | None = None,
        path: str = OBJECT_PATH, dest: str = BUS_NAME, iface: str = BUS_NAME,
    ) -> Any:
        result = self.conn.call_sync(
            dest, path, iface, method, args, GLib.VariantType(reply) if reply else None,
            Gio.DBusCallFlags.NO_AUTO_START, int(TIMEOUT * 1000), None,
        )
        return result.unpack()

    def inhibit(self, reason: str = "Playing video", path: str = OBJECT_PATH) -> int:
        return int(self.call("Inhibit", GLib.Variant("(ss)", ("test-app", reason)), "(u)", path)[0])

    def uninhibit(self, cookie: int, path: str = OBJECT_PATH) -> None:
        self.call("UnInhibit", GLib.Variant("(u)", (cookie,)), path=path)

    def has_owner(self, name: str) -> bool:
        return bool(self.call(
            "NameHasOwner", GLib.Variant("(s)", (name,)), "(b)",
            "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus",
        )[0])

    def close(self) -> None:
        if not self.conn.is_closed():
            self.conn.close_sync(None)


class Sandbox:
    def __init__(self, root: Path) -> None:
        self.root = root
        self.bin = root / "bin"
        self.runtime = root / "run"
        self.config = root / "config"
        for d in (self.bin, self.runtime, self.config / "xidle-inhibit"):
            d.mkdir(parents=True)
        for tool in ("xautolock", "xss-lock", "loginctl"):
            path = self.bin / tool
            path.write_text(FAKE_TOOL)
            path.chmod(0o755)
        self.fake_log = root / "fake.log"
        self.daemon_log = root / "daemon.log"
        self.state_file = self.runtime / STATE_FILE
        self.buses: list[Bus] = []
        self.clients: list[Client] = []
        self.daemons: list[subprocess.Popen[bytes]] = []
        self.bus = self.new_bus()
        self.set_backend("xautolock")

    def new_bus(self) -> Bus:
        bus = Bus(self.root / f"bus{len(self.buses)}")
        self.buses.append(bus)
        return bus

    def client(self, bus: Bus | None = None) -> Client:
        client = Client(bus or self.bus)
        self.clients.append(client)
        return client

    def set_backend(self, backend: str | None) -> None:
        config = self.config / "xidle-inhibit" / "config"
        if backend is None:
            config.unlink(missing_ok=True)
        else:
            config.write_text(f"backend = {backend}\n")

    def environ(self, bus: Bus) -> dict[str, str]:
        env = dict(os.environ)
        for var in ("DISPLAY", "WAYLAND_DISPLAY"):
            env.pop(var, None)
        env.update(
            PATH=str(self.bin), DBUS_SESSION_BUS_ADDRESS=bus.address, FAKE_LOG=str(self.fake_log),
            XDG_RUNTIME_DIR=str(self.runtime), XDG_CONFIG_HOME=str(self.config),
        )
        return env

    def start(self, *args: str, bus: Bus | None = None, wait: bool = True) -> subprocess.Popen[bytes]:
        """Start the daemon; with wait, return once it has taken the name and set the locker."""
        bus = bus or self.bus
        with open(self.daemon_log, "ab") as log:
            proc = subprocess.Popen(
                [sys.executable, str(SCRIPT), *args], env=self.environ(bus), stdout=log, stderr=subprocess.STDOUT,
            )
        self.daemons.append(proc)
        if wait:
            probe = Client(bus)
            try:
                # Method calls are handled after start() returns, so a reply means startup is done.
                def answers() -> bool:
                    assert proc.poll() is None, f"daemon exited:\n{self.log()}"
                    try:
                        probe.call("GetActive", reply="(b)")
                        return True
                    except GLib.Error:
                        return False
                wait_for(answers, "the daemon to own the name")
            finally:
                probe.close()
        return proc

    def stop(self, sig: int = signal.SIGTERM) -> int:
        proc = self.daemons[-1]
        proc.send_signal(sig)
        return proc.wait(TIMEOUT)

    def calls(self) -> list[str]:
        try:
            return self.fake_log.read_text().splitlines()
        except FileNotFoundError:
            return []

    def wait_calls(self, expected: list[str]) -> None:
        wait_for(lambda: self.calls() == expected, f"calls {expected}, got {self.calls()}")

    def log(self) -> str:
        return self.daemon_log.read_text()

    def close(self) -> None:
        for proc in self.daemons:
            if proc.poll() is None:
                proc.kill()
                proc.wait()
        for client in self.clients:
            client.close()
        for bus in self.buses:
            bus.stop()


@pytest.fixture
def sb(tmp_path: Path) -> Iterator[Sandbox]:
    sandbox = Sandbox(tmp_path)
    yield sandbox
    sandbox.close()


# --- blocking and unblocking ---------------------------------------------------


def test_blocks_once_and_unblocks_after_last_release(sb: Sandbox) -> None:
    sb.start(*DELAY)
    assert sb.calls() == [ENABLE]  # known state at startup
    a = sb.client()
    first, second = a.inhibit(), a.inhibit()
    assert second == first + 1
    assert sb.calls() == [ENABLE, DISABLE]
    a.uninhibit(first)
    time.sleep(SETTLE)
    assert sb.calls() == [ENABLE, DISABLE]
    a.uninhibit(second)
    sb.wait_calls([ENABLE, DISABLE, ENABLE])


def test_burst_within_release_delay_stays_blocked(sb: Sandbox) -> None:
    sb.start(*DELAY)
    a = sb.client()
    a.uninhibit(a.inhibit())
    a.inhibit()
    time.sleep(SETTLE)
    assert sb.calls() == [ENABLE, DISABLE]


def test_both_object_paths_share_cookies(sb: Sandbox) -> None:
    sb.start(*DELAY)
    a = sb.client()
    cookie = a.inhibit(path="/ScreenSaver")
    assert sb.calls() == [ENABLE, DISABLE]
    a.uninhibit(cookie, path=OBJECT_PATH)
    sb.wait_calls([ENABLE, DISABLE, ENABLE])


def test_ignored_reason_does_not_block(sb: Sandbox) -> None:
    sb.start(*DELAY)
    sb.client().inhibit(reason="Playing audio")
    assert sb.calls() == [ENABLE]
    assert "reason ignored" in sb.log()


def test_count_all_blocks_on_any_reason(sb: Sandbox) -> None:
    sb.start(*DELAY, "--count-all")
    sb.client().inhibit(reason="Playing audio")
    assert sb.calls() == [ENABLE, DISABLE]


def test_uninhibit_from_another_sender_is_ignored(sb: Sandbox) -> None:
    sb.start(*DELAY)
    a, b = sb.client(), sb.client()
    b.uninhibit(a.inhibit())
    time.sleep(SETTLE)
    assert sb.calls() == [ENABLE, DISABLE]
    assert "unknown cookie, ignored" in sb.log()


def test_sender_leaving_the_bus_releases_its_inhibits(sb: Sandbox) -> None:
    sb.start(*DELAY)
    a, b = sb.client(), sb.client()
    a.inhibit()
    a.inhibit()
    b.inhibit(reason="Playing audio")  # another sender's cookie stays
    a.close()
    sb.wait_calls([ENABLE, DISABLE, ENABLE])
    assert sb.log().count("left the bus") == 2


def test_xss_lock_backend_runs_no_locker_commands(sb: Sandbox) -> None:
    sb.set_backend("xss-lock")
    sb.start(*DELAY)
    a = sb.client()
    a.uninhibit(a.inhibit())
    time.sleep(SETTLE)
    assert sb.calls() == []
    assert sb.stop() == 0
    assert sb.calls() == []


@pytest.mark.parametrize(("backend", "command"), [
    ("xautolock", "xautolock -locknow"),
    ("xss-lock", "loginctl lock-session"),
])
def test_lock_and_stub_methods(sb: Sandbox, backend: str, command: str) -> None:
    sb.set_backend(backend)
    sb.start()
    a = sb.client()
    before = sb.calls()
    a.call("Lock")
    assert sb.calls() == [*before, command]
    assert a.call("GetActive", reply="(b)") == (False,)
    assert a.call("GetSessionIdleTime", reply="(u)") == (0,)


def test_stop_unblocks(sb: Sandbox) -> None:
    sb.start(*DELAY)
    sb.client().inhibit()
    assert sb.stop() == 0
    assert sb.calls() == [ENABLE, DISABLE, ENABLE]


# --- restoring after a restart -------------------------------------------------


def test_restart_restores_inhibits(sb: Sandbox) -> None:
    sb.start(*DELAY)
    a = sb.client()
    old = a.inhibit()
    assert sb.stop() == 0
    sb.start(*DELAY)
    assert sb.calls() == [ENABLE, DISABLE, ENABLE, DISABLE]
    new = a.inhibit()
    assert new == old + 1  # numbering continues, so a late UnInhibit can't hit a new cookie
    a.uninhibit(old)
    a.uninhibit(new)
    sb.wait_calls([ENABLE, DISABLE, ENABLE, DISABLE, ENABLE])


def test_restore_skips_senders_that_left(sb: Sandbox) -> None:
    sb.start()
    a, probe = sb.client(), sb.client()
    a.inhibit()
    sb.stop()
    a.close()
    wait_for(lambda: not probe.has_owner(a.name), "the client to leave the bus")
    sb.start()
    assert sb.calls() == [ENABLE, DISABLE, ENABLE, ENABLE]
    assert f"({a.name} left the bus)" in sb.log()


def test_no_restore_outside_window(sb: Sandbox) -> None:
    sb.start()
    sb.client().inhibit()
    sb.stop()
    sb.start("--restore-window", "0")
    assert sb.calls()[-1] == ENABLE
    assert "over the 0s window" in sb.log()


def test_no_restore_from_another_bus(sb: Sandbox) -> None:
    sb.start()
    sb.client().inhibit()
    sb.stop()
    other = sb.new_bus()
    sb.start(bus=other)
    assert sb.calls()[-1] == ENABLE
    assert "saved on another session bus" in sb.log()
    assert sb.client(other).inhibit() == 1  # cookie numbering is per bus


def test_crash_restores_by_mtime(sb: Sandbox) -> None:
    sb.start()
    sb.client().inhibit()
    sb.stop(signal.SIGKILL)
    assert json.loads(sb.state_file.read_text())["stopped_at"] is None
    sb.start()
    assert sb.calls() == [ENABLE, DISABLE, DISABLE]


def test_old_crash_is_not_restored(sb: Sandbox) -> None:
    sb.start()
    sb.client().inhibit()
    sb.stop(signal.SIGKILL)
    past = time.time() - 60
    os.utime(sb.state_file, (past, past))
    sb.start()
    assert sb.calls()[-1] == ENABLE
    assert "over the 30s window" in sb.log()


def test_unreadable_state_file_is_ignored(sb: Sandbox) -> None:
    sb.state_file.write_text("{not json")
    sb.start()
    assert sb.calls() == [ENABLE]
    assert "Ignoring unreadable state file" in sb.log()
    assert json.loads(sb.state_file.read_text())["cookies"] == []


# --- startup and cleanup -------------------------------------------------------


def test_exits_2_if_name_is_owned(sb: Sandbox) -> None:
    other = sb.client()
    do_not_queue = 4
    reply = other.call(
        "RequestName", GLib.Variant("(su)", (BUS_NAME, do_not_queue)), "(u)",
        "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus",
    )
    assert reply == (1,)  # primary owner
    proc = sb.start(wait=False)
    assert proc.wait(TIMEOUT) == 2
    assert sb.calls() == []
    assert not sb.state_file.exists()


@pytest.mark.parametrize(("backend", "xautolock_installed", "expected"), [
    ("xautolock", True, [ENABLE]),
    ("'xss-lock'", True, []),
    (None, True, [ENABLE]),  # both installed: xautolock
    (None, False, []),  # only xss-lock
    ("bogus", False, []),
])
def test_cleanup(sb: Sandbox, backend: str | None, xautolock_installed: bool, expected: list[str]) -> None:
    sb.set_backend(backend)
    if not xautolock_installed:
        (sb.bin / "xautolock").unlink()
    result = subprocess.run(
        [sys.executable, str(SCRIPT), "--cleanup"], env=sb.environ(sb.bus),
        capture_output=True, text=True, timeout=TIMEOUT,
    )
    assert result.returncode == 0
    assert sb.calls() == expected
    if backend == "bogus":
        assert "unknown backend 'bogus'" in result.stdout