DroidDeck/tools/tests/test_desktop_open.py

import contextlib
import os
from pathlib import Path
import runpy
import shutil
import struct
import tempfile
import threading
import time
import unittest
from unittest import mock
import zlib

BIN = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin"
OPEN = runpy.run_path(str(BIN / "droiddeck-open"))["main"].__globals__
RUN = runpy.run_path(str(BIN / "droiddeck-proton-run"))["main"].__globals__


def elf(machine, extra=b""):
    head = bytearray(64)
    head[0:4], head[4], head[5], head[18] = b"\x7fELF", 2, 1, machine
    head[8:8 + len(extra)] = extra
    return bytes(head)


class Temp(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())
        self.addCleanup(shutil.rmtree, self.tmp, True)

    def write(self, relative, data, mode=0o644):
        path = self.tmp / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data if isinstance(data, bytes) else data.encode())
        path.chmod(mode)
        return path


class KindTest(Temp):
    def test_each_file_is_told_by_its_content_then_its_name(self):
        cases = {
            "game.x86_64": (elf(62), "elf"),
            "Tool.AppImage": (elf(183, b"AI\x02"), "appimage"),
            "renamed": (elf(62, b"AI\x02"), "appimage"),
            "setup.exe": (b"MZ" + bytes(62), "windows"),
            "launcher.bat": (b"@echo off\r\n", "windows"),
            "no-extension": (b"MZ" + bytes(62), "windows"),
            "d3d11.dll": (b"MZ" + bytes(62), None),
            "start.sh": (b"echo hi\n", "script"),
            "run": (b"#!/usr/bin/env python3\n", "script"),
            "notes.txt": (b"hello\n", None),
        }
        for name, (data, expected) in cases.items():
            with self.subTest(name):
                self.assertEqual(OPEN["kind"](str(self.write(name, data))), expected)

    def test_a_script_names_its_interpreter_or_gets_bash(self):
        self.assertEqual(OPEN["interpreter"](str(self.write("a", b"#!/usr/bin/env  python3 -u\r\nx\n"))),
                         ["/usr/bin/env", "python3", "-u"])
        self.assertEqual(OPEN["interpreter"](str(self.write("b", b"echo\n"))), ["/bin/bash"])


class PlacementTest(Temp):
    def setUp(self):
        super().setUp()
        self.gpu = self.write("droiddeck-gpu", b"#!/bin/sh\n", 0o755)
        patcher = mock.patch.dict(OPEN, {"GPU": str(self.gpu), "unattended": lambda env: False})
        patcher.start()
        self.addCleanup(patcher.stop)
        self.desktop = {"BL_DESKTOP": "1"}

    def placed(self, what, mode, env=None):
        return OPEN["placed"](["cmd"], what, mode, self.desktop if env is None else env)

    def test_programs_from_the_desktop_go_to_the_gpu(self):
        for what in ("windows", "elf", "appimage", "x86 script", "folder"):
            with self.subTest(what):
                self.assertEqual(self.placed(what, "run"), [str(self.gpu), "cmd"])

    def test_a_plain_script_and_a_window_stay_on_the_desktop(self):
        self.assertEqual(self.placed("script", "run"), ["cmd"])
        self.assertEqual(self.placed("windows", "window"), ["cmd"])

    def test_away_from_the_desktop_nothing_moves(self):
        self.assertEqual(self.placed("windows", "run", {}), ["cmd"])
        self.assertEqual(self.placed("windows", "run", dict(self.desktop, GAMESCOPE_WAYLAND_DISPLAY="x")), ["cmd"])

    def test_konsole_takes_what_asks_for_a_terminal(self):
        with mock.patch.object(OPEN["shutil"], "which", return_value="/usr/bin/konsole"):
            self.assertEqual(self.placed("script", "terminal"), ["konsole", "--hold", "-e", "cmd"])
            self.assertEqual(self.placed("windows", "terminal"), [str(self.gpu), "cmd"])

    def test_an_exec_stays_where_its_output_is_read(self):
        self.assertEqual(self.placed("elf", "exec"), ["cmd"])
        self.assertEqual(self.placed("windows", "exec"), [str(self.gpu), "cmd"])
        with mock.patch.dict(OPEN, {"unattended": lambda env: True}):
            self.assertEqual(self.placed("elf", "exec"), [str(self.gpu), "cmd"])


class UnattendedTest(Temp):
    def with_stdout(self, path, env):
        saved = os.dup(1)
        try:
            with open(path, "w") as target:
                os.dup2(target.fileno(), 1)
                return OPEN["unattended"](env)
        finally:
            os.dup2(saved, 1)
            os.close(saved)

    def test_only_output_nobody_reads_counts(self):
        log = self.write("session/desktop.log", b"")
        env = {"BL_DEBUG_DIR": str(log.parent)}
        self.assertTrue(self.with_stdout(os.devnull, env))
        self.assertTrue(self.with_stdout(str(log), env))
        self.assertFalse(self.with_stdout(str(self.tmp / "out.txt"), env))
        self.assertFalse(self.with_stdout(str(log), {}))
        read, write = os.pipe()
        saved = os.dup(1)
        try:
            os.dup2(write, 1)
            self.assertFalse(OPEN["unattended"](env))
        finally:
            os.dup2(saved, 1)
            for fd in (saved, read, write):
                os.close(fd)


