"""installer/ bookkeeping with fake steps: fresh install, resume after a failure or cancel, transient steps, update with a newer release, totk.ini and the event stream. No game files, no tools; stdlib only. Run: python tests/installer_test.py """ import http.client import io import json import re import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from installer import ini, runner, state, steps # noqa: E402 from installer.events import Reporter # noqa: E402 from installer.layout import Layout # noqa: E402 class Fake(runner.Step): """A step whose output is /.out holding its key; dir is the temp folder for a transient step.""" def __init__(self, id, transient=False, always=False, previous=None): self.id, self.transient, self.always, self.previous = id, transient, always, previous def out(self, ctx): return (ctx.temp if self.transient else ctx.dest) / f'{self.id}.out' def key(self, ctx): upstream = ctx.keys[self.previous] if self.previous else '' return f'{self.id}:{ctx.options.inputs.get(self.id, "v1")}:{upstream}' def current(self, ctx, record): p = self.out(ctx) return p.is_file() and p.read_text() == ctx.keys[self.id] def verify(self, ctx, record): ctx.options.verified.append(self.id) return ctx.options.verify_ok.get(self.id, True) def run(self, ctx): o = ctx.options o.ran.append(self.id) if self.id in ctx.forced: o.forced.append(self.id) if self.id == 'check': ctx.verify_kept(self.id) failure = o.fail.pop(self.id, None) if failure == 'error': raise runner.InstallError('fake.failed', step=self.id) if failure == 'cancel': raise KeyboardInterrupt if self.id == 'cleanup': for p in ctx.temp.glob('*.out'): p.unlink() return {} p = self.out(ctx) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(ctx.keys[self.id]) return {'note': self.id} def fake_steps(): return [Fake('check', always=True), Fake('extract'), Fake('lift', transient=True, previous='extract'), Fake('compile', previous='lift'), Fake('prepare', previous='compile'), Fake('cleanup', always=True)] class Bed: """A dest and temp folder plus one run of the fake steps per call.""" def __init__(self, base): self.dest, self.temp = base / 'dest', base / 'temp' self.temp.mkdir() def run(self, release='1.0', inputs=None, fail=None, verify_ok=None, verify=False): layout = Layout(root=Path('.'), dev=True, release=release, python=Path('python'), tools=Path('tools'), recomp=Path('recomp'), llvm=Path('llvm'), deps=Path('deps'), runtime=Path('rt'), runtime_lib=Path('lib'), ninja=Path('ninja'), hactool=Path('hactool'), xlate=Path('xlate')) options = SimpleNamespace(inputs=inputs or {}, fail=dict(fail or {}), verify_ok=verify_ok or {}, ran=[], verified=[], forced=[]) out = io.StringIO() ctx = runner.Context(layout, self.dest, self.temp, Reporter(out=out), options, state=state.load(self.dest), verify=verify) mode = ctx.mode result = runner.run_steps(fake_steps(), ctx) events = [json.loads(line) for line in out.getvalue().splitlines()] return SimpleNamespace(result=result, ran=options.ran, verified=options.verified, forced=options.forced, events=events, mode=mode, state=state.load(self.dest)) @staticmethod def skipped(r): return {e['step']: e['reason'] for e in r.events if e['type'] == 'step' and e['state'] == 'skipped'} ALL = ['check', 'extract', 'lift', 'compile', 'prepare', 'cleanup'] class RunnerTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.bed = Bed(Path(self.tmp.name)) def tearDown(self): self.tmp.cleanup() def test_fresh_install_runs_everything_and_records_it(self): r = self.bed.run() self.assertEqual((r.result, r.mode, r.ran), ('ok', 'install', ALL)) self.assertEqual(set(r.state['steps']), {'extract', 'lift', 'compile', 'prepare'}) self.assertEqual(r.state['steps']['compile']['note'], 'compile') self.assertEqual(r.state['release'], '1.0') started = [e['step'] for e in r.events if e['type'] == 'step' and e['state'] == 'started'] done = [e['step'] for e in r.events if e['type'] == 'step' and e['state'] == 'done'] self.assertEqual((started, done), (ALL, ALL)) def test_rerun_skips_finished_steps_and_unneeded_transient(self): self.bed.run() r = self.bed.run() self.assertEqual((r.result, r.mode, r.ran), ('ok', 'resume', ['check', 'cleanup'])) self.assertEqual(self.bed.skipped(r), {'extract': 'current', 'lift': 'not_needed', 'compile': 'current', 'prepare': 'current'}) self.assertEqual(r.verified, []) # same release: no full checks def test_resume_after_failure(self): r = self.bed.run(fail={'compile': 'error'}) self.assertEqual((r.result, r.ran), ('failed', ['check', 'extract', 'lift', 'compile'])) error = [e for e in r.events if e['type'] == 'error'][0] self.assertEqual((error['id'], error['step'], error['params']), ('fake.failed', 'compile', {'step': 'compile'})) self.assertEqual(set(r.state['steps']), {'extract', 'lift'}) r = self.bed.run() # lift's output is still in the temp folder (no cleanup ran), so only compile and later run. self.assertEqual((r.result, r.mode, r.ran), ('ok', 'resume', ['check', 'compile', 'prepare', 'cleanup'])) def test_resume_after_cancel_with_transient_output_gone(self): r = self.bed.run(fail={'prepare': 'cancel'}) self.assertEqual(r.result, 'cancelled') self.assertNotIn('prepare', r.state['steps']) for p in self.bed.temp.glob('*.out'): p.unlink() r = self.bed.run() self.assertEqual(r.ran, ['check', 'prepare', 'cleanup']) self.assertEqual(self.bed.skipped(r)['lift'], 'not_needed') def test_failed_check_writes_nothing(self): r = self.bed.run(fail={'check': 'error'}) self.assertEqual(r.result, 'failed') self.assertFalse(self.bed.dest.exists()) def test_removed_output_runs_again(self): self.bed.run() (self.bed.dest / 'compile.out').unlink() r = self.bed.run() self.assertEqual(r.ran, ['check', 'lift', 'compile', 'cleanup']) def test_update_with_new_lifter_keeps_extraction(self): self.bed.run(release='1.0') r = self.bed.run(release='1.1', inputs={'lift': 'v2'}) self.assertEqual((r.result, r.mode), ('ok', 'update')) self.assertEqual(r.ran, ['check', 'lift', 'compile', 'prepare', 'cleanup']) self.assertEqual(r.verified, ['extract']) # kept from another release: full check self.assertEqual(r.state['steps']['extract']['release'], '1.0') self.assertEqual(r.state['steps']['compile']['release'], '1.1') self.assertEqual(r.state['release'], '1.1') def test_update_that_changes_nothing_skips_the_rebuild(self): self.bed.run(release='1.0') r = self.bed.run(release='1.1') self.assertEqual(r.ran, ['check', 'cleanup']) self.assertEqual(r.verified, ['extract', 'compile', 'prepare']) def test_update_with_changed_kept_files_extracts_again(self): self.bed.run(release='1.0') r = self.bed.run(release='1.1', verify_ok={'extract': False}) self.assertEqual(r.ran, ['check', 'extract', 'cleanup']) self.assertEqual(r.forced, ['extract']) # the step knows not to trust what it left behind warnings = [e for e in r.events if e['type'] == 'warning'] self.assertEqual([(w['id'], w['params']) for w in warnings], [('state.output_changed', {'step': 'extract'})]) def test_verify_option_checks_same_release(self): self.bed.run() r = self.bed.run(verify=True) self.assertEqual(r.verified, ['extract', 'compile', 'prepare']) def test_state_errors(self): self.bed.dest.mkdir() (self.bed.dest / state.NAME).write_text('{not json') with self.assertRaises(state.StateError) as e: state.load(self.bed.dest) self.assertEqual(e.exception.id, 'state.unreadable') (self.bed.dest / state.NAME).write_text(json.dumps({'format': 99, 'steps': {}})) with self.assertRaises(state.StateError) as e: state.load(self.bed.dest) self.assertEqual(e.exception.id, 'state.newer_format') class ReleaseFilesTest(unittest.TestCase): def test_release_ships_every_tool_the_steps_run(self): sys.path.insert(0, str(ROOT / 'tools')) import make_release text = (ROOT / 'installer' / 'steps.py').read_text(encoding='utf-8') used = set(re.findall(r"layout\.tools / '(\w+\.py)'", text)) for names in re.findall(r"layout\.tools / n for n in \((.*?)\)", text, re.S): used |= set(re.findall(r"'(\w+\.py)'", names)) self.assertIn('cheats.py', used) self.assertEqual(used - set(make_release.TOOLS_FILES), set()) class RenamedVersionTest(unittest.TestCase): """An install the setup before 0.3.1 recorded as 1.1.1 is the 1.1.0 update (#332).""" def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) root = Path(__file__).resolve().parents[1] layout = Layout(root=root, dev=True, release='9.9', python=Path('python'), tools=root / 'tools', recomp=root / 'recomp', llvm=Path('llvm'), deps=Path('deps'), runtime=Path('rt'), runtime_lib=Path('lib'), ninja=Path('ninja'), hactool=Path('hactool'), xlate=Path('xlate')) state = {'format': 1, 'release': '0.2.8', 'version': '1.1.1', 'inputs': {}, 'steps': {'extract': {'key': 'recorded by 0.2.8', 'release': '0.2.8'}}} out = io.StringIO() self.ctx = runner.Context(layout, Path(tmp.name), Path(tmp.name), Reporter(out=out), SimpleNamespace(), state=state) def test_old_name_is_read_as_1_1_0(self): steps.migrate_state(self.ctx) self.assertEqual(steps.game_version(self.ctx), '1.1.0') self.assertEqual(steps.version_tag(steps.game_version(self.ctx)), 'v110') self.assertEqual(steps.expected_update(self.ctx)['display_version'], '1.1.0') self.assertTrue(self.ctx.layout.patches(steps.game_version(self.ctx)).is_file()) def test_extraction_is_kept_under_the_new_name(self): steps.migrate_state(self.ctx) self.assertEqual(self.ctx.state['steps']['extract']['key'], steps.Extract().key(self.ctx)) self.assertEqual(self.ctx.state['steps']['extract']['release'], '0.2.8') # full check of kept output runs def test_other_versions_are_untouched(self): self.ctx.state['version'] = '1.4.3' steps.migrate_state(self.ctx) self.assertEqual(self.ctx.state['version'], '1.4.3') self.assertEqual(self.ctx.state['steps']['extract']['key'], 'recorded by 0.2.8') class LiftKeyTest(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.dest = Path(tmp.name) layout = SimpleNamespace(recomp=self.dest / 'no-recomp', tools=self.dest / 'no-tools', patches=lambda version: self.dest / 'no-patches') self.ctx = SimpleNamespace(dest=self.dest, layout=layout, keys={'extract': 'x'}, state={'version': '1.4.3'}) def key(self): return steps.Lift().key(self.ctx) def add_cheats(self, name, text, mod='Cheats', where='cheats'): folder = self.dest / 'mods' / mod / where folder.mkdir(parents=True, exist_ok=True) (folder / name).write_text(text) def test_extracting_main_keeps_the_key(self): before = self.key() (self.dest / 'exefs').mkdir() (self.dest / 'exefs' / 'main').write_bytes(bytes(0x48)) self.assertEqual(self.key(), before) def test_cheat_files_change_the_key(self): before = self.key() self.add_cheats('277178B7DBA1B6D4.txt', '[a]') added = self.key() self.assertNotEqual(added, before) self.add_cheats('277178B7DBA1B6D4.txt', '[b]') edited = self.key() self.assertNotEqual(edited, added) self.add_cheats('277178B7DBA1B6D4.txt', '[c]', mod='Other', where='atmosphere/contents/0100F2C0115B6000/cheats') self.assertNotEqual(self.key(), edited) def test_exefs_patch_files_change_the_key(self): before = self.key() self.add_cheats('dynamic-fps.pchtxt', '@nsobid-277178B7DBA1B6D4', mod='Fps', where='exefs') added = self.key() self.assertNotEqual(added, before) self.add_cheats('277178B7DBA1B6D4.ips', 'PATCHEOF', mod='Ips', where='atmosphere/exefs_patches/x') both = self.key() self.assertNotEqual(both, added) self.add_cheats('main.npdm', 'not a patch', mod='Fps', where='exefs') self.assertEqual(self.key(), both) def test_renaming_a_mod_folder_changes_the_key(self): # tools/cheats.py and tools/exefs_patches.py name a patch set after "/": a renamed # folder needs a lift even though its files are the same self.add_cheats('277178B7DBA1B6D4.txt', '[a]', mod='MyNXCheats') self.add_cheats('dynamic-fps.pchtxt', '@nsobid-277178B7DBA1B6D4', mod='Fps', where='exefs') before = self.key() mods = self.dest / 'mods' (mods / 'MyNXCheats').rename(mods / 'MyNXCheats 1.4.3') cheats_renamed = self.key() self.assertNotEqual(cheats_renamed, before) (mods / 'Fps').rename(mods / 'Fps 2') self.assertNotEqual(self.key(), cheats_renamed) def test_status_and_other_files_keep_the_key(self): self.add_cheats('277178B7DBA1B6D4.txt', '[a]') before = self.key() (self.dest / 'build-cache').mkdir() (self.dest / 'build-cache' / 'cheats.status').write_text('written by lift') self.add_cheats('readme.md', 'not a cheat file') (self.dest / 'mods' / 'Cheats' / 'romfs').mkdir() (self.dest / 'mods' / 'Cheats' / 'romfs' / 'a.txt').write_text('a RomFS file') self.assertEqual(self.key(), before) class LiftRunTest(unittest.TestCase): """Lift.run with fake tools: the ExeFS patch stage writes the status and data files, the lift lists the bytes it read, and the data writes are checked against that list after it; a failed lift or check drops both files.""" def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) base = Path(tmp.name) self.dest, temp = base / 'install', base / 'temp' self.dest.mkdir() temp.mkdir() layout = SimpleNamespace(recomp=ROOT / 'recomp', tools=ROOT / 'tools', patches=lambda version: base / 'patches-v143.txt') self.ctx = SimpleNamespace(dest=self.dest, temp=temp, layout=layout, keys={'lift': 'k'}, state={'version': '1.4.3'}, reporter=Reporter(out=io.StringIO())) self.addCleanup(self.ctx.reporter.close) mod = self.dest / 'mods' / 'SQ' / 'exefs' mod.mkdir(parents=True) (mod / '277178B7DBA1B6D4.ips').write_bytes(b'IPS32EEOF') self.calls, self.fail = [], set() saved = steps.run_tool self.addCleanup(lambda: setattr(steps, 'run_tool', saved)) steps.run_tool = self.run_tool def run_tool(self, ctx, step, script, args, parse=None, cwd=None): args = [str(a) for a in args] name = Path(script).name + (' --folded' if '--folded' in args else '') self.calls.append((name, args)) log = ctx.temp / 'tool.log' log.write_text('') if name in self.fail: return 1, log if name == 'exefs_patches.py': for flag in ('--status', '--data', '--out'): p = Path(args[args.index(flag) + 1]) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(flag) elif name == 'gen.py': out = Path(args[args.index('--out') + 1]) / 'main' out.mkdir(parents=True, exist_ok=True) (out / 'folded.txt').write_text('# test\n') return 0, log def outputs(self): return [p.is_file() for p in (steps.exefs_status(self.ctx), steps.exefs_data(self.ctx))] def test_data_file_and_the_check_after_the_lift(self): steps.Lift().run(self.ctx) self.assertEqual([c[0] for c in self.calls], ['exefs_patches.py', 'gen.py', 'exefs_patches.py --folded']) build = self.calls[0][1] self.assertEqual(build[build.index('--data') + 1], str(steps.exefs_data(self.ctx))) check = self.calls[2][1] self.assertEqual(check[check.index('--folded') + 1], str(self.ctx.temp / 'gen' / 'main' / 'folded.txt')) self.assertEqual(check[check.index('--status') + 1], str(steps.exefs_status(self.ctx))) self.assertEqual(self.outputs(), [True, True]) def test_failed_lift_drops_the_data_file(self): self.fail = {'gen.py'} with self.assertRaises(steps.InstallError): steps.Lift().run(self.ctx) self.assertEqual(self.outputs(), [False, False]) def test_failed_check_drops_both_files(self): self.fail = {'exefs_patches.py --folded'} steps.Lift().run(self.ctx) self.assertEqual(self.outputs(), [False, False]) self.assertIn('warn.exefs_skipped', self.ctx.reporter.out.getvalue()) class IniTest(unittest.TestCase): EXAMPLE = '[display]\n# vk_present = mailbox\n\n[language]\n# locale = en-US\n# lang = en-US\n# x = 1\n' def test_example_line_is_activated(self): out = ini.set_language(self.EXAMPLE, 'ru') self.assertIn('\nlang = ru\n', out) self.assertNotIn('# lang', out) self.assertIn('# locale = en-US', out) def test_active_line_is_rewritten_and_the_rest_kept(self): text = '[display]\nvk_present = fifo\n[language]\n LANG = ja \nfps60 = 0\n' out = ini.set_language(text, 'de') self.assertEqual(out, '[display]\nvk_present = fifo\n[language]\n lang = de\nfps60 = 0\n') self.assertEqual(ini.get_language(out), 'de') def test_missing_section_is_added(self): self.assertEqual(ini.set_language('fps60 = 0\n', 'fr'), 'fps60 = 0\n\n[language]\nlang = fr\n') self.assertEqual(ini.set_language('[language]\nlocale = de-DE\n', 'fr'), '[language]\nlang = fr\nlocale = de-DE\n') def test_old_chinese_codes_are_written_as_the_canonical_ones(self): self.assertEqual(ini.canonical_language('zh-CN'), 'zh-Hans') self.assertEqual(ini.canonical_language('zh-TW'), 'zh-Hant') self.assertEqual(ini.canonical_language('ja'), 'ja') self.assertNotIn('zh-CN', ini.LANGUAGES) self.assertNotIn('zh-TW', ini.LANGUAGES) out = ini.set_language(self.EXAMPLE, 'zh-CN') self.assertEqual(ini.get_language(out), 'zh-Hans') self.assertEqual(ini.get_language(ini.set_language(out, 'zh-TW')), 'zh-Hant') def test_write_keeps_user_lines(self): with tempfile.TemporaryDirectory() as d: d = Path(d) example = d / 'totk.ini.example' example.write_text(self.EXAMPLE) path = ini.write(d / '', 'en-GB', example) self.assertEqual(ini.get_language(path.read_text()), 'en-GB') path.write_text(path.read_text().replace('# vk_present = mailbox', 'vk_present = fifo')) ini.write(d, 'ja', example) text = path.read_text() self.assertIn('vk_present = fifo', text) self.assertEqual(ini.get_language(text), 'ja') class CleanupTest(unittest.TestCase): """The real Cleanup step on a finished install whose romfs/ is intact, emptied or gone (#107).""" def setUp(self): self.steps = steps self.tmp = tempfile.TemporaryDirectory() base = Path(self.tmp.name) self.dest, temp = base / 'dest', base / 'temp' (self.dest / 'romfs').mkdir(parents=True) temp.mkdir() image = b'romfs image' (self.dest / 'romfs' / 'romfs.bin').write_bytes(image) digest = steps.extract.sha256(self.dest / 'romfs' / 'romfs.bin') (self.dest / 'romfs' / 'index.json').write_text(json.dumps({'romfs_size': len(image), 'romfs_sha256': digest})) self.expected = {'romfs': {'size': len(image), 'sha256': digest}} self.ctx = SimpleNamespace(reporter=Reporter(out=io.StringIO(), log_path=base / 'x.log'), dest=self.dest, temp=temp, lang='en-US', state={}, layout=SimpleNamespace(ini_example=ROOT / 'totk.ini.example', dev=False, python=base / 'python' / 'python.exe', root=base / 'release')) self.saved = steps.expected_update steps.expected_update = lambda ctx: self.expected def tearDown(self): self.steps.expected_update = self.saved self.ctx.reporter.close() self.tmp.cleanup() def run_cleanup(self): try: self.steps.Cleanup().run(self.ctx) return 'ok' except self.steps.InstallError as e: return e.id def test_intact_romfs_passes(self): self.assertEqual(self.run_cleanup(), 'ok') def test_keeps_build_and_writes_rebuild_command(self): for name in ('gen', 'build', 'check'): (self.ctx.temp / name).mkdir() self.assertEqual(self.run_cleanup(), 'ok') self.assertTrue((self.ctx.temp / 'gen').is_dir() and (self.ctx.temp / 'build').is_dir()) self.assertFalse((self.ctx.temp / 'check').exists()) lines = (self.dest / 'build-cache' / 'rebuild.txt').read_text(encoding='utf-8').splitlines() self.assertEqual(lines[1:], ['-c', steps.REBUILD_BOOT, str(self.ctx.layout.root), '--root', str(self.ctx.layout.root), '--rebuild', str(self.dest)]) def test_emptied_romfs_fails(self): for p in (self.dest / 'romfs').iterdir(): p.unlink() self.assertEqual(self.run_cleanup(), 'cleanup.romfs_unreadable') def test_missing_romfs_fails(self): self.steps.remove(self.dest / 'romfs') self.assertEqual(self.run_cleanup(), 'cleanup.romfs_unreadable') def test_romfs_of_another_size_fails(self): (self.dest / 'romfs' / 'romfs.bin').write_bytes(b'short') self.assertEqual(self.run_cleanup(), 'cleanup.romfs_unreadable') class FreeSpaceTest(unittest.TestCase): def test_missing_folders_use_the_nearest_existing_parent(self): with tempfile.TemporaryDirectory() as d: parent = Path(d) path = parent / 'missing' / 'install' with mock.patch.object(steps.shutil, 'disk_usage', return_value=SimpleNamespace(free=1234)) as usage: self.assertEqual(steps.free_bytes(path), (1234, parent.stat().st_dev)) usage.assert_called_once_with(parent) self.assertFalse(path.parent.exists()) def test_missing_root_reports_the_filesystem_error(self): path = Path(ROOT.anchor) / 'missing' / 'install' root = Path(path.anchor) error = FileNotFoundError(2, 'No such file or directory', str(root)) # Bound the failed existence checks so the old parent walk cannot hang the test suite. with mock.patch.object(Path, 'exists', side_effect=[False] * (len(path.parents) + 1)): with mock.patch.object(steps.shutil, 'disk_usage', side_effect=error) as usage: with self.assertRaises(FileNotFoundError) as raised: steps.free_bytes(path) self.assertIs(raised.exception, error) usage.assert_called_once_with(root) class BaseGameCheckTest(unittest.TestCase): """Issue #261: an update NSP chosen as the base game was reported as "not Tears of the Kingdom".""" BASE, UPDATE = '0100F2C0115B6000', '0100F2C0115B6800' @staticmethod def nca(content_type, title_id): return ({'name': 'x.nca', 'nca_offset': 0, 'size': 0xc00, 'content_type': content_type, 'title_id': title_id}, b'') def test_update_package_names_the_update(self): # An update NSP keeps the base title id on its Program NCA; only its Meta NCA carries the update id. records = [self.nca(0, self.BASE), self.nca(1, self.UPDATE)] with self.assertRaises(ValueError): steps.program_nca(records) self.assertEqual(steps.base_misuse(records), 'check.base_is_update') def test_combined_package_is_named(self): records = [self.nca(0, self.BASE), self.nca(1, self.BASE), self.nca(1, self.UPDATE)] self.assertEqual(steps.base_misuse(records), 'check.base_combined') def test_plain_base_game_passes(self): records = [self.nca(0, self.BASE), self.nca(1, self.BASE), self.nca(3, self.BASE)] self.assertIsNone(steps.base_misuse(records)) self.assertEqual(steps.program_nca(records)[0]['title_id'], self.BASE) class FontsTest(unittest.TestCase): """The fonts step: keyed by language and firmware, a no-op without firmware, writes dest/shared-fonts.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() base = Path(self.tmp.name) self.dest, temp, self.firmware = base / 'dest', base / 'temp', base / 'firmware' temp.mkdir() self.firmware.mkdir() (self.firmware / 'a.nca').write_bytes(b'nca') self.keys = base / 'prod.keys' self.keys.write_text('header_key = 00\n') self.ctx = SimpleNamespace(dest=self.dest, temp=temp, lang='zh-Hans', state={}, layout=SimpleNamespace(tools=ROOT / 'tools', root=ROOT), reporter=Reporter(out=io.StringIO()), options=SimpleNamespace(lang='zh-Hans', firmware=self.firmware, keys=str(self.keys))) self.saved_run_tool = steps.run_tool def tearDown(self): steps.run_tool = self.saved_run_tool self.ctx.reporter.close() self.tmp.cleanup() def fake_tool(self, files=None, code=0): """Stand-in for extract_shared_fonts.py: writes the requested files where --output points.""" def run(ctx, step, script, args): out = Path(args[args.index('--output') + 1]) out.mkdir(parents=True) for name in steps.SHARED_FONT_FILES if files is None else files: (out / name).write_bytes(b'bfttf') (out / 'manifest.json').write_text('{}') log = ctx.temp / 'fonts.log' log.write_text('error line\n') return code, log steps.run_tool = run def test_all_steps_order(self): self.assertEqual([s.id for s in steps.all_steps()], ['check', 'fonts', 'extract', 'lift', 'compile', 'prepare', 'pc_prompts', 'dlss', 'cleanup']) def test_extracts_into_the_install_folder(self): self.fake_tool() self.assertEqual(steps.Fonts().run(self.ctx), {'files': 6}) self.assertTrue(steps.shared_fonts_ready(self.dest)) self.assertEqual(sorted(p.name for p in (self.dest / 'shared-fonts').iterdir()), sorted((*steps.SHARED_FONT_FILES, 'manifest.json'))) def test_noop_without_firmware(self): self.ctx.options.firmware = None self.fake_tool() self.assertEqual(steps.Fonts().run(self.ctx), {}) self.assertFalse((self.dest / 'shared-fonts').exists()) def test_noop_for_other_languages(self): self.ctx.lang, self.ctx.options.lang = 'en-US', 'en-US' self.fake_tool() self.assertEqual(steps.Fonts().run(self.ctx), {}) self.assertFalse((self.dest / 'shared-fonts').exists()) def test_failed_extraction_keeps_the_old_folder(self): self.fake_tool() steps.Fonts().run(self.ctx) self.fake_tool(code=1) with self.assertRaises(steps.InstallError) as e: steps.Fonts().run(self.ctx) self.assertEqual(e.exception.id, 'fonts.failed') self.assertTrue(steps.shared_fonts_ready(self.dest)) def test_incomplete_output_is_an_error(self): self.fake_tool(files=steps.SHARED_FONT_FILES[:1]) with self.assertRaises(steps.InstallError) as e: steps.Fonts().run(self.ctx) self.assertEqual(e.exception.id, 'fonts.incomplete') def test_current_tracks_language_and_output(self): fonts = steps.Fonts() self.assertFalse(fonts.current(self.ctx, {})) self.fake_tool() fonts.run(self.ctx) self.assertTrue(fonts.current(self.ctx, {})) self.ctx.options.lang = 'en-US' self.assertTrue(fonts.current(self.ctx, {})) self.ctx.options.lang = 'zh-Hans' (self.dest / 'shared-fonts' / '0-standard.bfttf').unlink() self.assertFalse(fonts.current(self.ctx, {})) def test_key_changes_with_language_and_firmware(self): fonts = steps.Fonts() zh = fonts.key(self.ctx) self.ctx.options.lang = 'en-US' self.assertNotEqual(zh, fonts.key(self.ctx)) self.ctx.options.lang = 'zh-Hans' (self.firmware / 'b.nca').write_bytes(b'other') self.assertNotEqual(zh, fonts.key(self.ctx)) def test_registered_child_is_used(self): nested = self.firmware / 'registered' nested.mkdir() (nested / 'c.nca').write_bytes(b'nca') self.assertEqual(steps.firmware_dir(self.firmware), nested) class EventsTest(unittest.TestCase): def test_lines_are_json_and_the_log_is_english(self): with tempfile.TemporaryDirectory() as d: out = io.StringIO() rep = Reporter(out=out, log_path=Path(d) / 'x.log') rep.step('check', 'started') rep.progress('extract', 'copy', 0, 10, 'bytes') rep.progress('extract', 'copy', 5, 10, 'bytes') # throttled rep.progress('extract', 'copy', 10, 10, 'bytes') # last one always goes out rep.error('check.disk_space', {'path': 'D:\\x', 'need_gb': 40, 'free_gb': 3}, step='check') rep.close() events = [json.loads(line) for line in out.getvalue().splitlines()] self.assertEqual([e['type'] for e in events], ['step', 'progress', 'progress', 'error']) self.assertEqual([e.get('done') for e in events[1:3]], [0, 10]) self.assertTrue(all('t' in e for e in events)) log = (Path(d) / 'x.log').read_text(encoding='utf-8') self.assertIn('40 GB needed, 3 GB free', log) self.assertIn('[check.disk_space]', log) def test_catalog_has_every_id_the_code_uses(self): catalog = json.loads((ROOT / 'installer' / 'text' / 'en.json').read_text(encoding='utf-8')) used = set() for path in (ROOT / 'installer').glob('*.py'): text = path.read_text(encoding='utf-8') used |= set(re.findall(r"InstallError\('([a-z_.]+)'", text)) used |= set(re.findall(r"warning\('([a-z_.]+)'", text)) used |= {f'phase.{s}.{p}' for s, p in re.findall(r"progress\('(\w+)', '(\w+)'[,)]", text)} self.assertEqual(sorted(i for i in used if i not in catalog), []) @unittest.skipUnless(steps.os.name == 'nt', 'DLSS is Windows-only') class DlssTest(unittest.TestCase): """The DLSS step: off without --dlss, needs an RTX GPU, keeps a player's library, verifies the download.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() base = Path(self.tmp.name) self.dest, temp = base / 'dest', base / 'temp' self.dest.mkdir() temp.mkdir() self.src = base / 'nvngx_dlss.dll' self.src.write_bytes(b'dlss' * 1000) self.saved = steps.DLSS_URL, steps.DLSS_SIZE, steps.DLSS_SHA256 steps.DLSS_URL = self.src.as_uri() # urllib reads file: URLs as it reads https: ones steps.DLSS_SIZE, steps.DLSS_SHA256 = 4000, steps.file_sha256(self.src) rtx = {'devices': [{'name': 'NVIDIA GeForce RTX 4080 SUPER', 'api': [1, 4, 0], 'type': 2}]} self.ctx = SimpleNamespace(dest=self.dest, temp=temp, host=rtx, reporter=Reporter(out=io.StringIO()), options=SimpleNamespace(dlss=True)) def tearDown(self): steps.DLSS_URL, steps.DLSS_SIZE, steps.DLSS_SHA256 = self.saved self.ctx.reporter.close() self.tmp.cleanup() def warnings(self): return [json.loads(line)['id'] for line in self.ctx.reporter.out.getvalue().splitlines() if json.loads(line)['type'] == 'warning'] def test_off_without_the_option(self): self.ctx.options.dlss = False self.assertEqual(steps.Dlss().run(self.ctx), {}) self.assertFalse((self.dest / 'nvngx_dlss.dll').exists()) self.assertNotEqual(steps.Dlss().key(self.ctx), steps.Dlss().key(SimpleNamespace(options=SimpleNamespace(dlss=True)))) def test_downloads_and_checks_the_library(self): rec = steps.Dlss().run(self.ctx) self.assertEqual(rec, {'installed': steps.DLSS_VERSION}) self.assertEqual((self.dest / 'nvngx_dlss.dll').read_bytes(), self.src.read_bytes()) self.assertFalse(list(self.ctx.temp.iterdir())) self.assertTrue(steps.Dlss().current(self.ctx, rec)) (self.dest / 'nvngx_dlss.dll').unlink() self.assertFalse(steps.Dlss().current(self.ctx, rec)) def test_a_wrong_file_is_not_installed_and_tried_again(self): steps.DLSS_SHA256 = '0' * 64 rec = steps.Dlss().run(self.ctx) self.assertEqual(rec, {'failed': True}) self.assertFalse((self.dest / 'nvngx_dlss.dll').exists()) self.assertEqual(self.warnings(), ['warn.dlss_download_failed']) self.assertFalse(steps.Dlss().current(self.ctx, rec)) def test_a_cut_connection_warns(self): class Cut(io.BytesIO): # a response whose connection drops mid-file def read(self, n=-1): raise http.client.IncompleteRead(b'dl', 3998) with mock.patch.object(steps.urllib.request, 'urlopen', return_value=Cut()): self.assertEqual(steps.Dlss().run(self.ctx), {'failed': True}) self.assertFalse((self.dest / 'nvngx_dlss.dll').exists()) self.assertFalse(list(self.ctx.temp.iterdir())) self.assertEqual(self.warnings(), ['warn.dlss_download_failed']) def test_keeps_the_players_library(self): (self.dest / 'nvngx_dlss.dll').write_bytes(b'newer') self.assertEqual(steps.Dlss().run(self.ctx), {'kept': True}) self.assertEqual((self.dest / 'nvngx_dlss.dll').read_bytes(), b'newer') self.assertEqual(self.warnings(), ['warn.dlss_kept']) def test_skipped_without_an_rtx_gpu(self): self.ctx.host = {'devices': [{'name': 'AMD Radeon RX 7800 XT', 'api': [1, 3, 0], 'type': 2}]} self.assertEqual(steps.Dlss().run(self.ctx), {'skipped': 'no_rtx'}) self.assertEqual(self.warnings(), []) if __name__ == '__main__': unittest.main()