mirror of
https://github.com/tangerineturtle1/NX2navi.git
synced 2026-11-15 19:19:58 -04:00
c403547c8b
NX2navi-Setup.exe installs the launcher into %LOCALAPPDATA%\Programs\NX2navi without administrator rights, with shortcuts and an Apps entry. The launcher finds the player's .nup, downloads the build tools itself (the MSYS2 UCRT64 packages toolchain.lock.json pins, checked by sha256 and unpacked without pacman, and zbic from PyPI) and runs tools/make_game.py, whose steps it shows with their progress. Its window follows the game's options menu: colours sampled from it, the menu's filigree, the white glowing cursor, the Triforce in the title band and as the programs' icon, and all text in the game's Cetra Display, decoded from the player's own dump. Mouse, keyboard and an XInput pad drive it. Updates come from GitHub releases (NX2navi-update.json): the new version is downloaded, checked and staged beside the running one, the stub restarts into it and falls back to the previous version when it fails to start, and the game is rebuilt without the dump. make_game.py now keys its exefs and romfs steps on the game files alone, has --rebuild and --events, compiles only the files whose source, headers, flags or compiler changed, and replaces ocarina.exe only once the new one links. tools/make_launcher.py builds the release on Windows or Linux. Tested: tests/launcher_test.py (19 tests), the real toolchain download and install, and under Wine 11 the setup, the stub, the window on the bundled Windows Python, an update from a local feed by clicking Updates, the rollback of a version that fails to start, and the downloaded clang compiling and linking a program against the game's libraries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Merged onto the current port: make_game.py keeps the launcher's step keys, cached link and in-place swap, and its own runtime manifest (windres), AMD's FidelityFX library and the shader preparation, which becomes a step of its own ("shaders") that the launcher shows in its install, rebuild and update jobs. make_launcher.py ships the shader tools and, with --ffx, AMD's library. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
679 lines
30 KiB
Python
679 lines
30 KiB
Python
"""launcher/ and tools/make_game.py's bookkeeping, without game files or Windows.
|
|
|
|
Run: python tests/launcher_test.py (Python 3.14 for the zstd tests; the window test needs tkinter, numpy and a display)
|
|
|
|
- versions, settings
|
|
- toolchain: the package database, dependency resolution and an install from a local fake repository
|
|
- net: downloads that resume, check their sha256 and stop on cancel
|
|
- feed: GitHub releases (stable, preview, private with a token) and a plain manifest URL, served locally
|
|
- updater: staging a release, switching, rollback, pruning, refusing unsafe archives
|
|
- gamefiles: the .bfotf font wrapping and zstd dictionaries, on a font of this system made into a fake RomFS
|
|
- make_game.py: a synthetic .nup through the check, exefs and romfs steps with --events, skipped steps on a second
|
|
run, --rebuild without the game files, and the object cache of the compile step with the system's C compiler
|
|
- jobs: an install driven by a fake make_game.py
|
|
- the window: --selftest draws every screen (skipped without a display)
|
|
"""
|
|
import hashlib
|
|
import http.server
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / 'tools'))
|
|
|
|
from launcher import build, feed, gamefiles, jobs, net, settings, toolchain, updater, versions # noqa: E402
|
|
from launcher.context import Context # noqa: E402
|
|
from launcher.paths import Layout # noqa: E402
|
|
|
|
try:
|
|
from compression import zstd as ZSTD
|
|
except ImportError:
|
|
ZSTD = None
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
class Server:
|
|
"""A local HTTP server for a folder, with extra routes (path -> (status, headers, body)) and Range support."""
|
|
|
|
def __init__(self, folder, routes=None):
|
|
folder, routes = Path(folder), routes if routes is not None else {}
|
|
self.hits = []
|
|
server = self
|
|
|
|
class H(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
server.hits.append((self.path, dict(self.headers)))
|
|
if self.path in routes:
|
|
status, headers, body = routes[self.path]
|
|
self.send_response(status)
|
|
for k, v in headers.items():
|
|
self.send_header(k, v)
|
|
self.send_header('Content-Length', str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
p = folder / self.path.lstrip('/').split('?')[0]
|
|
if not p.is_file():
|
|
self.send_error(404)
|
|
return
|
|
data = p.read_bytes()
|
|
rng = self.headers.get('Range')
|
|
if rng and rng.startswith('bytes='):
|
|
start = int(rng[6:].split('-')[0])
|
|
if start >= len(data):
|
|
self.send_error(416)
|
|
return
|
|
self.send_response(206)
|
|
self.send_header('Content-Range', f'bytes {start}-{len(data) - 1}/{len(data)}')
|
|
data = data[start:]
|
|
else:
|
|
self.send_response(200)
|
|
self.send_header('Content-Length', str(len(data)))
|
|
self.end_headers()
|
|
try:
|
|
self.wfile.write(data)
|
|
except ConnectionError: # the client stopped reading (the cancel test)
|
|
pass
|
|
|
|
self.httpd = http.server.ThreadingHTTPServer(('127.0.0.1', 0), H)
|
|
self.url = f'http://127.0.0.1:{self.httpd.server_address[1]}/'
|
|
threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
|
|
|
|
def close(self):
|
|
self.httpd.shutdown()
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def tar_bytes(files, compress=True):
|
|
"""A tar (zstd-compressed when the stdlib has zstd, else gzip) of {name: bytes}."""
|
|
buf = io.BytesIO()
|
|
mode = 'w:gz' if not (compress and ZSTD) else 'w'
|
|
with tarfile.open(fileobj=buf, mode=mode) as t:
|
|
for name, data in files.items():
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(data)
|
|
t.addfile(info, io.BytesIO(data))
|
|
data = buf.getvalue()
|
|
return ZSTD.compress(data) if compress and ZSTD else data
|
|
|
|
|
|
def desc(name, version, filename, data, depends=(), provides=()):
|
|
lines = ['%FILENAME%', filename, '', '%NAME%', name, '', '%VERSION%', version, '', '%CSIZE%', str(len(data)), '',
|
|
'%ISIZE%', str(len(data) * 3), '', '%SHA256SUM%', sha(data), '']
|
|
if depends:
|
|
lines += ['%DEPENDS%', *depends, '']
|
|
if provides:
|
|
lines += ['%PROVIDES%', *provides, '']
|
|
return '\n'.join(lines).encode()
|
|
|
|
|
|
def fake_repo(folder, damage=None):
|
|
"""A pacman repository with the ten root packages (each holding one of make_game.py's required files), one
|
|
dependency and one name provided by another package."""
|
|
P = toolchain.PREFIX
|
|
files = {'clang': {'ucrt64/bin/clang.exe': b'clang', 'ucrt64/bin/clang++.exe': b'clang++'},
|
|
'sdl3': {'ucrt64/bin/SDL3.dll': b'sdl', 'ucrt64/lib/libSDL3.dll.a': b'implib'},
|
|
'vulkan-memory-allocator': {'ucrt64/include/vma/vk_mem_alloc.h': b'// vma'},
|
|
'vulkan-loader': {'ucrt64/lib/libvulkan-1.dll.a': b'implib'},
|
|
'llvm-libs': {'ucrt64/bin/libLLVM.dll': b'llvm', 'ucrt64/share/doc/llvm/README': b'skipped'}}
|
|
deps = {'clang': [P + 'llvm-libs=23.1', P + 'gcc'], 'sdl3': [P + 'vulkan']}
|
|
entries = {}
|
|
for root in list(toolchain.ROOTS) + ['llvm-libs']:
|
|
name = P + root
|
|
fn = f'{name}-1.0-1-any.pkg.tar.zst'
|
|
data = tar_bytes({'.PKGINFO': b'pkg', **files.get(root, {f'ucrt64/share/{root}/x': b'x'})})
|
|
if damage == root:
|
|
(folder / fn).write_bytes(data[:-3] + b'bad')
|
|
else:
|
|
(folder / fn).write_bytes(data)
|
|
entries[f'{name}-1.0-1/desc'] = desc(name, '1.0-1', fn, data, deps.get(root, ()),
|
|
[P + 'vulkan'] if root == 'vulkan-loader' else ())
|
|
(folder / 'ucrt64.db').write_bytes(tar_bytes(entries))
|
|
|
|
|
|
class Tmp(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path(tempfile.mkdtemp(prefix='nx2navi-test-'))
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
self.layout = Layout(root=self.tmp / 'launcher', app=ROOT).ensure()
|
|
|
|
|
|
# ---------------------------------------------------------------- versions, settings
|
|
|
|
class VersionTest(unittest.TestCase):
|
|
def test_order(self):
|
|
self.assertTrue(versions.newer('0.3.11', '0.3.10'))
|
|
self.assertTrue(versions.newer('0.4', '0.3.99'))
|
|
self.assertFalse(versions.newer('0.3.10+abc', '0.3.10'))
|
|
self.assertTrue(versions.newer('1.0.0', '1.0.0-rc1'))
|
|
self.assertEqual(versions.display('0.3.11+abc1234'), '0.3.11')
|
|
with self.assertRaises(ValueError):
|
|
versions.parse('banana')
|
|
|
|
|
|
class SettingsTest(Tmp):
|
|
def test_roundtrip_and_reset(self):
|
|
s = settings.Settings(self.layout.settings)
|
|
self.assertTrue(s['auto_check'])
|
|
s['auto_check'], s['game_dir'], s['channel'] = False, 'D:/g', 'preview'
|
|
s.save()
|
|
t = settings.Settings(self.layout.settings)
|
|
self.assertEqual((t['auto_check'], t['game_dir'], t['channel']), (False, 'D:/g', 'preview'))
|
|
t.reset()
|
|
self.assertEqual((t['auto_check'], t['game_dir'], t['channel']), (True, 'D:/g', 'stable'))
|
|
|
|
def test_bad_types_ignored(self):
|
|
self.layout.settings.write_text(json.dumps({'auto_check': 'yes', 'channel': 'preview', 'junk': 1}))
|
|
s = settings.Settings(self.layout.settings)
|
|
self.assertEqual((s['auto_check'], s['channel']), (True, 'preview'))
|
|
self.assertNotIn('junk', s)
|
|
|
|
|
|
# ---------------------------------------------------------------- toolchain
|
|
|
|
class ToolchainTest(Tmp):
|
|
def test_resolve_and_install(self):
|
|
repo = self.tmp / 'repo'
|
|
repo.mkdir()
|
|
fake_repo(repo)
|
|
srv = Server(repo)
|
|
self.addCleanup(srv.close)
|
|
lock = toolchain.make_lock(srv.url)
|
|
names = [p['name'] for p in lock['packages']]
|
|
self.assertEqual(names[0], toolchain.PREFIX + 'clang')
|
|
self.assertIn(toolchain.PREFIX + 'llvm-libs', names) # a dependency with a version constraint
|
|
self.assertEqual(names.count(toolchain.PREFIX + 'vulkan-loader'), 1) # 'vulkan' resolves to its provider
|
|
self.assertFalse(toolchain.is_current(self.layout, lock))
|
|
phases = []
|
|
toolchain.install(self.layout, lock, progress=lambda ph, d, t: phases.append(ph), log=lambda m: None)
|
|
self.assertTrue(toolchain.is_current(self.layout, lock))
|
|
self.assertTrue((self.layout.msys / 'bin' / 'clang.exe').is_file())
|
|
self.assertFalse((self.layout.msys / 'share' / 'doc').exists()) # documentation left out
|
|
self.assertIn('download', phases)
|
|
self.assertIn('unpack', phases)
|
|
hits = len(srv.hits)
|
|
toolchain.install(self.layout, lock, log=lambda m: None) # cached package files: no download
|
|
self.assertEqual(len(srv.hits), hits)
|
|
self.assertEqual(toolchain.lock_digest(lock), toolchain.installed(self.layout)['digest'])
|
|
|
|
def test_damaged_package_refused(self):
|
|
repo = self.tmp / 'repo'
|
|
repo.mkdir()
|
|
fake_repo(repo, damage='opus')
|
|
srv = Server(repo)
|
|
self.addCleanup(srv.close)
|
|
lock = toolchain.make_lock(srv.url)
|
|
with self.assertRaises(toolchain.ToolchainError):
|
|
toolchain.install(self.layout, lock, log=lambda m: None)
|
|
self.assertFalse(toolchain.is_current(self.layout, lock))
|
|
|
|
def test_wheel_tag(self):
|
|
tag = toolchain.wheel_tag()[0]
|
|
self.assertTrue(tag.startswith(f'cp{sys.version_info.major}{sys.version_info.minor}-'))
|
|
|
|
|
|
# ---------------------------------------------------------------- net
|
|
|
|
class NetTest(Tmp):
|
|
def test_download_resume_and_checks(self):
|
|
data = os.urandom(3 * net.CHUNK + 123)
|
|
(self.tmp / 'f.bin').write_bytes(data)
|
|
srv = Server(self.tmp)
|
|
self.addCleanup(srv.close)
|
|
dest = self.tmp / 'out' / 'f.bin'
|
|
dest.parent.mkdir()
|
|
(dest.parent / 'f.bin.part').write_bytes(data[:1000]) # a part file from a cut connection
|
|
net.download(srv.url + 'f.bin', dest, sha256=sha(data), size=len(data))
|
|
self.assertEqual(dest.read_bytes(), data)
|
|
self.assertEqual(srv.hits[-1][1].get('Range'), 'bytes=1000-')
|
|
with self.assertRaises(net.DownloadError):
|
|
net.download(srv.url + 'f.bin', self.tmp / 'out' / 'g.bin', sha256='0' * 64)
|
|
self.assertFalse((self.tmp / 'out' / 'g.bin').exists())
|
|
with self.assertRaises(net.Cancelled):
|
|
net.download(srv.url + 'f.bin', self.tmp / 'out' / 'h.bin', cancel=lambda: True)
|
|
with self.assertRaises(net.DownloadError):
|
|
net.download(srv.url + 'missing.bin', self.tmp / 'out' / 'm.bin')
|
|
|
|
|
|
# ---------------------------------------------------------------- feed and updater
|
|
|
|
def make_app_zip(folder, ver, extra=None):
|
|
app = folder / f'app-{ver}'
|
|
(app / 'launcher').mkdir(parents=True)
|
|
(app / 'launcher' / 'start.pyw').write_text('# test\n')
|
|
(app / 'VERSION').write_text(ver + '\n')
|
|
z = folder / f'NX2navi-app-{ver}.zip'
|
|
with zipfile.ZipFile(z, 'w') as zf:
|
|
for p in app.rglob('*'):
|
|
if p.is_file():
|
|
zf.write(p, p.relative_to(app).as_posix())
|
|
for name, data in (extra or {}).items():
|
|
zf.writestr(name, data)
|
|
return z
|
|
|
|
|
|
def manifest_for(z, ver):
|
|
return {'format': 1, 'version': ver, 'notes': '- One\n- Two',
|
|
'app': {'name': z.name, 'size': z.stat().st_size, 'sha256': sha(z.read_bytes())}}
|
|
|
|
|
|
class FeedTest(Tmp):
|
|
def serve_github(self):
|
|
self.routes = {}
|
|
srv = Server(self.tmp, self.routes)
|
|
self.addCleanup(srv.close)
|
|
return srv
|
|
|
|
def publish(self, releases):
|
|
self.routes['/repos/o/r/releases?per_page=20'] = (200, {'Content-Type': 'application/json'},
|
|
json.dumps(releases).encode())
|
|
|
|
def release(self, srv, tag, ver, pre=False, manifest=True):
|
|
z = make_app_zip(self.tmp, ver)
|
|
m = self.tmp / f'm-{ver}.json'
|
|
m.write_text(json.dumps(manifest_for(z, ver)))
|
|
assets = [{'name': z.name, 'browser_download_url': srv.url + z.name, 'url': srv.url + z.name}]
|
|
if manifest:
|
|
assets.append({'name': feed.MANIFEST, 'browser_download_url': srv.url + m.name, 'url': srv.url + m.name})
|
|
return {'tag_name': tag, 'prerelease': pre, 'draft': False, 'body': 'notes', 'html_url': 'x', 'assets': assets}
|
|
|
|
def test_github_channels(self):
|
|
srv = self.serve_github()
|
|
self.publish([self.release(srv, 'v0.4.0-rc1', '0.4.0-rc1', pre=True),
|
|
self.release(srv, 'v0.3.12', '0.3.12', manifest=False), self.release(srv, 'v0.3.11', '0.3.11')])
|
|
f = {'type': 'github', 'repo': 'o/r', 'api': srv.url.rstrip('/')}
|
|
self.assertEqual(feed.latest(f, 'stable').version, '0.3.11') # 0.3.12 lacks a manifest
|
|
self.assertEqual(feed.latest(f, 'preview').version, '0.4.0-rc1')
|
|
r = feed.latest(f, 'stable', token='secret')
|
|
self.assertEqual(r.headers.get('Authorization'), 'Bearer secret')
|
|
self.assertTrue(any(h.get('Authorization') == 'Bearer secret' for _, h in srv.hits))
|
|
self.assertEqual(feed.plain_notes('## Head\n- **One**\n* `Two`\n<!-- x -->'), ['Head', 'One', 'Two'])
|
|
|
|
def test_url_feed_and_update(self):
|
|
z = make_app_zip(self.tmp, '0.9.0')
|
|
(self.tmp / feed.MANIFEST).write_text(json.dumps(manifest_for(z, '0.9.0')))
|
|
srv = Server(self.tmp)
|
|
self.addCleanup(srv.close)
|
|
rel = feed.latest({'type': 'url', 'url': srv.url + feed.MANIFEST})
|
|
self.assertEqual(rel.version, '0.9.0')
|
|
self.layout.apps.mkdir(parents=True)
|
|
(self.layout.apps / 'current.txt').write_text('0.8.0\n')
|
|
old = self.layout.apps / '0.8.0' / 'launcher'
|
|
old.mkdir(parents=True)
|
|
(old / 'start.pyw').write_text('')
|
|
for stale in ('0.7.0', '0.6.0'):
|
|
(self.layout.apps / stale).mkdir()
|
|
folder = updater.stage_app(self.layout, rel)
|
|
self.assertTrue((folder / 'launcher' / 'start.pyw').is_file())
|
|
updater.switch_to(self.layout, '0.9.0')
|
|
self.assertEqual((updater.current(self.layout), updater.previous(self.layout)), ('0.9.0', '0.8.0'))
|
|
updater.prune(self.layout)
|
|
self.assertEqual(sorted(p.name for p in self.layout.apps.iterdir() if p.is_dir()), ['0.8.0', '0.9.0'])
|
|
self.assertTrue(updater.rollback(self.layout))
|
|
self.assertEqual(updater.current(self.layout), '0.8.0')
|
|
self.assertFalse(updater.rollback(self.layout)) # nothing further back
|
|
|
|
def test_unsafe_or_wrong_archives(self):
|
|
z = make_app_zip(self.tmp, '1.0.0', extra={'../evil.txt': 'x'})
|
|
with self.assertRaises(updater.UpdateError):
|
|
updater.safe_extract(z, self.tmp / 'x')
|
|
self.assertFalse((self.tmp / 'evil.txt').exists())
|
|
z2 = make_app_zip(self.tmp, '1.0.1')
|
|
m = manifest_for(z2, '1.0.2') # the archive says 1.0.1
|
|
(self.tmp / feed.MANIFEST).write_text(json.dumps(m))
|
|
srv = Server(self.tmp)
|
|
self.addCleanup(srv.close)
|
|
rel = feed.Release(m, assets={z2.name: srv.url + z2.name})
|
|
with self.assertRaises(updater.UpdateError):
|
|
updater.stage_app(self.layout, rel)
|
|
self.assertFalse((self.layout.apps / '1.0.2').exists())
|
|
|
|
|
|
# ---------------------------------------------------------------- gamefiles
|
|
|
|
def sarc(files):
|
|
names = b''
|
|
offs = []
|
|
for n in files:
|
|
offs.append(len(names))
|
|
b = n.encode() + b'\0'
|
|
names += b + b'\0' * (-len(b) % 4)
|
|
data, ents = b'', b''
|
|
for (n, d), no in zip(files.items(), offs):
|
|
start = len(data)
|
|
data += d
|
|
ents += struct.pack('<IIII', 0, 0x01000000 | (no // 4), start, len(data))
|
|
data += b'\0' * (-len(data) % 8)
|
|
sfat = b'SFAT' + struct.pack('<HHI', 0xC, len(files), 0x65) + ents
|
|
sfnt = b'SFNT' + struct.pack('<HH', 8, 0) + names
|
|
base = 0x14 + len(sfat) + len(sfnt)
|
|
head = b'SARC' + struct.pack('<HHIIHH', 0x14, 0xFEFF, base + len(data), base, 0x100, 0)
|
|
return head + sfat + sfnt + data
|
|
|
|
|
|
def bfotf(otf, key=0x1234ABCD):
|
|
body = otf + b'\0' * (-len(otf) % 4)
|
|
words = struct.unpack(f'>{len(body) // 4}I', body)
|
|
return struct.pack('>II', 0x7F9A0218 ^ key, len(otf) ^ key) + struct.pack(f'>{len(words)}I',
|
|
*(w ^ key for w in words))
|
|
|
|
|
|
@unittest.skipUnless(ZSTD, 'needs compression.zstd (Python 3.14)')
|
|
class FontTest(Tmp):
|
|
FONT = Path('/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf')
|
|
|
|
@unittest.skipUnless(FONT.is_file(), 'needs DejaVu Sans')
|
|
def test_font_from_romfs_image(self):
|
|
otf = self.FONT.read_bytes()
|
|
self.assertEqual(gamefiles.bfotf_to_otf(bfotf(otf)), otf)
|
|
self.assertIsNone(gamefiles.bfotf_to_otf(b'\0' * 64))
|
|
samples = [otf[i:i + 512] for i in range(0, min(len(otf), 400000), 997)]
|
|
d = ZSTD.train_dict(samples, 16384)
|
|
packed = ZSTD.compress(bfotf(otf), zstd_dict=d)
|
|
self.assertNotEqual(ZSTD.get_frame_info(packed).dictionary_id, 0)
|
|
files = {'Pack/ZsDic.pack.zs': ZSTD.compress(sarc({'zs.zsdic': d.dict_content})),
|
|
gamefiles.FONTS['title']: packed}
|
|
romfs = self.tmp / 'game' / 'romfs'
|
|
romfs.mkdir(parents=True)
|
|
entries, blob = [], b''
|
|
for path, data in sorted(files.items()):
|
|
blob += b'\0' * (-len(blob) % 16)
|
|
entries.append({'path': path, 'offset': len(blob), 'size': len(data)})
|
|
blob += data
|
|
(romfs / 'romfs.bin').write_bytes(blob)
|
|
(romfs / 'index.json').write_text(json.dumps({'files': entries}))
|
|
got = gamefiles.extract_fonts(self.layout, self.tmp / 'game')
|
|
self.assertEqual(got['title'][1], 'DejaVu Sans')
|
|
self.assertEqual(Path(got['title'][0]).read_bytes(), otf)
|
|
self.assertEqual(gamefiles.cached_fonts(self.layout)['title'][1], 'DejaVu Sans')
|
|
|
|
|
|
# ---------------------------------------------------------------- make_game.py
|
|
|
|
def nso(build_id=b'\0' * 16, size=0x300):
|
|
head = bytearray(0x100)
|
|
head[0:4] = b'NSO0'
|
|
segs = [(0x100 + i * size, i * 0x10000, size) for i in range(3)]
|
|
for i, (fo, mo, sz) in enumerate(segs):
|
|
struct.pack_into('<III', head, 0x10 + i * 0x10, fo, mo, sz)
|
|
struct.pack_into('<I', head, 0x60 + i * 4, sz)
|
|
head[0x40:0x50] = build_id
|
|
return bytes(head) + b''.join(bytes([i + 1]) * size for i in range(3))
|
|
|
|
|
|
def write_nup(path, exefs, romfs, title_id=0x0400486024DEC000):
|
|
"""A .nup (tools/nup.py's format) with one base package: exefs {name: bytes}, romfs {path: bytes}."""
|
|
import nup as nupmod
|
|
pool = bytearray()
|
|
|
|
def name(s):
|
|
off = len(pool)
|
|
pool.extend(s.encode())
|
|
return off, len(s.encode())
|
|
|
|
dirs, files = [], []
|
|
|
|
def add_tree(tree):
|
|
"""tree: {name: bytes | dict}; returns the directory index."""
|
|
idx = len(dirs)
|
|
dirs.append(None)
|
|
kids = []
|
|
for n, v in sorted(tree.items()):
|
|
if isinstance(v, dict):
|
|
kids.append(('d', add_tree(v), n))
|
|
else:
|
|
files.append([n, v])
|
|
kids.append(('f', len(files) - 1, n))
|
|
dirs[idx] = kids
|
|
return idx
|
|
|
|
root = add_tree({'exefs': exefs, 'romfs': _nest(romfs)})
|
|
nd = len(dirs)
|
|
dir_rows, child_list = [], []
|
|
for kids in dirs:
|
|
first = len(child_list)
|
|
for kind, i, _n in kids:
|
|
child_list.append(i if kind == 'd' else nd + i)
|
|
dir_rows.append((first, len(kids)))
|
|
dir_names = {root: ''}
|
|
for kids in dirs:
|
|
for kind, i, n in kids:
|
|
if kind == 'd':
|
|
dir_names[i] = n
|
|
exefs_dir = next(i for k, i, n in dirs[root] if k == 'd' and n == 'exefs')
|
|
romfs_dir = next(i for k, i, n in dirs[root] if k == 'd' and n == 'romfs')
|
|
toc_head = 0x30 + nupmod.PACKAGE.size + nd * nupmod.DIRECTORY.size + len(files) * nupmod.FILE.size + \
|
|
4 * len(child_list)
|
|
blob_off = nupmod.SUPER.size
|
|
data = bytearray()
|
|
file_rows = []
|
|
for n, v in files:
|
|
no, nl = name(n)
|
|
file_rows.append((no, nl, len(v), blob_off + len(data), len(v), hashlib.sha256(v).digest(), 0))
|
|
data.extend(v)
|
|
dir_name_rows = [name(dir_names[i]) for i in range(nd)]
|
|
toc_off = blob_off + len(data)
|
|
toc = bytearray(struct.pack('<IIQQQQQ', 0x01434F54, 1, nd, len(files), toc_head, len(pool), 0))
|
|
toc += nupmod.PACKAGE.pack(title_id, root, exefs_dir, romfs_dir, 0, 0, 0, 0, 0, 0, 0, b'\0' * 20)
|
|
for i in range(nd):
|
|
toc += nupmod.DIRECTORY.pack(*dir_name_rows[i], *dir_rows[i], 0)
|
|
for no, nl, size, off, stored, digest, codec in file_rows:
|
|
toc += nupmod.FILE.pack(no, nl, size, off, stored, digest, codec, 0, 0, b'\0' * 12)
|
|
toc += struct.pack(f'<{len(child_list)}I', *child_list)
|
|
toc += pool
|
|
total = toc_off + len(toc)
|
|
sup = nupmod.SUPER.pack(b'NUP\x01', 0x80, 1, toc_off, len(toc), hashlib.sha256(bytes(toc)).digest(), blob_off,
|
|
total, 1, 0, b'\0' * 16, b'\0' * 32)
|
|
Path(path).write_bytes(sup + bytes(data) + bytes(toc))
|
|
|
|
|
|
def _nest(flat):
|
|
tree = {}
|
|
for p, v in flat.items():
|
|
parts = p.split('/')
|
|
node = tree
|
|
for part in parts[:-1]:
|
|
node = node.setdefault(part, {})
|
|
node[parts[-1]] = v
|
|
return tree
|
|
|
|
|
|
class MakeGameTest(Tmp):
|
|
def setUp(self):
|
|
super().setUp()
|
|
import make_game
|
|
self.mg = make_game
|
|
exefs = {'main': nso(bytes.fromhex(make_game.MAIN_BUILD_ID)), 'sdk': nso(), 'subsdk0': nso(), 'rtld': nso(),
|
|
'main.npdm': b'META' + b'\0' * 0x100}
|
|
self.romfs = {'a.txt': b'hello', 'dir/b.bin': os.urandom(5000), 'dir/sub/c': b'c' * 33}
|
|
self.nup = self.tmp / 'game.nup'
|
|
write_nup(self.nup, exefs, self.romfs)
|
|
self.msys = self.tmp / 'msys'
|
|
for need in ('bin/clang.exe', 'bin/clang++.exe', 'bin/SDL3.dll', 'lib/libSDL3.dll.a'):
|
|
(self.msys / need).parent.mkdir(parents=True, exist_ok=True)
|
|
(self.msys / need).write_bytes(b'')
|
|
(self.msys / 'include' / 'vma').mkdir(parents=True)
|
|
self.layout.pydeps.mkdir(parents=True, exist_ok=True)
|
|
(self.layout.pydeps / 'zbic.py').write_text('def decompress(b):\n return b\n')
|
|
self.out = self.tmp / 'game out'
|
|
|
|
def run_mg(self, *args):
|
|
events, lines = [], []
|
|
run = build.Run(build.command(self.layout, list(args)), events.append, lines.append)
|
|
code = run.wait()
|
|
return code, events, lines
|
|
|
|
def test_pipeline_rebuild_and_keys(self):
|
|
code, ev, lines = self.run_mg('--nup', self.nup, '--out', self.out, '--msys', self.msys, '--until', 'romfs')
|
|
self.assertEqual(code, 0, '\n'.join(lines))
|
|
states = [(e['step'], e['state']) for e in ev if e['ev'] == 'step']
|
|
self.assertEqual(states, [('check', 'start'), ('check', 'done'), ('exefs', 'start'), ('exefs', 'done'),
|
|
('romfs', 'start'), ('romfs', 'done')])
|
|
self.assertTrue(any(e['ev'] == 'progress' and e['step'] == 'romfs' and e['done'] == e['total'] for e in ev))
|
|
index = json.loads((self.out / 'romfs' / 'index.json').read_text())
|
|
blob = (self.out / 'romfs' / 'romfs.bin').read_bytes()
|
|
for e in index['files']:
|
|
self.assertEqual(blob[e['offset']:e['offset'] + e['size']], self.romfs[e['path']])
|
|
self.assertEqual(e['offset'] % 16, 0)
|
|
flat = (self.out / 'exefs' / 'main').read_bytes()
|
|
self.assertEqual(flat[0x40:0x50].hex(), self.mg.MAIN_BUILD_ID)
|
|
# a second run with the same input skips both steps
|
|
code, ev, _ = self.run_mg('--nup', self.nup, '--out', self.out, '--msys', self.msys, '--until', 'romfs')
|
|
self.assertEqual([e['state'] for e in ev if e['ev'] == 'step' and e['step'] in ('exefs', 'romfs')],
|
|
['skip', 'skip'])
|
|
# --rebuild needs no game files: it checks the unpacked ExeFS and keeps both steps
|
|
self.nup.rename(self.tmp / 'moved.nup')
|
|
code, ev, lines = self.run_mg('--rebuild', '--out', self.out, '--msys', self.msys, '--until', 'romfs')
|
|
self.assertEqual(code, 0, '\n'.join(lines))
|
|
self.assertEqual([e['state'] for e in ev if e['ev'] == 'step' and e['step'] in ('exefs', 'romfs')],
|
|
['skip', 'skip'])
|
|
# the exefs and romfs keys do not depend on the kit; the lift key does
|
|
st = json.loads((self.out / 'build-cache' / 'state.json').read_text())
|
|
self.assertEqual(st['steps']['exefs'], self.mg.step_key('exefs', st['input']))
|
|
self.assertNotIn(self.mg.kit_version(), st['steps']['romfs'])
|
|
|
|
def test_rebuild_without_build(self):
|
|
code, ev, lines = self.run_mg('--rebuild', '--out', self.tmp / 'empty', '--msys', self.msys)
|
|
self.assertEqual(code, 1)
|
|
self.assertTrue(any(e['ev'] == 'error' and 'rebuild' in e['message'] for e in ev))
|
|
|
|
def test_wrong_version_refused(self):
|
|
bad = self.tmp / 'bad.nup'
|
|
write_nup(bad, {'main': nso(b'\x11' * 16), 'sdk': nso(), 'subsdk0': nso(), 'rtld': nso(),
|
|
'main.npdm': b'META' + b'\0' * 0x100}, {'a': b'a'})
|
|
code, ev, _ = self.run_mg('--nup', bad, '--check')
|
|
self.assertEqual(code, 1)
|
|
self.assertTrue(any(e['ev'] == 'error' and '1.1.1' in e['message'] for e in ev))
|
|
|
|
def test_parse_deps(self):
|
|
text = 'C:/o/x.o: C:/src/x.c C:/inc/a\\ b.h \\\n C:/inc/c.h\n'
|
|
self.assertEqual(self.mg.parse_deps(text), ['C:/src/x.c', 'C:/inc/a b.h', 'C:/inc/c.h'])
|
|
|
|
@unittest.skipUnless(shutil.which('cc'), 'needs a C compiler')
|
|
def test_object_cache(self):
|
|
src = self.tmp / 'src'
|
|
src.mkdir()
|
|
(src / 'h.h').write_text('#define V 1\n')
|
|
(src / 'a.c').write_text('#include "h.h"\nint a(void) { return V; }\n')
|
|
(src / 'b.c').write_text('int b(void) { return 2; }\n')
|
|
obj = self.tmp / 'obj'
|
|
obj.mkdir()
|
|
srcs = [(src / 'a.c', obj / 'a.o'), (src / 'b.c', obj / 'b.o')]
|
|
|
|
def cmd_for(s, d):
|
|
return ['cc', '-c', '-I', str(src), '-o', str(d), str(s)]
|
|
|
|
def go():
|
|
cache = self.mg.ObjectCache(obj / 'objects.json')
|
|
return self.mg.compile_all(srcs, cmd_for, cache, 2, dict(os.environ), salt='cc')
|
|
self.assertEqual(go()[:2], (2, 0))
|
|
self.assertEqual(go()[:2], (0, 2))
|
|
(src / 'h.h').write_text('#define V 3\n') # a header: only a.c includes it
|
|
self.assertEqual(go()[:2], (1, 1))
|
|
(src / 'b.c').write_text('int b(void) { return syntax error }\n')
|
|
compiled, reused, bad = go()
|
|
self.assertEqual((compiled, len(bad)), (1, 1))
|
|
self.assertNotIn('b.o', json.loads((obj / 'objects.json').read_text()))
|
|
|
|
|
|
# ---------------------------------------------------------------- jobs
|
|
|
|
FAKE_MAKE_GAME = r'''
|
|
import json, sys, time
|
|
def ev(**k): print('@@' + json.dumps(k), flush=True)
|
|
if '--fail' in open(sys.argv[0] + '.mode').read():
|
|
ev(ev='step', step='check', state='start'); ev(ev='error', message='this is not game version 1.1.1'); sys.exit(1)
|
|
for s in ('check', 'exefs', 'romfs', 'lift', 'compile', 'shaders'):
|
|
ev(ev='step', step=s, state='start')
|
|
if s == 'compile':
|
|
ev(ev='progress', step=s, done=5, total=10, unit='files'); ev(ev='phase', step=s, phase='link')
|
|
print('working on', s, flush=True)
|
|
ev(ev='step', step=s, state='done', seconds=0.1)
|
|
ev(ev='end', exe='x')
|
|
'''
|
|
|
|
|
|
class JobTest(Tmp):
|
|
def make_ctx(self, mode):
|
|
app = self.tmp / 'app'
|
|
(app / 'tools').mkdir(parents=True)
|
|
(app / 'VERSION').write_text('0.3.10\n')
|
|
script = app / 'tools' / 'make_game.py'
|
|
script.write_text(FAKE_MAKE_GAME)
|
|
(app / 'tools' / 'make_game.py.mode').write_text(mode)
|
|
ctx = Context(Layout(root=self.tmp / 'root', app=app))
|
|
return ctx
|
|
|
|
def patch_tools(self):
|
|
saved = toolchain.load_lock, toolchain.is_current, toolchain.ensure_zbic
|
|
toolchain.load_lock = lambda layout, version='dev': {'packages': []}
|
|
toolchain.is_current = lambda layout, lock: True
|
|
toolchain.ensure_zbic = lambda *a, **k: None
|
|
self.addCleanup(lambda: setattr(toolchain, 'load_lock', saved[0]) or
|
|
setattr(toolchain, 'is_current', saved[1]) or setattr(toolchain, 'ensure_zbic', saved[2]))
|
|
|
|
def test_install_job(self):
|
|
self.patch_tools()
|
|
ctx = self.make_ctx('ok')
|
|
j = jobs.InstallJob(ctx, self.tmp / 'g.nup', self.tmp / 'game').run_inline()
|
|
self.assertEqual(j.result, 'ok', j.error)
|
|
self.assertEqual([s.state for s in j.steps], [jobs.SKIPPED] + [jobs.DONE] * 7) # tools ... shaders, ready
|
|
self.assertEqual(j.step('check').value, 'Version 1.1.1')
|
|
self.assertTrue(any('working on lift' in ln for ln in j.lines))
|
|
self.assertEqual(j.overall(), 1.0)
|
|
|
|
def test_failed_job(self):
|
|
self.patch_tools()
|
|
ctx = self.make_ctx('--fail')
|
|
j = jobs.InstallJob(ctx, self.tmp / 'g.nup', self.tmp / 'game').run_inline()
|
|
self.assertEqual(j.result, 'failed')
|
|
self.assertIn('1.1.1', j.error)
|
|
self.assertEqual(j.step('check').state, jobs.FAILED)
|
|
|
|
|
|
# ---------------------------------------------------------------- the window
|
|
|
|
def have_display():
|
|
if os.name == 'nt':
|
|
return True
|
|
if not os.environ.get('DISPLAY'):
|
|
return False
|
|
try:
|
|
import tkinter # noqa: F401
|
|
import numpy # noqa: F401
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
|
|
@unittest.skipUnless(have_display(), 'needs tkinter, numpy and a display')
|
|
class WindowTest(Tmp):
|
|
def test_selftest(self):
|
|
r = subprocess.run([sys.executable, '-m', 'launcher', '--root', str(self.tmp / 'w'), '--selftest'],
|
|
cwd=ROOT, capture_output=True, text=True, timeout=180)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertIn('8 screens drawn', r.stdout)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|