class CommandTest(Temp):
    def test_each_kind_gets_its_runner(self):
        exe, script = str(self.tmp / "a.exe"), str(self.tmp / "go.sh")
        self.write("go.sh", b"#!/bin/sh\n")
        self.assertEqual(OPEN["command"](exe, ["-x"], "windows"), ([OPEN["PROTON_RUN"], exe, "-x"], str(self.tmp)))
        self.assertEqual(OPEN["command"](exe, [], "windows", True)[0], [OPEN["PROTON_RUN"], "--window", exe])
        self.assertEqual(OPEN["command"](script, ["1"], "script")[0], ["/bin/sh", script, "1"])
        self.assertEqual(OPEN["command"](script, [], "x86 script")[0],
                         [OPEN["FEX"], "run", "--mode", "on", "--for", str(self.tmp), "--", "/bin/sh", script])

    def test_a_native_program_on_noexec_storage_runs_from_a_copy(self):
        program = self.write("storage/tool", elf(183))
        x86 = self.write("storage/game.x86_64", elf(62))
        with mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "noexec": lambda path: True}):
            copy = OPEN["runnable"](str(program))
            self.assertNotEqual(copy, str(program))
            self.assertEqual(Path(copy).read_bytes(), elf(183))
            self.assertTrue(os.access(copy, os.X_OK))
            self.assertEqual(OPEN["runnable"](str(program)), copy)
            self.assertEqual(OPEN["runnable"](str(x86)), str(x86))
        with mock.patch.dict(OPEN, {"noexec": lambda path: False}):
            self.assertEqual(OPEN["runnable"](str(program)), str(program))


class InstallTest(Temp):
    def test_the_run_entries_and_defaults_are_written_once(self):
        OPEN["install"](str(self.tmp))
        entry = (self.tmp / "droiddeck-open.desktop").read_text()
        self.assertIn("Exec=/usr/local/bin/droiddeck-open %f", entry)
        self.assertIn("application/x-ms-dos-executable", entry)
        self.assertIn("Exec=/usr/local/bin/droiddeck-open --terminal %f",
                      (self.tmp / "droiddeck-open-terminal.desktop").read_text())
        defaults = (self.tmp / "mimeapps.list").read_text()
        self.assertIn("application/x-shellscript=droiddeck-open.desktop\n", defaults)
        self.assertIn("application/x-msi=droiddeck-open.desktop;droiddeck-open-window.desktop;\n", defaults)
        stamp = (self.tmp / "mimeapps.list").stat().st_mtime_ns
        time.sleep(0.01)
        OPEN["install"](str(self.tmp))
        self.assertEqual((self.tmp / "mimeapps.list").stat().st_mtime_ns, stamp)


class ImportTest(Temp):
    def test_the_app_unpacks_what_the_desktop_cannot(self):
        launch = self.tmp / "launch"
        launch.mkdir()
        folder = self.tmp / "imported/eden"
        (folder / "app").mkdir(parents=True)

        def app():
            request = launch / "desktop-import-0123456789abcdef.request"
            while not request.exists():
                time.sleep(0.05)
            text = request.read_text()
            request.unlink()
            self.assertEqual(text, "id=0123456789abcdef\nkind=appimage\npath=/root/Storage/Eden.AppImage\n")
            (launch / "desktop-import-0123456789abcdef.result").write_text("ok %s\n" % folder)

        worker = threading.Thread(target=app)
        worker.start()
        with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), mock.patch.dict(OPEN, {"notify": lambda *a: None}):
            self.assertEqual(OPEN["imported"]("appimage", "/root/Storage/Eden.AppImage", "0123456789abcdef", "Eden"), str(folder))
        worker.join()
        self.assertEqual(list(launch.iterdir()), [])

    def test_the_apps_answer_is_the_error(self):
        launch = self.tmp / "launch"
        launch.mkdir()
        (launch / "desktop-import-0123456789abcdef.result").write_text("error Not enough space\n")
        with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), mock.patch.dict(OPEN, {"notify": lambda *a: None}):
            with self.assertRaisesRegex(RuntimeError, "^Not enough space$"):
                OPEN["imported"]("appimage", "/x.AppImage", "0123456789abcdef", "x")

    def test_outside_a_session_it_says_how_to_add_one(self):
        with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": ""}):
            with self.assertRaisesRegex(RuntimeError, "Add an app"):
                OPEN["imported"]("folder", "/x/start.sh", "0123456789abcdef", "x")

    def test_starts_of_one_image_unpack_it_once(self):
        image = self.write("Tool.AppImage", elf(183, b"AI\x02"))
        unpacks = []

        def extract(cmd, cwd, stdout):
            unpacks.append(cwd)
            time.sleep(0.2)
            os.makedirs(os.path.join(cwd, "squashfs-root"))
            open(os.path.join(cwd, "squashfs-root", "AppRun"), "w").close()
            return 0

        found = []
        none = {what: (str(self.tmp / "none"), made) for what, (_, made) in OPEN["IMPORTS"].items()}
        with mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "IMPORTS": none, "notify": lambda *a: None}), \
                mock.patch.object(OPEN["subprocess"], "call", extract):
            starts = [threading.Thread(target=lambda: found.append(OPEN["appimage"](str(image)))) for _ in range(3)]
            for start in starts:
                start.start()
            for start in starts:
                start.join()
        self.assertEqual(len(unpacks), 1)
        self.assertEqual(len(set(found)), 1)
        self.assertTrue(os.path.isfile(os.path.join(found[0], "app", "AppRun")))
        self.assertEqual(Path(found[0], "arch").read_text(), "arm64\n")

    def test_an_earlier_import_is_found_by_its_key(self):
        self.write("user/eden/desktop-import", "0123456789abcdef\n/root/Storage/Eden.AppImage")
        (self.tmp / "user/eden/app").mkdir()
        self.write("user/other/desktop-import", "fedcba9876543210\n/root/Storage/Other.AppImage")
        self.write("apps/game/desktop-import", "fedcba9876543210\n/root/Storage/Game/start.sh")
        self.write("apps/game/entry", "start.sh")
        imports = {"appimage": (str(self.tmp / "user"), "app"), "folder": (str(self.tmp / "apps"), "entry")}
        with mock.patch.dict(OPEN, {"IMPORTS": imports}):
            self.assertEqual(OPEN["import_of"]("0123456789abcdef", "appimage"), str(self.tmp / "user/eden"))
            self.assertIsNone(OPEN["import_of"]("fedcba9876543210", "appimage"))
            self.assertEqual(OPEN["import_of"]("fedcba9876543210", "folder"), str(self.tmp / "apps/game"))


