mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-11 02:51:10 +00:00
K11MCH1
fcbd45453f
seed_install read the app's manifest from every library before checking that it was the linked DroidDeck install, so a manifest in another library that is not UTF-8 raised past the caller and stopped routing for later games. It now returns before reading anything but the linked manifest. An untouched manifest from an older version was migrated even while steamapps/downloading/<appid> existed; that migration now waits until Steam is not staging the app, like the rewrite of a queued import.
689 lines
36 KiB
Python
689 lines
36 KiB
Python
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import zlib
|
|
from unittest.mock import Mock, patch
|
|
|
|
BIN = Path(__file__).resolve().parents[1] / 'linuxfs/overlay/usr/local/bin'
|
|
|
|
|
|
def load(name):
|
|
loader = importlib.machinery.SourceFileLoader(name, str(BIN / name))
|
|
spec = importlib.util.spec_from_loader(name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
imports = load('droiddeck-steam-games')
|
|
shortcuts = load('droiddeck-steam-shortcuts')
|
|
library = load('droiddeck-steam-library')
|
|
|
|
|
|
class ImportsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.steam = self.root / 'Steam'
|
|
self.acct = self.steam / 'userdata/123'
|
|
self.folder = self.root / 'Games/Example'
|
|
self.folder.mkdir(parents=True)
|
|
self.exe = self.folder / 'Example.exe'
|
|
self.exe.write_bytes(b'game')
|
|
self.game = dict(name='Example', exe=str(self.exe), folder=str(self.folder),
|
|
dir=str(self.folder), appid=0x87654321)
|
|
|
|
def snapshot(self, owned=True, account='123'):
|
|
imports.save_json(self.acct / 'config' / imports.STATE,
|
|
dict(account=account, owned={'42': owned},
|
|
candidates={str(self.folder): 42},
|
|
sources={str(self.folder): imports.source_stamp(self.game)}))
|
|
|
|
def test_owned_uses_real_id_and_registers_installed_files(self):
|
|
self.snapshot()
|
|
games, routes = imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual([], games)
|
|
self.assertEqual({str(self.game['appid']): 42}, routes)
|
|
manifest = self.steam / 'steamapps/appmanifest_42.acf'
|
|
self.assertIn('"StateFlags" "4"', manifest.read_text())
|
|
self.assertIn('"Universe" "1"', manifest.read_text())
|
|
self.assertEqual(self.folder, (self.steam / 'steamapps/common/DroidDeck-42').resolve())
|
|
self.assertEqual(b'game', self.exe.read_bytes())
|
|
|
|
def test_store_installs_stay_shortcuts_even_when_owned_on_steam(self):
|
|
self.snapshot()
|
|
game = dict(self.game, store='amazon')
|
|
games, routes = imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual([game], games)
|
|
self.assertEqual({}, routes)
|
|
self.assertFalse((self.steam / 'steamapps/appmanifest_42.acf').exists())
|
|
self.assertFalse(os.path.lexists(self.steam / 'steamapps/common/DroidDeck-42'))
|
|
|
|
def test_store_installs_imported_earlier_are_released(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
unrelated = self.steam / 'steamapps/appmanifest_7.acf'
|
|
unrelated.write_text('"AppState"\n{\n "appid" "7"\n "installdir" "Other"\n}\n')
|
|
games, routes = imports.route(self.steam, self.acct, [dict(self.game, store='epic')])
|
|
self.assertEqual({}, routes)
|
|
self.assertEqual(1, len(games))
|
|
self.assertFalse((self.steam / 'steamapps/appmanifest_42.acf').exists())
|
|
self.assertFalse(os.path.lexists(self.steam / 'steamapps/common/DroidDeck-42'))
|
|
self.assertTrue(unrelated.is_file())
|
|
self.assertEqual(b'game', self.exe.read_bytes())
|
|
|
|
def test_watch_never_looks_up_store_installs(self):
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([dict(self.game, store='gog')]))
|
|
with patch.object(imports, 'identify') as identify:
|
|
imports.watch(str(self.steam), str(listing))
|
|
identify.assert_not_called()
|
|
|
|
def test_native_programs_stay_shortcuts_and_are_never_looked_up(self):
|
|
self.snapshot()
|
|
game = dict(self.game, native=True)
|
|
self.assertEqual(([game], {}), imports.route(self.steam, self.acct, [game]))
|
|
self.assertFalse((self.steam / 'steamapps/appmanifest_42.acf').exists())
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([game]))
|
|
with patch.object(imports, 'identify') as identify:
|
|
imports.watch(str(self.steam), str(listing))
|
|
identify.assert_not_called()
|
|
|
|
def test_migrates_only_unchanged_legacy_import_manifests(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
path = self.steam / 'steamapps/appmanifest_42.acf'
|
|
current = path.read_text()
|
|
legacy = (current.replace(' "AutoUpdateBehavior" "1"\n', '').replace(' "Universe" "1"\n', '')
|
|
.replace('"StateFlags" "4"', '"StateFlags" "2"'))
|
|
path.write_text(legacy)
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(current, path.read_text())
|
|
changed = legacy.replace('"StateFlags" "2"', '"StateFlags" "6"')
|
|
path.write_text(changed)
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(changed, path.read_text())
|
|
|
|
def test_unresolved_titles_are_retried_after_network_recovers(self):
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([self.game]))
|
|
from unittest.mock import Mock
|
|
idle = imports.downloads_idle
|
|
result = Mock(returncode=0, stdout='DROIDDECK_OWNERSHIP=' + json.dumps(dict(account='123', owned={'42': True})))
|
|
with patch.object(imports, 'identify', side_effect=[None, 42]) as identify, \
|
|
patch.object(imports, 'request_restart'), \
|
|
patch.object(imports, 'ask_restart', return_value=True), \
|
|
patch.object(imports, 'time', Mock(monotonic=Mock(side_effect=[0, 301]),
|
|
sleep=Mock(side_effect=[None, RuntimeError('stop')]))), \
|
|
patch.object(imports.subprocess, 'run', return_value=result), \
|
|
patch('builtins.print'):
|
|
with self.assertRaisesRegex(RuntimeError, 'stop'):
|
|
imports.watch(self.steam, listing)
|
|
self.assertEqual(2, identify.call_count)
|
|
snapshot = imports.read_json(self.acct / 'config' / imports.STATE, {})
|
|
self.assertTrue(snapshot['owned']['42'])
|
|
|
|
def run_watch(self, beats, routes=None, busy=False, answer=True, started=False):
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([self.game]))
|
|
if routes is not None:
|
|
imports.save_json(self.acct / 'config/.droiddeck-routes.json', routes)
|
|
if busy:
|
|
(self.steam / 'steamapps/downloading/7').mkdir(parents=True)
|
|
from unittest.mock import Mock
|
|
idle = imports.downloads_idle
|
|
result = Mock(returncode=0, stdout='DROIDDECK_OWNERSHIP=' + json.dumps(dict(account='123', owned={'42': True})))
|
|
with patch.object(imports, 'identify', return_value=42), \
|
|
patch.object(imports, 'game_running', return_value=False), \
|
|
patch.object(imports, 'request_restart') as restart, \
|
|
patch.object(imports, 'ask_restart', return_value=answer) as asked, \
|
|
patch.object(imports, 'downloads_idle', side_effect=lambda root: idle(root) and not (started and asked.called)), \
|
|
patch.object(imports, 'time', Mock(monotonic=Mock(side_effect=list(range(0, 30 * beats, 30))),
|
|
sleep=Mock(side_effect=[None] * (beats - 1) + [RuntimeError('stop')]))), \
|
|
patch.object(imports.subprocess, 'run', return_value=result), \
|
|
patch('builtins.print'):
|
|
with self.assertRaisesRegex(RuntimeError, 'stop'):
|
|
imports.watch(self.steam, listing)
|
|
self.asked = asked
|
|
return restart
|
|
|
|
def test_newly_owned_game_restarts_the_client_once(self):
|
|
self.assertEqual(1, self.run_watch(3).call_count)
|
|
|
|
def test_an_import_that_did_not_take_restarts_only_once_across_sessions(self):
|
|
self.assertEqual(1, self.run_watch(3).call_count)
|
|
self.assertEqual(0, self.run_watch(3).call_count)
|
|
self.assertEqual([42], imports.read_json(self.acct / 'config' / imports.TRIED, None))
|
|
|
|
def test_later_skips_the_restart_and_does_not_ask_again(self):
|
|
self.assertEqual(0, self.run_watch(3, answer=False).call_count)
|
|
self.assertEqual([['Example']], [c.args[0] for c in self.asked.call_args_list])
|
|
self.assertEqual(0, self.run_watch(3).call_count)
|
|
self.assertEqual(0, self.asked.call_count)
|
|
|
|
def test_a_download_started_while_asking_leaves_the_import_for_the_next_start(self):
|
|
self.assertEqual(0, self.run_watch(3, started=True).call_count)
|
|
self.assertEqual(1, self.asked.call_count)
|
|
|
|
def test_the_app_answers_the_restart_question(self):
|
|
home = self.root / 'home'
|
|
folder = home / '.cache/droiddeck'
|
|
replies = []
|
|
def sleep(_):
|
|
ask = json.loads((folder / 'steam-import.json').read_text())
|
|
replies.append(ask['games'])
|
|
(folder / 'steam-import.answer').write_text(json.dumps(dict(token=ask['token'], choice=choice)))
|
|
for choice, expected in (('now', True), ('later', False)):
|
|
with patch.dict(os.environ, HOME=str(home)):
|
|
self.assertEqual(expected, imports.ask_restart(['Example'], clock=iter(range(100)).__next__, sleep=sleep))
|
|
self.assertFalse((folder / 'steam-import.json').exists())
|
|
self.assertFalse((folder / 'steam-import.answer').exists())
|
|
self.assertEqual([['Example'], ['Example']], replies)
|
|
|
|
def test_no_answer_means_later(self):
|
|
with patch.dict(os.environ, HOME=str(self.root / 'home')):
|
|
self.assertFalse(imports.ask_restart(['Example'], timeout=5, clock=iter(range(100)).__next__, sleep=lambda _: None))
|
|
|
|
def test_a_tried_import_that_took_is_forgotten(self):
|
|
imports.save_json(self.acct / 'config' / imports.TRIED, [42])
|
|
self.assertEqual(0, self.run_watch(2, routes={'1': 42}).call_count)
|
|
self.assertEqual([], imports.read_json(self.acct / 'config' / imports.TRIED, None))
|
|
|
|
def test_imported_game_does_not_restart_the_client(self):
|
|
self.assertEqual(0, self.run_watch(2, routes={'1': 42}).call_count)
|
|
|
|
def test_restart_waits_for_downloads(self):
|
|
self.assertEqual(0, self.run_watch(2, busy=True).call_count)
|
|
|
|
def test_unowned_unknown_and_different_account_stay_shortcuts(self):
|
|
for owned, account in [(False, '123'), (True, '456')]:
|
|
self.snapshot(owned, account)
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
(self.acct / 'config' / imports.STATE).unlink()
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertFalse((self.steam / 'steamapps').exists())
|
|
|
|
def winnative_manifest(self, flags='4', installdir='Example'):
|
|
path = self.folder.parent / imports.WINNATIVE / 'appmanifest_42.acf'
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text('"AppState"\n{\n\t"appid"\t\t"42"\n\t"StateFlags"\t\t"%s"\n\t"installdir"\t\t"%s"\n'
|
|
'\t"buildid"\t\t"7"\n\t"InstalledDepots"\n\t{\n\t\t"43"\n\t\t{\n\t\t\t"manifest"\t\t"99"\n'
|
|
'\t\t\t"size"\t\t"4"\n\t\t}\n\t}\n}\n' % (flags, installdir))
|
|
return path
|
|
|
|
def test_winnative_manifest_carries_installed_build(self):
|
|
source = self.winnative_manifest()
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
text = (self.steam / 'steamapps/appmanifest_42.acf').read_text()
|
|
self.assertEqual(source.read_text().replace('"Example"', '"DroidDeck-42"'), text)
|
|
|
|
def test_winnative_manifest_identifies_folder(self):
|
|
self.winnative_manifest()
|
|
with patch.object(imports.urllib.request, 'urlopen', side_effect=AssertionError):
|
|
self.assertEqual(42, imports.identify(self.game))
|
|
|
|
def test_incomplete_winnative_manifest_is_ignored(self):
|
|
for flags, installdir in [('1026', 'Example'), ('4', 'Other')]:
|
|
self.winnative_manifest(flags, installdir)
|
|
self.assertIsNone(imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
|
|
def test_depot_config_records_installed_depots(self):
|
|
config = self.folder / '.DepotDownloader/depot.config'
|
|
config.parent.mkdir()
|
|
config.write_text(json.dumps({'installedManifestIDs': {'43': 99}}))
|
|
self.assertIsNone(imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
(self.folder / '.download_complete').touch()
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
text = (self.steam / 'steamapps/appmanifest_42.acf').read_text()
|
|
self.assertIn('"InstalledDepots"\n {\n "43"\n {\n "manifest" "99"', text)
|
|
self.assertTrue(text.endswith(' }\n}\n'))
|
|
config.write_text(json.dumps({'installedManifestIDs': {'43': 0x7FFFFFFFFFFFFFFF}}))
|
|
self.assertIsNone(imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
|
|
def test_compressed_protobuf_depot_config_records_installed_depots(self):
|
|
config = self.folder / '.DepotDownloader/depot.config'
|
|
config.parent.mkdir()
|
|
(self.folder / '.download_complete').touch()
|
|
entry = bytes([0x08, 43, 0x10, 99])
|
|
packed = zlib.compressobj(wbits=-15)
|
|
config.write_bytes(packed.compress(bytes([0x0A, len(entry)]) + entry) + packed.flush())
|
|
self.assertIn('"43"\n {\n "manifest" "99"', imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
config.write_bytes(b'not a config')
|
|
self.assertIsNone(imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
(config.parent / 'completed').mkdir()
|
|
(config.parent / 'completed/43_98.manifest').write_bytes(b'x')
|
|
self.assertIn('"manifest" "98"', imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
config.write_text(json.dumps({'installedManifestIDs': {}}))
|
|
self.assertIn('"manifest" "98"', imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
|
|
def test_completed_manifests_record_installed_depots(self):
|
|
completed = self.folder / '.DepotDownloader/completed'
|
|
completed.mkdir(parents=True)
|
|
(self.folder / '.download_complete').touch()
|
|
self.assertIsNone(imports.installed_build(self.folder, 42, 'DroidDeck-42'))
|
|
(completed / '43_98.manifest').write_bytes(b'old')
|
|
(completed / '43_99.manifest').write_bytes(b'new')
|
|
(completed / '44_7.manifest').write_bytes(b'x')
|
|
os.utime(completed / '43_98.manifest', (1, 1))
|
|
text = imports.installed_build(self.folder, 42, 'DroidDeck-42')
|
|
self.assertIn('"43"\n {\n "manifest" "99"', text)
|
|
self.assertIn('"44"\n {\n "manifest" "7"', text)
|
|
self.assertNotIn('"98"', text)
|
|
|
|
def test_bare_import_manifest_gains_installed_build(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.winnative_manifest()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertIn('"buildid"\t\t"7"', (self.steam / 'steamapps/appmanifest_42.acf').read_text())
|
|
|
|
def steam_fresh_install(self, downloaded='0'):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
path = self.steam / 'steamapps/appmanifest_42.acf'
|
|
path.write_text('"AppState"\n{\n\t"appid"\t\t"42"\n\t"StateFlags"\t\t"6"\n\t"installdir"\t\t"DroidDeck-42"\n'
|
|
'\t"buildid"\t\t"0"\n\t"BytesDownloaded"\t\t"%s"\n\t"BytesStaged"\t\t"0"\n'
|
|
'\t"InstalledDepots"\n\t{\n\t}\n}\n' % downloaded)
|
|
self.winnative_manifest()
|
|
return path
|
|
|
|
def test_queued_fresh_install_takes_installed_build(self):
|
|
path = self.steam_fresh_install()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertIn('"buildid"\t\t"7"', path.read_text())
|
|
|
|
def test_import_without_a_build_updates_only_when_launched(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
path = self.steam / 'steamapps/appmanifest_42.acf'
|
|
current = path.read_text()
|
|
self.assertIn(' "AutoUpdateBehavior" "1"\n', current)
|
|
path.write_text(current.replace(' "AutoUpdateBehavior" "1"\n', ''))
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(current, path.read_text())
|
|
|
|
def queued_bare_install(self, extra):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
path = self.steam / 'steamapps/appmanifest_42.acf'
|
|
path.write_text('"AppState"\n{\n\t"appid"\t\t"42"\n\t"StateFlags"\t\t"6"\n\t"installdir"\t\t"DroidDeck-42"\n'
|
|
'\t"buildid"\t\t"0"\n\t"TargetBuildID"\t\t"9"\n%s\t"InstalledDepots"\n\t{\n\t}\n'
|
|
'\t"UserConfig"\n\t{\n\t}\n}\n' % extra)
|
|
return path
|
|
|
|
def test_queued_import_without_a_build_waits_for_a_launch(self):
|
|
path = self.queued_bare_install('\t"AutoUpdateBehavior"\t\t"0"\n')
|
|
before = path.read_text()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(before.replace('"AutoUpdateBehavior"\t\t"0"', '"AutoUpdateBehavior"\t\t"1"'), path.read_text())
|
|
path = self.queued_bare_install('')
|
|
before = path.read_text()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(before[:-2] + '\t"AutoUpdateBehavior"\t\t"1"\n}\n', path.read_text())
|
|
|
|
def test_queued_import_keeps_a_chosen_update_priority(self):
|
|
path = self.queued_bare_install('\t"AutoUpdateBehavior"\t\t"2"\n')
|
|
before = path.read_text()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(before, path.read_text())
|
|
|
|
def test_an_untouched_import_is_left_alone_while_steam_stages_it(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
path = self.steam / 'steamapps/appmanifest_42.acf'
|
|
older = path.read_text().replace(' "AutoUpdateBehavior" "1"\n', '')
|
|
path.write_text(older)
|
|
(self.steam / 'steamapps/downloading/42').mkdir(parents=True)
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(older, path.read_text())
|
|
|
|
def test_a_manifest_in_another_library_is_not_read(self):
|
|
other = self.root / 'Other'
|
|
(other / 'steamapps').mkdir(parents=True)
|
|
(other / 'steamapps/appmanifest_42.acf').write_bytes(b'"AppState"\n{\n "name" "\xff\xfe"\n}\n')
|
|
(self.steam / 'steamapps').mkdir(parents=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text('"libraryfolders"\n{\n\t"1"\n\t{\n\t\t"path"\t\t"%s"\n\t}\n}\n' % other)
|
|
self.snapshot()
|
|
games, routes = imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual({str(self.game['appid']): 42}, routes)
|
|
self.assertEqual(b'"AppState"\n{\n "name" "\xff\xfe"\n}\n', (other / 'steamapps/appmanifest_42.acf').read_bytes())
|
|
|
|
def test_started_import_without_a_build_is_left_to_steam(self):
|
|
for extra, staging in [('\t"BytesDownloaded"\t\t"5"\n\t"AutoUpdateBehavior"\t\t"0"\n', False),
|
|
('\t"AutoUpdateBehavior"\t\t"0"\n', True)]:
|
|
path = self.queued_bare_install(extra)
|
|
if staging:
|
|
(self.steam / 'steamapps/downloading/42').mkdir(parents=True)
|
|
before = path.read_text()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(before, path.read_text())
|
|
|
|
def test_started_fresh_install_is_left_to_steam(self):
|
|
for downloaded, staging in [('5', False), ('0', True)]:
|
|
path = self.steam_fresh_install(downloaded)
|
|
if staging:
|
|
(self.steam / 'steamapps/downloading/42').mkdir(parents=True)
|
|
before = path.read_text()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual(before, path.read_text())
|
|
|
|
def test_links_and_snapshots_from_the_old_mount_carry_over(self):
|
|
old = self.root / 'old-mount'
|
|
with patch.object(imports, 'LIBRARY', str(self.root / 'Games')), \
|
|
patch.object(imports, 'LEGACY_LIBRARY', str(old)):
|
|
legacy_game = dict(self.game, folder=str(old / 'Example'), exe=str(old / 'Example/Example.exe'))
|
|
stamp = imports.source_stamp(self.game)
|
|
imports.save_json(self.acct / 'config' / imports.STATE,
|
|
dict(account='123', owned={'42': True}, candidates={legacy_game['folder']: 42},
|
|
sources={legacy_game['folder']: [stamp[0], legacy_game['exe']] + stamp[2:]}))
|
|
link = self.steam / 'steamapps/common/DroidDeck-42'
|
|
link.parent.mkdir(parents=True)
|
|
link.symlink_to(legacy_game['folder'], target_is_directory=True)
|
|
games, routes = imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual([], games)
|
|
self.assertEqual({str(self.game['appid']): 42}, routes)
|
|
self.assertEqual(str(self.folder), imports.os.readlink(link))
|
|
|
|
def test_existing_manifest_is_never_overwritten(self):
|
|
manifest = self.steam / 'steamapps/appmanifest_42.acf'
|
|
manifest.parent.mkdir(parents=True)
|
|
manifest.write_text('Steam owns this file')
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual('Steam owns this file', manifest.read_text())
|
|
|
|
def test_import_collision_falls_back_without_touching_files(self):
|
|
occupied = self.steam / 'steamapps/common/DroidDeck-42'
|
|
occupied.mkdir(parents=True)
|
|
self.snapshot()
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertTrue(occupied.is_dir())
|
|
self.assertFalse(occupied.is_symlink())
|
|
|
|
def test_removed_folder_does_not_import(self):
|
|
self.snapshot()
|
|
self.exe.unlink()
|
|
self.folder.rmdir()
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
|
|
def test_absent_folder_sets_its_install_aside_until_it_returns(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
manifest = self.steam / 'steamapps/appmanifest_42.acf'
|
|
manifest.write_text(manifest.read_text().replace('}', ' "buildid" "99"\n}'))
|
|
state = manifest.read_text()
|
|
away = self.root / 'unplugged'
|
|
self.folder.rename(away)
|
|
with patch('builtins.print'):
|
|
self.assertEqual(([], {}), imports.route(self.steam, self.acct, []))
|
|
self.assertFalse(manifest.exists())
|
|
self.assertEqual(state, (self.steam / 'steamapps' / imports.PARKED / manifest.name).read_text())
|
|
away.rename(self.folder)
|
|
self.assertEqual(([], {str(self.game['appid']): 42}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertEqual(state, manifest.read_text())
|
|
self.assertFalse((self.steam / 'steamapps' / imports.PARKED / manifest.name).exists())
|
|
|
|
def test_steam_installs_are_never_set_aside(self):
|
|
manifest = self.steam / 'steamapps/appmanifest_42.acf'
|
|
manifest.parent.mkdir(parents=True)
|
|
manifest.write_text('"AppState"\n{\n "appid" "42"\n "installdir" "Example"\n}\n')
|
|
link = self.steam / 'steamapps/common/DroidDeck-42'
|
|
link.parent.mkdir(parents=True)
|
|
link.symlink_to(self.root / 'missing', target_is_directory=True)
|
|
imports.route(self.steam, self.acct, [])
|
|
self.assertTrue(manifest.is_file())
|
|
|
|
def test_install_in_another_library_is_used_as_is(self):
|
|
other = self.root / 'Games/PC'
|
|
(other / 'steamapps').mkdir(parents=True)
|
|
(other / 'steamapps/appmanifest_42.acf').write_text('Steam installed it here')
|
|
vdf = self.steam / 'steamapps/libraryfolders.vdf'
|
|
vdf.parent.mkdir(parents=True)
|
|
vdf.write_text('"libraryfolders"\n{\n\t"1"\n\t{\n\t\t"path"\t\t"%s"\n\t}\n}\n' % other)
|
|
self.snapshot()
|
|
self.assertEqual(([], {str(self.game['appid']): 42}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertFalse((self.steam / 'steamapps/appmanifest_42.acf').exists())
|
|
self.assertFalse(os.path.lexists(self.steam / 'steamapps/common/DroidDeck-42'))
|
|
|
|
def test_imported_game_stays_imported_when_ownership_is_unconfirmed(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
for owned, account in [(False, '123'), (True, '456')]:
|
|
self.snapshot(owned, account)
|
|
self.assertEqual(([], {str(self.game['appid']): 42}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertTrue((self.steam / 'steamapps/appmanifest_42.acf').is_file())
|
|
|
|
def test_unlisted_card_library_is_not_another_clients_install(self):
|
|
card = self.root / 'card'
|
|
(card / 'steamapps').mkdir(parents=True)
|
|
(card / 'steamapps/appmanifest_42.acf').write_text('the other client installed it here')
|
|
self.snapshot()
|
|
with patch.object(imports, 'LIBRARY', str(card)):
|
|
self.assertEqual(([], {str(self.game['appid']): 42}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assertTrue((self.steam / 'steamapps/appmanifest_42.acf').is_file())
|
|
self.assertEqual(self.folder, (self.steam / 'steamapps/common/DroidDeck-42').resolve())
|
|
|
|
def test_replaced_game_does_not_inherit_cached_ownership(self):
|
|
self.snapshot()
|
|
self.exe.write_bytes(b'a different game')
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
|
|
def test_duplicate_owned_folders_share_one_steam_entry(self):
|
|
self.snapshot()
|
|
games, routes = imports.route(self.steam, self.acct, [self.game, self.game])
|
|
self.assertEqual([], games)
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual(1, len(list((self.steam / 'steamapps').glob('*.acf'))))
|
|
|
|
def test_explicit_appid_does_not_need_network(self):
|
|
(self.folder / 'steam_appid.txt').write_text('42\n')
|
|
with patch.object(imports.urllib.request, 'urlopen', side_effect=AssertionError):
|
|
self.assertEqual(42, imports.identify(self.game))
|
|
|
|
def test_spacewar_appid_names_no_game(self):
|
|
(self.folder / 'steam_appid.txt').write_text('480\n')
|
|
with patch.object(imports.urllib.request, 'urlopen', side_effect=OSError):
|
|
self.assertIsNone(imports.identify(self.game))
|
|
|
|
def import_as(self, appid):
|
|
self.assertTrue(imports.seed_install(self.steam, self.game, appid))
|
|
self.assertTrue((self.steam / f'steamapps/appmanifest_{appid}.acf').is_file())
|
|
|
|
def snapshot_names(self, appid, owned):
|
|
imports.save_json(self.acct / 'config' / imports.STATE,
|
|
dict(account='123', owned={str(appid): owned},
|
|
candidates={str(self.folder): appid},
|
|
sources={str(self.folder): imports.source_stamp(self.game)}))
|
|
|
|
def assert_not_imported_as(self, appid):
|
|
self.assertFalse((self.steam / f'steamapps/appmanifest_{appid}.acf').exists())
|
|
self.assertFalse(os.path.lexists(self.steam / f'steamapps/common/DroidDeck-{appid}'))
|
|
|
|
def test_spacewar_import_is_released_and_never_made_again(self):
|
|
self.import_as(480)
|
|
self.snapshot_names(480, owned=True)
|
|
with patch('builtins.print'):
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assert_not_imported_as(480)
|
|
self.assertEqual(b'game', self.exe.read_bytes())
|
|
|
|
def test_import_follows_the_app_the_files_name_now(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.snapshot_names(43, owned=True)
|
|
with patch('builtins.print'):
|
|
self.assertEqual(([], {str(self.game['appid']): 43}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assert_not_imported_as(42)
|
|
self.assertEqual(self.folder, (self.steam / 'steamapps/common/DroidDeck-43').resolve())
|
|
|
|
def test_unowned_new_appid_leaves_a_shortcut(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.snapshot_names(43, owned=False)
|
|
with patch('builtins.print'):
|
|
self.assertEqual(([self.game], {}), imports.route(self.steam, self.acct, [self.game]))
|
|
self.assert_not_imported_as(42)
|
|
self.assert_not_imported_as(43)
|
|
|
|
def test_import_stays_while_the_new_identity_is_unknown(self):
|
|
self.snapshot()
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
for candidates in ({}, {str(self.folder): 43}):
|
|
imports.save_json(self.acct / 'config' / imports.STATE,
|
|
dict(account='123', owned={'43': True}, candidates=candidates,
|
|
sources={str(self.folder): ['stale']}))
|
|
imports.route(self.steam, self.acct, [self.game])
|
|
self.assertTrue((self.steam / 'steamapps/appmanifest_42.acf').is_file())
|
|
|
|
def test_manifest_identity_uses_install_directory(self):
|
|
folder = self.root / 'second/steamapps/common/Custom Name'
|
|
folder.mkdir(parents=True)
|
|
(folder.parent.parent / 'appmanifest_42.acf').write_text('"AppState" { "installdir" "Custom Name" }')
|
|
game = dict(self.game, folder=str(folder), exe=str(folder / 'game.exe'))
|
|
with patch.object(imports.urllib.request, 'urlopen', side_effect=AssertionError):
|
|
self.assertEqual(42, imports.identify(game))
|
|
|
|
def test_a_spacewar_manifest_names_no_game(self):
|
|
folder = self.root / 'second/steamapps/common/PEAK'
|
|
folder.mkdir(parents=True)
|
|
(folder.parent.parent / 'appmanifest_480.acf').write_text('"AppState" { "installdir" "PEAK" }')
|
|
(folder / 'steam_appid.txt').write_text('3527290\n')
|
|
game = dict(self.game, folder=str(folder), exe=str(folder / 'PEAK.exe'))
|
|
with patch.object(imports.urllib.request, 'urlopen', side_effect=AssertionError):
|
|
self.assertEqual(3527290, imports.identify(game))
|
|
|
|
def test_a_store_search_answering_spacewar_names_no_game(self):
|
|
with patch.object(imports.urllib.request, 'urlopen') as request:
|
|
request.return_value.__enter__.return_value.read.return_value = json.dumps(
|
|
{'items': [{'id': 480, 'name': 'Spacewar'}]}).encode()
|
|
self.assertIsNone(imports.identify(dict(self.game, name='Spacewar')))
|
|
|
|
def test_store_requires_unique_exact_title(self):
|
|
cases = [([{'id': 42, 'name': 'Example Deluxe'}], None),
|
|
([{'id': 42, 'name': 'Example'}, {'id': 43, 'name': 'Example'}], None),
|
|
([{'id': 42, 'name': 'EXAMPLE'}], 42)]
|
|
for items, expected in cases:
|
|
with patch.object(imports.urllib.request, 'urlopen') as request:
|
|
request.return_value.__enter__.return_value.read.return_value = json.dumps({'items': items}).encode()
|
|
self.assertEqual(expected, imports.identify(self.game))
|
|
|
|
def test_store_country_and_edition_suffixes(self):
|
|
for folder, store in [('FINAL FANTASY VII REMAKE', 'FINAL FANTASY VII REMAKE INTERGRADE'),
|
|
('FINAL FANTASY XV', 'FINAL FANTASY XV WINDOWS EDITION')]:
|
|
with patch.object(imports.urllib.request, 'urlopen') as request:
|
|
request.return_value.__enter__.return_value.read.return_value = json.dumps({'items': [{'id': 42, 'name': store}]}).encode()
|
|
self.assertEqual(42, imports.identify(dict(self.game, name=folder)))
|
|
self.assertIn('cc=US&', request.call_args.args[0])
|
|
|
|
def test_edition_matching_rejects_ambiguous_titles_and_soundtracks(self):
|
|
with patch.object(imports.urllib.request, 'urlopen') as request:
|
|
request.return_value.__enter__.return_value.read.return_value = json.dumps({'items': [
|
|
{'id': 42, 'name': 'Example WINDOWS EDITION'},
|
|
{'id': 43, 'name': 'Example INTERGRADE'},
|
|
{'id': 44, 'name': 'Example Soundtrack'},
|
|
]}).encode()
|
|
self.assertIsNone(imports.identify(self.game))
|
|
|
|
def test_corrupt_shortcuts_are_preserved(self):
|
|
path = self.acct / 'config/shortcuts.vdf'
|
|
path.parent.mkdir(parents=True)
|
|
original = b'corrupt'
|
|
path.write_bytes(original)
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([self.game]))
|
|
run = subprocess.run(['python3', str(BIN / 'droiddeck-steam-shortcuts'), str(self.steam), str(listing)], capture_output=True)
|
|
self.assertEqual(1, run.returncode)
|
|
self.assertEqual(original, path.read_bytes())
|
|
|
|
def test_user_shortcuts_survive_owned_promotion(self):
|
|
self.snapshot()
|
|
path = self.acct / 'config/shortcuts.vdf'
|
|
manual = dict(shortcuts.shortcut(self.game), tags={'0': 'Personal'}, AppName='Manual')
|
|
path.write_bytes(shortcuts.write({'shortcuts': {'0': manual, '1': shortcuts.shortcut(self.game)}}))
|
|
listing = self.root / 'games.json'
|
|
listing.write_text(json.dumps([self.game]))
|
|
subprocess.run(['python3', str(BIN / 'droiddeck-steam-shortcuts'), str(self.steam), str(listing)], check=True, capture_output=True)
|
|
result = shortcuts.parse(path.read_bytes())['shortcuts']
|
|
self.assertEqual([manual], list(result.values()))
|
|
self.assertEqual({str(self.game['appid']): 42}, json.loads((path.parent / '.droiddeck-routes.json').read_text()))
|
|
|
|
|
|
class LibraryRenameTest(unittest.TestCase):
|
|
def setUp(self):
|
|
temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temp.cleanup)
|
|
self.steam = Path(temp.name)
|
|
self.vdf = self.steam / 'steamapps/libraryfolders.vdf'
|
|
self.vdf.parent.mkdir()
|
|
|
|
def entry(self, index, path):
|
|
return '\t"%d"\n\t{\n\t\t"path"\t\t"%s"\n\t\t"contentid"\t\t"%d"\n\t}\n' % (index, path, 7 + index)
|
|
|
|
def test_old_mount_keeps_its_entry_under_the_new_path(self):
|
|
self.vdf.write_text('"libraryfolders"\n{\n' + self.entry(0, '/root') + self.entry(1, '/mnt/bannerlator-sd') + '}\n')
|
|
with patch('builtins.print'):
|
|
library.rename(str(self.steam), '/mnt/bannerlator-sd', '/mnt/droiddeck-sd')
|
|
text = self.vdf.read_text()
|
|
self.assertIn('"path"\t\t"/mnt/droiddeck-sd"\n\t\t"contentid"\t\t"8"', text)
|
|
self.assertNotIn('bannerlator-sd', text)
|
|
|
|
def test_old_mount_is_dropped_when_new_path_is_registered(self):
|
|
self.vdf.write_text('"libraryfolders"\n{\n' + self.entry(0, '/mnt/droiddeck-sd') + self.entry(1, '/mnt/bannerlator-sd') + '}\n')
|
|
with patch('builtins.print'):
|
|
library.rename(str(self.steam), '/mnt/bannerlator-sd', '/mnt/droiddeck-sd')
|
|
text = self.vdf.read_text()
|
|
self.assertEqual(1, text.count('"path"'))
|
|
self.assertIn('/mnt/droiddeck-sd', text)
|
|
|
|
|
|
class SteamProbeTest(unittest.TestCase):
|
|
def test_probe_reads_account_and_licenses_and_releases_handles(self):
|
|
import ctypes as C
|
|
from unittest.mock import Mock
|
|
objects, callbacks, calls = [], [], []
|
|
|
|
def interface(slots):
|
|
table = (C.c_void_p * 24)()
|
|
for slot, result, args, callback in slots:
|
|
fn = C.CFUNCTYPE(result, C.c_void_p, *args)(callback)
|
|
callbacks.append(fn)
|
|
table[slot] = C.cast(fn, C.c_void_p).value
|
|
obj = C.pointer(C.cast(table, C.POINTER(C.c_void_p)))
|
|
objects.extend([table, obj])
|
|
return C.cast(obj, C.c_void_p).value
|
|
|
|
user = interface([
|
|
(1, C.c_bool, [], lambda _: True),
|
|
(2, C.c_uint64, [], lambda _: imports.BASE_ID + 123),
|
|
])
|
|
apps = interface([(6, C.c_bool, [C.c_uint32], lambda _, appid: appid == 42)])
|
|
client = interface([
|
|
(0, C.c_int, [], lambda _: 7),
|
|
(1, C.c_bool, [C.c_int], lambda _, pipe: calls.append(('pipe', pipe)) or True),
|
|
(2, C.c_int, [C.c_int], lambda _, pipe: 9),
|
|
(4, None, [C.c_int, C.c_int], lambda _, pipe, handle: calls.append(('user', pipe, handle))),
|
|
(5, C.c_void_p, [C.c_int, C.c_int, C.c_char_p], lambda _, handle, pipe, version: user if version == b'SteamUser021' else 0),
|
|
(15, C.c_void_p, [C.c_int, C.c_int, C.c_char_p], lambda _, handle, pipe, version: apps if version == b'STEAMAPPS_INTERFACE_VERSION008' else 0),
|
|
])
|
|
library = Mock()
|
|
library.CreateInterface.return_value = client
|
|
with patch.object(imports.C, 'CDLL', return_value=library):
|
|
result = imports.probe('/Steam', [42, 43])
|
|
self.assertEqual('123', result['account'])
|
|
self.assertEqual({'42': True, '43': False}, result['owned'])
|
|
self.assertEqual([('user', 7, 9), ('pipe', 7)], calls)
|
|
library.CreateInterface.assert_called_once_with(b'SteamClient020', None)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|