mirror of
https://github.com/valentderah/totk-pc-recomp.git
synced 2026-11-11 01:41:02 +00:00
748 lines
35 KiB
Python
748 lines
35 KiB
Python
"""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 <dir>/<id>.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 "<mod folder>/<cheat or file>": 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', '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()
|