class GameFolderTest(Temp):
    def setUp(self):
        super().setUp()
        self.storage = self.tmp / "storage"
        patcher = mock.patch.dict(OPEN, {"noexec": lambda path: True, "storage_root": lambda folder: str(self.storage)})
        patcher.start()
        self.addCleanup(patcher.stop)

    def folder(self, path):
        return OPEN["game_folder"](str(self.storage / path))

    def test_a_native_games_folder_on_noexec_storage_is_copied_in(self):
        self.write("storage/Download/Game/start.sh", b"#!/bin/sh\n./bin/game.aarch64\n")
        self.write("storage/Download/Game/bin/game.aarch64", elf(183))
        self.write("storage/Download/Godot/Game.arm64", elf(183))
        self.write("storage/Download/Godot/libgodot.so", elf(183))
        self.write("storage/MyGame/start.sh", b"./game\n")
        self.write("storage/MyGame/game", elf(183))
        self.assertEqual(self.folder("Download/Game/start.sh"), str(self.storage / "Download/Game"))
        self.assertEqual(self.folder("Download/Godot/Game.arm64"), str(self.storage / "Download/Godot"))
        self.assertEqual(self.folder("MyGame/start.sh"), str(self.storage / "MyGame"))

    def test_x86_games_run_where_they_lie(self):
        self.write("storage/Download/Game/start.sh", b"#!/bin/sh\n./bin/game.x86_64\n")
        self.write("storage/Download/Game/bin/game.x86_64", elf(62))
        self.write("storage/Download/Unity/Game.x86_64", elf(62))
        self.write("storage/Download/Unity/UnityPlayer.so", elf(62))
        self.write("storage/Download/Old/game", elf(3))
        for path in ("Download/Game/start.sh", "Download/Unity/Game.x86_64", "Download/Old/game"):
            with self.subTest(path):
                self.assertIsNone(self.folder(path))

    def test_what_runs_where_it_lies_is_left_there(self):
        self.write("storage/Download/Scripts/hello.sh", b"echo hi\n")
        self.write("storage/Download/Tool/tool", elf(183))
        self.write("storage/Download/start.sh", b"./Tool/tool\n")
        self.write("storage/start.sh", b"./Download/Tool/tool\n")
        self.write("storage/Download/Deep/a/b/c/d/lib.so", elf(183))
        self.write("storage/Download/Deep/start.sh", b"echo\n")
        for path in ("Download/Scripts/hello.sh", "Download/Tool/tool", "Download/start.sh", "start.sh", "Download/Deep/start.sh"):
            with self.subTest(path):
                self.assertIsNone(self.folder(path))
        self.write("internal/Game/start.sh", b"./game\n")
        self.write("internal/Game/game", elf(183))
        with mock.patch.dict(OPEN, {"noexec": lambda path: False}):
            self.assertIsNone(OPEN["game_folder"](str(self.tmp / "internal/Game/start.sh")))

    def test_the_app_copies_the_folder_in_once(self):
        script = self.write("storage/Download/Game/start.sh", b"./game\n")
        launch = self.tmp / "launch"
        launch.mkdir()
        copy = self.tmp / "apps/game"
        copy.mkdir(parents=True)
        (copy / "entry").write_text("start.sh")
        requests = []

        def app():
            while not requests:
                for request in launch.glob("desktop-import-*.request"):
                    requests.append(request.read_text())
                    request.unlink()
                    (launch / request.name.replace(".request", ".result")).write_text("ok %s\n" % copy)
                time.sleep(0.05)

        worker = threading.Thread(target=app)
        worker.start()
        imports = {what: (str(self.tmp / "none"), made) for what, (_, made) in OPEN["IMPORTS"].items()}
        with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), \
                mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "IMPORTS": imports, "notify": lambda *a: None}):
            self.assertEqual(OPEN["command"](str(script), ["-x"], "folder"), ([OPEN["SCRIPT_RUN"], str(copy), "-x"], str(script.parent)))
        worker.join()
        self.assertRegex(requests[0], "^id=[0-9a-f]{16}\nkind=folder\npath=%s\n$" % script)


class ExecTest(Temp):
    def setUp(self):
        super().setUp()
        self.storage = self.tmp / "storage"
        self.write("storage/Download/Game/start.sh", b"#!/bin/sh\n./game.x86_64\n", 0o755)
        self.write("storage/Download/Game/game.x86_64", elf(62), 0o755)
        self.write("storage/Download/Game/helper.bin", elf(183), 0o755)
        self.write("storage/Download/Game/libhelper.so", elf(183))
        self.copied = []
        env = {k: v for k, v in os.environ.items() if k not in ("BL_DESKTOP", "BL_BINFMT", "GAMESCOPE_WAYLAND_DISPLAY")}
        for patcher in (mock.patch.dict(OPEN, {"noexec": lambda path: True, "storage_root": lambda folder: str(self.storage),
                                               "copied_in": lambda path: self.copied.append(path) or "/opt/droiddeck-apps/game",
                                               "steam_title": lambda path: None, "notify": lambda *a: None,
                                               "CACHE": str(self.tmp / "cache")}),
                        mock.patch.dict(os.environ, env, clear=True),
                        mock.patch.object(OPEN["os"], "chdir"),
                        mock.patch.object(OPEN["os"], "execv", side_effect=self.execv),
                        mock.patch.object(OPEN["subprocess"], "call", side_effect=self.call)):
            patcher.start()
            self.addCleanup(patcher.stop)
        self.ran = []

    def execv(self, program, cmd):
        self.ran.append(cmd)
        raise SystemExit(0)

    def call(self, cmd, *args, **kwargs):
        self.ran.append(cmd)
        return 0

    def open(self, *argv, env=None):
        os.environ.update(env or {})
        with contextlib.suppress(SystemExit):
            OPEN["main"](list(argv))
        return self.ran[-1]

    def test_an_x86_program_a_running_program_starts_runs_under_fex_where_it_lies(self):
        program = str(self.storage / "Download/Game/game.x86_64")
        for env in ({}, {"BL_BINFMT": "1"}):
            with self.subTest(env=env):
                self.assertEqual(self.open("--exec", program, "-x", env=env), [OPEN["FEX"], "run", "--mode", "auto", "--", program, "-x"])
        self.assertEqual(self.copied, [])

    def test_an_in_place_hand_off_keeps_the_callers_working_folder(self):
        program = str(self.storage / "Download/Game/game.x86_64")
        with mock.patch("os.chdir") as chdir:
            self.open("--exec", program, env={"BL_BINFMT": "1"})
            chdir.assert_not_called()
            self.open("--exec", str(self.storage / "Download/Game/helper.bin"))
            chdir.assert_called_once()

    def test_a_native_program_a_running_program_starts_still_gets_its_folder_copied_in(self):
        program = str(self.storage / "Download/Game/helper.bin")
        self.assertEqual(self.open("--exec", program, "-y"), [OPEN["SCRIPT_RUN"], "/opt/droiddeck-apps/game", "-y"])
        self.assertEqual(self.copied, [program])

    def test_opening_an_x86_program_from_the_desktop_still_copies_its_folder_in(self):
        program = str(self.storage / "Download/Game/game.x86_64")
        self.assertEqual(self.open(program, env={"BL_DESKTOP": "1"}), [OPEN["SCRIPT_RUN"], "/opt/droiddeck-apps/game"])
        self.assertEqual(self.copied, [program])


