Daniele Fucini
047139cb72
Look for discord-ipc sockets in the Flatpak and Snap runtime folders too. When a game moves to another process (e.g. at Steam launch), keep its first start time and don't resend the activity. Bump version to 0.2.
396 lines
14 KiB
Python
396 lines
14 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
import struct
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import discord_game_presence as dgp
|
|
|
|
GENSHIN: dict[str, Any] = {
|
|
"id": "762434991303950386",
|
|
"name": "Genshin Impact",
|
|
"executables": [{"is_launcher": False, "name": "genshinimpact.exe", "os": "win32"}],
|
|
}
|
|
NEVERWINTER: dict[str, Any] = {
|
|
"id": "406645401437863936",
|
|
"name": "Neverwinter",
|
|
"executables": [
|
|
{"is_launcher": False, "name": "neverwinter/live/gameclient.exe", "os": "win32"},
|
|
{"is_launcher": False, "name": "neverwinter.exe", "os": "win32"},
|
|
],
|
|
"third_party_skus": [{"distributor": "xbox", "id": "BV40D6Q97PP8"}, {"distributor": "steam", "id": "109600"}],
|
|
}
|
|
STO = {
|
|
"id": "448384255131779073",
|
|
"name": "Star Trek Online",
|
|
"executables": [{"is_launcher": False, "name": "star trek online/live/gameclient.exe", "os": "win32"}],
|
|
}
|
|
GMOD = {
|
|
"id": "1",
|
|
"name": "Garry's Mod",
|
|
"executables": [{"arguments": "-game garrysmod", "is_launcher": False, "name": ">hl2.exe", "os": "win32"}],
|
|
}
|
|
LAUNCHER = {
|
|
"id": "2",
|
|
"name": "Launcher Game",
|
|
"executables": [{"is_launcher": True, "name": "launcher.exe", "os": "win32"}],
|
|
}
|
|
LINUX_TOOL = {
|
|
"id": "3",
|
|
"name": "Linux Tool",
|
|
"executables": [{"is_launcher": False, "name": "tool", "os": "linux"}],
|
|
}
|
|
APPS: list[dict[str, Any]] = [GENSHIN, NEVERWINTER, STO, GMOD, LAUNCHER, LINUX_TOOL]
|
|
|
|
REAPER = [
|
|
"/home/u/.local/share/Steam/ubuntu12_32/reaper",
|
|
"SteamLaunch",
|
|
"AppId=109600",
|
|
"--",
|
|
"/home/u/.local/share/Steam/steamapps/common/Proton 9.0 (Beta)/proton",
|
|
"waitforexitandrun",
|
|
]
|
|
|
|
|
|
def index(config: dgp.Config | None = None) -> dgp.GameIndex:
|
|
return dgp.GameIndex(APPS, config or dgp.Config())
|
|
|
|
|
|
def name(app: dgp.App | None) -> str | None:
|
|
return app.name if app else None
|
|
|
|
|
|
# matching
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("argv0", "expected"),
|
|
[
|
|
("GenshinImpact.exe", "Genshin Impact"),
|
|
("Z:\\home\\u\\Steam\\steamapps\\common\\Cryptic Studios\\Neverwinter.exe", "Neverwinter"),
|
|
("C:/Games/Neverwinter/Live/GameClient.exe", "Neverwinter"),
|
|
("C:/Games/Star Trek Online/Live/GameClient.exe", "Star Trek Online"),
|
|
("Z:/Games/Cryptic Studios/Neverwinter/Live/x64/GameClient.exe", None), # not a suffix
|
|
("GameClient.exe", None),
|
|
("NotNeverwinter.exe", None), # suffix only at a '/' boundary
|
|
("C:\\windows\\system32\\services.exe", None),
|
|
("launcher.exe", None), # is_launcher entries are skipped
|
|
("/usr/bin/tool", None), # linux entries are off by default
|
|
],
|
|
)
|
|
def test_match_exe(argv0: str, expected: str | None) -> None:
|
|
assert name(index().match([argv0])) == expected
|
|
|
|
|
|
def test_linux_executables_option() -> None:
|
|
assert name(index(dgp.Config(linux_executables=True)).match(["/usr/bin/tool"])) == "Linux Tool"
|
|
|
|
|
|
def test_arguments_required() -> None:
|
|
assert name(index().match(["hl2.exe", "-game", "garrysmod"])) == "Garry's Mod"
|
|
assert index().match(["hl2.exe", "-game", "tf"]) is None
|
|
|
|
|
|
def test_steam_app_id() -> None:
|
|
assert name(index().match(REAPER)) == "Neverwinter"
|
|
assert index().match([REAPER[0], "SteamLaunch", "AppId=999"]) is None
|
|
# the game path after "--" or in another process doesn't count
|
|
assert index().match(["steam.exe", "SteamLaunch", "AppId=109600"]) is None
|
|
assert index().match([REAPER[0], "SteamLaunch", "--", "AppId=109600"]) is None
|
|
|
|
|
|
def test_ignore_by_id_and_name() -> None:
|
|
config = dgp.Config(ignore=frozenset({"762434991303950386", "neverwinter"}))
|
|
assert index(config).match(["GenshinImpact.exe"]) is None
|
|
assert index(config).match(REAPER) is None
|
|
|
|
|
|
def test_extra_executables_win() -> None:
|
|
config = dgp.Config(executables={"genshinimpact.exe": "406645401437863936", "my/game.exe": "555"})
|
|
assert name(index(config).match(["GenshinImpact.exe"])) == "Neverwinter"
|
|
assert index(config).match(["Z:\\x\\My\\Game.exe"]) == dgp.App("555", "555")
|
|
|
|
|
|
# /proc scanning
|
|
|
|
|
|
def add_proc(proc: Path, pid: int, argv: list[str], ticks: int, comm: str = "x", pad: int = 0) -> None:
|
|
d = proc / str(pid)
|
|
d.mkdir()
|
|
(d / "cmdline").write_bytes(("\0".join(argv) + "\0" * (pad + 1)).encode() if argv else b"")
|
|
fields = ["S"] + ["0"] * 18 + [str(ticks)] + ["0"] * 10
|
|
(d / "stat").write_text(f"{pid} ({comm}) {' '.join(fields)}\n")
|
|
|
|
|
|
@pytest.fixture
|
|
def proc(tmp_path: Path) -> Path:
|
|
root = tmp_path / "proc"
|
|
root.mkdir()
|
|
(root / "stat").write_text("cpu 1 2 3\nbtime 1000000\n")
|
|
(root / "self").mkdir()
|
|
return root
|
|
|
|
|
|
def test_scan(proc: Path) -> None:
|
|
hz = os.sysconf("SC_CLK_TCK")
|
|
add_proc(proc, 10, ["GenshinImpact.exe", "-popupwindow"], 50 * hz, comm="GenshinImpact.e", pad=200)
|
|
add_proc(proc, 20, REAPER, 30 * hz, comm="reaper) (x")
|
|
add_proc(proc, 21, ["Z:\\Steam\\Cryptic Studios\\Neverwinter.exe"], 40 * hz) # same app, newer
|
|
add_proc(proc, 30, [], 0) # kernel thread
|
|
add_proc(proc, 40, ["/usr/bin/bash"], 0)
|
|
(proc / "50").mkdir() # exited mid-scan
|
|
games = dgp.scan(index(), proc)
|
|
assert games == [
|
|
dgp.Game(dgp.App(NEVERWINTER["id"], "Neverwinter"), 20, 1000030),
|
|
dgp.Game(dgp.App(GENSHIN["id"], "Genshin Impact"), 10, 1000050),
|
|
]
|
|
|
|
|
|
def test_choose() -> None:
|
|
a = dgp.Game(dgp.App("1", "A"), 1, 100)
|
|
b = dgp.Game(dgp.App("2", "B"), 2, 200)
|
|
assert dgp.choose([], None) is None
|
|
assert dgp.choose([a, b], None) == a
|
|
assert dgp.choose([a, b], b) == b
|
|
# same app on a newer process: keep the first start so the timer doesn't restart
|
|
moved = dgp.Game(dgp.App("2", "B"), 3, 205)
|
|
assert dgp.choose([a, moved], b) == dgp.Game(dgp.App("2", "B"), 3, 200)
|
|
assert dgp.choose([a], b) == a
|
|
|
|
|
|
# config
|
|
|
|
|
|
def test_config_missing(tmp_path: Path) -> None:
|
|
assert dgp.load_config(tmp_path / "none.toml") == dgp.Config()
|
|
|
|
|
|
def test_config_values(tmp_path: Path) -> None:
|
|
path = tmp_path / "config.toml"
|
|
path.write_text('interval = 2\nignore = ["Some Game"]\n[executables]\n"Dir\\\\Game.EXE" = "123"\n')
|
|
config = dgp.load_config(path)
|
|
assert config.interval == 2.0
|
|
assert config.ignore == frozenset({"some game"})
|
|
assert config.executables == {"dir/game.exe": "123"}
|
|
|
|
|
|
def test_example_config_loads() -> None:
|
|
dgp.load_config(Path(__file__).parent.parent / "config.example.toml")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"text",
|
|
["interval = 0", "interval = true", 'ignore = "x"', '[executables]\n"a.exe" = "abc"', "foo = 1", "interval ="],
|
|
)
|
|
def test_config_errors(tmp_path: Path, text: str) -> None:
|
|
path = tmp_path / "config.toml"
|
|
path.write_text(text)
|
|
with pytest.raises(ValueError):
|
|
dgp.load_config(path)
|
|
|
|
|
|
# game list cache
|
|
|
|
|
|
def test_cache_missing_downloads(tmp_path: Path) -> None:
|
|
path = tmp_path / "c" / "detectable.json"
|
|
assert dgp.refresh_cache(path, lambda: json.dumps(APPS).encode())
|
|
assert dgp.read_cache(path) == APPS
|
|
|
|
|
|
def test_cache_fresh_not_downloaded(tmp_path: Path) -> None:
|
|
path = tmp_path / "detectable.json"
|
|
path.write_text("[]")
|
|
|
|
def fail() -> bytes:
|
|
raise AssertionError("downloaded")
|
|
|
|
assert not dgp.refresh_cache(path, fail)
|
|
|
|
|
|
@pytest.mark.parametrize("data", [b"{}", b"not json", OSError("offline")])
|
|
def test_cache_stale_kept_on_failure(tmp_path: Path, data: bytes | OSError) -> None:
|
|
path = tmp_path / "detectable.json"
|
|
path.write_text("[]")
|
|
old = time.time() - dgp.CACHE_MAX_AGE - 10
|
|
os.utime(path, (old, old))
|
|
|
|
def fetch() -> bytes:
|
|
if isinstance(data, OSError):
|
|
raise data
|
|
return data
|
|
|
|
assert not dgp.refresh_cache(path, fetch)
|
|
assert dgp.read_cache(path) == []
|
|
|
|
|
|
def test_read_cache_bad(tmp_path: Path) -> None:
|
|
assert dgp.read_cache(tmp_path / "none.json") is None
|
|
(tmp_path / "bad.json").write_text("{")
|
|
assert dgp.read_cache(tmp_path / "bad.json") is None
|
|
|
|
|
|
# Discord IPC
|
|
|
|
|
|
class FakeDiscord:
|
|
"""Discord's IPC socket: answers the handshake and SET_ACTIVITY, records what it got."""
|
|
|
|
def __init__(self, path: Path, error: str | None = None) -> None:
|
|
self.server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.server.bind(str(path))
|
|
self.server.listen()
|
|
self.error = error
|
|
self.received: list[tuple[int, dict[str, Any]]] = []
|
|
self.clients: list[socket.socket] = []
|
|
self.closed = threading.Event()
|
|
threading.Thread(target=self._serve, daemon=True).start()
|
|
|
|
def _serve(self) -> None:
|
|
while True:
|
|
try:
|
|
conn, _ = self.server.accept()
|
|
except OSError:
|
|
return
|
|
self.clients.append(conn)
|
|
threading.Thread(target=self._client, args=(conn,), daemon=True).start()
|
|
|
|
@staticmethod
|
|
def send(conn: socket.socket, op: int, payload: dict[str, Any]) -> None:
|
|
data = json.dumps(payload).encode()
|
|
conn.sendall(struct.pack("<II", op, len(data)) + data)
|
|
|
|
def _client(self, conn: socket.socket) -> None:
|
|
with conn:
|
|
while True:
|
|
header = conn.recv(8, socket.MSG_WAITALL)
|
|
if len(header) < 8:
|
|
self.closed.set()
|
|
return
|
|
op, size = struct.unpack("<II", header)
|
|
payload = json.loads(conn.recv(size, socket.MSG_WAITALL))
|
|
self.received.append((op, payload))
|
|
if op == dgp.OP_HANDSHAKE:
|
|
self.send(conn, dgp.OP_FRAME, {"cmd": "DISPATCH", "evt": "READY", "data": {}})
|
|
elif op == dgp.OP_FRAME and self.error:
|
|
reply = {"evt": "ERROR", "nonce": payload["nonce"], "data": {"message": self.error}}
|
|
self.send(conn, dgp.OP_FRAME, reply)
|
|
elif op == dgp.OP_FRAME:
|
|
self.send(conn, dgp.OP_FRAME, {"cmd": "SET_ACTIVITY", "evt": None, "nonce": payload["nonce"]})
|
|
|
|
def stop(self) -> None:
|
|
self.server.close()
|
|
for conn in self.clients:
|
|
try:
|
|
conn.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass # the client already left
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime_dir() -> Iterator[Path]:
|
|
# AF_UNIX paths are limited to 107 bytes: tmp_path plus the Flatpak subdir can exceed it
|
|
with tempfile.TemporaryDirectory(prefix="dgp-", dir="/tmp") as d:
|
|
yield Path(d)
|
|
|
|
|
|
def test_ipc_set_activity(runtime_dir: Path) -> None:
|
|
discord = FakeDiscord(runtime_dir / "discord-ipc-0")
|
|
ipc = dgp.DiscordIPC.connect("123", runtime_dir)
|
|
ipc.set_activity(1700000000)
|
|
assert discord.received[0] == (dgp.OP_HANDSHAKE, {"v": 1, "client_id": "123"})
|
|
op, payload = discord.received[1]
|
|
assert op == dgp.OP_FRAME
|
|
assert payload["cmd"] == "SET_ACTIVITY"
|
|
assert payload["args"] == {"pid": os.getpid(), "activity": {"timestamps": {"start": 1700000000}}}
|
|
assert ipc.alive()
|
|
ipc.close()
|
|
assert discord.closed.wait(2)
|
|
discord.stop()
|
|
|
|
|
|
def test_ipc_later_socket(runtime_dir: Path) -> None:
|
|
FakeDiscord(runtime_dir / "discord-ipc-3")
|
|
dgp.DiscordIPC.connect("123", runtime_dir).close()
|
|
|
|
|
|
@pytest.mark.parametrize("subdir", ["app/com.discordapp.Discord", "snap.discord"])
|
|
def test_ipc_sandbox_socket(runtime_dir: Path, subdir: str) -> None:
|
|
(runtime_dir / subdir).mkdir(parents=True)
|
|
discord = FakeDiscord(runtime_dir / subdir / "discord-ipc-0")
|
|
dgp.DiscordIPC.connect("123", runtime_dir).close()
|
|
assert discord.received[0] == (dgp.OP_HANDSHAKE, {"v": 1, "client_id": "123"})
|
|
discord.stop()
|
|
|
|
|
|
def test_ipc_no_socket(runtime_dir: Path) -> None:
|
|
with pytest.raises(FileNotFoundError):
|
|
dgp.DiscordIPC.connect("123", runtime_dir)
|
|
|
|
|
|
def test_ipc_error_reply(runtime_dir: Path) -> None:
|
|
FakeDiscord(runtime_dir / "discord-ipc-0", error="bad app")
|
|
ipc = dgp.DiscordIPC.connect("123", runtime_dir)
|
|
with pytest.raises(dgp.IPCError, match="bad app"):
|
|
ipc.set_activity(1)
|
|
|
|
|
|
def test_ipc_ping_and_disconnect(runtime_dir: Path) -> None:
|
|
discord = FakeDiscord(runtime_dir / "discord-ipc-0")
|
|
ipc = dgp.DiscordIPC.connect("123", runtime_dir)
|
|
FakeDiscord.send(discord.clients[0], dgp.OP_PING, {"n": 1})
|
|
time.sleep(0.1)
|
|
assert ipc.alive()
|
|
time.sleep(0.1)
|
|
assert discord.received[-1] == (dgp.OP_PONG, {"n": 1})
|
|
discord.stop()
|
|
time.sleep(0.1)
|
|
assert not ipc.alive()
|
|
|
|
|
|
def test_presence(runtime_dir: Path) -> None:
|
|
genshin = dgp.Game(dgp.App("1", "Genshin Impact"), 10, 100)
|
|
neverwinter = dgp.Game(dgp.App("2", "Neverwinter"), 20, 200)
|
|
presence = dgp.Presence(runtime_dir)
|
|
presence.update(genshin) # Discord not running: no error, retried later
|
|
assert presence.shown is None
|
|
|
|
discord = FakeDiscord(runtime_dir / "discord-ipc-0")
|
|
presence.update(genshin)
|
|
assert presence.shown == genshin
|
|
presence.update(genshin) # unchanged: nothing sent
|
|
presence.update(dgp.Game(genshin.app, 11, 100)) # another process, same timer: nothing sent
|
|
assert len(discord.received) == 2
|
|
|
|
presence.update(neverwinter) # another app needs a new connection
|
|
assert presence.shown == neverwinter
|
|
assert [p.get("client_id") for op, p in discord.received if op == dgp.OP_HANDSHAKE] == ["1", "2"]
|
|
|
|
presence.update(None)
|
|
assert presence.ipc is None and presence.shown is None
|
|
discord.stop()
|
|
|
|
|
|
def test_presence_reconnects(runtime_dir: Path) -> None:
|
|
game = dgp.Game(dgp.App("1", "Genshin Impact"), 10, 100)
|
|
path = runtime_dir / "discord-ipc-0"
|
|
discord = FakeDiscord(path)
|
|
presence = dgp.Presence(runtime_dir)
|
|
presence.update(game)
|
|
discord.stop()
|
|
path.unlink()
|
|
time.sleep(0.1)
|
|
presence.update(game) # Discord gone
|
|
assert presence.shown is None
|
|
discord = FakeDiscord(path)
|
|
presence.update(game) # Discord back
|
|
assert presence.shown == game
|
|
discord.stop()
|