mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-13 03:51:10 +00:00
Kurt Himebauch
a1dfb96708
* fix(steam): share the Steam app's prefix with shortcuts of the same install * fix(steam): prefer the matching install's save prefix * fix(steam): tolerate an unreadable library when linking a shortcut prefix A library folder that cannot be listed made the shortcut writer exit, so no added game was registered. * fix(steam): repair a dangling shortcut prefix link and keep Steam library installs to their own prefix * fix(steam): only change shortcut prefix links the importer created * fix(steam): record a prefix link only when the importer created it
855 lines
46 KiB
Python
855 lines
46 KiB
Python
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
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 steam_install(self, appid=42):
|
|
library = self.root / 'Library'
|
|
folder = library / 'steamapps/common/Example'
|
|
folder.mkdir(parents=True)
|
|
(folder / 'Example.exe').write_bytes(b'game')
|
|
(library / f'steamapps/appmanifest_{appid}.acf').write_text(f'"AppState"\n{{\n"appid" "{appid}"\n"installdir" "Example"\n}}\n')
|
|
saves = library / f'steamapps/compatdata/{appid}/pfx/drive_c/users/steamuser'
|
|
saves.mkdir(parents=True)
|
|
(saves / 'save.dat').write_bytes(b'cloud')
|
|
return dict(self.game, folder=str(folder), exe=str(folder / 'Example.exe'), dir=str(folder)), library
|
|
|
|
def shortcut_prefix(self):
|
|
return self.steam / 'steamapps/compatdata' / str(self.game['appid'] & 0xFFFFFFFF)
|
|
|
|
def link_shortcut_prefix(self, target, owned=True):
|
|
(self.steam / 'steamapps/compatdata').mkdir(parents=True, exist_ok=True)
|
|
self.shortcut_prefix().symlink_to(target)
|
|
if owned:
|
|
imports.save_json(self.steam / 'steamapps' / imports.PREFIX_LINKS,
|
|
{str(self.shortcut_prefix()): str(target)})
|
|
|
|
def test_shortcut_of_a_steam_install_shares_the_steam_apps_prefix(self):
|
|
game, library = self.steam_install()
|
|
(self.steam / 'steamapps').mkdir(parents=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
games, routes = imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(([game], {}), (games, routes))
|
|
self.assertEqual(library / 'steamapps/compatdata/42', self.shortcut_prefix().resolve())
|
|
self.assertEqual(b'cloud', (self.shortcut_prefix() / 'pfx/drive_c/users/steamuser/save.dat').read_bytes())
|
|
|
|
def test_matching_install_prefix_precedes_an_older_prefix_in_another_library(self):
|
|
game, library = self.steam_install()
|
|
(self.steam / 'steamapps').mkdir(parents=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
older = self.steam / 'steamapps/compatdata/42/pfx/drive_c/users/steamuser'
|
|
older.mkdir(parents=True)
|
|
(older / 'save.dat').write_bytes(b'old')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(library / 'steamapps/compatdata/42', self.shortcut_prefix().resolve())
|
|
self.assertEqual(b'cloud', (self.shortcut_prefix() / 'pfx/drive_c/users/steamuser/save.dat').read_bytes())
|
|
self.assertEqual(b'old', (older / 'save.dat').read_bytes())
|
|
|
|
def test_an_existing_shortcut_prefix_is_never_replaced(self):
|
|
game, library = self.steam_install()
|
|
(self.steam / 'steamapps').mkdir(parents=True, exist_ok=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
local = self.shortcut_prefix() / 'pfx/drive_c'
|
|
local.mkdir(parents=True)
|
|
(local / 'local.dat').write_bytes(b'local')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(self.shortcut_prefix().is_symlink())
|
|
self.assertEqual(b'local', (local / 'local.dat').read_bytes())
|
|
self.assertEqual(b'cloud', (library / 'steamapps/compatdata/42/pfx/drive_c/users/steamuser/save.dat').read_bytes())
|
|
|
|
def test_shortcuts_without_a_manifest_identity_keep_their_own_prefix(self):
|
|
game, library = self.steam_install()
|
|
(self.steam / 'steamapps').mkdir(parents=True, exist_ok=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
imports.route(self.steam, self.acct, [self.game, dict(game, folder=str(self.folder.parent / 'Other'))])
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
|
|
def test_a_manifest_without_a_steam_prefix_links_nothing(self):
|
|
game, library = self.steam_install()
|
|
shutil.rmtree(library / 'steamapps/compatdata/42')
|
|
(self.steam / 'steamapps').mkdir(parents=True, exist_ok=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
|
|
def test_an_older_prefix_in_another_library_is_not_linked_for_a_steam_install(self):
|
|
game, library = self.steam_install()
|
|
shutil.rmtree(library / 'steamapps/compatdata/42')
|
|
(self.steam / 'steamapps').mkdir(parents=True, exist_ok=True)
|
|
(self.steam / 'steamapps/libraryfolders.vdf').write_text(f'"libraryfolders"\n{{\n"1" {{ "path" "{library}" }}\n}}\n')
|
|
(self.steam / 'steamapps/compatdata/42/pfx/drive_c').mkdir(parents=True)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
|
|
def test_a_dangling_shortcut_prefix_link_is_replaced(self):
|
|
game, library = self.steam_install()
|
|
self.link_shortcut_prefix(self.root / 'moved')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(library / 'steamapps/compatdata/42', self.shortcut_prefix().resolve())
|
|
|
|
def test_a_dangling_shortcut_prefix_link_is_removed_without_a_replacement(self):
|
|
game, library = self.steam_install()
|
|
shutil.rmtree(library / 'steamapps/compatdata/42')
|
|
self.link_shortcut_prefix(self.root / 'moved')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
|
|
def test_a_link_the_user_made_is_never_changed(self):
|
|
game, library = self.steam_install()
|
|
for target in (self.root / 'moved', library / 'steamapps/compatdata/42', self.folder):
|
|
if os.path.lexists(self.shortcut_prefix()):
|
|
self.shortcut_prefix().unlink()
|
|
self.link_shortcut_prefix(target, owned=False)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(str(target), os.readlink(self.shortcut_prefix()))
|
|
|
|
def test_a_user_link_to_the_matching_prefix_is_never_recorded_or_removed(self):
|
|
game, library = self.steam_install()
|
|
target = library / 'steamapps/compatdata/42'
|
|
self.link_shortcut_prefix(target, owned=False)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(str(target), os.readlink(self.shortcut_prefix()))
|
|
self.assertNotIn(str(self.shortcut_prefix()), imports.read_json(self.steam / 'steamapps' / imports.PREFIX_LINKS, {}))
|
|
shutil.rmtree(target)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertTrue(os.path.islink(self.shortcut_prefix()))
|
|
self.assertEqual(str(target), os.readlink(self.shortcut_prefix()))
|
|
|
|
def test_our_link_survives_a_library_that_is_temporarily_missing(self):
|
|
game, library = self.steam_install()
|
|
imports.route(self.steam, self.acct, [game])
|
|
hidden = self.root / 'Hidden'
|
|
library.rename(hidden)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertTrue(os.path.islink(self.shortcut_prefix()))
|
|
hidden.rename(library)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertEqual(library / 'steamapps/compatdata/42', self.shortcut_prefix().resolve())
|
|
self.assertEqual(b'cloud', (self.shortcut_prefix() / 'pfx/drive_c/users/steamuser/save.dat').read_bytes())
|
|
|
|
def test_our_link_is_dropped_once_its_prefix_is_deleted(self):
|
|
game, library = self.steam_install()
|
|
imports.route(self.steam, self.acct, [game])
|
|
shutil.rmtree(library / 'steamapps/compatdata/42')
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
self.assertEqual({}, imports.read_json(self.steam / 'steamapps' / imports.PREFIX_LINKS, None))
|
|
|
|
def test_a_replaced_link_record_is_forgotten_when_the_user_makes_a_real_prefix(self):
|
|
game, library = self.steam_install()
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.shortcut_prefix().unlink()
|
|
(self.shortcut_prefix() / 'pfx/drive_c').mkdir(parents=True)
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertFalse(self.shortcut_prefix().is_symlink())
|
|
self.assertEqual({}, imports.read_json(self.steam / 'steamapps' / imports.PREFIX_LINKS, None))
|
|
|
|
def test_a_shortcut_prefix_with_its_own_saves_is_reported_and_kept(self):
|
|
game, library = self.steam_install()
|
|
local = self.shortcut_prefix() / 'pfx/drive_c'
|
|
local.mkdir(parents=True)
|
|
with patch('builtins.print') as printed:
|
|
imports.route(self.steam, self.acct, [game])
|
|
self.assertTrue(local.is_dir())
|
|
self.assertIn('keeps its own prefix', printed.call_args_list[-1][0][0])
|
|
|
|
def test_a_winnative_game_shares_the_prefix_the_steam_client_uses(self):
|
|
self.winnative_manifest()
|
|
(self.steam / 'steamapps/libraryfolders.vdf').parent.mkdir(parents=True, exist_ok=True)
|
|
saves = self.steam / 'steamapps/compatdata/42/pfx/drive_c/users/steamuser'
|
|
saves.mkdir(parents=True)
|
|
(saves / 'save.dat').write_bytes(b'cloud')
|
|
imports.share_prefix(self.steam, self.game)
|
|
self.assertEqual(self.steam / 'steamapps/compatdata/42', self.shortcut_prefix().resolve())
|
|
|
|
def test_an_unreadable_library_does_not_stop_the_routing(self):
|
|
with patch.object(Path, 'glob', side_effect=PermissionError):
|
|
games, routes = imports.route(self.steam, self.acct, [self.game])
|
|
self.assertEqual([self.game], games)
|
|
self.assertFalse(os.path.lexists(self.shortcut_prefix()))
|
|
|
|
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()
|