def appinfo(apps):
    strings = []

    def key(name):
        if name not in strings:
            strings.append(name)
        return struct.pack("<I", strings.index(name))

    def section(items):
        out = b""
        for name, value in items.items():
            if isinstance(value, dict):
                out += b"\x00" + key(name) + section(value)
            else:
                out += b"\x01" + key(name) + value.encode() + b"\0"
        return out + b"\x08"

    body = b""
    for app, info in apps.items():
        data = section({"appinfo": dict(info, appid=str(app))})
        header = struct.pack("<IIQ", 2, 0, 0) + bytes(20) + struct.pack("<I", 1) + bytes(20)
        body += struct.pack("<II", app, len(header) + len(data)) + header + data
    body += struct.pack("<I", 0)
    table = struct.pack("<I", len(strings)) + b"".join(s.encode() + b"\0" for s in strings)
    return struct.pack("<IIq", 0x07564429, 1, 16 + len(body)) + body + table


STARDEW = {"common": {"type": "Game", "oslist": "windows,macos,linux"}, "config": {"launch": {
    "0": {"executable": "Stardew Valley.exe", "type": "default", "config": {"oslist": "windows"}},
    "1": {"executable": "StardewValley", "type": "default", "config": {"oslist": "linux"}},
    "2": {"executable": ".\\Tools\\Setup.exe", "type": "option1"},
}}}


TOOL_VDF = '''"compatibilitytools"
{
  "compat_tools"
  {
    "%s"
    {
      "install_path" "."
      "display_name" "%s"
      "from_oslist" "%s"
      "to_oslist" "linux"
    }
  }
}
'''

CONFIG = '''"InstallConfigStore"
{
  "Software"
  {
    "Valve"
    {
      "Steam"
      {
        "CompatToolMapping"
        {
          "0"
          {
            "name" "droiddeck-proton-arm64"
            "config" ""
            "priority" "75"
          }
          "413150"
          {
            "name" "droiddeck-proton-11-arm64"
            "config" ""
            "priority" "250"
          }
        }
      }
    }
  }
}
'''


