"""Test compositor lock confirmation and helper failure when IPC disappears.

Usage: QML2_IMPORT_PATH=<qt5compat>/lib/qt-6/qml python3 scripts/session-lock-check.py <auth-bundle>
Requires qs and sway on PATH. Starts only a private, headless compositor.
"""

import os, subprocess as sp, tempfile, time, shutil, sys
from pathlib import Path

root = Path(tempfile.mkdtemp(prefix="fred-wayland-check."))
root.chmod(0o700)
(root / "sway.conf").write_text(
    "output HEADLESS-1 mode 1280x720\nseat seat0 fallback true\n"
)
env = {
    **os.environ,
    "HOME": str(root),
    "XDG_RUNTIME_DIR": str(root),
    "WLR_BACKENDS": "headless",
    "WLR_RENDERER": "pixman",
    "WLR_LIBINPUT_NO_DEVICES": "1",
    "QT_QUICK_BACKEND": "software",
}
env.pop("WAYLAND_DISPLAY", None)
env.pop("DISPLAY", None)
qs = shutil.which("qs")
assert qs and shutil.which("sway"), "qs and sway must be on PATH"
log = open(root / "sway.log", "w")
sway = sp.Popen(
    [shutil.which("sway"), "-c", str(root / "sway.conf")],
    env=env,
    stdout=log,
    stderr=log,
)
lock = None
try:
    for _ in range(100):
        sockets = [p for p in root.glob("wayland-*") if not str(p).endswith(".lock")]
        if sockets:
            break
        if sway.poll() is not None:
            raise RuntimeError("sway failed")
        time.sleep(0.1)
    assert sockets
    env["WAYLAND_DISPLAY"] = sockets[0].name
    env["QT_QPA_PLATFORM"] = "wayland"
    locklog = open(root / "lock.log", "w")
    lock = sp.Popen(
        [qs, "-p", str(Path(sys.argv[1]) / "lock")],
        env=env,
        stdout=locklog,
        stderr=locklog,
    )
    for _ in range(100):
        result = sp.run(
            [qs, "ipc", "--pid", str(lock.pid), "call", "lock", "secure"],
            env=env,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0 and result.stdout.strip() == "true":
            break
        if lock.poll() is not None:
            raise RuntimeError("lock failed")
        time.sleep(0.1)
    else:
        raise AssertionError(result.stdout + result.stderr)
    print("Wayland compositor confirms secure lock", flush=True)
    mock = root / "bin"
    mock.mkdir()
    (mock / "systemctl").write_text(
        '#!/bin/sh\ncase "$*" in *show*) echo ' + str(lock.pid) + ";; esac\n"
    )
    (mock / "systemctl").chmod(0o700)
    (mock / "qs").symlink_to(qs)
    testenv = {**env, "PATH": str(mock) + ":" + os.environ["PATH"]}
    result = sp.run(
        ["bash", str(Path(__file__).with_name("lock-session.sh"))],
        env=testenv,
        capture_output=True,
        text=True,
        timeout=20,
    )
    assert result.returncode == 0, result.stderr
    print("Lock helper returns success only with secure confirmation", flush=True)
    lock.terminate()
    lock.wait()
    result = sp.run(
        ["bash", str(Path(__file__).with_name("lock-session.sh"))],
        env=testenv,
        capture_output=True,
        text=True,
        timeout=20,
    )
    assert result.returncode != 0 and "did not confirm" in result.stderr, result
    print("Lock helper fails when lock IPC disappears", flush=True)
    (mock / "qs").unlink()
    (mock / "qs").write_text('#!/bin/sh\ncase "$*" in *secure*) echo false;; esac\n')
    (mock / "qs").chmod(0o700)
    result = sp.run(
        ["bash", str(Path(__file__).with_name("lock-session.sh"))],
        env=testenv,
        capture_output=True,
        text=True,
        timeout=20,
    )
    assert result.returncode != 0 and "did not confirm" in result.stderr, result
    print(
        "Lock helper fails when IPC works but the compositor never confirms coverage",
        flush=True,
    )
finally:
    if lock and lock.poll() is None:
        lock.terminate()
        lock.wait()
    sway.terminate()
    sway.wait()
    log.close()
    for name in ["lock.log", "sway.log"]:
        if (root / name).exists():
            print(name, (root / name).read_text()[-4000:])