Files
DroidDeck/tools/tests/test_steam_install.py
T
MaxsTechReview 4c29c6a61e Fetch the Steam client again when it never shows up (#225)
A client stopped part way through updating itself can be left unable to
draw anything. Its files and our install stamp are still on disk, so the
installer never looked at it again and every session sat on "Waiting for
Steam" with nothing retried.

The session now adds a line to package/droiddeck-unconfirmed each time it
starts the client, and the app removes the file on the client's first
frame. After two starts in a row with nothing shown, the installer:
- drops Steam's update manifests, the stamp, .crash and steam.pid
- deletes package zips whose sha1 no longer matches their name
- downloads the client again over the old one

steamapps, config and userdata are left alone. An automatic repair runs
once per failure, until a start is seen again.

The Steam page's Client settings get a "Repair Steam client" item that
queues the same repair for the next start. The request is only cleared
once the download completes, so an interrupted repair is retried.
2026-10-04 21:33:16 -04:00

179 lines
8.4 KiB
Python

import hashlib
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
import zipfile
SCRIPT = Path(__file__).resolve().parents[1] / 'linuxfs/overlay/usr/local/bin/droiddeck-steam-install'
class SteamInstallTest(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.home = self.root / 'home'
self.home.mkdir()
self.data = self.root / 'data'
self.steam = self.data / 'Steam'
self.cdn = self.root / 'cdn'
self.cdn.mkdir()
self.requests = self.root / 'requests'
self.bin = self.root / 'bin'
self.bin.mkdir()
# Only the network is faked: exercise real manifest parsing, checksums and zip extraction.
curl = self.bin / 'curl'
curl.write_text('''#!/bin/bash
set -eu
printf '%s\\n' "$2" >> "$TEST_REQUESTS"
cp "$TEST_CDN/${2##*/}" "$4"
''')
curl.chmod(0o755)
# macOS has shasum; the guest and Linux CI have sha256sum.
if shutil.which('sha256sum') is None:
sha = self.bin / 'sha256sum'
sha.write_text('#!/bin/bash\nexec shasum -a 256 "$@"\n')
sha.chmod(0o755)
if shutil.which('sha1sum') is None:
sha1 = self.bin / 'sha1sum'
sha1.write_text('#!/bin/bash\nexec shasum -a 1 "$@"\n')
sha1.chmod(0o755)
self.env = dict(os.environ, HOME=str(self.home), XDG_DATA_HOME=str(self.data),
PATH=str(self.bin) + os.pathsep + os.environ['PATH'],
TEST_CDN=str(self.cdn), TEST_REQUESTS=str(self.requests))
self.env.pop('BL_STEAM_CHANNEL', None)
for channel in ('publicbeta', 'steamdeck_publicbeta'):
archive = self.cdn / (channel + '.zip')
with zipfile.ZipFile(archive, 'w') as package:
package.writestr('steamrtarm64/steam', channel)
package.writestr('linuxarm64/wrapper', 'wrapper')
digest = hashlib.sha256(archive.read_bytes()).hexdigest()
(self.cdn / ('steam_client_' + channel + '_linuxarm64')).write_text(
'"linuxarm64"\n{\n\t"version" "123456"\n'
'\t"client_linuxarm64_linuxarm64"\n\t{\n'
f'\t\t"file" "{archive.name}"\n\t\t"sha2" "{digest}"\n'
'\t}\n}\n')
def install(self, channel=None):
env = self.env.copy()
if channel is not None:
env['BL_STEAM_CHANNEL'] = channel
return subprocess.run(['bash', str(SCRIPT)], env=env, text=True, capture_output=True)
def assert_installed(self, channel):
result = self.install(channel)
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual(channel + '\n', (self.steam / 'package/beta').read_text())
self.assertEqual('123456\n', (self.steam / 'package/droiddeck-installed').read_text())
client = self.steam / 'steamrtarm64/steam'
self.assertTrue(os.access(client, os.X_OK))
self.assertEqual(channel, client.read_text())
self.assertEqual((self.steam / 'linuxarm64').resolve(), (self.home / '.steam/sdkarm64').resolve())
self.assertEqual([
'https://client-update.fastly.steamstatic.com/steam_client_' + channel + '_linuxarm64',
'https://client-update.fastly.steamstatic.com/' + channel + '.zip',
], self.requests.read_text().splitlines())
def test_unset_channel_downloads_deck_beta(self):
result = self.install()
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual('steamdeck_publicbeta\n', (self.steam / 'package/beta').read_text())
self.assertIn('/steam_client_steamdeck_publicbeta_linuxarm64\n', self.requests.read_text())
def test_explicit_deck_beta_downloads_matching_client(self):
self.assert_installed('steamdeck_publicbeta')
def test_explicit_public_beta_downloads_matching_client(self):
self.assert_installed('publicbeta')
def test_existing_install_is_left_for_steam_to_update(self):
self.assert_installed('steamdeck_publicbeta')
requests = self.requests.read_text()
result = self.install('publicbeta')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual(requests, self.requests.read_text())
self.assertEqual('steamdeck_publicbeta\n', (self.steam / 'package/beta').read_text())
def test_failed_download_has_no_ready_marker_and_can_retry(self):
archive = self.cdn / 'publicbeta.zip'
original = archive.read_bytes()
archive.write_bytes(b'corrupt download')
self.assertNotEqual(0, self.install('publicbeta').returncode)
self.assertFalse((self.steam / 'package/droiddeck-installed').exists())
archive.write_bytes(original)
self.assertEqual(0, self.install('publicbeta').returncode)
self.assertEqual('publicbeta\n', (self.steam / 'package/beta').read_text())
def break_client(self, starts):
self.assert_installed('steamdeck_publicbeta')
self.requests.unlink()
package = self.steam / 'package'
(self.steam / 'steamrtarm64/steam').write_text('half updated')
(package / 'steam_client_steamdeck_publicbeta_linuxarm64.manifest').write_text('manifest')
(package / 'steam_client_steamdeck_publicbeta_linuxarm64.installed').write_text('installed')
good = b'complete package'
(package / ('bins_linuxarm64.zip.vz.' + hashlib.sha1(good).hexdigest() + '_16')).write_bytes(good)
(package / ('bins_misc_linuxarm64.zip.vz.' + hashlib.sha1(b'whole').hexdigest() + '_5')).write_bytes(b'wh')
(package / ('strings_all.zip.' + hashlib.sha1(b'strings').hexdigest())).write_bytes(b'strings')
(self.steam / '.crash').write_text('')
(package / 'droiddeck-unconfirmed').write_text('start\n' * starts)
return package
def test_two_starts_with_nothing_shown_fetch_the_client_again(self):
package = self.break_client(2)
self.assert_installed('steamdeck_publicbeta')
self.assertEqual('repaired\n', (package / 'droiddeck-unconfirmed').read_text())
self.assertFalse((self.steam / '.crash').exists())
self.assertEqual([], sorted(p.name for p in package.glob('steam_client_*')))
self.assertEqual(['bins_linuxarm64.zip.vz.', 'strings_all.zip.'],
sorted(p.name[:p.name.rindex('.') + 1] for p in package.glob('*.zip.*')))
def test_one_start_with_nothing_shown_is_left_alone(self):
package = self.break_client(1)
self.assertEqual(0, self.install().returncode)
self.assertFalse(self.requests.exists())
self.assertEqual('half updated', (self.steam / 'steamrtarm64/steam').read_text())
self.assertTrue((package / 'steam_client_steamdeck_publicbeta_linuxarm64.installed').exists())
def test_a_repair_is_not_repeated_until_the_client_has_been_seen(self):
package = self.break_client(2)
self.assertEqual(0, self.install().returncode)
self.requests.unlink()
with (package / 'droiddeck-unconfirmed').open('a') as marker:
marker.write('start\nstart\nstart\n')
self.assertEqual(0, self.install().returncode)
self.assertFalse(self.requests.exists())
def test_requested_repair_runs_once_and_is_retried_until_it_completes(self):
package = self.break_client(0)
(package / 'droiddeck-unconfirmed').unlink()
request = self.home / '.bl-steam-repair'
request.write_text('1\n')
archive = self.cdn / 'steamdeck_publicbeta.zip'
original = archive.read_bytes()
archive.write_bytes(b'corrupt download')
self.assertNotEqual(0, self.install().returncode)
self.assertTrue(request.exists())
self.assertFalse((package / 'droiddeck-installed').exists())
archive.write_bytes(original)
self.requests.unlink()
self.assert_installed('steamdeck_publicbeta')
self.assertFalse(request.exists())
self.requests.unlink()
self.assertEqual(0, self.install().returncode)
self.assertFalse(self.requests.exists())
def test_invalid_channel_fails_before_any_download(self):
result = self.install('../unknown')
self.assertNotEqual(0, result.returncode)
self.assertIn('unsupported Steam channel', result.stderr)
self.assertFalse(self.requests.exists())
if __name__ == '__main__':
unittest.main()