class ProtonRunTest(Temp):
    def setUp(self):
        super().setUp()
        self.root = self.tmp / "Steam"
        for name, oslist in (("droiddeck-proton-arm64", "windows"), ("droiddeck-proton-11-arm64", "windows"),
                             ("droiddeck-fex", "linux")):
            folder = "compatibilitytools.d/%s" % name
            self.write("Steam/%s/compatibilitytool.vdf" % folder, TOOL_VDF % (name, name, oslist))
            self.write("Steam/%s/toolmanifest.vdf" % folder, '"manifest"\n{\n  "commandline" "/droiddeck-proton %verb%"\n}\n')
            self.write("Steam/%s/droiddeck-proton" % folder, b"#!/bin/sh\n", 0o755)
        self.write("Steam/config/config.vdf", CONFIG)
        patcher = mock.patch.dict(RUN, {"EXTRA_LIBRARIES": (), "SYSTEM_TOOLS": (), "STATE": str(self.tmp / "state.json")})
        patcher.start()
        self.addCleanup(patcher.stop)

    def game(self, appid, installdir, relative):
        self.write("Steam/steamapps/appmanifest_%s.acf" % appid, '"AppState"\n{\n  "appid" "%s"\n  "installdir" "%s"\n}\n'
                   % (appid, installdir))
        return self.write("Steam/steamapps/common/%s/%s" % (installdir, relative), b"MZ")

    def test_only_windows_tools_that_can_start_are_offered(self):
        found = RUN["tools"](str(self.root))
        self.assertEqual(sorted(found), ["droiddeck-proton-11-arm64", "droiddeck-proton-arm64"])
        self.assertEqual(found["droiddeck-proton-arm64"],
                         (str(self.root / "compatibilitytools.d/droiddeck-proton-arm64"), "droiddeck-proton"))

    def test_a_valve_depot_is_registered_through_the_droiddeck_wrapper(self):
        name = "Proton Experimental (ARM64)"
        folder = "steamapps/common/%s" % name
        shutil.rmtree(self.root / "compatibilitytools.d/droiddeck-proton-arm64")
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        self.write("Steam/%s/toolmanifest.vdf" % folder, '"manifest"\n{\n  "commandline" "/proton %verb%"\n}\n')
        self.write("Steam/%s/proton" % folder, b"#!/bin/sh\n", 0o755)
        self.write("Steam/%s/files/bin-arm64/proton" % folder, b"#!/bin/sh\n", 0o755)
        found = RUN["tools"](str(self.root))
        wrapper = self.root / "compatibilitytools.d/droiddeck-proton-arm64"
        self.assertEqual(found["droiddeck-proton-arm64"], (str(wrapper), "droiddeck-proton"))
        self.assertNotIn(name, found)

    def test_an_incomplete_valve_depot_is_not_registered(self):
        name = "Proton Experimental (ARM64)"
        folder = "steamapps/common/%s" % name
        shutil.rmtree(self.root / "compatibilitytools.d/droiddeck-proton-arm64")
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        self.write("Steam/%s/toolmanifest.vdf" % folder, '"manifest"\n{\n  "commandline" "/proton %verb%"\n}\n')
        self.write("Steam/%s/files/bin-arm64/proton" % folder, b"#!/bin/sh\n", 0o755)
        found = RUN["tools"](str(self.root))
        self.assertNotIn("droiddeck-proton-arm64", found)
        self.assertFalse((self.root / "compatibilitytools.d/droiddeck-proton-arm64").exists())

    def test_a_failed_wrapper_build_does_not_hide_existing_tools(self):
        shutil.rmtree(self.root / "compatibilitytools.d/droiddeck-proton-arm64")
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        folder = "steamapps/common/Proton Experimental (ARM64)"
        self.write("Steam/%s/proton" % folder, b"#!/bin/sh\n", 0o755)
        self.write("Steam/%s/files/bin-arm64/proton" % folder, b"#!/bin/sh\n", 0o755)
        build = mock.Mock(side_effect=OSError("read-only"))
        with mock.patch.dict(RUN["compat"](), {"build_tool": build}):
            found = RUN["tools"](str(self.root))
        build.assert_called_once()
        self.assertIn("droiddeck-proton-11-arm64", found)

    def test_a_runnable_depot_is_chosen_over_a_newer_incomplete_one(self):
        shutil.rmtree(self.root / "compatibilitytools.d/droiddeck-proton-arm64")
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        good = "Steam/steamapps/common/Proton Experimental (ARM64)"
        self.write("%s/proton" % good, b"#!/bin/sh\n", 0o755)
        self.write("%s/files/bin-arm64/proton" % good, b"#!/bin/sh\n", 0o755)
        broken = "Library/steamapps/common/Proton Experimental (ARM64)"
        self.write("Library/steamapps/libraryfolders.vdf", b"")
        self.write("%s/version" % broken, b"2\n")
        self.write("%s/files/bin-arm64/proton" % broken, b"#!/bin/sh\n", 0o755)
        self.write("Steam/steamapps/libraryfolders.vdf", '"path" "%s"' % (self.tmp / "Library"))
        compat = RUN["compat"]()
        source = compat["find_source"]
        self.assertEqual(source(str(self.root), runnable=True), str(self.tmp / good))

    def test_a_registered_wrapper_without_its_launcher_is_rebuilt(self):
        wrapper = self.root / "compatibilitytools.d/droiddeck-proton-arm64"
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        folder = "Steam/steamapps/common/Proton Experimental (ARM64)"
        self.write("%s/proton" % folder, b"#!/bin/sh\n", 0o755)
        self.write("%s/files/bin-arm64/proton" % folder, b"#!/bin/sh\n", 0o755)
        launcher = wrapper / RUN["compat"]()["LAUNCHER"]
        if launcher.exists():
            launcher.unlink()
        found = RUN["tools"](str(self.root))
        self.assertTrue(os.access(launcher, os.X_OK))
        self.assertIn("droiddeck-proton-arm64", found)

    def test_a_registered_wrapper_without_its_manifest_is_rebuilt(self):
        wrapper = self.root / "compatibilitytools.d/droiddeck-proton-arm64"
        self.write("Steam/steamapps/appmanifest_4427310.acf", '"AppState" { "appid" "4427310" }')
        folder = "Steam/steamapps/common/Proton Experimental (ARM64)"
        self.write("%s/proton" % folder, b"#!/bin/sh\n", 0o755)
        self.write("%s/files/bin-arm64/proton" % folder, b"#!/bin/sh\n", 0o755)
        manifest = wrapper / "toolmanifest.vdf"
        if manifest.exists():
            manifest.unlink()
        found = RUN["tools"](str(self.root))
        self.assertTrue(manifest.is_file())
        self.assertIn("droiddeck-proton-arm64", found)

    def test_steams_choice_for_the_title_then_its_default(self):
        available = RUN["tools"](str(self.root))
        current = RUN["mapping"](str(self.root))
        self.assertEqual(RUN["choose_tool"]("413150", current, available), ("droiddeck-proton-11-arm64", "Steam's choice for this title"))
        self.assertEqual(RUN["choose_tool"]("70", current, available), ("droiddeck-proton-arm64", "Steam's default"))
        self.assertEqual(RUN["choose_tool"]("70", {"0": "GE-Proton"}, available), ("droiddeck-proton-arm64", "the default ARM64 Proton"))
        self.assertEqual(RUN["choose_tool"]("70", {}, {}), (None, None))

    def test_a_removed_droiddeck_proton_build_is_replaced_by_the_newest_one(self):
        for version in ("11-12.2", "11-12.3"):
            folder = "compatibilitytools.d/DroidDeck-Proton %s" % version
            self.write("Steam/%s/compatibilitytool.vdf" % folder, TOOL_VDF % ("DroidDeck-Proton-" + version, version, "windows"))
            self.write("Steam/%s/toolmanifest.vdf" % folder, '"manifest"\n{\n  "commandline" "/proton %verb%"\n}\n')
            self.write("Steam/%s/proton" % folder, b"#!/bin/sh\n", 0o755)
        available = RUN["tools"](str(self.root))
        newest = "DroidDeck-Proton-11-12.3"
        self.assertEqual(RUN["choose_tool"]("70", {"0": "DroidDeck-Proton-11-12.1"}, available)[0], newest)
        self.assertEqual(RUN["choose_tool"]("70", {"70": "DroidDeck-Proton-11-12.1", "0": "droiddeck-proton-arm64"}, available)[0], newest)
        self.assertEqual(RUN["choose_tool"]("70", {"0": "DroidDeck-Proton-11-12.2"}, available), ("DroidDeck-Proton-11-12.2", "Steam's default"))
        self.assertEqual(RUN["choose_tool"]("70", {"0": "GE-Proton"}, available), ("droiddeck-proton-arm64", "the default ARM64 Proton"))
        (self.tmp / "state.json").write_text('{"default": "DroidDeck-Proton-11-12.1"}')
        self.assertEqual(RUN["choose_tool"]("70", {}, available)[0], newest)
        (self.tmp / "state.json").write_text('{"default": "droiddeck-proton-11-arm64"}')
        self.assertEqual(RUN["choose_tool"]("70", {}, available), ("droiddeck-proton-11-arm64", "the default ARM64 Proton"))

    def test_a_steam_games_program_uses_its_prefix(self):
        program = self.game("413150", "Stardew Valley", "Stardew Valley.exe")
        target = RUN["locate"](str(program), str(self.root))
        self.assertEqual((target.appid, target.kind, target.compatdata, target.install),
                         ("413150", "steam", str(self.root / "steamapps/compatdata/413150"),
                          str(self.root / "steamapps/common/Stardew Valley")))

    def test_a_tool_in_the_library_is_not_a_game(self):
        program = self.game("1493710", "Proton - Experimental", "files/bin/wine.exe")
        self.assertEqual(RUN["locate"](str(program), str(self.root)).kind, "own")

    def test_a_program_in_a_prefix_uses_that_prefix(self):
        program = self.write("Steam/steamapps/compatdata/2870/pfx/drive_c/Setup/setup.exe", b"MZ")
        target = RUN["locate"](str(program), str(self.root))
        self.assertEqual((target.appid, target.kind, target.compatdata),
                         ("2870", "prefix", str(self.root / "steamapps/compatdata/2870")))

    def test_any_other_program_gets_the_id_steam_would_give_its_shortcut(self):
        program = str(self.write("Download/winemine.exe", b"MZ"))
        target = RUN["locate"](program, str(self.root))
        appid = zlib.crc32(('"%s"winemine' % program).encode()) | 0x80000000
        self.assertEqual((target.appid, target.kind), (str(appid), "own"))
        env = RUN["environment"]({}, target, "/tool", str(self.root), window=True)
        self.assertEqual((env["SteamAppId"], env["STEAM_COMPAT_APP_ID"], env["SteamGameId"], env["PROTON_USE_WINED3D"]),
                         ("0", "0", str(appid << 32 | 0x02000000), "1"))
        self.assertEqual(env["STEAM_COMPAT_DATA_PATH"], str(self.root / ("steamapps/compatdata/%d" % appid)))

    def test_a_window_draws_with_the_desktops_software_gl(self):
        target = RUN["Target"]("2596113170", "/c", "/i", str(self.root), "fixed")
        session = {"MESA_LOADER_DRIVER_OVERRIDE": "zink", "GALLIUM_DRIVER": "zink", "LIBGL_KOPPER_DRI2": "true"}
        env = RUN["environment"](session, target, "/tool", str(self.root), window=True)
        self.assertEqual((env["LIBGL_ALWAYS_SOFTWARE"], env["GALLIUM_DRIVER"]), ("1", "llvmpipe"))
        self.assertNotIn("MESA_LOADER_DRIVER_OVERRIDE", env)
        self.assertNotIn("LIBGL_KOPPER_DRI2", env)
        self.assertEqual(RUN["environment"](session, target, "/tool", str(self.root))["GALLIUM_DRIVER"], "zink")

    def test_a_steam_game_keeps_its_own_id(self):
        target = RUN["Target"]("413150", "/c", "/i", str(self.root), "steam")
        env = RUN["environment"]({"PROTON_USE_WINED3D": "0"}, target, "/tool", str(self.root), window=True)
        self.assertEqual((env["SteamAppId"], env["SteamGameId"], env["PROTON_USE_WINED3D"]), ("413150", "413150", "0"))

    def test_an_installer_package_goes_to_msiexec(self):
        self.assertEqual(RUN["arguments"]("/root/x/setup.msi", ["/qn"]), ["msiexec", "/i", "Z:\\root\\x\\setup.msi", "/qn"])
        self.assertEqual(RUN["arguments"]("/root/x/a.exe", ["-w"]), ["/root/x/a.exe", "-w"])

    def test_a_virtual_desktop_runs_the_program_by_its_windows_path(self):
        pfx = "/root/Steam/steamapps/compatdata/2596113170"
        self.assertEqual(RUN["arguments"](pfx + "/pfx/drive_c/Program Files (x86)/Battle.net/Battle.net Launcher.exe", [], "1280x720", pfx),
                         ["explorer.exe", "/desktop=droiddeck,1280x720", "C:\\Program Files (x86)\\Battle.net\\Battle.net Launcher.exe"])
        self.assertEqual(RUN["arguments"]("/root/.cache/setup.exe", ["--lang=enUS"], "1920x1080", pfx),
                         ["explorer.exe", "/desktop=droiddeck,1920x1080", "Z:\\root\\.cache\\setup.exe", "--lang=enUS"])

    def test_a_games_own_program_is_left_to_steam(self):
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
        self.assertEqual(RUN["compat"]()["launch_programs"](STARDEW["config"]),
                         ["stardew valley.exe", "stardewvalley", "tools/setup.exe"])
        root = str(self.root)
        for relative, expected in (("Stardew Valley.exe", True), ("StardewValley", True), ("Tools/Setup.exe", True),
                                   ("createdump.exe", False)):
            with self.subTest(relative):
                program = str(self.game("413150", "Stardew Valley", relative))
                self.assertEqual(RUN["steam_launches"](RUN["locate"](program, root), program, root), expected)
        own = str(self.write("Download/Stardew Valley.exe", b"MZ"))
        self.assertFalse(RUN["steam_launches"](RUN["locate"](own, root), own, root))

    def test_the_desktop_hands_a_steam_game_to_steam(self):
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
        program = str(self.game("413150", "Stardew Valley", "Stardew Valley.exe"))
        with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
                mock.patch.dict(OPEN, {"PROTON_RUN": str(BIN / "droiddeck-proton-run")}):
            appid = OPEN["steam_title"](program)
            self.assertEqual(appid, "413150")
            self.assertEqual(OPEN["command"](program, [], "steam", appid=appid)[0], [OPEN["STEAM_LAUNCH"], "desktop", "413150"])
            self.assertEqual(OPEN["placed"](["launch"], "steam", "run", {"BL_DESKTOP": "1"}), ["launch"])
            self.assertIsNone(OPEN["steam_title"](str(self.write("Download/winemine.exe", b"MZ"))))

    def test_a_linux_steam_games_program_goes_to_steam_too(self):
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
        program = self.game("413150", "Stardew Valley", "StardewValley")
        program.write_bytes(elf(62))
        self.write("Steam/steamapps/common/Stardew Valley/lib64/libSDL2.so", elf(62))
        ran = []
        with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
                mock.patch.dict(OPEN, {"PROTON_RUN": str(BIN / "droiddeck-proton-run"), "noexec": lambda path: True,
                                       "placed": lambda cmd, *a: cmd, "say": lambda *a: None}), \
                mock.patch.object(OPEN["subprocess"], "call", lambda cmd: ran.append(cmd) or 0), contextlib.chdir(self.tmp):
            self.assertEqual(OPEN["main"]([str(program)]), 0)
            self.assertIsNone(OPEN["game_folder"](str(program)))
        self.assertEqual(ran, [[OPEN["STEAM_LAUNCH"], "desktop", "413150"]])

    def test_the_tool_is_started_as_steam_starts_it(self):
        program = self.game("413150", "Stardew Valley", "Stardew Valley.exe")
        calls = []
        with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
                mock.patch.object(RUN["os"], "execve", lambda *a: calls.append(a)), \
                mock.patch.object(RUN["os"], "chdir"), mock.patch.dict(RUN, {"say": lambda message: None}):
            RUN["main"]([str(program), "-fullscreen"])
        entry, argv, env = calls[0]
        self.assertEqual(entry, str(self.root / "compatibilitytools.d/droiddeck-proton-11-arm64/droiddeck-proton"))
        self.assertEqual(argv, [entry, "waitforexitandrun", str(program), "-fullscreen"])
        self.assertEqual(env["SteamAppId"], "413150")
        self.assertTrue((self.root / "steamapps/compatdata/413150").is_dir())

    def run_main(self, argv):
        calls, dirs = [], []
        with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
                mock.patch.object(RUN["os"], "execve", lambda *a: calls.append(a)), \
                mock.patch.object(RUN["os"], "chdir", dirs.append), mock.patch.dict(RUN, {"say": lambda message: None}):
            code = RUN["main"](argv)
        return code, calls, dirs

    def test_without_steam_a_steam_game_starts_its_default_windows_launch(self):
        info = {"common": {"type": "Game", "oslist": "windows,linux"}, "config": {"launch": {
            "0": {"executable": "Bin\\Game.exe", "arguments": "-dx11 \"-name=Two Words\"", "workingdir": "bin",
                  "type": "default", "config": {"oslist": "windows"}},
            "1": {"executable": "game.sh", "type": "default", "config": {"oslist": "linux"}},
            "2": {"executable": "Beta.exe", "type": "default", "config": {"oslist": "windows", "betakey": "beta"}},
            "3": {"executable": "Tools\\Editor.exe", "type": "option1"},
        }}}
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: info}))
        program = self.game("413150", "Some Game", "bin/game.exe")
        self.game("413150", "Some Game", "Tools/Editor.exe")
        code, calls, dirs = self.run_main(["--app", "413150", "--options", "-windowed"])
        self.assertIsNone(code)
        entry, argv, env = calls[0]
        self.assertEqual(argv[1:], ["waitforexitandrun", str(program), "-dx11", "-name=Two Words", "-windowed"])
        self.assertEqual((env["SteamAppId"], env["STEAM_COMPAT_DATA_PATH"]),
                         ("413150", str(self.root / "steamapps/compatdata/413150")))
        self.assertEqual(dirs, [str(program.parent)])

    def test_without_steam_a_game_steam_cannot_start_is_refused(self):
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
        self.assertEqual(self.run_main(["--app", "413150"])[0], 2)
        self.assertEqual(self.run_main(["--app", "3384071994"])[0], 2)
        self.assertEqual(self.run_main(["--app", "0"])[0], 2)
        self.assertEqual(self.run_main(["--app", "game"])[0], 2)
        linux = {"common": {"type": "Game"}, "config": {"launch": {"0": {"executable": "run.sh", "config": {"oslist": "linux"}}}}}
        self.write("Steam/appcache/appinfo.vdf", appinfo({70: linux}))
        self.game("70", "Linux Game", "run.sh")
        self.assertEqual(self.run_main(["--app", "70"])[0], 2)

    def test_without_steam_a_shortcut_reuses_a_signed_prefix_folder(self):
        bat = self.write("Games/Epic/Game/.droiddeck-launch.bat", b"@echo off")
        start = self.tmp / "Games/Epic/Game/Binaries"
        start.mkdir()
        signed = self.root / "steamapps/compatdata/-1561598272"
        signed.mkdir(parents=True)
        code, calls, dirs = self.run_main(["--app", "2733369024", "--dir", str(start), "--", str(bat)])
        self.assertIsNone(code)
        env = calls[0][2]
        self.assertEqual(env["STEAM_COMPAT_DATA_PATH"], str(signed))
        self.assertFalse((self.root / "steamapps/compatdata/2733369024").exists())

    def test_prefix_folder_prefers_initialized_unsigned_over_empty_signed(self):
        signed = self.root / "steamapps/compatdata/-1561598272"
        signed.mkdir(parents=True)
        initialized = self.root / "steamapps/compatdata/2733369024/pfx"
        initialized.mkdir(parents=True)

        self.assertEqual(RUN["prefix_folder"](str(self.root), 2733369024), str(initialized.parent))

    def test_prefix_folder_prefers_initialized_signed_over_empty_unsigned(self):
        unsigned = self.root / "steamapps/compatdata/2733369024"
        unsigned.mkdir(parents=True)
        initialized = self.root / "steamapps/compatdata/-1561598272/pfx"
        initialized.mkdir(parents=True)

        self.assertEqual(RUN["prefix_folder"](str(self.root), 2733369024), str(initialized.parent))

    def test_prefix_folder_prefers_unsigned_when_initialization_ties(self):
        unsigned = self.root / "steamapps/compatdata/2733369024/pfx"
        unsigned.mkdir(parents=True)
        signed = self.root / "steamapps/compatdata/-1561598272/pfx"
        signed.mkdir(parents=True)

        self.assertEqual(RUN["prefix_folder"](str(self.root), 2733369024), str(unsigned.parent))

    def test_prefix_folder_skips_numeric_names_too_long_to_convert(self):
        from unittest.mock import patch
        folder = self.root / "steamapps/compatdata"
        folder.mkdir(parents=True)
        with patch.object(RUN["os"], "listdir", return_value=["9" * 5000]):
            self.assertEqual(RUN["prefix_folder"](str(self.root), 2733369024),
                             str(folder / "2733369024"))

    def test_without_steam_a_shortcut_keeps_its_prefix_folder_and_launch_options(self):
        bat = self.write("Games/Epic/Game/.droiddeck-launch.bat", b"@echo off")
        start = self.tmp / "Games/Epic/Game/Binaries"
        start.mkdir()
        code, calls, dirs = self.run_main(["--app", "2733369024", "--dir", str(start), "--options",
                                           "PROTON_LOG=1 %command% -epicportal", "--", str(bat)])
        self.assertIsNone(code)
        entry, argv, env = calls[0]
        self.assertEqual(argv[1:], ["waitforexitandrun", str(bat), "-epicportal"])
        self.assertEqual((env["STEAM_COMPAT_DATA_PATH"], env["SteamAppId"], env["SteamGameId"], env["PROTON_LOG"]),
                         (str(self.root / "steamapps/compatdata/2733369024"), "0", str(2733369024 << 32 | 0x02000000), "1"))
        self.assertTrue((self.root / "steamapps/compatdata/2733369024").is_dir())
        self.assertEqual(dirs, [str(start)])

    def write_account(self):
        self.write("Steam/config/loginusers.vdf",
                   '"users"\n{\n\t"76561198352563669"\n\t{\n\t\t"MostRecent"\t\t"1"\n\t}\n}\n')
        return "Steam/userdata/392297941/config/"

    def test_without_steam_and_options_a_game_takes_its_steam_launch_options(self):
        config = self.write_account()
        self.write(config + "localconfig.vdf", '"UserLocalConfigStore"\n{\n\t"Software"\n\t{\n\t\t"Valve"\n\t\t{\n'
                   '\t\t\t"Steam"\n\t\t\t{\n\t\t\t\t"apps"\n\t\t\t\t{\n\t\t\t\t\t"413150"\n\t\t\t\t\t{\n'
                   '\t\t\t\t\t\t"cloud"\n\t\t\t\t\t\t{\n\t\t\t\t\t\t\t"LaunchOptions"\t\t"-wrong"\n\t\t\t\t\t\t}\n'
                   '\t\t\t\t\t\t"LaunchOptions"\t\t"DXVK_HUD=1 %command% -d3d9 \\"-name=Two Words\\""\n'
                   '\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t}\n}\n')
        shortcuts = runpy.run_path(str(BIN / "droiddeck-steam-shortcuts"))
        self.write(config + "shortcuts.vdf", shortcuts["write"]({"shortcuts": {
            "0": {"appid": -1561598272, "AppName": "Batman", "LaunchOptions": "-d3d9"},
            "1": {"appid": -1, "AppName": "Other", "LaunchOptions": "-x"},
        }}))
        root = str(self.root)
        self.assertEqual(RUN["steam_options"]("413150", root), 'DXVK_HUD=1 %command% -d3d9 "-name=Two Words"')
        self.assertEqual(RUN["steam_options"]("2733369024", root), "-d3d9")
        self.assertEqual(RUN["steam_options"]("70", root), "")
        self.assertEqual(RUN["steam_options"]("2733369025", root), "")
        info = {"common": {"type": "Game"}, "config": {"launch": {
            "0": {"executable": "Game.exe", "type": "default", "config": {"oslist": "windows"}}}}}
        self.write("Steam/appcache/appinfo.vdf", appinfo({413150: info}))
        program = self.game("413150", "Some Game", "Game.exe")
        code, calls, _ = self.run_main(["--app", "413150"])
        self.assertIsNone(code)
        _, argv, env = calls[0]
        self.assertEqual(argv[1:], ["waitforexitandrun", str(program), "-d3d9", "-name=Two Words"])
        self.assertEqual(env["DXVK_HUD"], "1")
        bat = self.write("Games/GOG/Batman/Batman.exe", b"MZ")
        code, calls, _ = self.run_main(["--app", "2733369024", "--", str(bat)])
        self.assertEqual(calls[0][1][1:], ["waitforexitandrun", str(bat), "-d3d9"])
        code, calls, _ = self.run_main(["--app", "2733369024", "--options", "", "--", str(bat)])
        self.assertEqual(calls[0][1][1:], ["waitforexitandrun", str(bat)])

    def test_launch_options_split_like_steam_does(self):
        split = RUN["split_options"]
        self.assertEqual(split(""), ({}, []))
        self.assertEqual(split("-windowed -w 1280"), ({}, ["-windowed", "-w", "1280"]))
        self.assertEqual(split("A=1 B=\"x y\" gamemoderun %command% -dx12"), ({"A": "1", "B": "x y"}, ["-dx12"]))
        self.assertEqual(split("\"unterminated -x"), ({}, ["\"unterminated", "-x"]))


if __name__ == "__main__":
    unittest.main()