mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-11 11:01:07 +00:00
1188 lines
73 KiB
Python
1188 lines
73 KiB
Python
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
from pathlib import Path
|
|
import runpy
|
|
import shutil
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
BIN = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin"
|
|
SYNC = runpy.run_path(str(BIN / "droiddeck-esync"))
|
|
COMPAT = runpy.run_path(str(BIN / "steam-compatibility"))
|
|
GLOBALS = SYNC["select"].__globals__
|
|
NTDLL = SYNC["NTDLL"]
|
|
WINESERVER = SYNC["WINESERVER"]
|
|
COPY = list(SYNC["DEFAULT_COPY"])
|
|
REAL_DIRS = SYNC["REAL_DIRS"]
|
|
RECORD = SYNC["RECORD"]
|
|
VALVE_LINE = "1789159687 experimental-11.0-20260910b-arm64"
|
|
GE_LINE = "1789520806 GE-Proton11-7"
|
|
EXPORTS = ("NtClose", "NtCreateEvent", "wine_server_call")
|
|
WATCHED = ("WINESERVER", "BL_SYNC_PACK", "WINEESYNC", "PROTON_NO_ESYNC", "BL_SYNC", "CUSTOM")
|
|
PROTON = ("#!/usr/bin/python3\nimport json, os, sys\n"
|
|
"print(json.dumps([sys.argv[0], sys.argv[1:], {k: os.environ.get(k) for k in %r}]))\n" % (WATCHED,))
|
|
MANIFEST = '"manifest" { "commandline" "/proton %verb%" }'
|
|
|
|
|
|
def sha(path):
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def elf(exported=(), weak=(), local=(), undefined=(), salt=b""):
|
|
strings = b"\0"
|
|
offsets = {}
|
|
for name in list(exported) + list(weak) + list(local) + list(undefined):
|
|
if name not in offsets:
|
|
offsets[name] = len(strings)
|
|
strings += name.encode() + b"\0"
|
|
symbols = [bytes(24)]
|
|
for group, binding, shndx in ((exported, 1, 7), (weak, 2, 7), (local, 0, 7), (undefined, 1, 0)):
|
|
symbols += [struct.pack("<IBBHQQ", offsets[name], binding << 4 | 2, 0, shndx, 0, 0) for name in group]
|
|
dynsym = b"".join(symbols)
|
|
body = dynsym + strings
|
|
body += bytes(-len(body) % 8)
|
|
header = b"\x7fELF" + bytes((2, 1, 1)) + bytes(9) + struct.pack(
|
|
"<HHIQQQIHHHHHH", 3, 183, 1, 0, 0, 64 + len(body), 0, 64, 0, 0, 64, 3, 0)
|
|
sections = (bytes(64) + struct.pack("<IIQQQQIIQQ", 0, 11, 2, 0, 64, len(dynsym), 2, 1, 8, 24)
|
|
+ struct.pack("<IIQQQQIIQQ", 0, 3, 2, 0, 64 + len(dynsym), len(strings), 0, 0, 1, 0))
|
|
return header + body + sections + salt
|
|
|
|
|
|
def make_tool(parent, name, line=VALVE_LINE, salt=b"", builtin=False, ge=False, decorated=False, fsync=False, ntsync=True):
|
|
tool = Path(parent) / name
|
|
for folder in ("files/lib/wine/aarch64-unix", "files/lib/wine/aarch64-windows", "files/bin-arm64", "files/share/wine"):
|
|
(tool / folder).mkdir(parents=True, exist_ok=True)
|
|
(tool / NTDLL).write_bytes(elf(EXPORTS, salt=salt))
|
|
(tool / WINESERVER).write_bytes(b"wineserver " + salt + (b" esync: up and running" if builtin else b"")
|
|
+ (b" fsync: up and running" if fsync else b"") + (b" /dev/ntsync" if ntsync else b""))
|
|
for rel in COPY:
|
|
(tool / rel).write_bytes(b"stock " + rel.encode() + salt)
|
|
(tool / "files/lib/wine/aarch64-unix/winevulkan.so").write_bytes(b"vulkan")
|
|
(tool / "files/lib/wine/aarch64-windows/ntdll.dll").write_bytes(b"pe")
|
|
os.symlink("wine", tool / "files/bin-arm64/msidb")
|
|
if ge:
|
|
os.symlink("bin-arm64", tool / "files/bin")
|
|
(tool / "version").write_text(line + "\n")
|
|
(tool / "proton").write_text(PROTON)
|
|
(tool / "proton").chmod(0o755)
|
|
if decorated:
|
|
(tool / "toolmanifest.vdf").write_text(MANIFEST)
|
|
(tool / "toolmanifest.vdf.droiddeck-orig").write_text(MANIFEST)
|
|
(tool / "toolmanifest.vdf.bannerlator-orig").write_text(MANIFEST)
|
|
(tool / "compatibilitytool.vdf").write_text("compat")
|
|
(tool / "droiddeck-proton-wrap").write_text("wrap")
|
|
(tool / "bannerlator-proton-wrap").write_text("wrap")
|
|
return tool
|
|
|
|
|
|
def make_pack(root, tool, pack_id, rev=1, source_match=False, stock=None, version=None, exports=None, complete=True, **extra):
|
|
tool = Path(tool)
|
|
pack_dir = Path(root) / "packs" / pack_id
|
|
(pack_dir / "files/lib/wine/aarch64-unix").mkdir(parents=True)
|
|
(pack_dir / "files/bin-arm64").mkdir(parents=True)
|
|
(pack_dir / NTDLL).write_bytes(b"patched ntdll " + pack_id.encode())
|
|
(pack_dir / WINESERVER).write_bytes(b"patched wineserver " + pack_id.encode())
|
|
line = (tool / "version").read_text().strip()
|
|
data = {
|
|
"format": 1, "id": pack_id, "flavor": "valve", "rev": rev, "version": version or line.split()[1],
|
|
"version_line": line, "stock": stock or {NTDLL: sha(tool / NTDLL), WINESERVER: sha(tool / WINESERVER)},
|
|
"exports": exports or SYNC["exports_hash"](tool / NTDLL), "source_match": source_match,
|
|
"files": {NTDLL: sha(pack_dir / NTDLL), WINESERVER: sha(pack_dir / WINESERVER)}, "copy": COPY,
|
|
"protocol": 931, "source": {"repo": "r", "ref": "t", "commit": "c", "wine_commit": "w", "sdk_image": "i"},
|
|
"patch": {"series": "valve-11", "sha256": "0" * 64},
|
|
}
|
|
data.update(extra)
|
|
(pack_dir / "pack.json").write_text(json.dumps(data))
|
|
if complete:
|
|
(pack_dir / ".complete").write_text("")
|
|
return data
|
|
|
|
|
|
def files_under(path):
|
|
return sorted(str(Path(folder, name).relative_to(path)) for folder, dirs, names in os.walk(path) for name in dirs + names)
|
|
|
|
|
|
def concurrent_worker(barrier, queue, root, tool, pack, steam):
|
|
barrier.wait()
|
|
try:
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
path = SYNC["ensure_dist"](root, tool, pack)
|
|
SYNC["reconcile"](steam, {"HOME": str(Path(root).parents[2]), "BL_SYNC_FALLBACK": "1"})
|
|
queue.put(path)
|
|
except Exception as error:
|
|
queue.put("error: %r" % error)
|
|
|
|
|
|
class SyncTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
self.home = self.tmp / "home"
|
|
self.home.mkdir()
|
|
self.root = self.home / ".local/share/droiddeck-esync"
|
|
self.tools = self.tmp / "tools"
|
|
self.tools.mkdir()
|
|
|
|
def store(self):
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
return self.root
|
|
|
|
def env(self, **extra):
|
|
env = {"HOME": str(self.home), "STEAM_COMPAT_DATA_PATH": "/compatdata/42", "BL_SYNC_FALLBACK": "1", "BL_SYNC_MIN_NOFILE": "1"}
|
|
env.update(extra)
|
|
return {name: value for name, value in env.items() if value is not None}
|
|
|
|
def select(self, tool, env, verb="waitforexitandrun"):
|
|
command = [str(Path(tool) / "proton"), verb, "game.exe"]
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stderr(output):
|
|
chosen, chosen_env = SYNC["select"](command, env)
|
|
return chosen, chosen_env, output.getvalue()
|
|
|
|
def dist_of(self, pack_id, tool):
|
|
return self.root / "dist" / SYNC["dist_name"](pack_id, os.path.realpath(tool))
|
|
|
|
def quiet(self, function, *args):
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
return function(*args)
|
|
|
|
|
|
class ExportsHashTest(SyncTestCase):
|
|
def test_keeps_defined_global_and_weak_names(self):
|
|
path = self.tmp / "lib.so"
|
|
path.write_bytes(elf(("beta", "alpha", "beta"), weak=("weak",), local=("hidden",), undefined=("imported",), salt=b"x"))
|
|
self.assertEqual(SYNC["exports_hash"](path), hashlib.sha256(b"alpha\nbeta\nweak").hexdigest())
|
|
other = self.tmp / "other.so"
|
|
other.write_bytes(elf(("alpha", "beta"), weak=("weak",), salt=b"different"))
|
|
self.assertEqual(SYNC["exports_hash"](other), SYNC["exports_hash"](path))
|
|
for data in (b"not an elf", elf(("alpha",))[:80], b"\x7fELF\x01\x01" + bytes(60)):
|
|
path.write_bytes(data)
|
|
with self.assertRaises(ValueError):
|
|
SYNC["exports_hash"](path)
|
|
|
|
@unittest.skipUnless(shutil.which("cc") and shutil.which("nm"), "needs cc and nm")
|
|
def test_matches_nm_on_a_compiled_library(self):
|
|
source = self.tmp / "lib.c"
|
|
source.write_text("int alpha(void) { return 1; }\n__attribute__((weak)) int beta(void) { return 2; }\n"
|
|
"static int gamma_(void) { return 3; }\nint delta = 4;\nint use(void) { return gamma_(); }\n"
|
|
"extern int outside(void);\nint call(void) { return outside(); }\n")
|
|
library = self.tmp / "lib.so"
|
|
if subprocess.run(["cc", "-shared", "-fPIC", "-o", str(library), str(source)], capture_output=True).returncode:
|
|
self.skipTest("cc cannot build a shared library here")
|
|
listing = subprocess.run(["nm", "-D", "--defined-only", str(library)], capture_output=True, text=True, check=True).stdout
|
|
names = {fields[2].split("@")[0] for fields in (line.split() for line in listing.splitlines())
|
|
if len(fields) == 3 and fields[1].isupper()}
|
|
self.assertIn("alpha", names)
|
|
self.assertNotIn("outside", names)
|
|
self.assertEqual(SYNC["exports_hash"](library), hashlib.sha256("\n".join(sorted(names)).encode()).hexdigest())
|
|
|
|
|
|
class ShadowTest(SyncTestCase):
|
|
def test_layout_links_and_real_files(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "GE-Proton11-7", GE_LINE, ge=True, decorated=True)
|
|
os.symlink(self.tools, self.tmp / "launch")
|
|
launched = self.tmp / "launch" / "GE-Proton11-7"
|
|
pack = make_pack(root, tool, "ge-GE-Proton11-7-1789520806-abcdefabcdef-r1")
|
|
dist = Path(SYNC["ensure_dist"](root, launched, pack))
|
|
self.assertEqual(dist, self.dist_of(pack["id"], tool))
|
|
self.assertEqual(dist.name, "%s~%s" % (pack["id"], hashlib.sha256(str(tool.resolve()).encode()).hexdigest()[:8]))
|
|
for folder in REAL_DIRS:
|
|
self.assertTrue((dist / folder).is_dir() and not (dist / folder).is_symlink(), folder)
|
|
real_files = {NTDLL, WINESERVER, *COPY, RECORD}
|
|
for rel in files_under(dist):
|
|
if rel not in REAL_DIRS and rel not in real_files:
|
|
self.assertTrue((dist / rel).is_symlink(), rel)
|
|
self.assertEqual(os.readlink(dist / "files/bin"), "bin-arm64")
|
|
self.assertEqual(os.readlink(dist / "files/bin-arm64/msidb"), "wine")
|
|
for rel in ("proton", "version", "files/share", "files/lib/wine/aarch64-windows", "files/lib/wine/aarch64-unix/winevulkan.so"):
|
|
self.assertEqual(os.readlink(dist / rel), str(launched / rel))
|
|
for rel in ("dist.lock", "files/steampipe_fixups_mtime"):
|
|
self.assertEqual(os.readlink(dist / rel), str(launched / rel))
|
|
self.assertFalse((tool / rel).exists())
|
|
for name in ("toolmanifest.vdf", "compatibilitytool.vdf", "droiddeck-proton-wrap", "toolmanifest.vdf.droiddeck-orig",
|
|
"bannerlator-proton-wrap", "toolmanifest.vdf.bannerlator-orig"):
|
|
self.assertFalse(os.path.lexists(dist / name), name)
|
|
pack_dir = root / "packs" / pack["id"]
|
|
for rel in (NTDLL, WINESERVER):
|
|
self.assertEqual((dist / rel).read_bytes(), (pack_dir / rel).read_bytes())
|
|
for rel in COPY:
|
|
self.assertEqual((dist / rel).read_bytes(), (tool / rel).read_bytes())
|
|
for rel in (NTDLL, WINESERVER, *COPY):
|
|
info = os.lstat(dist / rel)
|
|
self.assertTrue(stat.S_ISREG(info.st_mode), rel)
|
|
self.assertEqual(stat.S_IMODE(info.st_mode), 0o755, rel)
|
|
self.assertEqual(info.st_nlink, 1, rel)
|
|
self.assertEqual((dist / "files/bin/wineserver").read_bytes(), (pack_dir / WINESERVER).read_bytes())
|
|
self.assertEqual((dist / "files/bin-arm64/msidb").resolve(), (dist / "files/bin-arm64/wine").resolve())
|
|
record = json.loads((dist / RECORD).read_text())
|
|
self.assertEqual(record["pack"], pack["id"])
|
|
self.assertEqual(record["stock"], str(tool.resolve()))
|
|
self.assertEqual(record["tool"], str(launched))
|
|
self.assertEqual(record["version"], GE_LINE)
|
|
self.assertEqual(sorted(record["copies"]), sorted(COPY))
|
|
for rel in COPY:
|
|
info = os.stat(tool / rel)
|
|
self.assertEqual(record["copies"][rel], [info.st_size, info.st_mtime_ns, sha(tool / rel)])
|
|
inodes = {rel: os.lstat(dist / rel).st_ino for rel in (NTDLL, WINESERVER, *COPY)}
|
|
self.assertEqual(Path(SYNC["ensure_dist"](root, tool, pack)), dist)
|
|
self.assertEqual(inodes, {rel: os.lstat(dist / rel).st_ino for rel in inodes})
|
|
self.assertEqual(os.readlink(dist / "proton"), str(launched / "proton"))
|
|
self.assertEqual([name for name in os.listdir(root / "dist") if ".tmp-" in name], [])
|
|
|
|
def test_reconcile_follows_stock_changes(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
dist = Path(SYNC["ensure_dist"](root, tool, pack))
|
|
unix_wine = "files/lib/wine/aarch64-unix/wine"
|
|
preloader = "files/lib/wine/aarch64-unix/wine-preloader"
|
|
(tool / "files/new.txt").write_text("new")
|
|
(tool / "files/lib/wine/aarch64-unix/new.so").write_text("new")
|
|
(tool / "files/lib/wine/aarch64-unix/winevulkan.so").unlink()
|
|
(dist / "files/share").unlink()
|
|
os.symlink("/nonexistent", dist / "files/share")
|
|
(dist / "files/stray").symlink_to("/nowhere")
|
|
for rel in (NTDLL, WINESERVER, "files/bin-arm64/wine"):
|
|
os.chmod(dist / rel, 0o444)
|
|
os.chmod(dist / unix_wine, 0o555)
|
|
kept = os.lstat(dist / unix_wine).st_ino
|
|
(tool / "files/bin-arm64/wine").write_bytes(b"an updated stock wine loader")
|
|
(tool / preloader).unlink()
|
|
self.assertEqual(Path(SYNC["ensure_dist"](root, tool, pack)), dist)
|
|
self.assertEqual(os.readlink(dist / "files/new.txt"), str(tool / "files/new.txt"))
|
|
self.assertEqual(os.readlink(dist / "files/lib/wine/aarch64-unix/new.so"), str(tool / "files/lib/wine/aarch64-unix/new.so"))
|
|
self.assertFalse(os.path.lexists(dist / "files/lib/wine/aarch64-unix/winevulkan.so"))
|
|
self.assertFalse(os.path.lexists(dist / "files/stray"))
|
|
self.assertEqual(os.readlink(dist / "files/share"), str(tool / "files/share"))
|
|
self.assertEqual((dist / "files/bin-arm64/wine").read_bytes(), b"an updated stock wine loader")
|
|
for rel in (NTDLL, WINESERVER, "files/bin-arm64/wine"):
|
|
self.assertEqual(stat.S_IMODE(os.lstat(dist / rel).st_mode), 0o755, rel)
|
|
self.assertEqual((dist / NTDLL).read_bytes(), (root / "packs" / pack["id"] / NTDLL).read_bytes())
|
|
self.assertEqual(os.lstat(dist / unix_wine).st_ino, kept)
|
|
self.assertTrue((dist / preloader).is_file() and not (dist / preloader).is_symlink())
|
|
record = json.loads((dist / RECORD).read_text())
|
|
self.assertEqual(record["copies"]["files/bin-arm64/wine"][2], sha(tool / "files/bin-arm64/wine"))
|
|
self.assertIn(preloader, record["copies"])
|
|
|
|
def test_tampered_pack_falls_back_to_stock(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
(root / "packs" / pack["id"] / NTDLL).write_bytes(b"tampered")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertNotIn("WINESERVER", env)
|
|
self.assertIn("unusable", said)
|
|
self.assertIn(f"pack {pack['id']} is damaged; the app fetches it again", said)
|
|
self.assertFalse((root / "packs" / pack["id"] / ".complete").exists())
|
|
self.assertEqual([name for name in os.listdir(root / "dist") if not name.endswith(".lock")], [])
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertIn("no pack for Proton Experimental (ARM64)", said)
|
|
self.assertIn(pack["stock"][NTDLL], (root / "wanted.tsv").read_text())
|
|
|
|
def test_a_missing_tool_file_does_not_condemn_the_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
with mock.patch.dict(GLOBALS, {"ensure_dist": mock.Mock(side_effect=OSError("disk full"))}):
|
|
command, _, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertIn("unusable", said)
|
|
self.assertNotIn("damaged", said)
|
|
self.assertTrue((root / "packs" / pack["id"] / ".complete").exists())
|
|
|
|
def test_concurrent_ensure_dist_builds_one_dist(self):
|
|
root = self.store()
|
|
steam = self.tmp / "steam"
|
|
tool = make_tool(steam / "steamapps/common", "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
context = multiprocessing.get_context("fork")
|
|
barrier = context.Barrier(4)
|
|
queue = context.Queue()
|
|
workers = [context.Process(target=concurrent_worker, args=(barrier, queue, str(root), str(tool), pack, str(steam)))
|
|
for _ in range(4)]
|
|
with mock.patch.dict(GLOBALS, {"SD_LIBRARY": str(self.tmp / "sd")}):
|
|
for worker in workers:
|
|
worker.start()
|
|
results = [queue.get(timeout=60) for _ in workers]
|
|
for worker in workers:
|
|
worker.join(60)
|
|
dist = self.dist_of(pack["id"], tool)
|
|
self.assertEqual(results, [str(dist)] * 4)
|
|
self.assertEqual(sorted(os.listdir(root / "dist")), sorted([dist.name, dist.name + ".lock"]))
|
|
record = json.loads((dist / RECORD).read_text())
|
|
self.assertEqual(record["pack"], pack["id"])
|
|
for rel in (NTDLL, WINESERVER):
|
|
self.assertEqual(sha(dist / rel), pack["files"][rel])
|
|
for rel in COPY:
|
|
self.assertEqual(sha(dist / rel), sha(tool / rel))
|
|
self.assertEqual((root / "tools.tsv").read_text().count("\n"), 1)
|
|
self.assertFalse([name for name in os.listdir(root) if ".tmp-" in name])
|
|
|
|
def test_build_shadow_command_line(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "proton-cachyos-11", "1790000000 cachyos-11.0-1")
|
|
pack = make_pack(root, tool, "cachyos-11-r1")
|
|
out = self.tmp / "shadow" / "dist"
|
|
for _ in range(2):
|
|
result = subprocess.run([sys.executable, str(BIN / "droiddeck-esync"), "build-shadow", str(tool),
|
|
str(root / "packs" / pack["id"]), str(out)], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout.strip(), str(out))
|
|
self.assertEqual(os.listdir(out.parent), ["dist"])
|
|
self.assertEqual(json.loads((out / RECORD).read_text())["pack"], pack["id"])
|
|
self.assertEqual(os.readlink(out / "proton"), str(tool / "proton"))
|
|
status = subprocess.run([sys.executable, str(BIN / "droiddeck-esync"), "status"], capture_output=True, text=True,
|
|
env={**os.environ, "HOME": str(self.home)}, check=True).stdout
|
|
self.assertIn("packs: 1 installed", status)
|
|
usage = subprocess.run([sys.executable, str(BIN / "droiddeck-esync"), "build-shadow"], capture_output=True, text=True)
|
|
self.assertEqual(usage.returncode, 2)
|
|
|
|
|
|
class SelectTest(SyncTestCase):
|
|
def test_store_missing_is_inert(self):
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
before = files_under(self.tmp)
|
|
env = self.env()
|
|
command, chosen_env, said = self.select(tool, env)
|
|
self.assertEqual((command, chosen_env, said), ([str(tool / "proton"), "waitforexitandrun", "game.exe"], env, ""))
|
|
command, chosen_env, said = self.select(tool, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="1"))
|
|
self.assertEqual(chosen_env, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="0"))
|
|
self.assertIsNone(self.quiet(SYNC["reconcile"], str(self.tmp / "steam"), env))
|
|
self.assertEqual(SYNC["gc"](None), 0)
|
|
self.assertIsNone(SYNC["store_root"](env))
|
|
self.assertEqual(files_under(self.tmp), before)
|
|
|
|
def test_probe_prefixes_and_other_tools_are_left_alone(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
make_pack(root, tool, "valve-experimental-r1")
|
|
for prefix in ("", "/steam/compatdata/0", "/steam/compatdata/0/", "/steam/compatdata/0-1234"):
|
|
env = self.env(STEAM_COMPAT_DATA_PATH=prefix, PROTON_NO_NTSYNC="1")
|
|
self.assertEqual(self.select(tool, env), ([str(tool / "proton"), "waitforexitandrun", "game.exe"], env, ""))
|
|
env = self.env(PROTON_USE_ARM64="0")
|
|
self.assertEqual(self.select(tool, env)[:2], ([str(tool / "proton"), "waitforexitandrun", "game.exe"], env))
|
|
other = self.tmp / "not-proton"
|
|
other.mkdir()
|
|
self.assertEqual(self.select(other, self.env())[1], self.env())
|
|
self.assertFalse((root / "dist").exists())
|
|
self.assertFalse((root / "launches.log").exists())
|
|
|
|
def test_fallback_switch_off(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
make_pack(root, tool, "valve-experimental-r1")
|
|
for value in ("0", None, "yes"):
|
|
env = self.env(BL_SYNC_FALLBACK=value)
|
|
command, chosen_env, said = self.select(tool, env)
|
|
self.assertEqual((command[0], chosen_env), (str(tool / "proton"), SYNC["wineserver_only"](env)))
|
|
self.assertEqual(said, "droiddeck-esync: off; wineserver sync only\n")
|
|
self.assertFalse((root / "dist").exists())
|
|
self.assertFalse((root / "wanted.tsv").exists())
|
|
self.assertEqual([line.split("\t")[4] for line in (root / "launches.log").read_text().splitlines()], ["-"] * 3)
|
|
|
|
def test_pack_is_used_and_logged(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="1"))
|
|
self.assertEqual(command, [str(dist / "proton"), "waitforexitandrun", "game.exe"])
|
|
self.assertEqual(env["WINESERVER"], str(dist / WINESERVER))
|
|
self.assertEqual(env["BL_SYNC_PACK"], pack["id"])
|
|
self.assertEqual(env["WINEESYNC"], "1")
|
|
self.assertNotIn("PROTON_NO_ESYNC", env)
|
|
self.assertTrue(said.startswith("droiddeck-esync: pack %s for Proton Experimental (ARM64) (%s, exact match); "
|
|
"droiddeck-ntsync on, droiddeck-esync on, fds " % (pack["id"], VALVE_LINE)), said)
|
|
self.assertEqual(said.count("\n"), 1)
|
|
fields = (root / "launches.log").read_text().splitlines()[-1].split("\t")
|
|
self.assertEqual(fields[1:7], ["42", "waitforexitandrun", str(tool), pack["id"], "1", "1"])
|
|
self.assertRegex(fields[7], r"^(\d+|unlimited)/(\d+|unlimited)$")
|
|
command, env, said = self.select(tool, self.env(WINEESYNC="2"))
|
|
self.assertEqual((env["WINEESYNC"], command[0]), ("2", str(dist / "proton")))
|
|
self.assertIn("droiddeck-ntsync off", said)
|
|
for _ in range(SYNC["LOG_LINES"] + 5):
|
|
SYNC["log_launch"](str(root), command, env, str(tool), None)
|
|
self.assertEqual(len((root / "launches.log").read_text().splitlines()), SYNC["LOG_LINES"])
|
|
|
|
def test_fallback_switched_off_per_game(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
cases = ({"PROTON_NO_ESYNC": "1"}, {"STEAM_COMPAT_APP_ID": "2630"}, {"SteamAppId": "1060210"},
|
|
{"STEAM_COMPAT_DATA_PATH": "/steam/compatdata/414740"}, {"STEAM_COMPAT_APP_ID": "201510", "WINEESYNC": "1"},
|
|
{"STEAM_COMPAT_DATA_PATH": "/steam/compatdata/1233880/"}, {"WINEESYNC": "0"},
|
|
{"WINEESYNC": "0", "PROTON_NO_ESYNC": "0"})
|
|
for extra in cases:
|
|
command, env, _ = self.select(tool, self.env(**extra))
|
|
self.assertEqual((env["WINEESYNC"], env["PROTON_NO_ESYNC"], env["BL_SYNC_PACK"]), ("0", "1", pack["id"]), extra)
|
|
self.assertNotIn("PROTON_NO_FSYNC", env, extra)
|
|
self.assertEqual(command[0], str(self.dist_of(pack["id"], tool) / "proton"))
|
|
for extra in ({"PROTON_NO_ESYNC": "0"}, {"PROTON_NO_ESYNC": ""}, {"STEAM_COMPAT_APP_ID": "42", "SteamAppId": "2630"},
|
|
{"STEAM_COMPAT_APP_ID": "2630", "PROTON_NO_ESYNC": "0"}, {"STEAM_COMPAT_APP_ID": "2630", "PROTON_NO_ESYNC": ""}):
|
|
env = self.select(tool, self.env(**extra))[1]
|
|
self.assertEqual((env["WINEESYNC"], env["PROTON_NO_FSYNC"]), ("1", "1"), extra)
|
|
|
|
def test_fallback_comes_before_fsync(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
make_pack(root, tool, "valve-experimental-r1")
|
|
for extra, expected in (({}, "1"), ({"WINEESYNC": "2"}, "1"), ({"PROTON_NO_FSYNC": "0"}, "0"),
|
|
({"PROTON_NO_FSYNC": ""}, ""), ({"PROTON_NO_FSYNC": "1"}, "1")):
|
|
env = self.select(tool, self.env(**extra))[1]
|
|
self.assertEqual(env.get("PROTON_NO_FSYNC"), expected, extra)
|
|
stock = self.select(tool, self.env(BL_SYNC_FALLBACK="0"))[1]
|
|
self.assertEqual((stock["PROTON_NO_FSYNC"], stock["WINEFSYNC"]), ("1", "0"))
|
|
|
|
def test_open_file_limit_threshold(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
make_pack(root, tool, "valve-experimental-r1")
|
|
cases = (((1024, 4096), None, "0"), ((1024, 4096), "4096", "1"), ((1024, 4096), "4097", "0"),
|
|
((1024, 4096), "junk", "0"), ((1024, 8192), None, "1"), ((1024, -1), "999999", "1"))
|
|
for limits, minimum, expected in cases:
|
|
with mock.patch("resource.getrlimit", return_value=limits):
|
|
command, env, said = self.select(tool, self.env(BL_SYNC_MIN_NOFILE=minimum))
|
|
self.assertEqual(env["WINEESYNC"], expected, (limits, minimum))
|
|
self.assertIn("fds 1024/unlimited", said)
|
|
|
|
def test_proton_no_ntsync_turns_emulation_off(self):
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
self.assertEqual(self.select(tool, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="1"))[1]["BL_SYNC"], "0")
|
|
self.assertEqual(self.select(tool, self.env(PROTON_NO_NTSYNC="1"))[1]["BL_SYNC"], "0")
|
|
self.assertEqual(self.select(tool, self.env(PROTON_NO_NTSYNC="0", BL_SYNC="1"))[1]["BL_SYNC"], "1")
|
|
root = self.store()
|
|
_, env, _ = self.select(tool, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="1"))
|
|
self.assertEqual(env["BL_SYNC"], "0")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
_, env, said = self.select(tool, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="1"))
|
|
self.assertEqual((env["BL_SYNC"], env["BL_SYNC_PACK"]), ("0", pack["id"]))
|
|
self.assertIn("droiddeck-ntsync off", said)
|
|
|
|
def test_internal_failure_launches_stock(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
make_pack(root, tool, "valve-experimental-r1")
|
|
|
|
def broken(*args, **kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
for name in ("measure", "load_packs", "ensure_dist", "find_pack"):
|
|
with mock.patch.dict(GLOBALS, {name: broken}):
|
|
command, env, said = self.select(tool, self.env(PROTON_NO_NTSYNC="1"))
|
|
self.assertEqual(command[0], str(tool / "proton"), name)
|
|
self.assertEqual(env, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="0"), name)
|
|
self.assertIn("boom", said)
|
|
|
|
def test_droiddeck_ntsync_stands_in_without_a_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="0"))
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertEqual(env, self.env(BL_SYNC="1"))
|
|
self.assertTrue(said.endswith("; stock binaries with droiddeck-ntsync\n"), said)
|
|
self.assertEqual((root / "launches.log").read_text().split("\n")[-2].split("\t")[6], "1")
|
|
command, env, said = self.select(tool, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="1"))
|
|
self.assertEqual(env, self.env(PROTON_NO_NTSYNC="1", BL_SYNC="0"))
|
|
self.assertTrue(said.endswith("; stock binaries\n"), said)
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
(root / "packs" / pack["id"] / NTDLL).write_bytes(b"tampered")
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="0"))
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertEqual(env, self.env(BL_SYNC="1"))
|
|
self.assertIn("unusable", said)
|
|
self.assertTrue(said.endswith("; stock binaries with droiddeck-ntsync\n"), said)
|
|
|
|
def test_droiddeck_fsync_stands_in_without_a_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)", fsync=True)
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="0"))
|
|
self.assertEqual(command, [str(tool / "proton"), "waitforexitandrun", "game.exe"])
|
|
self.assertEqual(env, self.env(BL_SYNC="0", BL_FSYNC="1", WINEFSYNC="1"))
|
|
self.assertEqual(said, "droiddeck-fsync: no pack for Proton Experimental (ARM64) (%s, ntdll %s); "
|
|
"stock binaries with Proton's fsync\n" % (VALVE_LINE, sha(tool / NTDLL)[:12]))
|
|
fields = (root / "launches.log").read_text().splitlines()[-1].split("\t")
|
|
self.assertEqual((fields[4], fields[6], fields[8]), ("-", "0", "1"))
|
|
self.assertEqual(len((root / "wanted.tsv").read_text().splitlines()), 1)
|
|
for extra in ({"BL_SYNC": "1"}, {"PROTON_NO_FSYNC": "1"}, {"STEAM_COMPAT_APP_ID": "2630"}, {"SteamAppId": "1233880"}):
|
|
command, env, said = self.select(tool, self.env(**extra))
|
|
self.assertNotIn("BL_FSYNC", env, extra)
|
|
self.assertEqual(env["BL_SYNC"], "1", extra)
|
|
self.assertTrue(said.endswith("; stock binaries with droiddeck-ntsync\n"), (extra, said))
|
|
for extra in ({"STEAM_COMPAT_APP_ID": "2630", "PROTON_NO_FSYNC": "0"}, {"PROTON_NO_FSYNC": ""}, {"BL_SYNC": "0"}):
|
|
self.assertEqual(self.select(tool, self.env(**extra))[1]["BL_FSYNC"], "1", extra)
|
|
env = self.select(tool, self.env(PROTON_NO_FSYNC="1", PROTON_NO_NTSYNC="1"))[1]
|
|
self.assertEqual((env["BL_SYNC"], "BL_FSYNC" in env), ("0", False))
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
(root / "packs" / pack["id"] / NTDLL).write_bytes(b"tampered")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((command[0], env["BL_FSYNC"]), (str(tool / "proton"), "1"))
|
|
self.assertIn("\ndroiddeck-fsync: pack %s unusable for Proton Experimental (ARM64) (" % pack["id"], said)
|
|
self.assertTrue(said.endswith("; stock binaries with Proton's fsync\n"), said)
|
|
|
|
def test_droiddeck_fsync_first(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)", fsync=True)
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
for extra in ({}, {"BL_SYNC_FALLBACK": None}, {"BL_SYNC_FALLBACK": "0"}, {"BL_FSYNC": "0"}):
|
|
command, env, said = self.select(tool, self.env(BL_FSYNC_FIRST="1", **extra))
|
|
self.assertEqual(command[0], str(tool / "proton"), extra)
|
|
self.assertEqual((env["BL_FSYNC"], env["BL_SYNC"], env["WINEFSYNC"]), ("1", "0", "1"), extra)
|
|
self.assertNotIn("BL_SYNC_PACK", env, extra)
|
|
self.assertEqual(said, "droiddeck-fsync: chosen for Proton Experimental (ARM64); stock binaries with Proton's fsync\n")
|
|
self.assertFalse((root / "wanted.tsv").exists())
|
|
command, env, said = self.select(tool, self.env(BL_FSYNC_FIRST="1", BL_SYNC="1"))
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"], env["BL_SYNC"]), (str(self.dist_of(pack["id"], tool) / "proton"), pack["id"], "1"))
|
|
self.assertNotIn("BL_FSYNC", env)
|
|
command, env, said = self.select(tool, self.env(BL_FSYNC_FIRST="1", BL_FSYNC="1", STEAM_COMPAT_APP_ID="414740"))
|
|
self.assertEqual((env["BL_FSYNC"], env["BL_SYNC_PACK"], env["PROTON_NO_ESYNC"]), ("0", pack["id"], "1"))
|
|
command, env, said = self.select(tool, self.env(BL_FSYNC_FIRST="1", BL_FSYNC="1", BL_SYNC_FALLBACK="0", PROTON_NO_FSYNC="1"))
|
|
self.assertEqual((command[0], env["BL_FSYNC"], said), (str(tool / "proton"), "0", "droiddeck-esync: off\n"))
|
|
plain = make_tool(self.tools, "GE-Proton11-9", "1790000000 GE-Proton11-9", builtin=True, ge=True)
|
|
command, env, said = self.select(plain, self.env(BL_FSYNC_FIRST="1"))
|
|
self.assertEqual((command[0], said), (str(plain / "proton"), "droiddeck-esync: GE-Proton11-9 has esync built in\n"))
|
|
self.assertNotIn("BL_FSYNC", env)
|
|
|
|
def test_fsync_takes_over_when_esync_is_off_for_a_game(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)", fsync=True)
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
for extra in ({"PROTON_NO_ESYNC": "1"}, {"WINEESYNC": "0"}):
|
|
command, env, said = self.select(tool, self.env(**extra))
|
|
self.assertEqual((command[0], env["BL_FSYNC"]), (str(tool / "proton"), "1"), extra)
|
|
self.assertEqual(said, "droiddeck-fsync: esync is off for this game; stock binaries with Proton's fsync\n")
|
|
with mock.patch("resource.getrlimit", return_value=(1024, 4096)):
|
|
env = self.select(tool, self.env(BL_SYNC_MIN_NOFILE=None))[1]
|
|
self.assertEqual(env["BL_FSYNC"], "1")
|
|
for extra in ({"STEAM_COMPAT_APP_ID": "2630"}, {"PROTON_NO_ESYNC": "1", "PROTON_NO_FSYNC": "1"}, {"WINEESYNC": "0", "BL_SYNC": "1"}):
|
|
command, env, said = self.select(tool, self.env(**extra))
|
|
self.assertEqual((env["BL_SYNC_PACK"], env["WINEESYNC"]), (pack["id"], "0"), extra)
|
|
self.assertNotIn("BL_FSYNC", env, extra)
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((env["BL_SYNC_PACK"], env["WINEESYNC"], env["PROTON_NO_FSYNC"]), (pack["id"], "1", "1"))
|
|
self.assertNotIn("BL_FSYNC", env)
|
|
|
|
def test_the_hash_cache_remembers_fsync(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)", fsync=True)
|
|
self.select(tool, self.env())
|
|
cache = (root / "hash-cache.tsv").read_text().splitlines()
|
|
rows = {line.split("\t")[0]: line.split("\t") for line in cache}
|
|
server = rows[os.path.realpath(tool / WINESERVER)]
|
|
self.assertEqual((len(server), server[7], server[9], server[10]), (11, "0", "1", "1"))
|
|
self.assertEqual(rows[os.path.realpath(tool / NTDLL)][9], "0")
|
|
(root / "hash-cache.tsv").write_text("\n".join("\t".join(row[:9]) for row in rows.values()) + "\n")
|
|
self.assertEqual(SYNC["parse_cache"]((root / "hash-cache.tsv").read_text()), {})
|
|
self.assertEqual(self.select(tool, self.env())[1]["BL_FSYNC"], "1")
|
|
|
|
def test_esync_off_games_get_droiddeck_ntsync_on_a_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
for extra in ({"PROTON_NO_ESYNC": "1"}, {"STEAM_COMPAT_APP_ID": "2630"}, {"WINEESYNC": "0"}):
|
|
command, env, said = self.select(tool, self.env(**extra))
|
|
self.assertEqual((env["BL_SYNC_PACK"], env["WINEESYNC"], env["BL_SYNC"]), (pack["id"], "0", "1"), extra)
|
|
self.assertIn("droiddeck-ntsync on, droiddeck-esync off", said)
|
|
env = self.select(tool, self.env(PROTON_NO_ESYNC="1", PROTON_NO_NTSYNC="1"))[1]
|
|
self.assertEqual((env["WINEESYNC"], env["BL_SYNC"]), ("0", "0"))
|
|
env = self.select(tool, self.env(BL_SYNC="0"))[1]
|
|
self.assertEqual((env["WINEESYNC"], env["BL_SYNC"]), ("1", "0"))
|
|
|
|
def test_a_running_prefix_keeps_its_sync_choice(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)", fsync=True)
|
|
prefix = os.path.realpath("/compatdata/42/pfx")
|
|
command, first, _ = self.select(tool, self.env())
|
|
self.assertEqual(first["BL_FSYNC"], "1")
|
|
rows = json.loads((root / "prefixes.json").read_text())
|
|
self.assertEqual(rows[prefix]["env"]["BL_FSYNC"], "1")
|
|
self.assertIsNone(rows[prefix]["env"]["BL_SYNC_PACK"])
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
running = []
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: running.append(path) or True}):
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="0", WINEFSYNC=None))
|
|
self.assertEqual(running, [prefix])
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertEqual(env, self.env(BL_SYNC="0", BL_FSYNC="1", WINEFSYNC="1"))
|
|
self.assertEqual(said, "droiddeck-esync: Wine is still running in this prefix; keeping its sync choice (droiddeck-fsync)\n")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"]), (str(dist / "proton"), pack["id"]))
|
|
self.assertNotIn("BL_FSYNC", env)
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: True}):
|
|
command, env, said = self.select(tool, self.env(BL_FSYNC_FIRST="1"))
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"], env["WINESERVER"]), (str(dist / "proton"), pack["id"], str(dist / WINESERVER)))
|
|
self.assertNotIn("BL_FSYNC", env)
|
|
self.assertIn("(droiddeck-esync)", said)
|
|
other = make_tool(self.tools, "Proton 11.0 (ARM64)", salt=b"11", fsync=True)
|
|
self.assertEqual(self.select(other, self.env())[1]["BL_FSYNC"], "1")
|
|
shutil.rmtree(self.root / "dist")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"]), (str(dist / "proton"), pack["id"]))
|
|
self.assertNotIn("keeping", said)
|
|
(root / "prefixes.json").write_text("not json")
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: True}):
|
|
self.assertEqual(self.select(tool, self.env())[1]["BL_SYNC_PACK"], pack["id"])
|
|
self.assertEqual(json.loads((root / "prefixes.json").read_text())[prefix]["env"]["BL_SYNC_PACK"], pack["id"])
|
|
|
|
def test_a_running_prefix_drops_a_pack_that_no_longer_fits(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
self.assertEqual(self.select(tool, self.env())[0][0], str(dist / "proton"))
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: True}):
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"]), (str(dist / "proton"), pack["id"]))
|
|
self.assertIn("keeping its sync choice", said)
|
|
(tool / NTDLL).write_bytes(elf(EXPORTS, salt=b"updated in place"))
|
|
(tool / "version").write_text("1791000000 experimental-11.0-20261008-arm64\n")
|
|
command, env, said = self.select(tool, self.env(), verb="run")
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"]), (str(dist / "proton"), pack["id"]))
|
|
self.assertIn("keeping its sync choice", said)
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertNotIn("BL_SYNC_PACK", env)
|
|
self.assertNotIn("keeping", said)
|
|
self.assertIn("no pack for Proton Experimental (ARM64)", said)
|
|
other = make_tool(self.tools, "Proton 11.0 (ARM64)", "1788504814 proton-11.0-2c-arm64", salt=b"2c")
|
|
pack = make_pack(root, other, "valve-proton-11.0-2c-arm64-r1")
|
|
dist = self.dist_of(pack["id"], other)
|
|
self.assertEqual(self.select(other, self.env())[1]["BL_SYNC_PACK"], pack["id"])
|
|
rev2 = make_pack(root, other, "valve-proton-11.0-2c-arm64-r2", rev=2)
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: True}):
|
|
self.assertEqual(self.select(other, self.env(), verb="run")[1]["BL_SYNC_PACK"], pack["id"])
|
|
self.assertEqual(self.select(other, self.env())[1]["BL_SYNC_PACK"], rev2["id"])
|
|
for gone in (pack, rev2):
|
|
data = json.loads((root / "packs" / gone["id"] / "pack.json").read_text())
|
|
(root / "packs" / gone["id"] / "pack.json").write_text(json.dumps(dict(data, revoked=True)))
|
|
with mock.patch.dict(GLOBALS, {"server_running": lambda path: True}):
|
|
self.assertEqual(self.select(other, self.env(), verb="run")[1]["BL_SYNC_PACK"], rev2["id"])
|
|
command, env, said = self.select(other, self.env())
|
|
self.assertEqual(command[0], str(other / "proton"))
|
|
self.assertNotIn("keeping", said)
|
|
|
|
def test_droiddeck_ntsync_does_not_stand_in_when_esync_is_off(self):
|
|
self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
env = self.env(BL_SYNC="0")
|
|
env.pop("BL_SYNC_FALLBACK", None)
|
|
command, chosen, said = self.select(tool, env)
|
|
self.assertEqual(chosen, SYNC["wineserver_only"](env))
|
|
self.assertEqual(said, "droiddeck-esync: off; wineserver sync only\n")
|
|
|
|
def test_the_wineserver_tab_turns_proton_sync_off(self):
|
|
self.store()
|
|
tool = make_tool(self.tools, "GE-Proton11-9", "1790000000 GE-Proton11-9", builtin=True, ge=True, fsync=True)
|
|
command, env, said = self.select(tool, self.env(BL_SYNC_FALLBACK="0"))
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertEqual({name: env.get(name) for name in ("PROTON_NO_ESYNC", "PROTON_NO_FSYNC", "PROTON_NO_NTSYNC", "WINEESYNC", "WINEFSYNC", "BL_SYNC")},
|
|
{"PROTON_NO_ESYNC": "1", "PROTON_NO_FSYNC": "1", "PROTON_NO_NTSYNC": "1", "WINEESYNC": "0", "WINEFSYNC": "0", "BL_SYNC": "0"})
|
|
env = self.select(tool, self.env(BL_SYNC_FALLBACK="0", PROTON_NO_FSYNC="0"))[1]
|
|
self.assertEqual((env["PROTON_NO_FSYNC"], "WINEFSYNC" in env, env["PROTON_NO_ESYNC"]), ("0", False, "1"))
|
|
self.assertEqual(SYNC["describe"](SYNC["wineserver_only"]({})), "wineserver")
|
|
|
|
def test_the_ntsync_tab_needs_a_wine_with_ntsync(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton 11.0 (ARM64)", "1789000000 proton-11.0-3", fsync=True, ntsync=False)
|
|
command, env, said = self.select(tool, self.env(BL_SYNC="1"))
|
|
self.assertEqual((command[0], env["BL_SYNC"], env["BL_FSYNC"], env["WINEFSYNC"]), (str(tool / "proton"), "0", "1", "1"))
|
|
self.assertIn("droiddeck-ntsync: Proton 11.0 (ARM64) has no ntsync; trying droiddeck-fsync\n", said)
|
|
plain = make_tool(self.tools, "Proton 10.0 (ARM64)", "1780000000 proton-10.0-1", salt=b"p", ntsync=False)
|
|
command, env, said = self.select(plain, self.env(BL_SYNC="1"))
|
|
self.assertEqual((env["BL_SYNC"], "BL_FSYNC" in env), ("0", False))
|
|
self.assertTrue(said.endswith("; stock binaries (this Wine has no ntsync)\n"), said)
|
|
pack = make_pack(root, plain, "valve-10-r1")
|
|
command, env, said = self.select(plain, self.env(PROTON_NO_ESYNC="1"))
|
|
self.assertEqual((env["BL_SYNC_PACK"], env["WINEESYNC"], env.get("BL_SYNC")), (pack["id"], "0", None))
|
|
|
|
def test_built_in_esync_still_falls_back_when_esync_is_off(self):
|
|
self.store()
|
|
tool = make_tool(self.tools, "GE-Proton11-9", "1790000000 GE-Proton11-9", builtin=True, ge=True, fsync=True)
|
|
for extra in ({"PROTON_NO_ESYNC": "1"}, {"WINEESYNC": "0"}, {"STEAM_COMPAT_APP_ID": "2630", "PROTON_NO_FSYNC": "0"}):
|
|
command, env, said = self.select(tool, self.env(**extra))
|
|
self.assertEqual((command[0], env["BL_FSYNC"]), (str(tool / "proton"), "1"), extra)
|
|
plain = make_tool(self.tools, "GE-Proton11-8", "1789900000 GE-Proton11-8", salt=b"g", builtin=True, ge=True)
|
|
for extra in ({"PROTON_NO_ESYNC": "1"}, {"STEAM_COMPAT_APP_ID": "2630"}):
|
|
command, env, said = self.select(plain, self.env(**extra))
|
|
self.assertEqual((env["BL_SYNC"], env["PROTON_NO_ESYNC"], env["WINEESYNC"]), ("1", "1", "0"), extra)
|
|
self.assertTrue(said.endswith("; stock binaries with droiddeck-ntsync\n"), (extra, said))
|
|
command, env, said = self.select(plain, self.env())
|
|
self.assertEqual(said, "droiddeck-esync: GE-Proton11-8 has esync built in\n")
|
|
|
|
def test_a_stock_fallback_replaces_the_recorded_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
command, env, said = self.select(tool, self.env())
|
|
prefix = os.path.realpath("/compatdata/42/pfx")
|
|
rows = json.loads((root / "prefixes.json").read_text())
|
|
self.assertEqual(rows[prefix]["env"]["BL_SYNC_PACK"], pack["id"])
|
|
stock = [str(tool / "proton"), "waitforexitandrun", "game.exe"]
|
|
self.quiet(SYNC["record"], stock, self.env())
|
|
rows = json.loads((root / "prefixes.json").read_text())
|
|
self.assertEqual((rows[prefix]["command"], rows[prefix]["env"]["BL_SYNC_PACK"]), (stock[0], None))
|
|
|
|
def test_built_in_esync_never_uses_a_pack(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "GE-Proton11-9", "1790000000 GE-Proton11-9", builtin=True, ge=True)
|
|
make_pack(root, tool, "ge-builtin-r1")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertNotIn("WINEESYNC", env)
|
|
self.assertEqual(said, "droiddeck-esync: GE-Proton11-9 has esync built in\n")
|
|
self.assertFalse((root / "wanted.tsv").exists())
|
|
self.assertFalse((root / "dist").exists())
|
|
|
|
def test_droiddeck_proton_has_esync_built_in(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "DroidDeck-Proton 11-7.5", "1791000000 DroidDeck-Proton-11-7.5", builtin=True, ge=True, fsync=True)
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertNotIn("BL_SYNC_PACK", env)
|
|
self.assertEqual(said, "droiddeck-esync: DroidDeck-Proton 11-7.5 has esync built in\n")
|
|
self.assertFalse((root / "wanted.tsv").exists())
|
|
self.assertFalse((root / "dist").exists())
|
|
|
|
def test_wanted_rows_are_deduplicated(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "GE-Proton11-7", GE_LINE, ge=True)
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertEqual(said, "droiddeck-esync: no pack for GE-Proton11-7 (%s, ntdll %s); stock binaries with droiddeck-ntsync\n"
|
|
% (GE_LINE, sha(tool / NTDLL)[:12]))
|
|
self.assertEqual(env["BL_SYNC"], "1")
|
|
wanted = root / "wanted.tsv"
|
|
rows = [line.split("\t") for line in wanted.read_text().splitlines()]
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0][:5], [sha(tool / NTDLL), sha(tool / WINESERVER), "GE-Proton11-7",
|
|
SYNC["exports_hash"](tool / NTDLL), str(tool)])
|
|
self.assertTrue(rows[0][5].isdigit())
|
|
inode = os.stat(wanted).st_ino
|
|
self.select(tool, self.env())
|
|
self.assertEqual(os.stat(wanted).st_ino, inode)
|
|
copy = self.tmp / "elsewhere" / "GE-Proton11-7"
|
|
shutil.copytree(tool, copy, symlinks=True)
|
|
self.select(copy, self.env())
|
|
rows_after = [line.split("\t") for line in wanted.read_text().splitlines()]
|
|
self.assertEqual(len(rows_after), 1)
|
|
self.assertEqual((rows_after[0][4], rows_after[0][5]), (str(copy), rows[0][5]))
|
|
other = make_tool(self.tools, "GE-Proton11-8", "1790000001 GE-Proton11-8", salt=b"8", ge=True)
|
|
self.select(other, self.env())
|
|
self.assertEqual(len(wanted.read_text().splitlines()), 2)
|
|
|
|
|
|
class MatchingTest(SyncTestCase):
|
|
def test_exact_beats_source_then_rev_then_id(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton 11.0 (ARM64)", "1789000000 proton-11.0-3")
|
|
other = {NTDLL: "1" * 64, WINESERVER: "2" * 64}
|
|
find = SYNC["find_pack"]
|
|
exact1 = make_pack(root, tool, "a-exact-r1", rev=1)
|
|
source9 = make_pack(root, tool, "b-source-r9", rev=9, source_match=True, stock=other)
|
|
measured = SYNC["measure"](tool)
|
|
self.assertIs(find([exact1, source9], measured), exact1)
|
|
exact2 = make_pack(root, tool, "c-exact-r2", rev=2)
|
|
self.assertIs(find([exact1, source9, exact2], measured), exact2)
|
|
exact2b = make_pack(root, tool, "d-exact-r2", rev=2)
|
|
self.assertIs(find([exact2b, exact1, exact2, source9], measured), exact2b)
|
|
source3 = make_pack(root, tool, "e-source-r3", rev=3, source_match=True, stock=other)
|
|
self.assertIs(find([source3, source9], measured), source9)
|
|
self.assertIsNone(find([make_pack(root, tool, "f-nosource", rev=5, stock=other)], measured))
|
|
self.assertIsNone(find([make_pack(root, tool, "g-exports", source_match=True, stock=other, exports="3" * 64)], measured))
|
|
self.assertIsNone(find([make_pack(root, tool, "h-version", source_match=True, stock=other, version="proton-11.0-4")], measured))
|
|
self.assertIsNone(find([dict(exact2, revoked=True)], measured))
|
|
loaded = {pack["id"] for pack in SYNC["load_packs"](root)}
|
|
self.assertEqual(loaded, {"a-exact-r1", "b-source-r9", "c-exact-r2", "d-exact-r2", "e-source-r3", "f-nosource",
|
|
"g-exports", "h-version"})
|
|
self.assertEqual(SYNC["find_pack"](SYNC["load_packs"](root), SYNC["measure"](tool))["id"], "d-exact-r2")
|
|
|
|
def test_identical_releases_prefer_the_pack_built_for_the_installed_one(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton 11.0 (ARM64)", "1787950357 proton-11.0-2b-arm64")
|
|
find = SYNC["find_pack"]
|
|
own = make_pack(root, tool, "valve-proton-11.0-2b-arm64-r1")
|
|
newer = make_pack(root, tool, "valve-proton-11.0-2c-arm64-r1", version="proton-11.0-2c-arm64",
|
|
version_line="1788504814 proton-11.0-2c-arm64")
|
|
older = make_pack(root, tool, "valve-proton-11.0-2a-arm64-r1", version="proton-11.0-2a-arm64",
|
|
version_line="1787000000 proton-11.0-2a-arm64")
|
|
measured = SYNC["measure"](tool)
|
|
self.assertIs(find([newer, own], measured), own)
|
|
self.assertIsNone(find([newer], measured))
|
|
self.assertIsNone(SYNC["match_kind"](newer, measured))
|
|
self.assertIs(find([newer, older], measured), older)
|
|
own2 = make_pack(root, tool, "valve-proton-11.0-2b-arm64-r2", rev=2)
|
|
older2 = make_pack(root, tool, "valve-proton-11.0-2a-arm64-r2", rev=2, version="proton-11.0-2a-arm64",
|
|
version_line="1787000000 proton-11.0-2a-arm64")
|
|
self.assertIs(find([own, older2], measured), own)
|
|
self.assertIs(find([own, older2, own2], measured), own2)
|
|
self.assertEqual(find(SYNC["load_packs"](root), measured)["id"], "valve-proton-11.0-2b-arm64-r2")
|
|
unknown = make_tool(self.tools, "Proton Hotfix", "proton-hotfix", salt=b"hotfix")
|
|
sibling = make_pack(root, unknown, "valve-hotfix-sibling-r1", version="proton-11.0-2c-arm64",
|
|
version_line="1788504814 proton-11.0-2c-arm64")
|
|
self.assertIs(find([sibling], SYNC["measure"](unknown)), sibling)
|
|
|
|
def test_exports_are_measured_only_when_needed(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton 11.0 (ARM64)", "1789000000 proton-11.0-3")
|
|
exact = make_pack(root, tool, "a-exact-r1")
|
|
measured = SYNC["measure"](tool)
|
|
self.assertEqual(SYNC["find_pack"]([exact], measured), exact)
|
|
self.assertNotIn("exports", measured)
|
|
other = {NTDLL: "1" * 64, WINESERVER: "2" * 64}
|
|
unrelated = make_pack(root, tool, "b-source-other", source_match=True, stock=other, version="proton-10.0-1")
|
|
self.assertIsNone(SYNC["find_pack"]([unrelated], measured))
|
|
self.assertNotIn("exports", measured)
|
|
source = make_pack(root, tool, "c-source", source_match=True, stock=other)
|
|
self.assertEqual(SYNC["find_pack"]([source], measured), source)
|
|
self.assertEqual(measured["exports"], SYNC["exports_hash"](tool / NTDLL))
|
|
|
|
def test_only_complete_valid_packs_load(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton 11.0 (ARM64)")
|
|
make_pack(root, tool, "good")
|
|
make_pack(root, tool, "incomplete", complete=False)
|
|
make_pack(root, tool, "revoked", revoked=True)
|
|
make_pack(root, tool, "badcopy", copy=["../escape"])
|
|
make_pack(root, tool, "badrev", rev="1")
|
|
make_pack(root, tool, "badhash", files={NTDLL: "x", WINESERVER: "y"})
|
|
make_pack(root, tool, "ignored")
|
|
(root / "packs/ignored/.ignored").write_text("")
|
|
make_pack(root, tool, "nofiles")
|
|
(root / "packs/nofiles" / WINESERVER).unlink()
|
|
make_pack(root, tool, "renamed")
|
|
(root / "packs/renamed").rename(root / "packs/other-name")
|
|
make_pack(root, tool, "badformat", format=True)
|
|
make_pack(root, tool, ".hidden")
|
|
(root / "packs/broken").mkdir()
|
|
(root / "packs/broken/.complete").write_text("")
|
|
(root / "packs/broken/pack.json").write_text("{")
|
|
self.assertEqual([pack["id"] for pack in SYNC["load_packs"](root)], ["good"])
|
|
|
|
|
|
class ReconcileTest(SyncTestCase):
|
|
def test_states_tables_and_prebuilt_dists(self):
|
|
root = self.store()
|
|
steam = self.tmp / "steam"
|
|
sd = self.tmp / "sd"
|
|
valve = make_tool(steam / "steamapps/common", "Proton Experimental (ARM64)")
|
|
builtin = make_tool(sd / "steamapps/common", "Proton 11.0 (ARM64)", "1789000000 proton-11.0-3", salt=b"b", builtin=True)
|
|
tools = steam / "compatibilitytools.d"
|
|
ge = make_tool(tools, "GE-Proton11-7", GE_LINE, salt=b"ge", ge=True, decorated=True)
|
|
make_tool(tools, "droiddeck-proton-arm64", salt=b"own")
|
|
make_tool(tools, ".hidden", salt=b"hidden")
|
|
(make_tool(tools, "incomplete", salt=b"inc") / WINESERVER).unlink()
|
|
loose = make_tool(self.tmp / "loose", "GE-Proton10-1", "1700000000 GE-Proton10-1", salt=b"loose")
|
|
gone = make_tool(self.tmp / "gone", "GE-Proton9-1", "1600000000 GE-Proton9-1", salt=b"gone")
|
|
pack = make_pack(root, valve, "valve-experimental-r1")
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
self.select(loose, self.env())
|
|
self.select(gone, self.env())
|
|
shutil.rmtree(gone)
|
|
with mock.patch.dict(GLOBALS, {"SD_LIBRARY": str(sd)}):
|
|
summary = self.quiet(SYNC["reconcile"], str(steam), {"HOME": str(self.home)})
|
|
self.assertEqual(summary, "droiddeck-esync: 1 pack(s) in use, 2 wanted")
|
|
self.assertFalse((root / "dist").exists())
|
|
rows = {row[0]: row[1:] for row in (line.split("\t") for line in (root / "tools.tsv").read_text().splitlines())}
|
|
self.assertEqual(rows, {
|
|
str(valve): [VALVE_LINE, sha(valve / NTDLL), sha(valve / WINESERVER), "pack", pack["id"]],
|
|
str(builtin): ["1789000000 proton-11.0-3", sha(builtin / NTDLL), sha(builtin / WINESERVER), "builtin", "-"],
|
|
str(ge): [GE_LINE, sha(ge / NTDLL), sha(ge / WINESERVER), "wanted", "-"],
|
|
})
|
|
wanted = [line.split("\t") for line in (root / "wanted.tsv").read_text().splitlines()]
|
|
self.assertEqual(sorted(row[4] for row in wanted), sorted([str(loose), str(ge)]))
|
|
self.assertTrue((root / "hash-cache.tsv").is_file())
|
|
summary = self.quiet(SYNC["reconcile"], str(steam), {"HOME": str(self.home), "BL_SYNC_FALLBACK": "1"})
|
|
self.assertEqual(summary, "droiddeck-esync: 1 pack(s) in use, 2 wanted")
|
|
dist = self.dist_of(pack["id"], valve)
|
|
self.assertEqual(json.loads((dist / RECORD).read_text())["tool"], str(valve))
|
|
make_pack(root, ge, "ge-GE-Proton11-7-r1", flavor="ge")
|
|
make_pack(root, loose, "ge-GE-Proton10-1-r1", flavor="ge")
|
|
summary = self.quiet(SYNC["reconcile"], str(steam), {"HOME": str(self.home), "BL_SYNC_FALLBACK": "1"})
|
|
self.assertEqual(summary, "droiddeck-esync: 2 pack(s) in use, 0 wanted")
|
|
self.assertEqual((root / "wanted.tsv").read_text(), "")
|
|
self.assertTrue(self.dist_of("ge-GE-Proton11-7-r1", ge).is_dir())
|
|
self.assertIn("ge-GE-Proton11-7-r1", (root / "tools.tsv").read_text())
|
|
|
|
def test_droiddeck_proton_is_scanned_but_our_own_tools_are_not(self):
|
|
steam = self.tmp / "steam"
|
|
tools = steam / "compatibilitytools.d"
|
|
ours = make_tool(tools, "DroidDeck-Proton 11-7.5", "1791000000 DroidDeck-Proton-11-7.5", salt=b"dd", builtin=True, ge=True)
|
|
make_tool(tools, "droiddeck-proton-arm64", salt=b"own")
|
|
make_tool(tools, "droiddeck-proton-11-arm64", salt=b"own11")
|
|
with mock.patch.dict(GLOBALS, {"SD_LIBRARY": str(self.tmp / "sd")}):
|
|
self.assertEqual(SYNC["find_tools"](str(steam)), [str(ours)])
|
|
|
|
def test_steam_compatibility_reports_sync(self):
|
|
steam = self.tmp / "steam"
|
|
make_tool(steam / "steamapps/common", "Proton Experimental (ARM64)")
|
|
env = {"PATH": os.defpath, "HOME": str(self.home), "BL_SYNC_FALLBACK": "1"}
|
|
result = subprocess.run([sys.executable, str(BIN / "steam-compatibility"), str(steam)], env=env,
|
|
capture_output=True, text=True, check=True)
|
|
self.assertNotIn("droiddeck-esync", result.stdout + result.stderr)
|
|
self.assertFalse(self.root.exists())
|
|
root = self.store()
|
|
make_pack(root, steam / "steamapps/common/Proton Experimental (ARM64)", "valve-experimental-r1")
|
|
result = subprocess.run([sys.executable, str(BIN / "steam-compatibility"), str(steam)], env=env,
|
|
capture_output=True, text=True, check=True)
|
|
self.assertIn("steam-compatibility: droiddeck-esync: 1 pack(s) in use, 0 wanted\n", result.stdout)
|
|
self.assertTrue(self.dist_of("valve-experimental-r1", steam / "steamapps/common/Proton Experimental (ARM64)").is_dir())
|
|
|
|
|
|
def test_an_updated_proton_drops_its_pack_until_one_is_built_for_it(self):
|
|
root = self.store()
|
|
steam = self.tmp / "steam"
|
|
tool = make_tool(steam / "steamapps/common", "Proton Experimental (ARM64)")
|
|
pack = make_pack(root, tool, "valve-experimental-20261001-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
self.assertEqual(self.select(tool, self.env())[0][0], str(dist / "proton"))
|
|
(tool / NTDLL).write_bytes(elf(EXPORTS, salt=b"20261008"))
|
|
(tool / WINESERVER).write_bytes(b"wineserver 20261008 /dev/ntsync")
|
|
line = "1791000000 experimental-11.0-20261008-arm64"
|
|
(tool / "version").write_text(line + "\n")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual(command[0], str(tool / "proton"))
|
|
self.assertNotIn(pack["id"], said)
|
|
with mock.patch.dict(GLOBALS, {"SD_LIBRARY": str(self.tmp / "sd")}):
|
|
summary = self.quiet(SYNC["reconcile"], str(steam), {"HOME": str(self.home)})
|
|
self.assertEqual(summary, "droiddeck-esync: 0 pack(s) in use, 1 wanted")
|
|
rows = {row[0]: row[1:] for row in (line.split("\t") for line in (root / "tools.tsv").read_text().splitlines())}
|
|
self.assertEqual(rows, {str(tool): [line, sha(tool / NTDLL), sha(tool / WINESERVER), "wanted", "-"]})
|
|
wanted = [row.split("\t") for row in (root / "wanted.tsv").read_text().splitlines()]
|
|
self.assertEqual([(row[0], row[1], row[2], row[4]) for row in wanted],
|
|
[(sha(tool / NTDLL), sha(tool / WINESERVER), "experimental-11.0-20261008-arm64", str(tool))])
|
|
old = time.time() - 2 * SYNC["GC_GRACE"]
|
|
os.utime(dist, (old, old))
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: []}):
|
|
SYNC["gc"](root)
|
|
self.assertFalse(dist.exists())
|
|
new = make_pack(root, tool, "valve-experimental-20261008-r1")
|
|
command, env, said = self.select(tool, self.env())
|
|
self.assertEqual((command[0], env["BL_SYNC_PACK"]), (str(self.dist_of(new["id"], tool) / "proton"), new["id"]))
|
|
summary = self.quiet(SYNC["reconcile"], str(steam), {"HOME": str(self.home)})
|
|
self.assertEqual(summary, "droiddeck-esync: 1 pack(s) in use, 0 wanted")
|
|
|
|
|
|
class GcTest(SyncTestCase):
|
|
def test_removes_stale_dists_unless_in_use(self):
|
|
root = self.store()
|
|
keep = make_tool(self.tools, "keep", salt=b"keep")
|
|
uninstalled = make_tool(self.tools, "uninstalled", salt=b"uninstalled")
|
|
updated = make_tool(self.tools, "updated", salt=b"updated")
|
|
running = make_tool(self.tools, "running", salt=b"running")
|
|
recent = make_tool(self.tools, "recent", salt=b"recent")
|
|
dists = {}
|
|
for tool in (keep, uninstalled, updated, running, recent):
|
|
pack = make_pack(root, tool, "pack-" + tool.name)
|
|
dists[tool.name] = Path(SYNC["ensure_dist"](root, tool, pack))
|
|
for name in ("uninstalled", "running", "recent"):
|
|
shutil.rmtree(root / "packs" / ("pack-" + name))
|
|
(updated / NTDLL).write_bytes(elf(EXPORTS, salt=b"new build"))
|
|
old = time.time() - 2 * SYNC["GC_GRACE"]
|
|
for name in ("keep", "uninstalled", "updated", "running"):
|
|
os.utime(dists[name], (old, old))
|
|
stale_temp = root / "dist" / ".pack-x~00000000.tmp-123"
|
|
stale_temp.mkdir()
|
|
os.utime(stale_temp, (time.time() - 2 * SYNC["TEMP_AGE"],) * 2)
|
|
fresh_temp = root / "dist" / ".pack-y~00000000.tmp-456"
|
|
fresh_temp.mkdir()
|
|
(root / "dist" / "ghost~00000000.lock").write_text("")
|
|
cmdline = b"/usr/bin/python3\0" + str(dists["running"] / "proton").encode() + b"\0waitforexitandrun\0"
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: None}):
|
|
self.assertEqual(SYNC["gc"](root), 0)
|
|
self.assertTrue(dists["uninstalled"].exists())
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: [b"bash\0", cmdline]}):
|
|
self.assertEqual(SYNC["gc"](root), 2)
|
|
remaining = sorted(os.listdir(root / "dist"))
|
|
self.assertEqual(remaining, sorted([dists["keep"].name, dists["keep"].name + ".lock", dists["running"].name,
|
|
dists["running"].name + ".lock", dists["recent"].name, dists["recent"].name + ".lock",
|
|
fresh_temp.name]))
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: []}):
|
|
os.utime(dists["recent"], (old, old))
|
|
self.assertEqual(SYNC["gc"](root), 2)
|
|
self.assertEqual(sorted(os.listdir(root / "dist")), sorted([dists["keep"].name, dists["keep"].name + ".lock",
|
|
fresh_temp.name]))
|
|
|
|
def test_superseded_pack_revisions_are_collected(self):
|
|
root = self.store()
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
first = make_pack(root, tool, "valve-experimental-r1", rev=1)
|
|
first_dist = Path(SYNC["ensure_dist"](root, tool, first))
|
|
second = make_pack(root, tool, "valve-experimental-r2", rev=2)
|
|
second_dist = Path(SYNC["ensure_dist"](root, tool, second))
|
|
self.assertNotEqual(first_dist, second_dist)
|
|
old = time.time() - 2 * SYNC["GC_GRACE"]
|
|
for dist in (first_dist, second_dist):
|
|
os.utime(dist, (old, old))
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: [str(first_dist / "proton").encode() + b"\0run\0"]}):
|
|
self.assertEqual(SYNC["gc"](root), 0)
|
|
with mock.patch.dict(GLOBALS, {"process_cmdlines": lambda: []}):
|
|
self.assertEqual(SYNC["gc"](root), 1)
|
|
self.assertEqual(sorted(os.listdir(root / "dist")), [second_dist.name, second_dist.name + ".lock"])
|
|
self.assertTrue((root / "packs" / first["id"] / ".complete").is_file())
|
|
_, env, _ = self.select(tool, self.env())
|
|
self.assertEqual(env["BL_SYNC_PACK"], second["id"])
|
|
|
|
|
|
class LauncherTest(SyncTestCase):
|
|
def test_game_env_applies_sync_keys_for_every_verb(self):
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
config = self.home / ".config/droiddeck/game-environment.json"
|
|
config.parent.mkdir(parents=True)
|
|
config.write_text(json.dumps({"version": 1, "shared": {"PROTON_NO_NTSYNC": "1", "CUSTOM": "shared"},
|
|
"games": {"42": {"WINEESYNC": "3", "BL_SYNC_FALLBACK": "1"}}}))
|
|
env = {"PATH": os.defpath, "HOME": str(self.home), "STEAM_COMPAT_DATA_PATH": "/compatdata/42", "BL_SYNC": "1",
|
|
"BL_SYNC_MIN_NOFILE": "1"}
|
|
|
|
def launch(verb):
|
|
output = subprocess.run([sys.executable, str(BIN / "droiddeck-game-env"), str(tool / "proton"), verb, "a b"],
|
|
env=env, capture_output=True, text=True, check=True).stdout
|
|
return json.loads(output)
|
|
|
|
argv0, args, seen = launch("run")
|
|
self.assertEqual((argv0, args), (str(tool / "proton"), ["run", "a b"]))
|
|
self.assertEqual((seen["CUSTOM"], seen["BL_SYNC"], seen["WINEESYNC"]), (None, "0", "3"))
|
|
self.assertEqual(launch("waitforexitandrun")[2]["CUSTOM"], "shared")
|
|
root = self.store()
|
|
pack = make_pack(root, tool, "valve-experimental-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
for verb in ("run", "waitforexitandrun", "getcompatpath"):
|
|
argv0, args, seen = launch(verb)
|
|
self.assertEqual((argv0, args), (str(dist / "proton"), [verb, "a b"]))
|
|
self.assertEqual((seen["WINEESYNC"], seen["BL_SYNC_PACK"], seen["WINESERVER"]), ("3", pack["id"], str(dist / WINESERVER)))
|
|
|
|
def test_failed_dist_exec_keeps_the_sync_choice(self):
|
|
tool = make_tool(self.tools, "Proton Experimental (ARM64)")
|
|
pack = make_pack(self.store(), tool, "valve-experimental-r1")
|
|
dist = self.dist_of(pack["id"], tool)
|
|
game_env = runpy.run_path(str(BIN / "droiddeck-game-env"))
|
|
command = [str(tool / "proton"), "waitforexitandrun", "game.exe"]
|
|
env = self.env(STEAM_COMPAT_APP_ID="2630", PROTON_NO_NTSYNC="1", BL_SYNC="1", WINEESYNC="1")
|
|
calls = []
|
|
|
|
def execvpe(path, args, environment):
|
|
calls.append((path, list(args), dict(environment)))
|
|
if len(calls) == 1:
|
|
raise PermissionError("denied")
|
|
raise SystemExit(0)
|
|
|
|
with mock.patch.dict(os.environ, env, clear=True), mock.patch.object(sys, "argv", ["droiddeck-game-env"] + command), \
|
|
mock.patch("os.execvpe", execvpe), contextlib.redirect_stderr(io.StringIO()) as said:
|
|
with self.assertRaises(SystemExit):
|
|
game_env["main"]()
|
|
self.assertEqual([call[:2] for call in calls], [(str(dist / "proton"), [str(dist / "proton")] + command[1:]),
|
|
(str(tool / "proton"), command)])
|
|
self.assertEqual((calls[0][2]["WINEESYNC"], calls[0][2]["PROTON_NO_ESYNC"], calls[0][2]["BL_SYNC_PACK"]), ("0", "1", pack["id"]))
|
|
self.assertEqual(calls[1][2], self.env(STEAM_COMPAT_APP_ID="2630", PROTON_NO_NTSYNC="1", BL_SYNC="0", WINEESYNC="1",
|
|
vblank_mode="0"))
|
|
self.assertIn("could not start (", said.getvalue())
|
|
|
|
def test_generated_launchers_run_the_dist_proton(self):
|
|
root = self.store()
|
|
steam = self.tmp / "Steam"
|
|
tools = steam / "compatibilitytools.d"
|
|
depot = make_tool(steam / "steamapps/common", COMPAT["SOURCES"][0])
|
|
extra = make_tool(tools, "GE-Proton11-7", GE_LINE, salt=b"ge", ge=True)
|
|
(extra / "toolmanifest.vdf").write_text(MANIFEST)
|
|
COMPAT["build_tool"](str(tools / COMPAT["TOOL"]), str(depot))
|
|
COMPAT["adopt_extras"](str(tools))
|
|
packs = {depot: make_pack(root, depot, "valve-experimental-r1"), extra: make_pack(root, extra, "ge-GE-Proton11-7-r1", flavor="ge")}
|
|
wrappers = {depot: tools / COMPAT["TOOL"] / COMPAT["LAUNCHER"], extra: extra / COMPAT["EXTRA_WRAPPER"]}
|
|
env = {"PATH": os.defpath, "HOME": str(self.home), "STEAM_COMPAT_DATA_PATH": "/compatdata/42",
|
|
"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(steam), "BL_SYNC_FALLBACK": "1", "BL_SYNC_MIN_NOFILE": "1"}
|
|
for tool, wrapper in wrappers.items():
|
|
wrapper.write_text(wrapper.read_text().replace("/usr/local/bin/droiddeck-game-env", str(BIN / "droiddeck-game-env")))
|
|
dist = self.dist_of(packs[tool]["id"], tool)
|
|
result = subprocess.run([str(wrapper), "waitforexitandrun", "game with spaces.exe"], env=env, text=True, capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
argv0, args, seen = json.loads(result.stdout)
|
|
self.assertEqual((argv0, args), (str(dist / "proton"), ["waitforexitandrun", "game with spaces.exe"]))
|
|
self.assertEqual((seen["BL_SYNC_PACK"], seen["WINESERVER"], seen["WINEESYNC"]),
|
|
(packs[tool]["id"], str(dist / WINESERVER), "1"))
|
|
self.assertIn("droiddeck-esync: pack %s" % packs[tool]["id"], result.stderr)
|
|
for name in ("droiddeck-proton-wrap", "toolmanifest.vdf", "compatibilitytool.vdf", "toolmanifest.vdf.droiddeck-orig",
|
|
"bannerlator-proton-wrap", "toolmanifest.vdf.bannerlator-orig"):
|
|
self.assertFalse(os.path.lexists(dist / name), name)
|
|
result = subprocess.run([str(wrapper), "waitforexitandrun"], env={**env, "BL_SYNC_FALLBACK": "0"}, text=True, capture_output=True)
|
|
self.assertEqual(json.loads(result.stdout)[0], str(tool / "proton"), result.stderr)
|
|
|
|
def test_session_collects_sync_tables(self):
|
|
script = (BIN / "droiddeck-session").read_text()
|
|
start = script.index("collect_steam_logs() {")
|
|
function = script[start:script.index("\n}\n", start) + 3]
|
|
debug = self.tmp / "debug"
|
|
env = {"PATH": os.defpath, "HOME": str(self.home), "BL_DEBUG_DIR": str(debug)}
|
|
subprocess.run(["bash", "-c", "set -u\n" + function + "collect_steam_logs"], env=env, check=True, capture_output=True)
|
|
self.assertFalse((debug / "droiddeck-esync").exists())
|
|
root = self.store()
|
|
(root / "launches.log").write_text("launch\n")
|
|
(root / "tools.tsv").write_text("tool\n")
|
|
(root / "hash-cache.tsv").write_text("cache\n")
|
|
subprocess.run(["bash", "-c", "set -u\n" + function + "collect_steam_logs"], env=env, check=True, capture_output=True)
|
|
self.assertEqual(sorted(os.listdir(debug / "droiddeck-esync")), ["launches.log", "tools.tsv"])
|
|
|
|
|
|
class BundleTest(unittest.TestCase):
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
def release_env(self):
|
|
values = {}
|
|
for line in (self.ROOT / "tools/droiddeck-esync/release.env").read_text().splitlines():
|
|
key, _, value = line.partition("=")
|
|
values[key] = value
|
|
return values
|
|
|
|
def test_release_env_pins_a_components_bundle(self):
|
|
env = self.release_env()
|
|
self.assertEqual(set(env), {"SYNC_BUNDLE_REPO", "SYNC_BUNDLE_TAG", "SYNC_BUNDLE_ASSET", "SYNC_BUNDLE_SHA256"})
|
|
self.assertEqual(env["SYNC_BUNDLE_REPO"], "Droid-Deck/DroidDeck-Components")
|
|
self.assertEqual(env["SYNC_BUNDLE_TAG"], "droiddeck-esync-index")
|
|
self.assertRegex(env["SYNC_BUNDLE_ASSET"], r"^bundle-[0-9]{8}-[0-9]{6}\.tzst$")
|
|
self.assertRegex(env["SYNC_BUNDLE_SHA256"], r"^[0-9a-f]{64}$")
|
|
packs = (self.ROOT / "app/src/main/java/com/droiddeck/launcher/session/EsyncPacks.kt").read_text()
|
|
self.assertIn(f'const val REPO = "{env["SYNC_BUNDLE_REPO"]}"', packs)
|
|
self.assertIn('releases/download/{}/index.json"'.format(env["SYNC_BUNDLE_TAG"]), packs)
|
|
|
|
def test_builds_fetch_the_pinned_bundle(self):
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
self.skipTest("PyYAML is not installed")
|
|
apk = yaml.safe_load((self.ROOT / ".github/workflows/build.yml").read_text())
|
|
steps = {step.get("name"): step for step in apk["jobs"]["components"]["steps"]}
|
|
import json
|
|
recipes = json.loads((self.ROOT / "tools/components/recipes.json").read_text())
|
|
bundle = dict(steps["Bundle the droiddeck-esync packs"])
|
|
self.assertIn("prepare.py esync", bundle["run"])
|
|
bundle["run"] = (self.ROOT / recipes["esync"]["script"]).read_text()
|
|
self.assertNotIn("if", bundle)
|
|
self.assertIn("tools/droiddeck-esync/release.env", bundle["run"])
|
|
self.assertIn('release_url="https://github.com/${SYNC_BUNDLE_REPO}/releases/download/${SYNC_BUNDLE_TAG}"', bundle["run"])
|
|
self.assertIn('curl -fsSL --retry 5 --retry-all-errors --connect-timeout 20 -o sync-bundle.tzst "$release_url/$SYNC_BUNDLE_ASSET"', bundle["run"])
|
|
self.assertIn('curl -fsSL --retry 5 --retry-all-errors --connect-timeout 20 -o sync-index.json "$release_url/index.json"', bundle["run"])
|
|
self.assertIn("sha256sum -c -", bundle["run"])
|
|
self.assertNotIn("revoked.txt", bundle["run"])
|
|
stage = {"run": (self.ROOT / recipes["runtime"]["script"]).read_text()}
|
|
self.assertIn("tools/linuxfs/overlay/usr/local/bin/droiddeck-* ", stage["run"])
|
|
local = (self.ROOT / "tools/build_local.sh").read_text()
|
|
self.assertIn("tools/linuxfs/overlay/usr/local/bin/droiddeck-* ", local)
|
|
self.assertIn('"${SYNC_BUNDLE_REPO}"', local)
|
|
self.assertNotIn("revoked.txt", local)
|
|
self.assertFalse((self.ROOT / ".github/workflows/build-droiddeck-esync-packs.yml").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|