chore: merge main into the round 4 fixes

# Conflicts:
#	wiki/installer.md
This commit is contained in:
Abdessamad Derraz committed 2026-10-23 21:37:25 +02:00
commit b88f23947a
16 files changed
+1670 -224

No files matched your search

+1 -1
View File
@@ -11,7 +11,7 @@
$ErrorActionPreference = "Stop"
$defaultInstallUrl = "https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
$defaultInstallSha256 = "e2cc5f073a8226fe26b0513ebd25f00de6f3f613d8c9851473c2d30b53356f48"
$defaultInstallSha256 = "563ed14fea5ea54f5d6cb98311932293e714e9581014ea60c195325c20a587eb"
$maximumInstallerBytes = 2MB
$installerExit = 0
$installer = if ($PSScriptRoot) { Join-Path $PSScriptRoot "install.py" } else { $null }
+523 -115
View File
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -18,7 +18,7 @@ esac
TEMP_INSTALLER=""
TEMP_DIRECTORY=""
DEFAULT_INSTALL_URL="https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
DEFAULT_INSTALL_SHA256="e2cc5f073a8226fe26b0513ebd25f00de6f3f613d8c9851473c2d30b53356f48"
DEFAULT_INSTALL_SHA256="563ed14fea5ea54f5d6cb98311932293e714e9581014ea60c195325c20a587eb"
MAX_INSTALLER_BYTES=2097152
cleanup() {
+2 -5
View File
@@ -69,13 +69,10 @@ rb_open_log keep
# muOS: PortMaster's mod_muOS.txt points CUR_TTY at /dev/null, so there is no
# console to draw on. muxterm, the terminal muOS's Terminal core runs, shows
# the interface instead. It is told to leave the pad alone (an SDL hint) so
# that gptokeyb sends every button once, with this port's mapping: muxterm
# sends Y as Space, where the interface reads Y as Options. The hint allows
# one vendor and product id, 0001:0001, which no pad carries.
# the interface instead, told to leave the pad to gptokeyb (rb_muxterm).
RB_MUXTERM="${RB_MUXTERM:-/opt/muos/frontend/muxterm}"
if [ -x "$RB_MUXTERM" ] && [ -n "$RB_MAPPER" ]; then
RB_TERMAPP=(env SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT=0x0001/0x0001 "$RB_MUXTERM" --gl --)
rb_muxterm "$RB_MUXTERM"
fi
export SDL_GAMECONTROLLERCONFIG="$sdl_controllerconfig"
+5 -1
View File
@@ -10,9 +10,13 @@ connection, or from a pack copied to the root of the card without one.
It runs in muxterm, muOS's terminal. With PortMaster installed, gptokeyb2
turns every button into a key and muxterm is told to leave the pad alone, so
each button is sent once: D-pad to move, A or Start to choose, B or Select to
go back, X to tick a system, Y for the options, L1 and R1 to page. Hold
go back, X to tick a system, Y for the options. L1 and R1 do nothing in
muxterm, which keeps Page Up and Page Down for its own scrollback. Hold
Select, then press Start to quit. Without PortMaster, muxterm reads the pad
itself (Y then ticks a system) and Menu closes it.
With no button pressed on its first screen for 5 minutes, it closes by
itself: a pad the system cannot read then never leaves it stuck.
Not tried on a device yet: the icon, and whether muxterm honours SDL's
controller allow-list as SDL 2.30 does.
+3 -11
View File
@@ -46,17 +46,9 @@ if [ -x "$PM_DIR/gptokeyb2" ]; then
fi
if [ -x "$MUXTERM" ]; then
RB_ROUTES="app"
if [ -n "${RB_MAPPER:-}" ]; then
# muxterm reads the pad itself and sends Y as Space (MustardOS/terminal
# input.c), where the interface reads Y as Options. The SDL hint allows
# only vendor and product 0001:0001, which no pad carries, so muxterm
# leaves the pad to gptokeyb2 and every button is sent once.
RB_TERMAPP=(env SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT=0x0001/0x0001 "$MUXTERM" --gl --)
else
# No gptokeyb2: muxterm's own buttons are all there is.
RB_TERMAPP=("$MUXTERM" --gl --)
RB_PAD="muxterm"
fi
# With gptokeyb2, muxterm leaves the pad to it. Without, muxterm's own
# buttons are all there is.
rb_muxterm "$MUXTERM"
else
RB_ROUTES="vt"
RB_CONSOLE="${RB_CONSOLE:-/dev/tty0}"
+75 -12
View File
@@ -62,7 +62,7 @@ RB_MAPPER_NEEDED="${RB_MAPPER_NEEDED:-1}"
RB_PAD="${RB_PAD:-}"
RB_RUNTIME="python_3.11"
RB_RELEASES="github.com/Abdess/retrobios/releases/latest"
RB_PC_ROUTE="On a computer, with this SD card mounted, the same installer fills the card: install.py --platform NAME --dest BIOS-FOLDER-ON-THE-CARD. Instructions: abdess.github.io/retrobios/wiki/installer"
RB_PC_ROUTE="On a computer, with this SD card mounted, the same installer fills the card: install.py finds a muOS, Knulli or ROCKNIX card by itself, and any other with install.py --platform NAME --dest BIOS-FOLDER-ON-THE-CARD. Instructions: abdess.github.io/retrobios/wiki/installer"
RB_PYTHON="${RB_PYTHON:-}"
RB_PYTHONHOME="${RB_PYTHONHOME:-}"
RB_PYTHON_PROBLEM="${RB_PYTHON_PROBLEM:-}"
@@ -284,17 +284,17 @@ rb_online() {
}
# The system and BIOS folder install.py finds here, read without a network:
# detection is local, and the run stops where the file list would download.
# That download is sent to a loopback port nothing listens on, so it fails
# at once instead of retrying three times 30 s on a network already found
# missing.
# detection is local, and RETROBIOS_INSTALL_OFFLINE stops the run at once
# where the file list would download (reason offline), instead of three
# tries of 30 s on a network already found missing. Set for this run only:
# the interface's own run reads a pack, which makes no request either.
rb_detect() {
RB_PLATFORM=""
RB_PLATFORM_NAME=""
RB_BIOS_DIR=""
mkdir -p "$RB_STATE"
rb_pycmd
RETROBIOS_BASE_URL="http://127.0.0.1:9" rb_bounded "$RB_DETECT_LIMIT" \
RETROBIOS_INSTALL_OFFLINE=1 rb_bounded "$RB_DETECT_LIMIT" \
"${RB_PYCMD[@]}" "$RB_INSTALL" --check --events > "$RB_STATE/detect.jsonl" 2>> "$RB_LOG"
[ $? = 124 ] && rb_say "the detection run took over $RB_DETECT_LIMIT s and was stopped"
local kind a b c
@@ -415,6 +415,36 @@ rb_terminal_app() {
fi
}
# The only pad SDL may show muxterm: vendor 0, product 1, which no device
# can carry. SDL builds a GUID from the ids only when the vendor is not 0,
# else from the name, and reads a name GUID back as vendor 0 and product 0
# (SDL_CreateJoystickGUID and SDL_GetJoystickGUIDInfo, SDL_joystick.c), so
# a pad plugged in later is hidden too. The first list, 0001:0001, is the
# id gpio-keys gives the H700 pad on muOS before 2026-08-28: every button
# acted twice there.
RB_SDL_NO_PAD="0x0000/0x0001"
# muxterm, muOS's terminal, as the terminal app. It reads the pad itself
# (Y as Space, where the interface reads Y as Options), so with a mapper it
# is told to leave the pad alone and each button is sent once. muxterm also
# keeps PgUp and PgDn for its scrollback and never writes them to the
# interface (MustardOS/terminal input.c): L1 and R1 send nothing there,
# rather than scroll the interface out of sight. Run after rb_open_log.
rb_muxterm() {
if [ -z "${RB_MAPPER:-}" ]; then
RB_TERMAPP=("$1" --gl --)
RB_PAD="muxterm"
return 0
fi
RB_TERMAPP=(env "SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT=$RB_SDL_NO_PAD" "$1" --gl --)
if awk '$1 == "l1" || $1 == "r1" { print $1 " = \\\""; next } { print }' \
"$RB_GPTK" > "$RB_STATE/muxterm.gptk" 2>> "$RB_LOG"; then
RB_GPTK="$RB_STATE/muxterm.gptk"
else
rb_say "$RB_STATE/muxterm.gptk cannot be written: L1 and R1 scroll muxterm"
fi
}
# A console a terminal can be read from. PortMaster's mod_muOS.txt sets
# CUR_TTY to /dev/null, a character device that is no console at all. A
# console this user cannot open yet (rb_vt_enter opens it up) is judged by
@@ -581,10 +611,13 @@ rb_vt_leave() {
# wrote. So the target is a sentinel, a process this launcher owns whose
# command line alone carries the token: when it dies, the launcher asks the
# interface to stop, and the interface gives a running install its 2 s.
# It also ends once the launcher is gone, which a SIGKILL leaves no trap
# to say.
rb_stop_start() {
RB_STOP_TOKEN="retrobios-stop-${$}x"
bash -c 'exec {f}<> <(:); while :; do read -r -t 3600 -u "$f" _; done' \
"$RB_STOP_TOKEN" > /dev/null 2>&1 &
# shellcheck disable=SC2016 # expanded by the inner bash
bash -c 'exec {f}<> <(:); while [ -d "/proc/$1" ]; do read -r -t 5 -u "$f" _; done' \
"$RB_STOP_TOKEN" "$$" > /dev/null 2>&1 &
RB_STOP_PID=$!
}
@@ -628,8 +661,26 @@ rb_mapper_stop() {
rb_stop_end
}
# The mapper's own name, gptokeyb or gptokeyb2, past sudo and its options.
rb_mapper_name() {
local word
# shellcheck disable=SC2086 # RB_MAPPER is a command and its flags
for word in $RB_MAPPER; do
case "$word" in
sudo|-*|*=*) ;;
*) printf '%s\n' "${word##*/}"; return 0 ;;
esac
done
printf 'gptokeyb\n'
}
# "run" when the mapper stopped during the run: the interface just went.
rb_mapper_text() {
printf '%s\n' "The controller cannot be read: gptokeyb, which turns the buttons into keys, stopped as soon as it started. What it said is in $RB_LOG. $RB_PC_ROUTE"
if [ "${1:-}" = "run" ]; then
printf '%s\n' "The controller stopped responding: $(rb_mapper_name), which turns the buttons into keys, stopped during the run, so RetroBIOS closed. What it said is in $RB_LOG. Start RetroBIOS again."
return 0
fi
printf '%s\n' "The controller cannot be read: $(rb_mapper_name), which turns the buttons into keys, stopped as soon as it started. What it said is in $RB_LOG. $RB_PC_ROUTE"
}
# --- Messages -------------------------------------------------------------
@@ -816,6 +867,10 @@ rb_here() {
if [ "${RB_FALLBACK:-0}" != "1" ]; then
rb_message "$(tail -n 1 "$RB_STATE/tui.err") The messages are in $RB_LOG."
fi
elif [ "$RB_STOPPED" = "mapper" ]; then
# The interface disappeared: why, for RB_DEAD_WAIT seconds, since no
# button is left to close it.
rb_message "$(rb_mapper_text run)"
elif [ -n "$RB_STOPPED" ] || [ "$status" -ge 128 ]; then
# Ended by Start + Select or a signal: the player asked to leave, and no
# button may be left to close a message.
@@ -891,6 +946,9 @@ rb_launch() {
printf '%s\n' "$RB_STATE_PROBLEM" >&2
if rb_is_console "$RB_CONSOLE"; then
printf '\r\n%s\r\n' "$RB_STATE_PROBLEM" > "$RB_CONSOLE" 2>/dev/null
# No mapper runs yet to close it with a button: left in view a while
# before the menu draws over it.
rb_nap "$RB_DEAD_WAIT"
fi
return 1
fi
@@ -898,6 +956,10 @@ rb_launch() {
export RB_DIR RB_INSTALL RB_EVENTS RB_STATE RB_LOG RB_PACK_PATH RB_FALLBACK
export RB_OTHER_CARDS RB_BIOS_DEST RB_STOP_GRACE RB_DEAD_WAIT RB_DETECT_LIMIT
export RB_PYTHON RB_PYTHONHOME RB_PYTHON_PROBLEM RB_MESSAGE_WAIT RB_ROUTE_TABLE
# A mapper that runs but never opens the pad (one SDL has no mapping for)
# leaves no button to move the interface or end the run: it closes by
# itself after this many seconds on its first screen with nothing pressed.
export RETROBIOS_IDLE_EXIT="${RETROBIOS_IDLE_EXIT:-300}"
# The interface names the buttons of this route in its hints.
if [ -z "$RB_PAD" ] && [ -n "${RB_MAPPER:-}" ]; then
RB_PAD="gptokeyb"
@@ -931,9 +993,10 @@ if [ "${BASH_SOURCE[0]}" = "$0" ]; then
case "${1:-}" in
here)
shift
# The terminal app closed or the launcher signalled: the interface
# gets its time to stop.
trap 'rb_ui_stop; exit 143' TERM HUP
# The terminal app closed (muxterm's Menu) or the launcher signalled:
# the interface gets its time to stop, and the status says the run
# ended, so the launcher opens no other route after it.
trap 'rb_ui_stop; printf "143\n" > "$RB_STATE/status"; exit 143' TERM HUP
rb_here "$@"
exit $?
;;
+10 -2
View File
@@ -82,8 +82,16 @@ else
# shellcheck source=/dev/null
source "$control"
declare -F get_controls >/dev/null && get_controls
# It carries its own kill flag (-sudokill).
[ -n "${GPTOKEYB:-}" ] && mapper="$GPTOKEYB" kill_flag=""
# Its kill flag is -sudokill (control-gen.cpp:31), which kills through
# sudo, and ROCKNIX's sudo is a stub that runs nothing (busybox
# scripts/sudo): Start + Select would end the run only 3 s later,
# through the mapper's exit. -1 kills as the user it runs as, root.
if [ -n "${GPTOKEYB:-}" ]; then
mapper="$GPTOKEYB"
case "$mapper" in
*" -sudokill"|*" -1"|*" -k") mapper="${mapper% *}" ;;
esac
fi
ESUDO="$saved_esudo" controlfolder="$saved_folder"
fi
fi
+437 -57
View File
@@ -177,13 +177,100 @@ class ControllerMappings(unittest.TestCase):
targets = {a["trigger"]: a["target"] for a in keys["actions_player1"]}
self.assertEqual((targets["x"], targets["y"]), (["KEY_SPACE"], ["KEY_TAB"]))
# muxterm sends Y as Space on its own (MustardOS/terminal input.c):
# where gptokeyb runs beside it, muxterm is told to leave the pad.
# where gptokeyb runs beside it, muxterm is told to leave the pad
# (assert_muxterm_hides_every_pad).
for launcher in (PORTS / "muos" / "mux_launch.sh", PORTS / "RetroBIOS" / "RetroBIOS.sh"):
text = launcher.read_text()
self.assertIn("SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT=0x0001/0x0001", text)
self.assertIn("rb_muxterm ", launcher.read_text())
self.assertFalse((PORTS / "muos" / "muxterm.gptk").exists())
def sdl_vidpid(bus: int, vendor: int, product: int, version: int, name: str) -> tuple[int, int]:
"""The vendor and product SDL 2 reads back from the GUID it builds for
a device: SDL_CreateJoystickGUID, then SDL_GetJoystickGUIDInfo
(libsdl-org/SDL release-2.30.x src/joystick/SDL_joystick.c). A vendor of
0 makes the GUID carry the name instead of the ids."""
data = bytearray(16)
struct.pack_into("<HH", data, 0, bus, 0x5a5a)
if vendor:
struct.pack_into("<HHHHH", data, 4, vendor, 0, product, 0, version)
else:
raw = name.encode()[:11]
data[4:4 + len(raw)] = raw
words = struct.unpack("<8H", bytes(data))
if (words[0] < 0x20 or words[0] == 0xFF) and words[3] == 0 and words[5] == 0:
return words[2], words[4]
return 0, 0
def sdl_vidpid_list(hint: str) -> list[int]:
"""SDL_LoadVIDPIDListFromHint: every pair of 0x numbers."""
entries = []
for vendor, product in re.findall(r"0x([0-9a-fA-F]+)[^x]*?0x([0-9a-fA-F]+)", hint):
entries.append((int(vendor, 16) & 0xFFFF) << 16 | (int(product, 16) & 0xFFFF))
return entries
# Input devices a muOS device lists in /proc/bus/input/devices: the H700
# pad before 2026-08-28 (gpio-keys, 0001:0001, MustardOS internal
# f6a14e3121 device/rg35xx-plus/control/gamecontrollerdb/modern.txt), the
# muinput pad after it (0x756d), a USB and a Bluetooth pad, gptokeyb2's
# uinput keyboard, and pads whose vendor is 0.
PADS = [
(0x19, 0x0001, 0x0001, 0x0100, "Deeplay-keys"),
(0x19, 0x0001, 0x0001, 0x0100, "muOS-Keys"),
(0x06, 0x756d, 0x6b6d, 0x0001, "muOS-Keys"),
(0x03, 0x045e, 0x028e, 0x0114, "Microsoft X-Box 360 pad"),
(0x05, 0x057e, 0x2009, 0x8111, "Pro Controller"),
(0x06, 0x0000, 0x0000, 0x0000, "Fake Keyboard"),
]
PADS += [(bus, 0, product, 1, "pad"[:length] + "x" * max(0, length - 3))
for bus in (0x03, 0x05, 0x06, 0x19, 0xFF) for product in (0, 1, 0x2009)
for length in range(13)]
PADS += [(bus, vendor, product, 1, "pad")
for bus in (0x03, 0x05, 0x19) for vendor in (1, 0x0079, 0xFFFF)
for product in (0, 1, 0xFFFF)]
def assert_muxterm_hides_every_pad(test: unittest.TestCase, hint: str) -> None:
"""muxterm reads the pad itself (T/input.c:303-317) while gptokeyb2
sends its keys, so every button would act twice. The launchers set
SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT, which hides from SDL every
device it does not name (SDL_gamecontroller.c:2145-2150, applied to the
joystick layer by SDL_ShouldIgnoreJoystick, SDL_joystick.c:3006)."""
allowed = sdl_vidpid_list(hint)
# An empty list allows every device.
test.assertTrue(allowed, hint)
for pad in PADS:
vendor, product = sdl_vidpid(*pad)
with test.subTest(pad=pad):
test.assertNotIn(vendor << 16 | product, allowed)
def assert_muxterm_mapping(test: unittest.TestCase, args: list[str], shared: Path) -> None:
"""muxterm keeps PgUp and PgDn for its scrollback and never writes them
to the terminal (T/input.c:352-360): in muxterm, L1 and R1 send nothing,
so they cannot scroll the interface away. Every other button is
retrobios.gptk's."""
test.assertEqual(args[-2], "-c")
mapping = read_gptk(Path(args[-1]))
expected = read_gptk(shared)
expected["l1"] = expected["r1"] = ['\\"']
test.assertEqual(mapping, expected)
class TheSdlModel(unittest.TestCase):
def test_it_reads_the_ids_as_sdl_does(self):
self.assertEqual(sdl_vidpid(0x19, 1, 1, 0x100, "Deeplay-keys"), (1, 1))
self.assertEqual(sdl_vidpid(0x05, 0, 0x2009, 1, "pad"), (0, 0))
self.assertEqual(sdl_vidpid(0x05, 0, 0x2009, 1, "ab"), (0x6261, 0))
self.assertEqual(sdl_vidpid_list("0x0001/0x0001, 0x045e/0x028e"),
[0x00010001, 0x045e028e])
def test_the_first_hint_allowed_the_h700_pad(self):
with self.assertRaises(AssertionError):
assert_muxterm_hides_every_pad(unittest.TestCase(), "0x0001/0x0001")
class TheArchives(unittest.TestCase):
LAYOUT = {
build_port.PORT_ZIP: ("RetroBIOS.sh", "retrobios"),
@@ -300,6 +387,8 @@ if "--engine" in sys.argv:
signal.signal(signal.SIGINT, summary)
signal.signal(signal.SIGTERM, other)
with open(os.path.join(state, "engine.pid"), "w") as fh:
fh.write(str(os.getpid()))
note("engine.ready")
time.sleep(60)
sys.exit(0)
@@ -309,17 +398,32 @@ with open(calls, "a") as fh:
"stdout": tty(1), "term": os.environ.get("TERM", ""),
"pythonhome": os.environ.get("PYTHONHOME", ""),
"base_url": os.environ.get("RETROBIOS_BASE_URL", ""),
"offline": os.environ.get("RETROBIOS_INSTALL_OFFLINE", ""),
"idle_exit": os.environ.get("RETROBIOS_IDLE_EXIT"),
"pad": os.environ.get("RETROBIOS_PAD")}) + "\n")
if "--check" in sys.argv:
# install.py --check --events: detection is local, then the file list.
# With RETROBIOS_INSTALL_OFFLINE the file list stops the run at once
# with reason offline. Otherwise it is downloaded: a closed loopback
# port refuses at once, any other address is three tries of 30 s on a
# device without a network (FAKE_CURL, the probe that found it so).
offline = os.environ.get("RETROBIOS_INSTALL_OFFLINE", "") not in ("", "0")
base_url = os.environ.get("RETROBIOS_BASE_URL", "")
no_network = os.environ.get("FAKE_CURL", "0") in ("5", "6", "7", "28")
if os.environ.get("FAKE_DETECT_HANG"):
time.sleep(60)
print(json.dumps({"event": "start", "mode": "check"}), flush=True)
print(json.dumps({"event": "detect", "os": "linux", "os_name": "Fake OS",
"platforms": [{"platform": os.environ.get("FAKE_PLATFORM", "retroarch"),
"name": os.environ.get("FAKE_PLATFORM_NAME", "RetroArch"),
"bios_dir": "/roms/bios"}]}))
print(json.dumps({"event": "error", "stage": "manifest", "reason": "network",
"bios_dir": "/roms/bios"}]}), flush=True)
if no_network and not offline and not base_url.startswith("http://127.0.0.1"):
time.sleep(90)
print(json.dumps({"event": "error", "stage": "manifest",
"reason": "offline" if offline else "network",
"platform": os.environ.get("FAKE_PLATFORM", "retroarch"),
"message": "unreachable"}))
"message": "this run makes no network request (--offline)" if offline
else "unreachable"}))
print(json.dumps({"event": "end", "code": 1}))
sys.exit(1)
if os.environ.get("FAKE_REFUSE"):
@@ -329,6 +433,21 @@ if os.environ.get("FAKE_REFUSE"):
if os.environ.get("FAKE_RAW") and os.isatty(0):
import tty as _tty
_tty.setraw(0)
if os.environ.get("FAKE_DRAIN") and os.isatty(0):
# What the console still holds when the interface starts: bytes a
# question left behind are read as the interface's first keys.
# TCSANOW, as install.py sets its mode: TCSAFLUSH would discard them.
import select as _select
import termios as _termios
import tty as _tty
_tty.setraw(0, _termios.TCSANOW)
left = b""
while _select.select([0], [], [], 0.3)[0]:
chunk = os.read(0, 64)
if not chunk:
break
left += chunk
note("ui.input", left.hex())
if os.environ.get("FAKE_HOLD"):
# The interface up and waiting for a key until a signal ends it. With
# FAKE_ENGINE an install runs, and SIGTERM gives it 2 s to say what it
@@ -354,9 +473,14 @@ if os.environ.get("FAKE_HOLD"):
print("Error: " + os.environ["FAKE_TERM_ERROR"], file=sys.stderr)
sys.exit(128 + signum)
if not os.environ.get("FAKE_NO_HANDLER"):
if os.environ.get("FAKE_IGNORE_TERM"):
# An interface stuck where its signal handler never runs.
signal.signal(signal.SIGTERM, signal.SIG_IGN)
elif not os.environ.get("FAKE_NO_HANDLER"):
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGHUP, stop)
with open(os.path.join(state, "ui.pid"), "w") as fh:
fh.write(str(os.getpid()))
note("ui.ready")
time.sleep(60)
sys.exit(0)
@@ -371,12 +495,13 @@ PGREP = shutil.which("pgrep", path="/usr/bin:/bin") or "pgrep"
PKILL = shutil.which("pkill", path="/usr/bin:/bin") or "pkill"
# gptokeyb as PortsMaster/gptokeyb runs it: kill mode takes the argument
# after the flag (`-1`, `-k` or `-sudokill`) as its target. Start + Select (FAKE_PRESS, once the
# interface is up) is doKillMode in src/util.cpp: pkill -f on the target,
# which also matches gptokeyb's own command line (SDL turns that SIGTERM
# into an event it does not read there), 3 s, then pkill -9 -f and exit.
# Stopped otherwise by a SIGTERM, which it records. FAKE_MAPPER_DIES is a
# mapper that cannot start (no /dev/uinput).
# after the flag (`-1`, `-k` or `-sudokill`) as its target. Start + Select
# (FAKE_PRESS, once FAKE_PRESS_AT exists, by default once the interface is
# up) is doKillMode in src/util.cpp: pkill -f on the target, which also
# matches gptokeyb's own command line (SDL turns that SIGTERM into an event
# it does not read there), 3 s, then pkill -9 -f and exit. `-sudokill` runs
# both through sudo. Stopped otherwise by a SIGTERM, which it records.
# FAKE_MAPPER_DIES is a mapper that cannot start (no /dev/uinput).
FAKE_MAPPER = f'''#!/bin/bash
printf '%s\\n' "$@" > "$FAKE_STATE/mapper.args"
echo $$ > "$FAKE_STATE/mapper.pid"
@@ -385,32 +510,39 @@ if [ -n "${{FAKE_MAPPER_DIES:-}}" ]; then
echo "gptokeyb: cannot open /dev/uinput" >&2
exit 1
fi
name=""
name="" sudo=""
while [ $# -gt 0 ]; do
case "$1" in
-1|-k|-sudokill) name="$2"; shift 2 ;;
-1|-k) name="$2"; shift 2 ;;
-sudokill) name="$2" sudo="sudo"; shift 2 ;;
-c) shift 2 ;;
*) shift ;;
esac
done
killing=""
trap '[ -n "$killing" ] || {{ echo stopped > "$FAKE_STATE/mapper.stopped"; exit 0; }}' TERM
until [ -n "${{FAKE_PRESS:-}}" ] && [ -e "$FAKE_STATE/ui.ready" ]; do sleep 0.1; done
until [ -n "${{FAKE_PRESS:-}}" ] && [ -e "$FAKE_STATE/${{FAKE_PRESS_AT:-ui.ready}}" ]; do sleep 0.1; done
killing=1
echo "kill $name" >> "$FAKE_STATE/mapper.log"
{PKILL} -f "$name"
$sudo {PKILL} -f "$name"
sleep 3
if {PGREP} -f "$name" > /dev/null; then {PKILL} -9 -f "$name"; fi
if {PGREP} -f "$name" > /dev/null; then $sudo {PKILL} -9 -f "$name"; fi
exit 0
'''
# gptokeyb2 as PortsMaster/gptokeyb2 src/util.c kills: SIGKILL only, by
# process name (pkill -9 NAME) unless -X, then pgrep -f and kill -9 on every
# match but its own ancestry and children. It quits after a kill, and keeps
# running when it has no process name.
# running when it has no process name. FAKE_MAPPER_STAYS is gptokeyb2
# before 2026-10-05, which keeps running after a -X kill (an inner
# `int status` shadows the result, 23581b62ec src/util.c:486-519).
FAKE_GPTOKEYB2 = f'''#!/bin/bash
printf '%s\\n' "$@" > "$FAKE_STATE/mapper.args"
echo $$ > "$FAKE_STATE/mapper.pid"
if [ -n "${{FAKE_MAPPER_DIES:-}}" ]; then
echo "gptokeyb2: cannot open /dev/uinput" >&2
exit 1
fi
want_kill="" name=""
while [ $# -gt 0 ]; do
case "$1" in
@@ -421,7 +553,7 @@ while [ $# -gt 0 ]; do
esac
done
trap 'echo stopped > "$FAKE_STATE/mapper.stopped"; exit 0' TERM
until [ -n "${{FAKE_PRESS:-}}" ] && [ -e "$FAKE_STATE/ui.ready" ]; do sleep 0.1; done
until [ -n "${{FAKE_PRESS:-}}" ] && [ -e "$FAKE_STATE/${{FAKE_PRESS_AT:-ui.ready}}" ]; do sleep 0.1; done
echo "kill $name" >> "$FAKE_STATE/mapper.log"
if [ -z "$name" ]; then
while :; do sleep 0.1; done
@@ -442,9 +574,20 @@ if [ -n "$want_kill" ]; then
else
{PKILL} -9 "$name"
fi
if [ -n "${{FAKE_MAPPER_STAYS:-}}" ]; then
while :; do sleep 0.1; done
fi
exit 0
'''
# ROCKNIX's /usr/bin/sudo: a stub that runs nothing (ROCKNIX distribution
# packages/sysutils/busybox/scripts/sudo).
ROCKNIX_SUDO = """#!/bin/sh
printf '%s\\n' "sudo $*" >> "$FAKE_STATE/tools.log"
echo " You are 'root' so 'sudo' is not required or supported"
exit 1
"""
FAKE_RECORDER = """#!/bin/sh
printf '%s\\n' "$(basename "$0") $*" >> "$FAKE_STATE/tools.log"
"""
@@ -472,6 +615,8 @@ done
FAKE_TERMAPP = """#!/bin/sh
printf '%s\\n' "$(basename "$0") $*" >> "$FAKE_STATE/tools.log"
printf '%s\\n' "${SDL_GAMECONTROLLER_IGNORE_DEVICES_EXCEPT:-}" > "$FAKE_STATE/termapp.sdl"
printf '%s\\n' "${SDL_GAMECONTROLLER_IGNORE_DEVICES:-}" > "$FAKE_STATE/termapp.sdl_ignore"
echo $$ > "$FAKE_STATE/termapp.pid"
if [ -n "$FAKE_TERM_FAILS" ]; then
echo "$(basename "$0"): Can't open display" >&2
exit 1
@@ -540,7 +685,8 @@ class Device:
(self.root / "fb").write_text("640,480\n")
environ = {k: v for k, v in os.environ.items()
if k not in ("DISPLAY", "WAYLAND_DISPLAY", "RETROBIOS_BASE_URL",
"XAUTHORITY", "XDG_RUNTIME_DIR")}
"RETROBIOS_INSTALL_OFFLINE", "RETROBIOS_IDLE_EXIT",
"RETROBIOS_PAD", "XAUTHORITY", "XDG_RUNTIME_DIR")}
self.env = {
**environ,
"PATH": f"{self.bin}{os.pathsep}{os.environ['PATH']}",
@@ -621,7 +767,7 @@ class Device:
def _reap(self) -> None:
"""A fake the launcher failed to stop must not outlive the test."""
for name in ("mapper.pid",):
for name in ("mapper.pid", "ui.pid", "engine.pid"):
path = self.state / name
if path.exists():
try:
@@ -671,6 +817,29 @@ class Device:
self.test.assertEqual(termios.tcgetattr(self.slave), self.before)
def sentinels(token: str) -> list[int]:
"""The Start + Select sentinels of a run: `bash -c SCRIPT TOKEN`, where
the mapper carries the token after its own path."""
found = []
for entry in Path("/proc").iterdir():
if not entry.name.isdigit():
continue
try:
argv = (entry / "cmdline").read_bytes().split(b"\0")
except OSError:
continue
if len(argv) > 3 and argv[1] == b"-c" and argv[3] == token.encode():
found.append(int(entry.name))
return found
def wait_until(test: unittest.TestCase, check, timeout: float, what: str) -> None:
end = time.monotonic() + timeout
while not check():
test.assertLess(time.monotonic(), end, what)
time.sleep(0.1)
CONTROL_TXT = '''
directory="{roms}"
ESUDO=""
@@ -727,6 +896,7 @@ class ThePortMasterPort(unittest.TestCase):
installer = str(self.gamedir / "install.py")
self.assertEqual(call["argv0"], installer)
self.assertEqual(call["pad"], "gptokeyb")
self.assertEqual(call["idle_exit"], "300")
# Kill mode on the Start + Select sentinel, never on the installer's
# path, which the engine's command line carries too.
args = device.mapper_args()
@@ -734,6 +904,7 @@ class ThePortMasterPort(unittest.TestCase):
self.assertRegex(args[1], r"^retrobios-stop-\d+x$")
self.assertEqual(args[2:], ["-c", str(self.gamedir / "retrobios.gptk")])
device.assert_mapper_stopped()
self.assertEqual(sentinels(args[1]), [])
device.assert_console_restored()
# Reset when the console is taken and again when it is handed back.
self.assertEqual(device.screen.count("\x1bc"), 2)
@@ -790,9 +961,8 @@ class ThePortMasterPort(unittest.TestCase):
device.assert_console_restored()
def test_offline_says_so_at_once_and_bounds_the_detection(self):
"""The file-list download is three tries of 30 s: the detection run
is pointed at a closed loopback port and stopped after
RB_DETECT_LIMIT, with a line on the console from the start."""
"""A detection run that hangs is stopped after RB_DETECT_LIMIT, with
a line on the console from the start."""
device = self.device(FAKE_CURL="28")
pack = self.gamedir / "RetroArch_Lakka_v1.22.2_BIOS_Pack.zip"
pack.write_bytes(b"")
@@ -802,12 +972,35 @@ class ThePortMasterPort(unittest.TestCase):
screen = device.screen
self.assertIn("Checking the connection", screen)
self.assertIn("No connection. Looking for the BIOS folder", screen)
[check] = [c for c in device.calls() if "--check" in c["args"]]
self.assertEqual(check["base_url"], "http://127.0.0.1:9")
[call] = device.tui_calls()
# The detection was stopped: no platform, the first pack.
self.assertEqual(call["args"], ["--tui", "--from-pack", str(pack)])
self.assertEqual(call["base_url"], "")
log = (self.gamedir / "log.txt").read_text()
self.assertIn("the detection run took over 2 s and was stopped", log)
def test_the_detection_is_told_to_make_no_request(self):
"""RETROBIOS_INSTALL_OFFLINE: install.py detects, then stops where
the file list would download, at once and with reason offline
(install.py _network_allowed). The interface run does not inherit
it, and no address is borrowed for the detection."""
device = self.device(FAKE_CURL="7")
pack = self.gamedir / "RetroArch_Lakka_v1.22.2_BIOS_Pack.zip"
pack.write_bytes(b"")
proc = device.run(["bash", self.script])
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
self.assertLess(device.elapsed, 15)
[check] = [c for c in device.calls() if "--check" in c["args"]]
self.assertEqual(check["args"], ["--check", "--events"])
self.assertEqual((check["offline"], check["base_url"]), ("1", ""))
detected = (self.gamedir / "run" / "detect.jsonl").read_text().splitlines()
self.assertEqual([json.loads(line)["event"] for line in detected],
["start", "detect", "error", "end"])
self.assertEqual(json.loads(detected[2])["reason"], "offline")
[call] = device.tui_calls()
self.assertEqual(call["args"], ["--tui", "--from-pack", str(pack),
"--platform", "retroarch"])
self.assertEqual((call["offline"], call["base_url"]), ("", ""))
self.assertNotIn("took over", (self.gamedir / "log.txt").read_text())
def test_a_certificate_failure_is_not_offline(self):
# curl 60: the network answered, install.py reports the certificate.
@@ -955,9 +1148,11 @@ class ThePortMasterPort(unittest.TestCase):
[call] = device.tui_calls()
self.assertEqual(call["args"], ["--tui"])
self.assertEqual(device.tools(), [f"muxterm --gl -- bash {self.gamedir}/tui.txt here"])
self.assertEqual(device.lines("termapp.sdl"), ["0x0001/0x0001"])
self.assertEqual(device.mapper_args()[2:], ["-c", str(self.gamedir / "retrobios.gptk")])
[hint] = device.lines("termapp.sdl")
assert_muxterm_hides_every_pad(self, hint)
assert_muxterm_mapping(self, device.mapper_args(), self.gamedir / "retrobios.gptk")
self.assertEqual(call["pad"], "gptokeyb")
self.assertEqual(call["idle_exit"], "300")
self.assertEqual(self.dialog(device), ["FINISH"])
def test_on_muos_without_muxterm_the_log_says_why_the_screens_open(self):
@@ -1027,13 +1222,73 @@ class ThePortMasterPort(unittest.TestCase):
"""Nothing moves the interface once gptokeyb is gone: it is asked to
stop instead of waiting for a key forever."""
device = self.device()
proc = device.spawn(["bash", self.script], FAKE_HOLD="1")
proc = device.spawn(["bash", self.script], FAKE_HOLD="1", RB_DEAD_WAIT="1",
RB_MESSAGE_WAIT="300")
device.wait_for("ui.ready")
os.kill(int((device.state / "mapper.pid").read_text()), signal.SIGKILL)
self.assertEqual(proc.wait(timeout=20), 143)
self.assertEqual(device.lines("ui.log"), ["signal 15"])
# The interface disappears: the console says why, for RB_DEAD_WAIT.
screen = " ".join(device._drain().split())
self.assertIn("gptokeyb, which turns the buttons into keys, stopped during the run",
screen)
device.assert_console_restored()
def test_start_select_with_an_interface_that_ignores_sigterm(self):
"""An interface that does not stop within RB_STOP_GRACE is killed,
and the engine it started in its own session with it."""
device = self.device()
proc = device.run(["bash", self.script], FAKE_HOLD="1", FAKE_ENGINE="1",
FAKE_IGNORE_TERM="1", FAKE_PRESS="1", RB_STOP_GRACE="1",
RB_MESSAGE_WAIT="300")
self.assertEqual(proc.returncode, 137, proc.stderr.decode())
self.assertLess(device.elapsed, 15)
engine = int((device.state / "engine.pid").read_text())
wait_until(self, lambda: not Path(f"/proc/{engine}").exists(), 5,
"the engine outlived the interface")
log = (self.gamedir / "log.txt").read_text()
self.assertIn("the interface did not stop within 1 s", log)
device.assert_console_restored()
def test_an_interface_that_ignores_sigterm_without_pkill(self):
"""With no working pkill, the engines cannot be found: the interface
itself is still killed after RB_STOP_GRACE."""
device = self.device()
device.tool("pkill", "#!/bin/sh\necho 'pkill: applet not found' >&2\nexit 127\n")
proc = device.run(["bash", self.script], FAKE_HOLD="1", FAKE_IGNORE_TERM="1",
FAKE_PRESS="1", RB_STOP_GRACE="1", RB_MESSAGE_WAIT="300")
self.assertEqual(proc.returncode, 137, proc.stderr.decode())
self.assertLess(device.elapsed, 15)
ui = int((device.state / "ui.pid").read_text())
self.assertFalse(Path(f"/proc/{ui}").exists())
def test_the_sentinel_ends_when_the_launcher_is_killed(self):
"""SIGKILL leaves the launcher no trap: the sentinel ends by itself
once the launcher is gone."""
device = self.device()
proc = device.spawn(["bash", self.script], FAKE_HOLD="1")
device.wait_for("ui.ready")
token = device.mapper_args()[1]
self.assertEqual(len(sentinels(token)), 1)
proc.kill()
proc.wait(timeout=5)
wait_until(self, lambda: not sentinels(token), 10, "the sentinel outlived the launcher")
def test_a_closed_terminal_ends_the_run(self):
"""The terminal app closed (muxterm's Menu, foot's window): the run
inside it stops the interface and ends the run, and no other route
or screen opens after it."""
device = self.device(WAYLAND_DISPLAY="wayland-1")
device.tool("foot", FAKE_TERMAPP)
proc = device.spawn(["bash", self.script], FAKE_HOLD="1")
device.wait_for("ui.ready")
# FAKE_TERMAPP execs its command: its pid is the run inside it.
os.kill(int((device.state / "termapp.pid").read_text()), signal.SIGHUP)
self.assertEqual(proc.wait(timeout=20), 143)
self.assertEqual(device.lines("ui.log"), ["signal 15"])
self.assertEqual(self.dialog(device), ["FINISH"])
device.assert_mapper_stopped()
def test_sigterm_to_the_launcher_stops_the_interface_and_restores(self):
"""A firmware that signals the launcher alone: the trap stops the
interface, the mapper and puts the console back."""
@@ -1086,10 +1341,12 @@ class ThePortMasterPort(unittest.TestCase):
runtime = device.root / "runtime"
runtime.mkdir(mode=0o555)
self.addCleanup(runtime.chmod, 0o755)
proc = device.run(["bash", self.script])
proc = device.run(["bash", self.script], RB_DEAD_WAIT="2")
self.assertEqual(proc.returncode, 1)
self.assertEqual(device.calls(), [])
self.assertIn("the SD card may be read-only or full", " ".join(device.screen.split()))
# Left on the console long enough to read before the menu redraws.
self.assertGreaterEqual(device.elapsed, 2)
def test_the_previous_log_is_kept(self):
device = self.device()
@@ -1125,6 +1382,7 @@ class TheBatoceraPort(unittest.TestCase):
# evmapy reads RetroBIOS.sh.keys: no gptokeyb is started.
self.assertEqual(device.mapper_args(), [])
self.assertEqual(call["pad"], "evmapy")
self.assertEqual(call["idle_exit"], "300")
def test_the_x_cookie_goes_with_the_display(self):
"""S31emulationstation starts X with XAUTHORITY=/var/lib/.Xauthority:
@@ -1218,13 +1476,15 @@ class TheMuosApp(unittest.TestCase):
f"muxterm --gl -- bash {app}/tui.txt here"])
# muxterm leaves the pad to gptokeyb2, which sends each button once
# with the interface's mapping: Y is Tab (Options), not muxterm's Space.
self.assertEqual(device.lines("termapp.sdl"), ["0x0001/0x0001"])
[hint] = device.lines("termapp.sdl")
assert_muxterm_hides_every_pad(self, hint)
args = device.mapper_args()
self.assertEqual(args[0], "-X")
self.assertRegex(args[1], r"^retrobios-stop-\d+x$")
self.assertEqual(args[2:], ["-c", str(app / "retrobios.gptk")])
self.assertEqual(read_gptk(app / "retrobios.gptk")["y"], ["tab"])
assert_muxterm_mapping(self, args, app / "retrobios.gptk")
self.assertEqual(read_gptk(Path(args[-1]))["y"], ["tab"])
self.assertEqual(call["pad"], "gptokeyb")
self.assertEqual(call["idle_exit"], "300")
device.assert_mapper_stopped()
def test_muxterm_alone_without_gptokeyb2(self):
@@ -1233,6 +1493,7 @@ class TheMuosApp(unittest.TestCase):
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
[call] = device.tui_calls()
self.assertEqual(call["pad"], "muxterm")
self.assertEqual(call["idle_exit"], "300")
# muxterm then reads the pad itself.
self.assertEqual(device.lines("termapp.sdl"), [""])
self.assertEqual(device.mapper_args(), [])
@@ -1257,6 +1518,54 @@ class TheMuosApp(unittest.TestCase):
self.assertEqual(device.lines("ui.log"), ["signal 15"])
device.assert_console_restored()
def test_start_select_with_a_gptokeyb2_that_keeps_running(self):
"""gptokeyb2 before 2026-10-05 keeps running after its -X kill: only
the sentinel's end says Start + Select was pressed."""
device, app = self.device(muxterm=True)
proc = device.run([str(app / "mux_launch.sh"), str(app)], FAKE_HOLD="1",
FAKE_PRESS="1", FAKE_MAPPER_STAYS="1", RB_MESSAGE_WAIT="300")
self.assertEqual(proc.returncode, 143, proc.stderr.decode())
self.assertLess(device.elapsed, 10)
self.assertEqual(device.lines("ui.log"), ["signal 15"])
self.assertIn("Start + Select", (app / "log.txt").read_text())
device.assert_mapper_stopped()
def test_a_gptokeyb2_that_cannot_start_is_named(self):
device, app = self.device(muxterm=False)
proc = device.run([str(app / "mux_launch.sh"), str(app)], FAKE_MAPPER_DIES="1",
RB_CONSOLE=device.console, RB_DEAD_WAIT="1")
self.assertEqual(proc.returncode, 1)
self.assertEqual(device.tui_calls(), [])
screen = " ".join(device.screen.split())
self.assertIn("gptokeyb2, which turns the buttons into keys, stopped as soon as it "
"started", screen)
def test_an_arrow_at_the_question_leaves_nothing_behind(self):
"""An arrow is three bytes: the first answers the question, the rest
must not reach the interface as two more keys."""
device, app = self.device(muxterm=False)
_sd1, sd2, _pack = self.two_cards(device)
proc = device.run([str(app / "mux_launch.sh"), str(app)], keys="\x1b[B",
FAKE_CURL="7", FAKE_DRAIN="1", RB_CONSOLE=device.console,
RB_BIOS_DEST=str(sd2))
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
[call] = device.tui_calls()
self.assertNotIn("--from", call["args"])
self.assertEqual(device.lines("ui.input"), [""])
def test_start_select_at_the_question_starts_no_interface(self):
device, app = self.device(muxterm=False)
_sd1, sd2, _pack = self.two_cards(device)
proc = device.run([str(app / "mux_launch.sh"), str(app)], FAKE_CURL="7",
FAKE_PRESS="1", FAKE_PRESS_AT="calls.jsonl", RB_MESSAGE_WAIT="300",
RB_CONSOLE=device.console, RB_BIOS_DEST=str(sd2))
self.assertEqual(proc.returncode, 143, proc.stderr.decode())
self.assertLess(device.elapsed, 10)
self.assertEqual(device.tui_calls(), [])
# Never started: an interface started then stopped at once would
# end the same way, before Python records anything.
self.assertFalse((app / "run" / "tui.err").exists())
def test_on_the_console_without_muxterm(self):
device, app = self.device(muxterm=False)
proc = device.run([str(app / "mux_launch.sh"), str(app)], RB_CONSOLE=device.console)
@@ -1376,6 +1685,7 @@ class TheToolsEntry(unittest.TestCase):
[call] = device.tui_calls()
self.assertEqual((call["stdin"], call["term"]), (device.console, "linux"))
self.assertEqual(call["pad"], "gptokeyb")
self.assertEqual(call["idle_exit"], "300")
args = device.mapper_args()
self.assertEqual(args[0], "-1")
self.assertRegex(args[1], r"^retrobios-stop-\d+x$")
@@ -1454,26 +1764,11 @@ class TheToolsEntry(unittest.TestCase):
etc = device.root / "fs" / "etc"
etc.mkdir(parents=True)
(etc / "os-release").write_text('NAME="ROCKNIX"\n')
device.tool("sudo", ROCKNIX_SUDO)
def test_rocknix_runs_in_the_terminal_tools_opens(self):
device, tools = self.device()
self.rocknix(device)
proc = device.run(["bash", str(tools / "RetroBIOS.sh")], stdio_tty=True, TERM="xterm")
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
[call] = device.tui_calls()
self.assertEqual((call["stdin"], call["stdout"]), (device.console, device.console))
self.assertEqual(call["term"], "xterm")
self.assertEqual(call["pad"], "gptokeyb")
# The terminal app owns its window: no console reset, no font.
self.assertNotIn("\x1bc", device.screen)
self.assertEqual([t for t in device.tools() if t.startswith("setfont")], [])
def test_rocknix_loads_the_controller_list_its_own_tools_load(self):
"""control-gen_init.sh writes control.ini, which names the pad's
controller list and the GPTOKEYB to start (ROCKNIX fex-emu's Start
FEXConfig.sh). Its ESUDO is not this launcher's."""
device, tools = self.device()
self.rocknix(device)
def control_ini(self, device: Device) -> None:
"""control-gen_init.sh writes control.ini as control-gen.cpp:31
does: `ESUDOKILL="-sudokill"`, and GPTOKEYB with it."""
folder = device.root / "fs" / "storage" / ".config" / "gptokeyb"
shutil.copy(device.bin / "gptokeyb", device.root / "gptokeyb-rocknix")
device.tool("control-gen_init.sh", f'''#!/bin/sh
@@ -1490,15 +1785,57 @@ param_device="fake pad"
GPTOKEYB="$controlfolder/gptokeyb-rocknix $ESUDOKILL"
INI
''')
def test_rocknix_runs_in_the_terminal_tools_opens(self):
device, tools = self.device()
self.rocknix(device)
proc = device.run(["bash", str(tools / "RetroBIOS.sh")], stdio_tty=True, TERM="xterm")
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
args = device.mapper_args()
self.assertEqual(args[0], "-sudokill")
self.assertRegex(args[1], r"^retrobios-stop-\d+x$")
[call] = device.tui_calls()
self.assertEqual((call["stdin"], call["stdout"]), (device.console, device.console))
self.assertEqual(call["term"], "xterm")
self.assertEqual(call["pad"], "gptokeyb")
self.assertEqual(call["idle_exit"], "300")
# The terminal app owns its window: no console reset, no font.
self.assertNotIn("\x1bc", device.screen)
self.assertEqual([t for t in device.tools() if t.startswith("setfont")], [])
def test_rocknix_loads_the_controller_list_its_own_tools_load(self):
"""control-gen_init.sh writes control.ini, which names the pad's
controller list and the GPTOKEYB to start (ROCKNIX fex-emu's Start
FEXConfig.sh). Its ESUDO is not this launcher's."""
device, tools = self.device()
self.rocknix(device)
self.control_ini(device)
proc = device.run(["bash", str(tools / "RetroBIOS.sh")], stdio_tty=True, TERM="xterm")
self.assertEqual(proc.returncode, 0, proc.stderr.decode())
self.assertEqual(device.lines("mapper.args")[0], "-1")
self.assertRegex(device.mapper_args()[1], r"^retrobios-stop-\d+x$")
self.assertEqual(device.lines("mapper.db"), [f"{device.root}/gamecontrollerdb.txt"])
started = [t for t in device.tools() if t.startswith("sudo")]
self.assertEqual(started, [])
# Stopped without the sudo control.ini names.
device.assert_mapper_stopped()
def test_rocknix_start_select_with_its_controller_setup(self):
"""control.ini's `-sudokill` runs the kill through sudo, which on
ROCKNIX runs nothing: the mapper is started with -1, and Start +
Select ends the run at once instead of 3 s later through the
liveness watch."""
device, tools = self.device()
self.rocknix(device)
self.control_ini(device)
proc = device.run(["bash", str(tools / "RetroBIOS.sh")], stdio_tty=True, TERM="xterm",
FAKE_HOLD="1", FAKE_PRESS="1", RB_MESSAGE_WAIT="300")
self.assertEqual(proc.returncode, 143, proc.stderr.decode())
self.assertLess(device.elapsed, 10)
self.assertEqual(device.lines("ui.log"), ["signal 15"])
log = (tools / "retrobios" / "log.txt").read_text()
self.assertIn("Start + Select: the interface is asked to stop", log)
self.assertNotIn("the controller mapper stopped", log)
self.assertNotIn("You are 'root'", log)
device.assert_mapper_gone()
def test_rocknix_offline_names_no_pack_the_releases_lack(self):
"""No release pack spells ROCKNIX, and install.py refuses one named
for another system: the RetroArch pack on the card is not offered,
@@ -1527,6 +1864,28 @@ INI
device.assert_mapper_gone()
class TheLiveness(unittest.TestCase):
def test_a_zombie_is_not_running(self):
"""A process that exited and that its parent has not reaped still
has /proc/PID: rb_alive reads its state."""
pid = os.fork()
if pid == 0:
os._exit(0)
self.addCleanup(os.waitpid, pid, 0)
wait_until(self, lambda: Path(f"/proc/{pid}/stat").read_text().split(") ")[1][0] == "Z",
5, "no zombie")
proc = subprocess.run(
["bash", "-c", 'source "$1"; rb_alive "$2" && echo alive || echo gone', "bash",
str(PORTS / "shared" / "tui.txt"), str(pid)],
capture_output=True, text=True, check=False)
self.assertEqual(proc.stdout.strip(), "gone", proc.stderr)
proc = subprocess.run(
["bash", "-c", 'source "$1"; rb_alive "$$" && echo alive || echo gone', "bash",
str(PORTS / "shared" / "tui.txt")],
capture_output=True, text=True, check=False)
self.assertEqual(proc.stdout.strip(), "alive", proc.stderr)
class TheDocs(unittest.TestCase):
def test_the_offline_packs_named_are_the_ones_that_serve(self):
"""The wiki names the pack each firmware takes offline: the release
@@ -1552,6 +1911,27 @@ class TheDocs(unittest.TestCase):
self.assertIn(f"| `{value}` |", text)
self.assertIn(value, tui)
def handheld(self) -> str:
text = (REPO_ROOT / "wiki" / "installer.md").read_text(encoding="utf-8")
start = text.index("## On a handheld")
end = text.find("\n## ", start + 1)
return text[start:end if end != -1 else len(text)]
def test_the_idle_limit_is_documented(self):
"""Every launcher sets RETROBIOS_IDLE_EXIT for the interface: a
mapper that runs but never opens the pad leaves no button to quit."""
self.assertIn("`RETROBIOS_IDLE_EXIT`", self.handheld())
self.assertIn("RETROBIOS_IDLE_EXIT", (PORTS / "shared" / "tui.txt").read_text())
def test_no_page_buttons_are_promised_in_muxterm(self):
"""muxterm keeps PgUp and PgDn for its scrollback (T/input.c:352-360)."""
readme = " ".join((PORTS / "muos" / "README.md").read_text().split())
self.assertNotIn("L1 and R1 to page", readme)
self.assertIn("L1 and R1 do nothing in muxterm", readme)
rows = [line for line in self.handheld().splitlines() if line.startswith("| L1, R1 |")]
self.assertEqual(len(rows), 1)
self.assertIn("not in muxterm", rows[0])
def test_muos_paths_are_the_card_muos_mounts(self):
"""muOS mounts the first card at /mnt/mmc (MustardOS internal
mux_launch.sh, GET_VAR storage/rom/mount)."""
+78
View File
@@ -17,6 +17,7 @@ import subprocess
import sys
import tempfile
import threading
import time
import unittest
from types import SimpleNamespace
import unittest.mock
@@ -2372,6 +2373,48 @@ class FileListsAreRetried(unittest.TestCase):
self.assertIn(f"after {install.MAX_RETRIES} attempts", err.getvalue())
class ASilentServerIsBounded(unittest.TestCase):
"""A server that takes the connection and never answers held line mode
93 s: three waits of 30 s for a first byte that never came."""
def test_two_short_waits_then_the_reason(self):
import socket
import threading
listener = socket.socket()
listener.bind(("127.0.0.1", 0))
listener.listen(8)
self.addCleanup(listener.close)
held = []
def accept():
while True:
try:
conn, _addr = listener.accept()
except OSError:
return
held.append(conn)
threading.Thread(target=accept, daemon=True).start()
self.addCleanup(lambda: [conn.close() for conn in held])
port = listener.getsockname()[1]
err = io.StringIO()
start = time.monotonic()
with unittest.mock.patch.object(install, "LIST_TIMEOUT", 0.5), \
unittest.mock.patch.object(install, "_retry_delay", return_value=0), \
unittest.mock.patch.object(install, "MANIFEST_URL",
f"http://127.0.0.1:{port}/{{platform}}.json"), \
contextlib.redirect_stderr(err):
with self.assertRaises(SystemExit):
install.fetch_manifest("retroarch")
elapsed = time.monotonic() - start
self.assertLess(elapsed, 5)
self.assertEqual(len(held), install.STALLED_ATTEMPTS)
self.assertIn(f"after {install.STALLED_ATTEMPTS} attempts", err.getvalue())
# A download keeps its own, longer wait between chunks.
self.assertGreater(install.LIST_TIMEOUT, 1)
class TargetListFailureIsNotAnAnswer(unittest.TestCase):
"""A timeout is not 'no targets': it sent users to the full pack."""
@@ -3251,6 +3294,41 @@ class DataDirectoriesAreInstalled(unittest.TestCase):
self.assertEqual(counts["changed"], 1)
self.assertLess(Endless.served, 64 * 1024)
def test_a_damaged_member_is_a_reason_and_leaves_no_part(self):
"""zlib.error and EOFError from a damaged member ended the run on a
traceback, and a failed write left its .part beside the tree."""
import zlib
work = Path(_tempdir(self))
entry = self._entry()
entry["files"] = {"Sys/GC/font.bin": {"sha1": "1" * 40, "size": 4}}
for error in (zlib.error("invalid stored block lengths"), EOFError(),
OSError(28, "No space left on device")):
with self.subTest(error=type(error).__name__):
class Broken(io.RawIOBase):
def readable(self):
return True
def readinto(self, buffer, error=error):
raise error
def members(_archive, _entry):
yield "Sys/GC/font.bin", 4, lambda: io.BufferedReader(Broken())
with unittest.mock.patch.object(install, "_fetch_archive",
lambda url, d: Path(d) / "a"), \
unittest.mock.patch.object(install, "_archive_members", members):
counts = install.install_data_directory(entry, work)
self.assertTrue(counts["error"])
self.assertEqual(counts["installed"], 0)
self.assertEqual([p.name for p in work.rglob("*.part")], [])
def test_offline_keeps_its_reason(self):
work = Path(_tempdir(self))
with unittest.mock.patch.dict(os.environ, {install.OFFLINE_VARIABLE: "1"}):
counts = install.install_data_directory(self._entry(), work)
self.assertIsInstance(counts["error"], install.OfflineReason)
def test_a_tarball_maps_through_source_path(self):
import io as _io
import tarfile
+50
View File
@@ -288,6 +288,56 @@ class APlatformFileSettledByCoresAndSystems(unittest.TestCase):
self.assertEqual(self.served(["dsi"]), "c")
self.assertEqual(self.served(["nds"]), "b")
def test_declared_only_keeps_the_declaration_whose_system_remains(self):
"""--declared-only settled the slot on the first declaration before
the cores narrowed it: on a target without melonds it checked the
nintendo-dsi copy, the platform the nintendo-ds one."""
manifest = self.manifest()
for names, letter in ((["nds"], "b"), (["dsi"], "c"), (["melonds"], "c"),
(["desmume"], "b")):
with self.subTest(names=names):
files, omitted = install.keep_declared(manifest["files"], [])
files, _ = install.narrow_to_cores(manifest, names, files, omitted)
self.assertEqual(files[0]["sha1"][0], letter)
def test_the_declared_event_judges_the_copy_the_platform_checks(self):
"""melonds' copy is served at the slot, the platform checks its own:
the event and a --declared-only check give the same answer."""
import hashlib
tmp = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
self.addCleanup(shutil.rmtree, tmp, True)
payloads = {letter: letter.encode() * 3 for letter in "abc"}
manifest = self.manifest()
entry = manifest["files"][0]
def real(copy: dict, letter: str) -> None:
copy.update(sha1=hashlib.sha1(payloads[letter]).hexdigest(),
sha256=hashlib.sha256(payloads[letter]).hexdigest(), size=3)
real(entry, "a")
real(entry["slot_choices"][1], "a")
real(entry["declared_choices"][0], "c")
real(entry["declared_choices"][1], "b")
(tmp / "dsi_bios9.bin").write_bytes(payloads["c"])
def declared_event(files: list[dict]) -> dict:
_todo, fine, wrong = install.check_local(files, tmp)
with mock.patch.object(install.EVENTS, "emit") as emit:
install._emit_check("retroarch", manifest, tmp, files, fine, wrong, 0, 0)
return next(call.kwargs for call in emit.call_args_list
if call.args[0] == "declared")
for names in (None, ["melonds"], ["desmume", "melonds"]):
with self.subTest(names=names):
served = manifest["files"]
own, _ = install.keep_declared(manifest["files"], [])
if names is not None:
served, _ = install.narrow_to_cores(manifest, names)
own, _ = install.narrow_to_cores(manifest, names, own, [])
self.assertEqual(declared_event(served), declared_event(own))
self.assertEqual(declared_event(own)["present"], 1)
def test_the_boundary_checks_every_copy(self):
install._validate_manifest(self.manifest(), "retroarch")
bad_choices = (
+84
View File
@@ -17,7 +17,9 @@ import errno
import io
import json
import os
import re
import subprocess
import threading
import sys
import time
import unittest
@@ -518,6 +520,18 @@ class ACopyThatFails(FromFolder):
self.assertNotIn(install.NOT_WRITABLE, reason)
self.assertIsInstance(reason, install.NetworkReason)
def test_offline_the_fallback_keeps_the_offline_reason(self):
"""The copy failed, the download was refused by --offline: the
joined reason lost its kind, and the advice named neither."""
_unprivileged(self)
self.source.chmod(0)
self.addCleanup(self.source.chmod, 0o644)
with mock.patch.dict(os.environ, {install.OFFLINE_VARIABLE: "1"}):
_dest, ok, reason = install._download_one(self.entry, self.dest, copy=self.copy)
self.assertFalse(ok)
self.assertIsInstance(reason, install.OfflineReason)
self.assertIn("could not be read: Permission denied, and not fetched", reason)
def test_a_read_error_is_the_source_and_not_the_folder(self):
class Failing(io.BytesIO):
def read(self, *_args):
@@ -596,6 +610,28 @@ class UnreadableCopies(FromFolder):
self.assertIn(line, proc.stdout)
self.assertNotIn("notes.txt", proc.stdout)
def test_names_on_the_card_cannot_drive_the_terminal(self):
"""The plain output printed the paths it found as they are: a name
holding ESC cleared the screen or set the clipboard (OSC 52)."""
hostile = "x\x1b]52;c;eA==\x07\x1b[2J"
self.locked(f"{hostile}/scph5501.bin", US)
self.keep(f"{hostile}/gba_bios.bin", GBA)
(self.dest / f"pcsx{hostile}").mkdir(parents=True)
(self.dest / f"pcsx{hostile}" / "extra.bin").write_bytes(EXTRA)
for options in (("--check",), ("--dry-run",), ("--check", "--verbose"), ()):
with self.subTest(options=options):
proc = self.install(*options)
self.assertNotIn("Traceback", proc.stderr)
for stream in (proc.stdout, proc.stderr):
self.assertIsNone(re.search(r"[\x00-\x09\x0b-\x1f\x7f-\x9f]", stream),
stream)
self.assertIn("x?]52;c;eA==??[2J", proc.stdout)
proc = self.run_installer("--manifest", str(self.list), "--no-input",
"--from", str(self.root / hostile))
self.assertEqual(proc.returncode, 1)
self.assertNotIn("\x1b", proc.stderr)
self.assertIn("x?]52;c;eA==??[2J", proc.stderr)
def test_an_unreadable_copy_is_an_event(self):
path = self.locked("psx/scph5501.bin", US).resolve()
proc = self.install("--check", "--events")
@@ -843,6 +879,54 @@ class ManySameSizedFiles(FromFolder):
self.assertEqual(len(copies), 3)
self.assertLess(elapsed, 30, f"{elapsed:.1f} s")
def test_one_archive_of_thousands_of_members_is_opened_once(self):
"""Each member read reopened its ZIP and parsed its directory again:
4000 members of one size made 4000 parses of 4000 entries."""
count = 4000
_zip(self.old / "romset.zip", {f"m{n:04}.bin": b"m%015d" % n for n in range(count)})
wanted = [_entry(f"z{n}.bin", b"m%015d" % n, ["x"]) for n in (count - 1, 3, 1234)]
index = install.FolderIndex(self.old)
opened = []
real = zipfile.ZipFile
def counting(*args, **kwargs):
opened.append(args[0])
return real(*args, **kwargs)
start = time.monotonic()
with mock.patch.object(install.zipfile, "ZipFile", side_effect=counting):
copies = install.plan_local_copies(wanted, self.dest, None, [index])
self.assertEqual(len(copies), 3)
# The copy step reads members of the archive the plan opened,
# from threads.
results = []
threads = [threading.Thread(target=lambda e=e: results.append(
install._copy_local(e, self.dest, self.dest / e["dest"], copies[id(e)], None)))
for e in wanted]
self.dest.mkdir(parents=True, exist_ok=True)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
elapsed = time.monotonic() - start
self.assertEqual(sorted(ok for _d, ok, _r in results), [True] * 3, results)
self.assertEqual(len(opened), 1)
self.assertLess(elapsed, 20, f"{elapsed:.1f} s")
index.close()
self.assertEqual(index.open_archives(), 0)
def test_a_card_of_many_archives_holds_few_open(self):
count = install.OPEN_ARCHIVES * 3
for n in range(count):
_zip(self.old / f"sets/s{n:03}.zip", {f"m{n}.bin": b"n%015d" % n})
wanted = [_entry(f"z{n}.bin", b"n%015d" % n, ["x"]) for n in range(count)]
index = install.FolderIndex(self.old)
copies = install.plan_local_copies(wanted, self.dest, None, [index])
self.assertEqual(len(copies), count)
self.assertLessEqual(index.open_archives(), install.OPEN_ARCHIVES)
install.LOCAL.close_indexes()
self.assertEqual(index.open_archives(), 0)
def test_a_file_of_the_same_name_comes_first(self):
self.keep("a/other.bin", US)
self.keep("b/scph5501.bin", US)
+46
View File
@@ -496,6 +496,19 @@ class TheBoundaryChecksTheNewFields(unittest.TestCase):
with self.subTest(field=field, value=bad), self.assertRaises(ValueError):
install._validate_manifest(manifest, "retroarch")
def test_the_system_titles(self):
"""A title is read as text by --system and the interface: a list or
a number there raised mid-run instead of at the boundary."""
good = self._manifest()
good["system_titles"] = {"sony-playstation": "Sony PlayStation"}
install._validate_manifest(good, "retroarch")
for bad in (["x"], {"s": ["Sony"]}, {"s": None}, {"s": 5}, {"s": ""},
{"": "Sony"}, {"s": "x" * 257}, {"s" * 257: "Sony"}):
manifest = self._manifest()
manifest["system_titles"] = bad
with self.subTest(value=bad), self.assertRaises(ValueError):
install._validate_manifest(manifest, "retroarch")
class _QuietHandler(http.server.SimpleHTTPRequestHandler):
def log_message(self, format, *args): # noqa: A002 - stdlib signature
@@ -1491,5 +1504,38 @@ class TheTitleASetupPrints(unittest.TestCase):
self.assertEqual(install.system_aliases(manifest)["Sony PlayStation"], "arcade")
class SomethingElseAtADestination(unittest.TestCase):
"""A folder, or a file this user cannot read, at a destination ended
--check and the install on an IsADirectoryError traceback."""
def prepare(self, repo: FakeRepository) -> None:
(repo.dest / "scph5500.bin").mkdir(parents=True)
unreadable = repo.dest / "scph5501.bin"
unreadable.write_bytes(b"x")
unreadable.chmod(0)
@unittest.skipIf(os.name == "nt" or (hasattr(os, "geteuid") and os.geteuid() == 0),
"a mode of 0 stops no read on Windows or as root")
def test_each_says_why_in_the_check_and_the_install(self):
with FakeRepository() as repo:
self.prepare(repo)
proc = repo.run("--check")
self.assertEqual(proc.returncode, 1, proc.stderr)
self.assertNotIn("Traceback", proc.stderr)
self.assertIn(" wrong hash: scph5500.bin\n a folder sits where this "
"file goes: move it away, then run again", proc.stdout)
self.assertIn(" wrong hash: scph5501.bin\n could not be read "
"(Permission denied)", proc.stdout)
proc = repo.run("--yes")
self.assertEqual(proc.returncode, 1, proc.stderr)
self.assertNotIn("Traceback", proc.stderr)
self.assertIn("failed: scph5500.bin (a folder sits where this file goes",
proc.stdout)
self.assertIn("failed: scph5501.bin (file in place could not be kept aside",
proc.stdout)
self.assertTrue((repo.dest / "scph5500.bin").is_dir())
self.assertIn("gba_bios.bin", repo.installed())
if __name__ == "__main__":
unittest.main()
+313 -4
View File
@@ -10,6 +10,8 @@ installs it).
from __future__ import annotations
import argparse
import contextlib
import errno
import fcntl
import functools
import hashlib
@@ -17,6 +19,7 @@ import http.server
import io
import json
import os
import queue
import re
import select
import shutil
@@ -1095,6 +1098,13 @@ class PadRoutes(unittest.TestCase):
"Close the terminal"):
self.assertIn(words, muxterm)
self.assertNotIn("Start+Select", muxterm)
# muxterm keeps PgUp and PgDn for its scrollback, and L1 and R1
# send nothing: no paging is offered there, on a list or a pager.
for screen in ("systems", "result", "plan"):
with self.subTest(screen=screen):
text = help_text("muxterm", screen)
for words in ("L1", "R1", "PgUp", "PgDn", "Page"):
self.assertNotIn(words, text)
miyoo = help_text("miyoo", "systems")
for words in ("Start, Enter", "A, Space, Right/Left", "R2, Esc, Backspace",
"L2, Tab, o", "Sends no key here"):
@@ -1444,13 +1454,14 @@ class CleanupSurvivesABrokenTerminal(unittest.TestCase):
mock.patch.object(sys, "stderr", out):
code = install.run_tui(_args())
self.assertIs(sys.stdout, out)
self.assertEqual(code, 1)
# A window that answers EIO is a closed window: its signal's code,
# not a crash.
self.assertEqual(code, 129)
# The farewell found no terminal: what still flushes at exit goes to
# the null device, and the exit code stays the one returned.
self.assertEqual(os.fstat(screen_fd).st_rdev, os.stat(os.devnull).st_rdev)
# The crash keeps its log, marked kept, and nothing else of the run.
self.assertEqual(sorted(path.name for path in work.iterdir()),
["error.log", "kept", "owner"])
# No crash log is kept, and nothing else of the run.
self.assertFalse(work.exists())
self.assertEqual(termios.tcgetattr(slave), before)
for number, handler in handlers.items():
self.assertEqual(signal.getsignal(number), handler)
@@ -1800,6 +1811,31 @@ class WorkFolders(unittest.TestCase):
work, out = self.run_to(130)
self.assertFalse(work.exists())
def test_a_full_temp_disk_or_a_failing_pruner_still_opens(self):
real_write = Path.write_text
def full(path, *args, **kwargs):
if path.name == install.TUI_OWNER:
raise OSError(errno.ENOSPC, "No space left on device")
return real_write(path, *args, **kwargs)
with mock.patch.object(Path, "write_text", full):
work, out = self.run_to(0)
self.assertFalse(work.exists())
with mock.patch.object(install, "_prune_work_folders",
side_effect=OSError(errno.EACCES, "denied")):
work, out = self.run_to(0)
self.assertFalse(work.exists())
err = io.StringIO()
with mock.patch.object(install.tempfile, "mkdtemp",
side_effect=OSError(errno.ENOSPC, "No space left")), \
mock.patch.object(install, "tui_start", return_value=(fresh_state(), [])), \
mock.patch.object(install, "_open_local_sources"), \
mock.patch.object(install, "detect_os", return_value="linux"), \
mock.patch.object(sys, "stderr", err):
self.assertEqual(install.run_tui(_args()), 1)
self.assertIn("no work folder could be made", err.getvalue())
def test_a_stop_while_it_starts_ends_on_the_farewell(self):
"""Start+Select during the detection: the signal is the interface's,
never the default action that ended Python without a word."""
@@ -2095,6 +2131,33 @@ class AuditRoundThree(unittest.TestCase):
self.assertEqual(install.card_dest("retroarch", "retroarch", root),
("retroarch", root))
def test_p1_1_the_folder_named_is_compared_by_folders(self):
"""NOTMUOS/bios ended with MUOS/bios as text and was taken for a
muOS card's folder."""
root = self.card({"roms/": ""})
stray = root / "NOTMUOS" / "bios"
with self.assertRaises(SystemExit), \
contextlib.redirect_stderr(io.StringIO()) as err:
install.card_dest("muos", "retroarch", stray)
self.assertIn("is not a muOS card", err.getvalue())
self.assertEqual(install.card_dest("muos", "retroarch", root / "muos" / "BIOS"),
("retroarch", root / "muos" / "BIOS"))
def test_p1_1_a_card_two_systems_claim_says_so(self):
root = self.card({"MUOS/": "", "spruce/": ""})
self.assertIsNone(install.card_system(root))
with contextlib.redirect_stdout(io.StringIO()) as out:
self.assertEqual(install.card_dest("retroarch", "retroarch", root),
("retroarch", root))
lines = out.getvalue().splitlines()
self.assertEqual(len(lines), 1, lines)
self.assertIn("holds the files of muOS and spruceOS", lines[0])
# The system named among them decides.
with contextlib.redirect_stdout(io.StringIO()) as out:
self.assertEqual(install.card_dest("muos", "retroarch", root),
("retroarch", root / "MUOS/bios"))
self.assertIn("muOS card", out.getvalue())
# P1-2: Space is X on PortMaster and Y on muxterm.
def test_p1_2_space_on_install_opens_the_confirm_screen(self):
@@ -2455,12 +2518,147 @@ class AuditRoundThree(unittest.TestCase):
unknown_old = folder("unknown-old", None, False, 2 * week)
other = root / "something-else"
other.mkdir()
# A stray or hostile owner file is an unknown owner: a number past
# what a pid holds raised OverflowError, and 0 or -1 name a group.
hostile = [folder(f"hostile{index}", None, False, 60)
for index in range(3)]
for path, text in zip(hostile, ("9" * 40, "0", "-1")):
(path / install.TUI_OWNER).write_text(text)
self.assertIsNone(install._owner_alive(path))
removed = install._prune_work_folders(root, now)
self.assertEqual(sorted(removed), sorted([killed, failed_old, unknown_old]))
for path in hostile:
shutil.rmtree(path)
self.assertEqual(sorted(path.name for path in root.iterdir()),
sorted(path.name for path in (running, failed, unknown, other)))
class IdleExit(unittest.TestCase):
def test_the_variable(self):
for value, seconds in (("30", 30.0), (" 2.5 ", 2.5), ("", None), ("0", None),
("-4", None), ("soon", None), ("inf", None), ("nan", None)):
with self.subTest(value=value):
self.assertEqual(install.tui_idle_exit({install.TUI_IDLE_ENV: value}),
seconds)
self.assertIsNone(install.tui_idle_exit({}))
def test_the_farewell(self):
state = fresh_state()
state.idle_closed, state.exit_code = True, 0
self.assertEqual(install.tui_farewell(state),
["RetroBIOS: No button pressed, closing."])
class AuditRoundFour(unittest.TestCase):
"""tui_audit_3.md, each finding held where it was found."""
UNPLACED = {"reason": "no_save_directory", "label": "no save directory", "files": 2,
"required": 0}
def test_p1_1_files_left_out_of_the_save_directory_are_said(self):
state = fresh_state()
state.stack = ["setup"]
with mock.patch.object(install.Path, "exists", return_value=True):
install.tui_update(state, Key("enter"))
install.tui_receive(state, ("manifest", "retroarch", MANIFEST))
summary = {"event": "summary", "mode": "check", "installed": 0, "up_to_date": 0,
"pending": 5, "failed": [], "skipped": [self.UNPLACED], "left_out": [],
"kept": [], "interrupted": False}
for event in [*FILES, *SYSTEMS, CHECK, summary]:
install.tui_receive(state, ("event", state.run_id, event))
with mock.patch.object(install, "_tui_folder_facts", return_value={
"free": 50 * 1024 ** 3, "writable": True, "cores": []}):
install.tui_receive(state, ("ended", state.run_id, 1, []))
line = ("2 files read from RetroArch's save directory left out: its place is "
"not known. Name it with --save-dir.")
plan = " ".join(" ".join(install.tui_view(state, 120, 40)).split())
self.assertIn(line, plan)
state.stack.append("result")
state.code = 0
state.summary = {**summary, "mode": "install", "installed": 5}
self.assertIn(line, install.tui_outcome(state)[1])
class OfflineInTheInterface(unittest.TestCase):
"""--offline in the full screen: a "No connection" error offering a Try
again that can never work, and a plan counting downloads the run
forbids."""
def setUp(self):
patcher = mock.patch.dict(os.environ, {install.OFFLINE_VARIABLE: "1"})
patcher.start()
self.addCleanup(patcher.stop)
def test_no_list_names_the_ways_that_work(self):
inbox = queue.Queue()
install._tui_fetch("retroarch", inbox, io.StringIO())
kind, plat, title, lines, code = inbox.get_nowait()
self.assertEqual((kind, title, code), ("manifest_failed",
"No internet: offline mode", 1))
text = " ".join(lines)
for words in ("--manifest", "--from-pack", "without --offline"):
self.assertIn(words, text)
state = fresh_state()
state.stack = ["setup"]
with mock.patch.object(install.Path, "exists", return_value=True):
install.tui_update(state, Key("enter"))
install.tui_receive(state, (kind, plat, title, lines, code))
self.assertEqual(state.screen, "error")
self.assertNotIn("retry", [row.action for row in install.tui_rows(state)])
view = "\n".join(install.tui_view(state, 80, 24))
self.assertIn("No internet: offline mode", view)
self.assertNotIn("Try again", view)
def test_the_plan_offers_no_download(self):
state = checked_state()
text = " ".join(" ".join(install.tui_view(state, 120, 40)).split())
self.assertNotIn("to download", text)
self.assertIn("need the internet, which --offline does not use", text)
self.assertNotIn("install", [row.action for row in install.tui_rows(state)])
# A file found on this machine can still be copied.
state.found = [{"event": "found", "platform": "retroarch", "dest": FILES[0]["dest"],
"folder": "/media/old", "path": "/media/old/scph5500.bin",
"member": "", "reason": ""}]
text = " ".join(" ".join(install.tui_view(state, 120, 40)).split())
self.assertIn("1 found in old,", text)
self.assertIn("need the internet (--offline)", text)
self.assertEqual(install.tui_rows(state)[0].action, "install")
self.assertIn("--offline", install.tui_command(state))
class NarrowSystemNames(unittest.TestCase):
"""At 40 columns the size column took half the line: Game Boy, Game Boy
Advance and Game Boy Color all read "Nintendo Gam"."""
def test_the_name_comes_before_the_size(self):
state = checked_state()
names = {"nintendo-gb": "Nintendo Game Boy", "nintendo-gba": "Nintendo Game Boy Advance",
"nintendo-gbc": "Nintendo Game Boy Color",
"videoton-tvc": "Videoton TV Computer"}
state.manifest = {**state.manifest, "system_titles": names}
state.systems = [
{"event": "system", "platform": "retroarch", "system": system, "name": system,
"files": 30, "present": 0, "missing": 30, "wrong_hash": 0,
"bytes_needed": 30 * 1024 * 1024}
for system in names]
state.files = [{"event": "file", "platform": "retroarch", "dest": f"s{n}.bin",
"systems": list(names), "size": 1024, "state": "missing"}
for n in range(27)]
state.ticked = set(names)
state.stack.append("systems")
for cols, rows in ((40, 15), (40, 12), (52, 15)):
with self.subTest(size=f"{cols}x{rows}"):
text = "\n".join(install.tui_view(state, cols, rows))
for name in names.values():
self.assertIn(name, text)
self.assertNotIn("...", text)
for line in text.splitlines():
self.assertLessEqual(install.cell_width(line), cols)
# Wide, the full form stays.
self.assertIn("3 files, 30.0 MB + 27 shared",
"\n".join(install.tui_view(state, 120, 40)))
class Crash(unittest.TestCase):
"""An unexpected error ends on one sentence and a log, never a traceback."""
@@ -2620,6 +2818,29 @@ class GameDataRepository(FakeRepository):
return self
class SaveDirectoryRepository(FakeRepository):
"""The fake repository with one file read from RetroArch's save
directory, as Azahar's keys.txt and 105 others of the real list."""
KEYS = b"azahar keys"
def __enter__(self) -> SaveDirectoryRepository:
super().__enter__()
path = self.root / "install" / "retroarch.json"
manifest = json.loads(path.read_text())
(self.root / "bios" / "keys.txt").write_bytes(self.KEYS)
manifest["files"].append({
"dest": "Azahar/Azahar/sysdata/keys.txt", "root": "save",
"core_folder": "Azahar", "sha1": hashlib.sha1(self.KEYS).hexdigest(),
"sha256": hashlib.sha256(self.KEYS).hexdigest(), "size": len(self.KEYS),
"repo_path": "bios/keys.txt", "cores": ["azahar"],
"systems": ["nintendo-3ds"]})
manifest["save_directory"] = {"folder": "saves", "by_core": True}
manifest["base_destination"] = "system"
path.write_text(json.dumps(manifest))
return self
@unittest.skipUnless(pyte and pty and hasattr(pty, "fork"),
"screen tests need pyte (pip install pyte==0.8.2) and a pty")
class EndToEnd(unittest.TestCase):
@@ -2671,6 +2892,49 @@ class EndToEnd(unittest.TestCase):
# The cursor and the main screen come back before the result is printed.
self.assertIn(b"\x1b[?25h\x1b[?1049l", term.raw)
def test_the_save_directory_is_the_one_line_mode_finds(self):
"""The interface named the BIOS folder with --dest, so its runs never
read retroarch.cfg: keys.txt went to a guessed RetroArch/saves, or
was dropped without a word, where line mode writes it to the
savefile_directory RetroArch reads."""
import subprocess
def written(repo, saves: Path) -> list[str]:
return sorted([*repo.installed(), *(
"saves/" + p.relative_to(saves).as_posix()
for p in saves.rglob("*") if p.is_file())])
results = []
for mode in ("line", "interface"):
with self.subTest(mode=mode):
repo = self.repo(SaveDirectoryRepository)
home = Path(self.env["HOME"])
saves = home / "MySaves"
saves.mkdir()
cfg = home / ".config" / "retroarch" / "retroarch.cfg"
cfg.write_text(f'system_directory = "{repo.dest}"\n'
f'savefile_directory = "{saves}"\n')
if mode == "line":
proc = subprocess.run(
[sys.executable, str(REPO_ROOT / "install.py"), "--platform",
"retroarch", "--yes"], env=self.env, capture_output=True,
text=True, timeout=60, check=False, stdin=subprocess.DEVNULL)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("6 files installed", proc.stdout)
else:
term = self.start()
term.wait("Where do you play?")
term.send("\r")
screen = term.wait("files to download")
self.assertIn("6 files to download", screen)
term.send("\r")
term.wait("6 files installed.")
term.send("\r")
self.assertEqual(term.close(), 0)
results.append(written(repo, saves))
self.assertIn("saves/Azahar/Azahar/sysdata/keys.txt", results[-1])
self.assertEqual(results[0], results[1])
def test_one_system_from_the_checklist(self):
repo = self.repo()
term = self.start("--platform", "retroarch", "--dest", str(repo.dest), cols=40, rows=15)
@@ -2859,6 +3123,51 @@ class EndToEnd(unittest.TestCase):
self.assertTrue(os.WIFEXITED(status))
self.assertEqual(term.status, 129)
def test_closing_the_window_while_it_installs(self):
"""The changing screen fails to draw on the closed window: that is
the window's signal, not a crash with a log kept."""
repo = self.repo(GameDataRepository, slow=True)
scratch = repo.root / "scratch"
scratch.mkdir()
self.env["TMPDIR"] = str(scratch)
term = self.start("--platform", "retroarch", "--dest", str(repo.dest), "--jobs", "1")
term.wait("> Install")
term.send("\r")
term.wait("1 of 6 files")
os.close(term.fd)
end = time.monotonic() + 20
while time.monotonic() < end:
pid, status = os.waitpid(term.pid, os.WNOHANG)
if pid:
break
time.sleep(0.05)
else:
os.kill(term.pid, 9)
self.fail("still running once its window closed")
term.status = os.WEXITSTATUS(status)
self.assertEqual(term.status, 129)
self.assertEqual(list(scratch.glob("retrobios-tui-*/error.log")), [])
def test_no_key_on_the_first_screen_closes_it(self):
"""A mapper that never opens the pad left the first screen up for
good: a handheld then has no way out but the power button."""
self.repo()
self.env[install.TUI_IDLE_ENV] = "1"
term = self.start()
term.wait("Where do you play?")
self.assertEqual(term.close(), 0)
self.assertIn("No button pressed, closing.", term.after())
# A key pressed, the screen stays: the person is there.
term = self.start()
term.wait("Where do you play?")
term.send(DOWN)
end = time.monotonic() + 2.5
while time.monotonic() < end:
term._pump(0.05)
self.assertEqual(os.waitpid(term.pid, os.WNOHANG), (0, 0))
term.send("q")
self.assertEqual(term.close(), 130)
def test_a_desktop_names_its_keys_and_a_pad_its_buttons(self):
self.repo()
for env, bar in (({"DISPLAY": ":0"}, "Enter Continue Esc Quit ? Help"),
+7 -3
View File
@@ -83,7 +83,7 @@ It opens a menu:
- **Wrong folder?** Choose **Another setup or folder**, then browse to yours.
- **A handheld's SD card in a reader?** It is listed, as **muOS SD card**.
Arrows move, Enter chooses, Esc goes back. A controller's A and B do the same.
Arrows move, Enter chooses, Esc goes back. A handheld's A and B do the same.
## 3. Install
@@ -142,8 +142,12 @@ The message shows the command that installs Python. Run it, then run the
installer again.
A RetroPie 4.8 image has Python 3.7.3, and its update command finds nothing
newer. Run the command on a computer, with
`--platform retropie --dest \\RETROPIE\bios\`.
newer. Run it on a computer, the `bios` share open:
<!-- os-tabs windows: --platform retropie --dest \\RETROPIE\bios\ -->
```sh
curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh -s -- --platform retropie --dest <the bios share>
```
### Error: the downloaded installer does not match its expected fingerprint.
+35 -12
View File
@@ -226,7 +226,8 @@ below the BIOS folder: Azahar's keys, shared font and NAND files, Citra's
`sysdata`, Dolphin's `User` tree. The file list marks them `root: save`, and
the run writes them below the save directory, which it names before checking
files (`5 files read from RetroArch's save directory go to ...`). For a
RetroArch found by its `retroarch.cfg`, that is `savefile_directory` as
RetroArch found by its `retroarch.cfg` (with `--dest` too, when the folder is
the config's `system_directory`, as in the full screen), that is `savefile_directory` as
RetroArch expands it (`:` for its own folder, `~` for the home), `saves/`
beside the config when the value is `default`, empty, or names no folder, and
the core's own folder below it when `sort_savefiles_enable` is on, as it is by
@@ -236,7 +237,7 @@ the save directory beside it (`saves`, `/storage/savefiles` on Lakka).
applies, or when `savefiles_in_content_dir` or
`sort_savefiles_by_content_enable` keeps saves beside each game, those files
are left out and the run says why, like the other entries the list cannot
install.
install. The full screen says so on its plan and its result.
Batocera, KNULLI, Recalbox, RetroBat and RetroDECK write the save directory at
each launch, one folder per system (`by_system` in the file list). Their lists
@@ -1054,13 +1055,20 @@ muxterm, as the muOS app does.
| B, Select | Back, and quit on the first screen |
| X | Tick or untick a system |
| Y | Options |
| L1, R1 | Page up, page down |
| L1, R1 | Page up, page down (not in muxterm, which keeps them for its own scrollback) |
| Hold Select, then Start | Quit at once (PortMaster port, muOS, Tools) |
Start pressed first also chooses the line in focus: hold Select first. A
running install is then stopped the way B stops it, and the last screen says
what it wrote. On muOS without PortMaster, muxterm reads the pad itself: Y
ticks a system, Select shows its keyboard, and Menu quits.
ticks a system, Select shows its keyboard, and Menu quits. L1 and R1 do not
page there: muxterm keeps PgUp and PgDn for its own scrollback, so a list
moves with the D-pad.
Each launcher also sets `RETROBIOS_IDLE_EXIT` to 300: with no button pressed
on its first screen for 5 minutes, the interface closes by itself. A pad the
system has no mapping for sends nothing, and would otherwise leave the
console stuck on the interface with no way out.
Each launcher tells the interface which buttons reach it, in
`RETROBIOS_PAD`, so the hints can name them:
@@ -1072,7 +1080,12 @@ Each launcher tells the interface which buttons reach it, in
| `evmapy` | `RetroBIOS.sh.keys`, the same buttons as the table | the Batocera and Knulli port |
| `miyoo` | buttons that are already keys: A as Space, Start as Enter, L1 and R1 as E and T, L2 as Tab, R2 as Backspace | no launcher yet |
A launcher with no mapper at all (a keyboard) leaves it unset. On a Miyoo, B,
A launcher with no mapper at all (a keyboard) leaves it unset.
A launcher can also set `RETROBIOS_IDLE_EXIT` to a number of seconds: when
no button reaches the first screen for that long, before any choice, the
interface closes with `No button pressed, closing.` and exit status 0. A
mapper that never opened the pad would otherwise hold the device on that
screen. On a Miyoo, B,
X, Y and Select are Left Ctrl, Shift, Alt and Right Ctrl: pressed alone, a
modifier sends nothing to a terminal, so the interface goes back with R2
instead, and the bottom line says so.
@@ -1112,9 +1125,10 @@ names each route it tried, and `log.prev.txt` keeps the run before.
**From a computer.** Onion, MinUI, NextUI, Allium, GarlicOS and the Switch
have no Python 3 to run the installer. Put the card in a computer and run
the [one-line command](#one-line) there with
`--platform <platform> --dest <BIOS folder on the card>`. This works for
every system above too.
the [one-line command](#one-line) there: it finds a mounted Onion,
spruceOS, MinUI, muOS, Knulli or ROCKNIX card by itself. For a card it does
not recognise, add `--platform <platform> --dest <BIOS folder on the card>`.
This works for every system above too.
## Where the files come from
@@ -1229,13 +1243,15 @@ fields, the installer refuses to narrow rather than install everything.
|------|---------|
| `0` | everything asked for is installed or already up to date |
| `1` | a failure: the one-liner could not download or verify `install.py`, or found no Python 3.8 (nothing was run), an unusable base URL, an unknown platform or target, `--installed-cores` with no core found, a manifest that fails validation, a fetch that never succeeded, no platform to install for in a non-interactive run, some files failed while others installed, `--check` found files to fetch, a `--from-pack` pack that cannot be read, lacks a part or is for another platform, a `--from` folder that does not exist or cannot be read, or `--restore` could not put a file back |
| `2` | nothing was installed: every file the run tried failed (a BIOS folder that cannot be written, for one), the drive has too little free space for the download (checked before any folder is created), or the question about a folder that does not exist, named with `--dest` or typed at the folder question, was answered no. Also an option the installer refused before any download: an unknown option, `--jobs` outside 1 to 32, `--target` with `--every-board` or `--from-pack`. The message then starts with `install.py: error:`. With `--restore`, a folder that is not a dated folder inside `retrobios-backup` |
| `2` | nothing was installed: every file the run tried failed (a BIOS folder that cannot be written, for one), the drive has too little free space for the download (checked before any folder is created), or the question about a folder that does not exist, named with `--dest` or typed at the folder question, was answered no. Also an option the installer refused before any download: an unknown option, `--jobs` outside 1 to 32, `--target` with `--every-board` or `--from-pack`, `--target` or `--list-targets` with `--offline` (or `RETROBIOS_INSTALL_OFFLINE` set), which read a core list from the network. The message then starts with `install.py: error:`. With `--restore`, a folder that is not a dated folder inside `retrobios-backup` |
| `130` | stopped with Ctrl+C, or with `q`, Ctrl+C or Ctrl+D at a question. Run the same command again to carry on |
| `141` | the output was piped to a reader that closed early (`--list-platforms \| head`) |
The full screen ended by SIGTERM (Start+Select on a handheld) before any
install exits `143`. During an install it stops the install first, and exits
`130`.
`130`. Its window or terminal closed (SIGHUP) exits `129`, during an install
too: the install is stopped as Ctrl+C stops it, and the last lines go
nowhere, no one being left to read them.
`install.ps1` passes the installer's code on unchanged: run as a file it exits
with it, run through `irm | iex` it sets `$LASTEXITCODE` (an `exit` there
@@ -1287,7 +1303,9 @@ other schemas under `api/v1/schemas/`.
Schema 1 can gain a kind (`found` came with `--from`, `declared` with `--declared-only`), never a key of an
existing kind: a reader skips a kind it does not know. An option refused
before the run starts (`install.py: error:`, code 2) writes
no event at all. A program reads a missing `end` as that case.
no event at all. A program reads a missing `end` as that case, or as a
crash: an error the installer did not expect ends the process on a Python
traceback, on standard error, without an `end` line.
Five lines of a run of `--check --events --system sony-playstation` on a
RetroArch folder that holds only `scph5501.bin`, in the order written:
@@ -1316,6 +1334,8 @@ RetroArch folder that holds only `scph5501.bin`, in the order written:
| `FORCE_COLOR`, `CLICOLOR_FORCE` | Colour even when the output is a pipe or a file, set to any value but `0` or an empty one. `NO_COLOR` wins over both, and the progress line stays one line per file |
| `SSL_CERT_FILE`, `SSL_CERT_DIR` | The trusted certificates to check servers with, for a Python that finds none of its own |
| `RETROBIOS_PLAIN`, `ACCESSIBLE` | Set to anything but `0`: the questions come line by line, as with `--plain`, instead of the full-screen interface |
| `RETROBIOS_PAD` | The buttons a handheld launcher's mapper sends to the full screen (`gptokeyb`, `muxterm`, `evmapy`, `miyoo`), so its hints name them. See [On a handheld](#on-a-handheld-the-retrobios-port) |
| `RETROBIOS_IDLE_EXIT` | Seconds the full screen's first screen waits for a first button. With none pressed, it closes with exit status 0 |
| `HTTPS_PROXY`, `NO_PROXY` | Honoured for every download through the standard library's default proxy handling: one names the proxy, the other the hosts to reach directly |
## When something goes wrong
@@ -1330,7 +1350,10 @@ The installer fetches its manifest, then downloads only the repository or
`https://github.com` opens. Behind a proxy, set `HTTPS_PROXY` in the
environment. A connection that does not open within 4 seconds counts as
none, so a dropped network gives its error in about 15 seconds. Once a
connection is open, a slow transfer keeps the full timeout. A file that fails three attempts is counted and reported at the
connection is open, a slow transfer keeps the full timeout. The file list
is the exception: a server that takes the connection and then sends nothing
(a Wi-Fi sign-in page, a stuck proxy) gets 15 seconds per wait and two
attempts, about 30 seconds in all. A file that fails three attempts is counted and reported at the
end, and the next run fetches only what is still missing.
The reason is said in words: no internet connection or a name that could not