Initial commit with nexus updates and streamlined UI
This commit is contained in:
jschantli committed
commit
f780daecad
41 files changed
+4668
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,70 @@
|
||||
import hashlib
|
||||
import os
|
||||
import struct
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
from mhwmm.core.paks import path_hash, pak_name
|
||||
from mhwmm.core.paths import Paths
|
||||
|
||||
|
||||
def snapshot(root: Path) -> dict:
|
||||
"""Byte-level picture of a tree: files by hash, symlinks by target, and directories."""
|
||||
snap = {}
|
||||
for dirpath, dirnames, filenames in os.walk(root):
|
||||
for n in dirnames + filenames:
|
||||
p = Path(dirpath) / n
|
||||
rel = str(p.relative_to(root))
|
||||
if p.is_symlink():
|
||||
snap[rel] = ("l", os.readlink(p))
|
||||
elif p.is_dir():
|
||||
snap[rel] = ("d",)
|
||||
else:
|
||||
snap[rel] = ("f", hashlib.sha256(p.read_bytes()).hexdigest())
|
||||
return snap
|
||||
|
||||
|
||||
def make_game(root: Path, ceiling: int = 15) -> Path:
|
||||
game = root / "game"
|
||||
game.mkdir()
|
||||
for chain in ("main", "sub"):
|
||||
for n in range(1, ceiling + 1):
|
||||
(game / pak_name(chain, n)).write_bytes(f"vanilla-{chain}-{n}".encode())
|
||||
(game / "re_chunk_000.pak").write_bytes(b"base")
|
||||
(game / "MonsterHunterWilds.exe").write_bytes(b"exe")
|
||||
return game
|
||||
|
||||
|
||||
def write_tree(root: Path, files: dict[str, bytes | str]) -> Path:
|
||||
for rel, data in files.items():
|
||||
p = root / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_bytes(data if isinstance(data, bytes) else data.encode())
|
||||
return root
|
||||
|
||||
|
||||
def make_kpka(paths: list[str]) -> bytes:
|
||||
"""Minimal KPKA with entries keyed by the path hashes of `paths`."""
|
||||
out = b"KPKA" + struct.pack("<BBHII", 4, 0, 0, len(paths), 0)
|
||||
for p in paths:
|
||||
lo, up = path_hash(p)
|
||||
out += struct.pack("<IIQQQQQ", lo, up, 0, 0, 0, 0, 0)
|
||||
return out
|
||||
|
||||
|
||||
def make_zip(path: Path, files: dict[str, bytes | str]) -> Path:
|
||||
with zipfile.ZipFile(path, "w") as zf:
|
||||
for name, data in files.items():
|
||||
zf.writestr(name, data)
|
||||
return path
|
||||
|
||||
|
||||
class TmpCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._td = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._td.cleanup)
|
||||
self.tmp = Path(self._td.name)
|
||||
self.paths = Paths(self.tmp / "data", self.tmp / "config")
|
||||
self.paths.ensure()
|
||||
@@ -0,0 +1,211 @@
|
||||
import os
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from mhwmm.core import paks
|
||||
from mhwmm.core.deploy import DeployError, DeployMod, Deployer, build_desired, validate_dest
|
||||
from mhwmm.core.layout import classify
|
||||
from tests.helpers import TmpCase, make_game, make_kpka, snapshot, write_tree
|
||||
|
||||
|
||||
class DeployBase(TmpCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.game = make_game(self.tmp)
|
||||
self.baseline = paks.capture_baseline(self.game)
|
||||
paks.save_baseline(self.paths.baseline_file, self.baseline)
|
||||
|
||||
def mod(self, mid: str, files: dict, **kw) -> DeployMod:
|
||||
root = write_tree(self.paths.mods_dir / mid / "files", files)
|
||||
return DeployMod(mid, tuple(classify(root, **kw).entries))
|
||||
|
||||
def deployer(self, mode="symlink") -> Deployer:
|
||||
return Deployer(self.game, self.paths, mode)
|
||||
|
||||
def deploy(self, d: Deployer, mods):
|
||||
return d.apply(d.plan(build_desired(mods, self.baseline)))
|
||||
|
||||
|
||||
class RoundTrip(DeployBase):
|
||||
def _roundtrip(self, mode):
|
||||
# a pre-existing unmanaged file that a mod will shadow
|
||||
write_tree(self.game, {"natives/STM/shadowed.txt": "original"})
|
||||
before = snapshot(self.game)
|
||||
mods = [
|
||||
self.mod("a", {"natives/STM/a.bin": "A", "natives/STM/shadowed.txt": "mod",
|
||||
"reframework/autorun/a.lua": "print(1)", "a.pak": "P1"}),
|
||||
self.mod("b", {"natives/STM/deep/dir/b.bin": "B", "b.pak": "P2",
|
||||
"sub.sub_000.pak.patch_099.pak": "S"}),
|
||||
]
|
||||
d = self.deployer(mode)
|
||||
res = self.deploy(d, mods)
|
||||
self.assertGreater(res.added, 0)
|
||||
self.assertTrue((self.game / "natives/STM/deep/dir/b.bin").exists())
|
||||
self.assertEqual((self.game / "natives/STM/shadowed.txt").read_text(), "mod")
|
||||
self.assertTrue((self.game / "re_chunk_000.pak.patch_016.pak").exists())
|
||||
self.assertTrue((self.game / "re_chunk_000.pak.patch_017.pak").exists())
|
||||
self.assertTrue((self.game / "re_chunk_000.pak.sub_000.pak.patch_016.pak").exists())
|
||||
self.assertNotEqual(before, snapshot(self.game))
|
||||
d.undeploy_all()
|
||||
self.assertEqual(before, snapshot(self.game))
|
||||
self.assertEqual(d.manifest()["entries"], {})
|
||||
|
||||
def test_roundtrip_symlink(self):
|
||||
self._roundtrip("symlink")
|
||||
|
||||
def test_roundtrip_hardlink(self):
|
||||
self._roundtrip("hardlink")
|
||||
|
||||
def test_roundtrip_copy(self):
|
||||
self._roundtrip("copy")
|
||||
|
||||
def test_incremental_disable_prunes_only_our_dirs(self):
|
||||
(self.game / "natives").mkdir() # pre-existing, not ours
|
||||
before = snapshot(self.game)
|
||||
a = self.mod("a", {"natives/STM/x/a.bin": "A"})
|
||||
d = self.deployer()
|
||||
self.deploy(d, [a])
|
||||
self.deploy(d, [])
|
||||
self.assertEqual(before, snapshot(self.game))
|
||||
self.assertTrue((self.game / "natives").is_dir())
|
||||
|
||||
|
||||
class Conflicts(DeployBase):
|
||||
def test_later_mod_wins_and_reorder_flips(self):
|
||||
a = self.mod("a", {"natives/STM/shared.bin": "A"})
|
||||
b = self.mod("b", {"natives/STM/Shared.bin": "B"}) # case differs
|
||||
des = build_desired([a, b], self.baseline)
|
||||
self.assertEqual(len(des.conflicts), 1)
|
||||
self.assertEqual(des.conflicts[0].winner, "b")
|
||||
self.assertEqual(des.conflicts[0].losers, ["a"])
|
||||
des = build_desired([b, a], self.baseline)
|
||||
self.assertEqual(des.conflicts[0].winner, "a")
|
||||
d = self.deployer()
|
||||
self.deploy(d, [a, b])
|
||||
self.assertEqual((self.game / "natives/STM/Shared.bin").read_text(), "B")
|
||||
self.deploy(d, [b, a])
|
||||
self.assertEqual((self.game / "natives/STM/shared.bin").read_text(), "A")
|
||||
self.assertFalse((self.game / "natives/STM/Shared.bin").exists())
|
||||
|
||||
def test_conflicts_span_pak_and_loose(self):
|
||||
loose = "natives/STM/player/mod.bin"
|
||||
pak_a = make_kpka([loose, "natives/stm/other.bin"])
|
||||
pak_b = make_kpka(["natives/stm/other.bin"])
|
||||
a = self.mod("a", {"a.pak": pak_a})
|
||||
b = self.mod("b", {"b.pak": pak_b})
|
||||
c = self.mod("c", {loose: "x"})
|
||||
des = build_desired([a, b, c], self.baseline)
|
||||
kinds = {(x.kind, tuple(x.mods)): x for x in des.conflicts}
|
||||
self.assertIn(("pak-pak", ("a", "b")), kinds)
|
||||
self.assertEqual(kinds[("pak-pak", ("a", "b"))].winner, "b")
|
||||
pl = kinds[("pak-loose", ("a", "c"))]
|
||||
self.assertEqual(pl.winner, "c") # loose loader beats PAK contents
|
||||
|
||||
|
||||
class Safety(DeployBase):
|
||||
def test_modified_file_is_never_deleted(self):
|
||||
a = self.mod("a", {"natives/STM/a.bin": "A"})
|
||||
d = self.deployer("copy")
|
||||
self.deploy(d, [a])
|
||||
target = self.game / "natives/STM/a.bin"
|
||||
target.write_text("user edited")
|
||||
plan = d.plan(build_desired([], self.baseline))
|
||||
self.assertTrue(plan.warnings)
|
||||
d.apply(plan)
|
||||
self.assertEqual(target.read_text(), "user edited")
|
||||
self.assertEqual(d.manifest()["entries"], {})
|
||||
|
||||
def test_foreign_pak_is_never_deleted(self):
|
||||
d = self.deployer()
|
||||
stranger = self.game / "re_chunk_000.pak.patch_016.pak"
|
||||
stranger.write_bytes(b"not ours")
|
||||
self.deploy(d, [])
|
||||
d.undeploy_all()
|
||||
self.assertEqual(stranger.read_bytes(), b"not ours")
|
||||
|
||||
def test_reframework_data_never_managed(self):
|
||||
m = self.mod("a", {"reframework/data/settings.json": "x", "reframework/autorun/a.lua": "y"})
|
||||
self.assertTrue(all("data" not in e.dest for e in m.entries))
|
||||
self.assertIsNotNone(validate_dest("reframework/data/x.json"))
|
||||
self.assertIsNotNone(validate_dest("../evil"))
|
||||
self.assertIsNotNone(validate_dest("/etc/passwd"))
|
||||
self.assertIsNotNone(validate_dest("natives/../../x"))
|
||||
self.assertIsNotNone(validate_dest("MonsterHunterWilds.exe"))
|
||||
self.assertIsNone(validate_dest("natives/STM/a.bin"))
|
||||
|
||||
def test_rollback_on_midway_failure(self):
|
||||
good = self.mod("a", {"natives/STM/a.bin": "A", "natives/STM/z.bin": "Z"})
|
||||
d = self.deployer()
|
||||
plan = d.plan(build_desired([good], self.baseline))
|
||||
os.unlink(good.entries[-1].src) # last add will fail: source vanished
|
||||
before = snapshot(self.game)
|
||||
with self.assertRaises(DeployError):
|
||||
d.apply(plan)
|
||||
self.assertEqual(before, snapshot(self.game))
|
||||
self.assertEqual(d.manifest()["entries"], {})
|
||||
|
||||
def test_replacing_keeps_original_backup(self):
|
||||
write_tree(self.game, {"natives/STM/s.txt": "orig"})
|
||||
before = snapshot(self.game)
|
||||
d = self.deployer()
|
||||
self.deploy(d, [self.mod("a", {"natives/STM/s.txt": "A"})])
|
||||
self.deploy(d, [self.mod("a", {"natives/STM/s.txt": "A"}), self.mod("b", {"natives/STM/s.txt": "B"})])
|
||||
self.assertEqual((self.game / "natives/STM/s.txt").read_text(), "B")
|
||||
d.undeploy_all()
|
||||
self.assertEqual(before, snapshot(self.game))
|
||||
|
||||
def test_dll_choice(self):
|
||||
root = write_tree(self.paths.mods_dir / "d" / "files", {"thing.dll": "x"})
|
||||
self.assertEqual(len(classify(root).ambiguous), 1)
|
||||
self.assertEqual(classify(root, "plugin").entries[0].dest, "reframework/plugins/thing.dll")
|
||||
self.assertEqual(classify(root, "root").entries[0].dest, "thing.dll")
|
||||
self.assertTrue(classify(root, "skip").empty)
|
||||
|
||||
|
||||
class PakUpdate(DeployBase):
|
||||
def test_slots_start_above_ceiling_in_load_order(self):
|
||||
a = self.mod("a", {"a1.pak": "1", "a2.pak": "2"})
|
||||
b = self.mod("b", {"b.pak": "3"})
|
||||
slots = paks.allocate([(m.id, list(m.entries)) for m in (a, b)], self.baseline)
|
||||
self.assertEqual([(s.mod_id, s.number) for s in slots], [("a", 16), ("a", 17), ("b", 18)])
|
||||
|
||||
def test_title_update_reslots(self):
|
||||
d = self.deployer()
|
||||
a = self.mod("a", {"a.pak": "AAA", "natives/STM/a.bin": "A"})
|
||||
self.deploy(d, [a])
|
||||
self.assertEqual(d.owned_paks(), {"re_chunk_000.pak.patch_016.pak"})
|
||||
self.assertFalse(paks.detect_update(self.game, self.baseline, d.owned_paks()).updated)
|
||||
|
||||
# Steam ships patch_016 over our symlink and patch_017 on the sub chain
|
||||
(self.game / "re_chunk_000.pak.patch_016.pak").unlink()
|
||||
(self.game / "re_chunk_000.pak.patch_016.pak").write_bytes(b"vanilla-16")
|
||||
(self.game / "re_chunk_000.pak.sub_000.pak.patch_016.pak").write_bytes(b"vanilla-sub-16")
|
||||
rep = paks.detect_update(self.game, self.baseline, d.owned_paks())
|
||||
self.assertTrue(rep.updated)
|
||||
|
||||
d.apply(d.plan(d.desired_from_manifest(exclude_kind="pak")))
|
||||
self.assertEqual((self.game / "re_chunk_000.pak.patch_016.pak").read_bytes(), b"vanilla-16")
|
||||
self.assertTrue((self.game / "natives/STM/a.bin").is_symlink()) # loose files untouched
|
||||
|
||||
self.baseline = paks.capture_baseline(self.game, d.owned_paks())
|
||||
paks.save_baseline(self.paths.baseline_file, self.baseline)
|
||||
self.assertEqual(self.baseline.ceiling("main"), 16)
|
||||
self.assertEqual(self.baseline.ceiling("sub"), 16)
|
||||
self.deploy(d, [a])
|
||||
self.assertTrue((self.game / "re_chunk_000.pak.patch_017.pak").is_symlink())
|
||||
self.assertEqual(d.owned_paks(), {"re_chunk_000.pak.patch_017.pak"})
|
||||
d.undeploy_all()
|
||||
self.assertEqual((self.game / "re_chunk_000.pak.patch_016.pak").read_bytes(), b"vanilla-16")
|
||||
|
||||
def test_synthetic_patch_016_from_outside_is_detected(self):
|
||||
d = self.deployer()
|
||||
(self.game / "re_chunk_000.pak.patch_016.pak").write_bytes(b"steam")
|
||||
self.assertTrue(paks.detect_update(self.game, self.baseline, d.owned_paks()).updated)
|
||||
|
||||
def test_baseline_ceiling(self):
|
||||
self.assertEqual(self.baseline.ceiling("main"), 15)
|
||||
self.assertEqual(self.baseline.ceiling("sub"), 15)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,165 @@
|
||||
import io
|
||||
import json
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
from mhwmm.core import loader, nexus, steam
|
||||
from tests.helpers import TmpCase, make_zip
|
||||
|
||||
VDF = '''"libraryfolders"
|
||||
{
|
||||
\t"0"
|
||||
\t{
|
||||
\t\t"path"\t\t"/tmp/a b"
|
||||
\t\t"apps" { "2246340" "1" }
|
||||
\t}
|
||||
\t"1" { "path" "/mnt/games" }
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
class VDFTests(unittest.TestCase):
|
||||
def test_parse_and_case_insensitive_get(self):
|
||||
d = steam.parse_vdf(VDF)
|
||||
self.assertEqual(steam.vdf_get(d, "LibraryFolders", "1", "PATH"), "/mnt/games")
|
||||
|
||||
def test_roundtrip_dump(self):
|
||||
d = steam.parse_vdf(VDF)
|
||||
self.assertEqual(steam.parse_vdf(steam.dump_vdf(d)), d)
|
||||
|
||||
def test_escapes_and_comments(self):
|
||||
d = steam.parse_vdf('// hi\n"a" "x\\"y\\\\z"\n')
|
||||
self.assertEqual(d["a"], 'x"y\\z')
|
||||
|
||||
def test_errors(self):
|
||||
for bad in ('"a" {', '}', '{ "a" "b" }'):
|
||||
with self.assertRaises(steam.VDFError):
|
||||
steam.parse_vdf(bad)
|
||||
|
||||
def test_real_files_on_this_machine(self):
|
||||
root = Path.home() / ".steam/steam/steamapps"
|
||||
if not (root / "appmanifest_2246340.acf").is_file():
|
||||
self.skipTest("Monster Hunter Wilds not installed here")
|
||||
acf = steam.vdf_get(steam.read_vdf(root / "appmanifest_2246340.acf"), "AppState")
|
||||
self.assertEqual(acf["appid"], "2246340")
|
||||
self.assertTrue(acf["installdir"])
|
||||
self.assertTrue(steam.library_folders(root.parent))
|
||||
found = steam.find_app()
|
||||
self.assertIsNotNone(found)
|
||||
self.assertTrue((found.game_dir / "MonsterHunterWilds.exe").exists())
|
||||
|
||||
def test_launch_options(self):
|
||||
req = steam.REQUIRED_LAUNCH_OPTIONS
|
||||
self.assertTrue(steam.launch_options_ok(req))
|
||||
self.assertFalse(steam.launch_options_ok(""))
|
||||
self.assertFalse(steam.launch_options_ok("mangohud %command%"))
|
||||
self.assertEqual(steam.suggest_launch_options(""), req)
|
||||
self.assertEqual(steam.suggest_launch_options("mangohud %command%"),
|
||||
'WINEDLLOVERRIDES="dinput8.dll=n,b" mangohud %command%')
|
||||
self.assertEqual(steam.suggest_launch_options('WINEDLLOVERRIDES="winmm=n" %command%'),
|
||||
'WINEDLLOVERRIDES="winmm=n;dinput8.dll=n,b" %command%')
|
||||
self.assertEqual(steam.suggest_launch_options(req), req)
|
||||
|
||||
|
||||
class LocalConfigTests(TmpCase):
|
||||
def test_read_and_write_launch_options(self):
|
||||
lc = self.tmp / "localconfig.vdf"
|
||||
lc.write_text('"UserLocalConfigStore"\n{\n\t"Software"\n\t{\n\t\t"Valve"\n\t\t{\n\t\t\t"Steam"\n\t\t\t{\n'
|
||||
'\t\t\t\t"apps"\n\t\t\t\t{\n\t\t\t\t\t"2246340"\n\t\t\t\t\t{\n\t\t\t\t\t\t"LaunchOptions"\t\t"-x"\n'
|
||||
'\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t}\n}\n')
|
||||
self.assertEqual(steam.read_launch_options(lc), "-x")
|
||||
orig = steam.steam_running
|
||||
try:
|
||||
steam.steam_running = lambda: True
|
||||
with self.assertRaises(steam.SteamRunningError):
|
||||
steam.write_launch_options(lc, "y")
|
||||
self.assertEqual(steam.read_launch_options(lc), "-x")
|
||||
steam.steam_running = lambda: False
|
||||
bak = steam.write_launch_options(lc, steam.REQUIRED_LAUNCH_OPTIONS)
|
||||
finally:
|
||||
steam.steam_running = orig
|
||||
self.assertTrue(bak.exists())
|
||||
self.assertEqual(steam.read_launch_options(lc), steam.REQUIRED_LAUNCH_OPTIONS)
|
||||
|
||||
|
||||
class LoaderTests(TmpCase):
|
||||
def test_config_patch_preserves_other_keys(self):
|
||||
cfg = self.tmp / loader.CONFIG_NAME
|
||||
cfg.write_text("Foo=1\nlooseFileLoader_enabled=false\nBar=2\n")
|
||||
loader.enable_loose_loader(self.tmp)
|
||||
self.assertEqual(cfg.read_text(), "Foo=1\nlooseFileLoader_enabled=true\nBar=2\n")
|
||||
|
||||
def test_config_created_when_absent(self):
|
||||
loader.enable_loose_loader(self.tmp)
|
||||
self.assertTrue(loader.loose_loader_enabled(self.tmp))
|
||||
|
||||
def test_status_checks(self):
|
||||
st = loader.status(self.tmp, None)
|
||||
self.assertEqual([c.ok for c in st.checks], [False, False, False])
|
||||
(self.tmp / "dinput8.dll").write_bytes(b"x")
|
||||
loader.enable_loose_loader(self.tmp)
|
||||
st = loader.status(self.tmp, steam.REQUIRED_LAUNCH_OPTIONS)
|
||||
self.assertTrue(st.all_ok)
|
||||
|
||||
def test_install_from_release(self):
|
||||
blob = io.BytesIO()
|
||||
with zipfile.ZipFile(blob, "w") as zf:
|
||||
zf.writestr("dinput8.dll", b"MZfake")
|
||||
zf.writestr("reframework/x", b"ignored")
|
||||
rel = json.dumps({"tag_name": "nightly-1", "assets": [
|
||||
{"name": "MHWILDS.zip", "browser_download_url": "https://example/MHWILDS.zip"}]}).encode()
|
||||
fetch = lambda url: rel if url == loader.RELEASES_API else blob.getvalue()
|
||||
self.assertEqual(loader.install_reframework(self.tmp, fetch), "nightly-1")
|
||||
self.assertEqual((self.tmp / "dinput8.dll").read_bytes(), b"MZfake")
|
||||
self.assertFalse((self.tmp / "reframework").exists())
|
||||
|
||||
|
||||
class NexusTests(TmpCase):
|
||||
def test_parse_nxm_with_key_and_expires(self):
|
||||
l = nexus.parse_nxm("nxm://monsterhunterwilds/mods/123/files/456?key=AbC_-9&expires=1700000000&user_id=1")
|
||||
self.assertEqual((l.game, l.mod_id, l.file_id, l.key, l.expires),
|
||||
("monsterhunterwilds", 123, 456, "AbC_-9", "1700000000"))
|
||||
|
||||
def test_parse_rejects_garbage(self):
|
||||
for bad in ("https://x/mods/1/files/2", "nxm://g/mods/x/files/2", "nxm:///mods/1/files/2", "nxm://g/mods/1"):
|
||||
with self.assertRaises(nexus.NexusError, msg=bad):
|
||||
nexus.parse_nxm(bad)
|
||||
|
||||
def test_client_passes_key_and_records_limits(self):
|
||||
seen = {}
|
||||
|
||||
class Resp:
|
||||
headers = {"x-rl-daily-remaining": "99", "x-rl-daily-limit": "100",
|
||||
"x-rl-hourly-remaining": "9", "x-rl-hourly-limit": "10"}
|
||||
def read(self): return b'[{"URI": "https://cdn/x.zip"}]'
|
||||
|
||||
def opener(req):
|
||||
seen["url"], seen["headers"] = req.full_url, dict(req.header_items())
|
||||
return Resp()
|
||||
|
||||
c = nexus.NexusClient("SECRET", opener)
|
||||
out = c.download_links(nexus.parse_nxm("nxm://monsterhunterwilds/mods/1/files/2?key=K&expires=5"))
|
||||
self.assertEqual(out[0]["URI"], "https://cdn/x.zip")
|
||||
self.assertIn("/v1/games/monsterhunterwilds/mods/1/files/2/download_link.json?key=K&expires=5", seen["url"])
|
||||
self.assertEqual(seen["headers"]["Apikey"], "SECRET")
|
||||
self.assertIn("9/10 hourly", c.limits.summary())
|
||||
|
||||
def test_no_key_is_a_clear_error(self):
|
||||
with self.assertRaises(nexus.NexusError):
|
||||
nexus.NexusClient("").mod_info(nexus.parse_nxm("nxm://monsterhunterwilds/mods/1/files/2"))
|
||||
|
||||
def test_keystore_file_fallback_is_0600(self):
|
||||
ks = nexus.KeyStore(self.tmp / "secrets.json")
|
||||
nexus._secret_mod = lambda: None
|
||||
ks.set("abc")
|
||||
self.assertEqual(ks.get(), "abc")
|
||||
self.assertEqual((self.tmp / "secrets.json").stat().st_mode & 0o777, 0o600)
|
||||
|
||||
def test_download_refuses_http(self):
|
||||
with self.assertRaises(nexus.NexusError):
|
||||
nexus.download("http://x/y.zip", self.tmp)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,167 @@
|
||||
import unittest
|
||||
|
||||
from mhwmm.core import archive, layout
|
||||
from mhwmm.core.library import Library, LibraryError
|
||||
from mhwmm.core.manager import Manager
|
||||
from mhwmm.core.config import Settings
|
||||
from mhwmm.core.modinfo import parse_modinfo
|
||||
from mhwmm.core.profiles import Profiles
|
||||
from tests.helpers import TmpCase, make_game, make_zip, write_tree
|
||||
import zipfile
|
||||
|
||||
|
||||
class ArchiveTests(TmpCase):
|
||||
def test_zip_slip_rejected(self):
|
||||
z = make_zip(self.tmp / "evil.zip", {"../../evil.txt": "x"})
|
||||
with self.assertRaises(archive.ArchiveError):
|
||||
archive.extract(z, self.tmp / "out")
|
||||
self.assertFalse((self.tmp / "evil.txt").exists())
|
||||
|
||||
def test_absolute_and_drive_paths_rejected(self):
|
||||
for name in ("/etc/evil", "C:/evil", "a/../../evil", "..\\evil"):
|
||||
z = make_zip(self.tmp / "e.zip", {name: "x"})
|
||||
with self.assertRaises(archive.ArchiveError, msg=name):
|
||||
archive.extract(z, self.tmp / "out2")
|
||||
|
||||
def test_symlink_entry_rejected(self):
|
||||
p = self.tmp / "s.zip"
|
||||
with zipfile.ZipFile(p, "w") as zf:
|
||||
info = zipfile.ZipInfo("link")
|
||||
info.external_attr = 0o120777 << 16
|
||||
zf.writestr(info, "/etc/passwd")
|
||||
with self.assertRaises(archive.ArchiveError):
|
||||
archive.extract(p, self.tmp / "o")
|
||||
|
||||
def test_sniff_ignores_extension(self):
|
||||
z = make_zip(self.tmp / "actually.rar", {"a": "b"})
|
||||
self.assertEqual(archive.sniff(z), "zip")
|
||||
(self.tmp / "junk.zip").write_bytes(b"nope")
|
||||
with self.assertRaises(archive.ArchiveError):
|
||||
archive.sniff(self.tmp / "junk.zip")
|
||||
|
||||
def test_good_zip_extracts(self):
|
||||
z = make_zip(self.tmp / "ok.zip", {"natives/STM/a.bin": "A"})
|
||||
archive.extract(z, self.tmp / "ok")
|
||||
self.assertEqual((self.tmp / "ok/natives/STM/a.bin").read_text(), "A")
|
||||
|
||||
|
||||
class LayoutTests(TmpCase):
|
||||
def classify(self, files):
|
||||
return layout.classify(write_tree(self.tmp / f"t{len(list(self.tmp.iterdir()))}", files))
|
||||
|
||||
def dests(self, c):
|
||||
return sorted(e.dest for e in c.entries)
|
||||
|
||||
def test_natives_any_depth_case_insensitive(self):
|
||||
c = self.classify({"Wrapper/Inner/Natives/STM/x.bin": "1", "Wrapper/Inner/readme.txt": "r"})
|
||||
self.assertEqual(self.dests(c), ["natives/STM/x.bin"])
|
||||
|
||||
def test_bare_stm(self):
|
||||
self.assertEqual(self.dests(self.classify({"STM/a/b.bin": "1"})), ["natives/STM/a/b.bin"])
|
||||
|
||||
def test_reframework_preserved(self):
|
||||
c = self.classify({"reframework/autorun/a.lua": "1", "reframework/plugins/p.dll": "2",
|
||||
"reframework/data/keep.json": "3"})
|
||||
self.assertEqual(self.dests(c), ["reframework/autorun/a.lua", "reframework/plugins/p.dll"])
|
||||
self.assertTrue(c.notes)
|
||||
|
||||
def test_pak_near_root(self):
|
||||
c = self.classify({"paks/re_chunk_000.pak.sub_000.pak.patch_020.pak": "1", "x.pak": "2"})
|
||||
self.assertEqual(sorted((e.kind, e.chain) for e in c.entries), [("pak", "main"), ("pak", "sub")])
|
||||
|
||||
def test_loose_lua_and_ambiguous_dll(self):
|
||||
c = self.classify({"script.lua": "1", "mystery.dll": "2"})
|
||||
self.assertEqual(self.dests(c), ["reframework/autorun/script.lua"])
|
||||
self.assertEqual(len(c.ambiguous), 1)
|
||||
|
||||
def test_mixed_kinds_in_one_mod(self):
|
||||
c = self.classify({"natives/STM/a.bin": "1", "reframework/autorun/a.lua": "2", "m.pak": "3"})
|
||||
self.assertEqual({e.kind for e in c.entries}, {"loose", "reframework", "pak"})
|
||||
|
||||
def test_junk_only_is_empty(self):
|
||||
self.assertTrue(self.classify({"readme.txt": "hi", "modinfo.ini": "name=x"}).empty)
|
||||
|
||||
|
||||
class LibraryTests(TmpCase):
|
||||
def test_import_reads_modinfo_and_screenshot(self):
|
||||
z = make_zip(self.tmp / "Cool Mod-123-1-0-99.zip", {
|
||||
"modinfo.ini": "name=Cool Mod\nversion=1.2\nauthor=me\ncategory=Armor\ndescription=hello",
|
||||
"screenshot.png": b"\x89PNG",
|
||||
"natives/STM/a.bin": "A"})
|
||||
lib = Library(self.paths)
|
||||
mod = lib.import_source(z)
|
||||
self.assertEqual((mod.name, mod.version, mod.category), ("Cool Mod", "1.2", "Armor"))
|
||||
self.assertTrue(mod.screenshot_path and mod.screenshot_path.name == "screenshot.png")
|
||||
lib2 = Library(self.paths)
|
||||
lib2.scan()
|
||||
self.assertIn(mod.id, lib2.mods)
|
||||
self.assertEqual(lib2.categories(), ["Armor"])
|
||||
|
||||
def test_import_rejects_empty_and_leaves_no_debris(self):
|
||||
z = make_zip(self.tmp / "junk.zip", {"readme.txt": "x"})
|
||||
lib = Library(self.paths)
|
||||
with self.assertRaises(LibraryError):
|
||||
lib.import_source(z)
|
||||
self.assertEqual([p for p in self.paths.mods_dir.iterdir()], [])
|
||||
|
||||
def test_duplicate_names_get_unique_ids(self):
|
||||
lib = Library(self.paths)
|
||||
ids = {lib.import_source(make_zip(self.tmp / f"m{i}.zip", {"modinfo.ini": "name=Same", "STM/a": "x"})).id
|
||||
for i in range(3)}
|
||||
self.assertEqual(len(ids), 3)
|
||||
|
||||
def test_modinfo_tolerant(self):
|
||||
i = parse_modinfo("\ufeff[~ modinfo ~]\nName = X \n; c\nVersion=2\n")
|
||||
self.assertEqual((i.name, i.version), ("X", "2"))
|
||||
|
||||
|
||||
class ProfileTests(TmpCase):
|
||||
def test_profiles_roundtrip(self):
|
||||
p = Profiles(self.paths.profiles_file)
|
||||
p.reconcile(["a", "b", "c"])
|
||||
p.set_enabled("a", True)
|
||||
p.set_enabled("c", True)
|
||||
p.move("c", 0)
|
||||
p.create("Alt", copy_from="Default")
|
||||
p.switch("Alt")
|
||||
p.set_enabled("b", True)
|
||||
p.save()
|
||||
q = Profiles(self.paths.profiles_file)
|
||||
self.assertEqual(q.active, "Alt")
|
||||
self.assertEqual(q.data["Default"]["enabled"], ["a", "c"])
|
||||
q.switch("Default")
|
||||
self.assertEqual(q.enabled_in_order(), ["c", "a"])
|
||||
q.reconcile(["a", "b"]) # c vanished from the library
|
||||
self.assertEqual(q.enabled_in_order(), ["a"])
|
||||
|
||||
|
||||
class ManagerTests(TmpCase):
|
||||
def test_end_to_end_and_startup_reslot(self):
|
||||
game = make_game(self.tmp)
|
||||
s = Settings(game_dir=str(game))
|
||||
m = Manager(self.paths, s)
|
||||
self.assertTrue(m.startup().game_found)
|
||||
self.assertEqual(m.pak_summary()["main"]["ceiling"], 15)
|
||||
mod = m.import_source(make_zip(self.tmp / "a.zip", {"modinfo.ini": "name=A", "natives/STM/a.bin": "A", "a.pak": "P"}))
|
||||
m.set_enabled(mod.id, True)
|
||||
self.assertTrue((game / "natives/STM/a.bin").is_symlink())
|
||||
self.assertEqual(m.pak_summary()["main"]["mod_slots"], [16])
|
||||
self.assertFalse(m.pending())
|
||||
|
||||
# Steam patches while the manager is closed
|
||||
(game / "re_chunk_000.pak.patch_016.pak").unlink()
|
||||
(game / "re_chunk_000.pak.patch_016.pak").write_bytes(b"steam")
|
||||
m2 = Manager(self.paths, Settings(game_dir=str(game)))
|
||||
rep = m2.startup()
|
||||
self.assertTrue(any("re-slotted to 017+" in n for n in rep.notes), rep.notes)
|
||||
self.assertEqual((game / "re_chunk_000.pak.patch_016.pak").read_bytes(), b"steam")
|
||||
self.assertEqual(m2.pak_summary()["main"]["mod_slots"], [17])
|
||||
|
||||
m2.set_enabled(mod.id, False)
|
||||
self.assertFalse((game / "natives").exists())
|
||||
m2.delete_mod(mod.id)
|
||||
self.assertEqual(m2.library.mods, {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user