11 Commits
29 changed files with 4447 additions and 958 deletions
+1
View File
@@ -17,3 +17,4 @@ dist/
.idea/
Thumbs.db
.DS_Store
.venv/
+5 -3
View File
@@ -1,7 +1,9 @@
#!/usr/bin/env python3
"""Octo Updater entry point."""
from config import _relocate_legacy_data
from platform_ops import enable_dpi_awareness
from ui import OctoUpdaterApp
import sys
from src.config import _relocate_legacy_data
from src.platform_ops import enable_dpi_awareness
from src.ui import OctoUpdaterApp
def main() -> None:
enable_dpi_awareness()
+31
View File
@@ -0,0 +1,31 @@
Refactor v2
Copy these four files into src/:
torrent.py
client_sync.py
client_patch.py
mpq_patches.py
UI imports:
from .client_sync import VerifyWorker, UpdateWorker
from .torrent import stop_aria2c
If the UI uses optional MPQ helpers, import them from .mpq_patches.
Validated in this archive:
- all four generated modules compile
- clear_torrent_resume_state exists in torrent.py
- run_aria2c exists in torrent.py
- stop_aria2c and parse_aria_progress exist in torrent.py
- VerifyWorker and UpdateWorker exist in client_sync.py
- patch_exe exists in client_patch.py
- download_mpq_patch exists in mpq_patches.py
The original supplied torrent.py referenced some names without defining/importing
them there. This refactor assigns those existing project symbols as follows:
.mods: MODS_REGISTRY
.config: TWEAKS_DEFAULTS, load_tweaks_config
.client: _LOCALE_ASSERT_OFFSET, get_client_version, remove_wdb
If those names currently live in different sibling modules in your project,
change those import lines to match their actual owners.
View File
View File
+3 -3
View File
@@ -18,9 +18,9 @@ import queue
from functools import cache
from pathlib import Path
from app_log import log
from config import load_config, update_config
from platform_ops import display_info
from .app_log import log
from .config import load_config, update_config
from .platform_ops import display_info, ensure_dir
def remove_wdb(client_dir: str):
"""Delete the client's WDB folder (server-data cache, safe to drop)."""
+119
View File
@@ -0,0 +1,119 @@
"""Pristine WoW.exe caching and binary tweak application."""
import math
import os
import shutil
import struct
from .app_log import log
from .config import APP_DATA_DIR
from .client import _LOCALE_ASSERT_OFFSET, locale_patches, TWEAKS_DEFAULTS, load_tweaks_config
PRISTINE_WOW_PATH = os.path.join(APP_DATA_DIR, "base-WoW.exe")
def refresh_pristine_wow(client_dir: str):
"""Cache the freshly-synced (unpatched) WoW.exe as the pristine base, so a
later re-patch always starts from clean bytes."""
exe = os.path.join(client_dir, "WoW.exe")
try:
with open(exe, "rb") as f:
f.seek(_LOCALE_ASSERT_OFFSET)
pristine = f.read(1) == b"\xa1"
if pristine:
shutil.copyfile(exe, PRISTINE_WOW_PATH)
log("Cached pristine WoW.exe base.", "dim")
except OSError:
pass
def read_pristine_wow(client_dir: str) -> bytes:
"""The clean base to patch from: the cached pristine exe if present, else
the on-disk WoW.exe."""
if os.path.exists(PRISTINE_WOW_PATH):
with open(PRISTINE_WOW_PATH, "rb") as f:
return f.read()
with open(os.path.join(client_dir, "WoW.exe"), "rb") as f:
return f.read()
def build_tweaks(buf, tweaks: dict | None = None):
if tweaks is None:
tweaks = load_tweaks_config()
fov_deg = tweaks.get("fieldOfView", TWEAKS_DEFAULTS["fieldOfView"])
fov = fov_deg * (math.pi / 180.0)
flags = struct.unpack_from("<H", buf, 0x126)[0] | 0x20
nameplate = float(tweaks.get("nameplateRange", TWEAKS_DEFAULTS["nameplateRange"]))
far_clip = float(tweaks.get("farClip", TWEAKS_DEFAULTS["farClip"]))
frill = float(tweaks.get("frillDistance", TWEAKS_DEFAULTS["frillDistance"]))
cam_dist = float(tweaks.get("cameraDistance", TWEAKS_DEFAULTS["cameraDistance"]))
snd_bg = 0x27 if tweaks.get("soundInBackground", TWEAKS_DEFAULTS["soundInBackground"]) else 0x14
always_loot = tweaks.get("alwaysAutoLoot", TWEAKS_DEFAULTS["alwaysAutoLoot"])
locale = tweaks.get("locale", TWEAKS_DEFAULTS["locale"])
# fmt: off
return [
("gameLanguage", "bytes", None, locale_patches(locale)),
("largeAddress", "uint16", 0x126, flags),
("fieldOfView", "float", 0x4089b4, fov),
("cameraDistance", "float", 0x4089a4, cam_dist),
("farClip", "float", 0x40fed8, far_clip),
("frillDistance", "float", 0x467958, frill),
("nameplateRange", "float", 0x40c448, nameplate),
("soundInBackground", "int8", 0x3a4869, snd_bg),
("alwaysAutoLoot", "bytes", None, [
(0x0c1ecf, bytes([0x75 if always_loot else 0x74])),
(0x0c2b25, bytes([0x75 if always_loot else 0x74])),
]),
# cameraSkipFix is baked into the torrent's WoW.exe, so we don't
# apply it. skillUiGateHijack and octowowUrlAllowlist below are
# baked in too, but the official launcher still applies these 2
# specific patches, so we mirror it in case the Octo devs drop them
# from WoW.exe again.
("octowowUrlAllowlist", "bytes", None, [
(0x45ccd8, bytes([
0x6f,0x63,0x74,0x6f,0x77,0x6f,0x77,0x2e,0x73,0x74,
0x00,0x00,0x00,0x00,0x00,0x00,
])),
]),
("skillUiGateHijack", "bytes", None, [
(0x002ddf90, bytes([
0x55,0x8b,0xec,0x83,0xec,0x08,0x53,0x56,0x57,0x8b,0x3d,0x60,0xab,0xce,0x00,0x83,
0xff,0xff,0x89,0x55,0xfc,0x89,0x4d,0xf8,0x74,0x79,0x8b,0x75,0x08,0x8b,0x15,0x58,
0xab,0xce,0x00,0x8b,0xc7,0x23,0xc6,0x8d,0x04,0x40,0x8b,0x4c,0x82,0x08,0xf6,0xc1,
0x01,0x8d,0x44,0x82,0x04,0x75,0x04,0x85,0xc9,0x75,0x05,0x33,0xc9,0x8d,0x49,0x00,
0xf6,0xc1,0x01,0x75,0x4e,0x85,0xc9,0x74,0x4a,0x39,0x31,0x74,0x13,0x8b,0xc7,0x23,
0xc6,0x8d,0x04,0x40,0x8d,0x04,0x82,0x8b,0x00,0x03,0xc1,0x8b,0x48,0x04,0xeb,0xe0,
0x8b,0x59,0x1c,0x8b,0x71,0x18,0x33,0xff,0x85,0xdb,0x7e,0x27,0x8d,0x64,0x24,0x00,
0x8b,0x4e,0x0c,0x8b,0x56,0x08,0x6a,0x00,0x6a,0x00,0x51,0x8b,0x4d,0xf8,0x52,0x8b,
0x55,0xfc,0xe8,0xb9,0xfd,0xff,0xff,0x84,0xc0,0x75,0x13,0x47,0x83,0xc6,0x20,0x3b,
0xfb,0x7c,0xdd,0x5f,0x5e,0x33,0xc0,0x5b,0x8b,0xe5,0x5d,0xc2,0x04,0x00,0x5f,0x8b,
0xc6,0x5e,0x5b,0x8b,0xe5,0x5d,0xc2,0x04,0x00,0x90,0x90,0x90,0x90,0x90,0x90,0x90,
])),
]),
]
# fmt: on
def patch_exe(client_dir: str, tweaks: dict | None = None, log_fn=log):
exe = os.path.join(client_dir, "WoW.exe")
if not os.path.exists(exe):
raise RuntimeError(f"WoW.exe not found in {client_dir}")
log_fn("\nApplying binary tweaks to WoW.exe…")
# Patch the pristine (unpatched) base rather than the on-disk exe, so a
# re-patch (tweak or language change) never stacks on patched bytes.
buf = bytearray(read_pristine_wow(client_dir))
for label, kind, offset, value in build_tweaks(buf, tweaks):
log_fn(f" {label}", "dim")
if kind == "float":
struct.pack_into("<f", buf, offset, value)
elif kind == "int8":
struct.pack_into("<b", buf, offset, value)
elif kind == "uint16":
struct.pack_into("<H", buf, offset, value)
elif kind == "bytes":
for off, data in value:
buf[off: off + len(data)] = data
with open(exe, "wb") as f:
f.write(buf)
log_fn("WoW.exe patched.", "ok")
+305
View File
@@ -0,0 +1,305 @@
"""High-level client verification and update workers."""
import os
import queue
import shutil
from .client import get_client_version, remove_wdb, write_config_wtf
from .client_patch import PRISTINE_WOW_PATH, patch_exe, refresh_pristine_wow
from .config import load_config, update_config
from .torrent import (
clear_torrent_resume_state,
fetch_torrent,
recover_protected_files,
run_aria2c,
shield_protected_files,
torrent_all_selection,
torrent_selection,
torrent_tree_intact,
torrent_version,
unshield_protected_files,
)
def fmt_size(value: float) -> str:
value = float(value)
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
if abs(value) < 1024.0 or unit == "TiB":
return f"{value:.0f}{unit}" if unit == "B" else f"{value:.1f}{unit}"
value /= 1024.0
def fmt_speed(value: float) -> str:
return f"{fmt_size(value)}/s"
# Legacy leftovers the current client no longer ships. Locale data folders are
# matched by name; the old patch archives by name AND exact size, so a player
# mod that reused one of these names is never deleted.
_LOCALE_DATA_DIRS = {
"enus", "engb", "encn", "entw", "kokr", "frfr", "dede", "zhcn",
"zhtw", "eses", "esmx", "ruru", "ptbr", "ptpt", "itit",
}
_LEGACY_ARCHIVES = {
"patch-6.mpq": 451195806,
"patch-7.mpq": 175256564,
"patch-8.mpq": 484649870,
"patch-9.mpq": 506808141,
"patch-a.mpq": 241751337,
}
def prune_stale_client_files(client_dir: str, files) -> list:
"""Remove legacy Data/<locale>/ folders and known old-client patch MPQs the
current torrent no longer ships. Folders are removed by name (unless the torrent
still uses them); archives only when the name AND the exact size match a known
legacy one. Returns removed names."""
data_dir = os.path.join(client_dir, "Data")
if not os.path.isdir(data_dir):
return []
# what the current torrent puts directly in Data/ (.mpq files and subdirs)
expected, used_dirs = set(), set()
for parts, _length in files:
if not parts or parts[0] != "Data":
continue
if len(parts) == 2 and parts[1].lower().endswith(".mpq"):
expected.add(parts[1].lower())
elif len(parts) >= 3:
used_dirs.add(parts[1].lower())
removed = []
for name in os.listdir(data_dir):
lc, full = name.lower(), os.path.join(data_dir, name)
if os.path.isdir(full):
if lc in _LOCALE_DATA_DIRS and lc not in used_dirs:
try:
shutil.rmtree(full)
removed.append(name + "/")
except OSError:
pass
continue
if not lc.endswith(".mpq") or lc in expected:
continue
try:
size = os.path.getsize(full)
except OSError:
continue
if _LEGACY_ARCHIVES.get(lc) == size:
try:
os.remove(full)
removed.append(name)
except OSError:
pass
return removed
class VerifyWorker:
def __init__(self, out_dir: str, log_q: queue.Queue, prog_q: queue.Queue):
self.out_dir = out_dir
self.log_q = log_q
self.prog_q = prog_q
self._cancel = False
def cancel(self):
self._cancel = True
def log(self, msg, tag=""):
self.log_q.put((msg, tag))
def progress(self, value, label=""):
self.prog_q.put((value, label))
def run(self):
try:
self.log("Checking for updates…", "acct")
raw, files = fetch_torrent()
self.log("Checking game files…", "acct")
# aria2 selects by size (a patched WoW.exe keeps the torrent's size,
# so it's never flagged); need == files missing or wrong-sized.
ignore_speech = bool(load_config().get("ignore_speech", False))
need, _missing = torrent_selection(self.out_dir, files,
ignore_speech=ignore_speech)
have_exe = os.path.exists(os.path.join(self.out_dir, "WoW.exe"))
if have_exe and not need:
self.log("Everything is up to date!", "ok")
self.log_q.put(("__UP_TO_DATE__", ""))
else:
self.log("Update available.", "acct")
self.log_q.put(("__UPDATE_NEEDED__", ""))
except Exception as e:
self.log(f"Verification failed: {e}", "err")
self.log_q.put(("__UPDATE_NEEDED__", ""))
class UpdateWorker:
def __init__(self, out_dir: str, log_q: queue.Queue, prog_q: queue.Queue,
check_integrity: bool = False, overwrite_config: bool = False):
self.out_dir = out_dir
self.log_q = log_q
self.prog_q = prog_q
self._cancel = False
# Integrity mode: aria2 hash-checks every file's pieces and repairs
# them (catches same-size corruption size-based selection misses).
self.check_integrity = check_integrity
# Write a fresh Config.wtf on a reconcile.
self.overwrite_config = overwrite_config
def cancel(self):
self._cancel = True
def log(self, msg: str, tag: str = ""):
self.log_q.put((msg, tag))
def progress(self, value: float, label: str = "", status: str | None = None):
# status (when given) updates the big status line; None leaves it as-is.
self.prog_q.put((value, label, status))
def run(self):
# The selection is recomputed here from the live torrent so an
# interrupted sync always resumes against the current file set.
try:
self.log("\nStarting client sync…\n", "acct")
self.progress(0.0, "Preparing…")
raw, files = fetch_torrent()
version = torrent_version(raw)
# aria2's saved control state pins a torrent revision and its
# completed pieces. When the torrent was re-rolled (new identity) or
# the folder changed, that state is stale — an 'info hash mismatch'
# error, or skipped re-downloads. Clear it and (for a new revision)
# drop wrong-sized files so they re-fetch clean.
cfg = load_config()
stale = (cfg.get("active_torrent_hash") != version or
cfg.get("active_client_dir") != os.path.abspath(self.out_dir))
if stale:
clear_torrent_resume_state()
update_config(lambda c: c.update({
"active_torrent_hash": version,
"active_client_dir": os.path.abspath(self.out_dir)}))
# Config.wtf is user game config, not in the torrent — (re)write it
# on a reconcile (overwrite_config), or when missing.
cfg_wtf = os.path.join(self.out_dir, "WTF", "Config.wtf")
if self.overwrite_config or not os.path.exists(cfg_wtf):
write_config_wtf(self.out_dir)
# WoW.exe on disk is patched, so an integrity check always flags it
# and re-fetches its pieces over the network. Restore the cached
# pristine base first: the check then passes with no re-download when
# the client is unchanged (aria2 still repairs it if the torrent's
# WoW.exe genuinely changed). It's re-patched after the check.
wow_path = os.path.join(self.out_dir, "WoW.exe")
if (self.check_integrity and os.path.exists(PRISTINE_WOW_PATH)
and os.path.exists(wow_path)):
shutil.copyfile(PRISTINE_WOW_PATH, wow_path)
# Put back any file a prior run shielded but couldn't restore (crash
# mid-sync), so its version isn't stranded as a .octobak.
recover_protected_files(self.out_dir, files)
# speech.MPQ is left unverified/un-updated when the user keeps a
# custom one (Settings → Ignore speech.mpq).
ignore_speech = bool(load_config().get("ignore_speech", False))
if self.check_integrity:
# Full piece-hash verify + repair of every non-mod file — aria2
# re-hashes them and re-fetches only the bad/missing pieces.
need, missing = torrent_all_selection(
self.out_dir, files, ignore_speech=ignore_speech), False
else:
need, missing = torrent_selection(
self.out_dir, files, drop_mismatched=stale,
ignore_speech=ignore_speech)
# A file the user deleted leaves its pieces marked done in the
# control file, so aria2 skips it forever — clear resume state.
if missing:
clear_torrent_resume_state()
wow_downloaded = any(files[i - 1][0] == ["WoW.exe"] for i in need)
if need:
self.log(
(f"Verifying {len(need)} file(s) via torrent…"
if self.check_integrity
else f"Syncing {len(need)} file(s) via torrent…"), "acct")
# A reconcile runs two aria2 phases: it hash-checks every file
# (p["checking"]), then fetches the bad/missing pieces. Flip the
# status from "Verifying" to "Updating" between them so it doesn't
# read "Verifying" while files are actually being updated. Only
# sent on a phase change.
_phase = {"status": None}
def _prog(p):
label = f"{fmt_size(p['done'])} / {fmt_size(p['total'])}"
if p["bps"]:
label += " • " + fmt_speed(p["bps"])
frac = p["done"] / p["total"] if p["total"] else 0.0
status = None
if self.check_integrity:
want = ("Verifying game files…" if p["checking"]
else "Updating game files…")
if want != _phase["status"]:
_phase["status"] = status = want
self.progress(min(frac, 1.0), label, status)
# Excluded files (mod-owned + kept speech.MPQ) can still be
# rewritten by aria2 through a shared torrent piece. Move them
# aside for the sync and put them back after, so the Mods tab /
# custom speech stays authoritative.
shielded = shield_protected_files(self.out_dir, files,
ignore_speech)
try:
run_aria2c(self.out_dir, select_files=need,
check_integrity=self.check_integrity,
on_progress=_prog,
should_cancel=lambda: self._cancel,
log_fn=self.log)
finally:
for name in unshield_protected_files(shielded):
self.log(f" kept mod file: {name}", "dim")
else:
self.log("All game files already present.", "dim")
if self._cancel:
self.log("\nUpdate cancelled.", "err")
self.progress(0.0, "Cancelled")
self.log_q.put(("__ERROR__", ""))
return
self.progress(1.0, "Verifying…")
if not torrent_tree_intact(self.out_dir, files,
ignore_speech=ignore_speech):
self.log("\n✗ Download incomplete — click Update to finish.", "err")
self.log_q.put(("__ERROR__", ""))
return
self.log("\nDownload complete.", "ok")
remove_wdb(self.out_dir)
# Drop legacy leftovers the current torrent no longer ships
for gone in prune_stale_client_files(self.out_dir, files):
self.log(f"Removed legacy file: {gone}", "dim")
# Cache the fresh pristine exe, then patch WoW.exe from that
# clean base — but only when the sync actually (re)downloaded it.
refresh_pristine_wow(self.out_dir)
if wow_downloaded:
self.progress(1.0, "Patching…")
patch_exe(self.out_dir, log_fn=self.log)
else:
self.log("\nWoW.exe unchanged — skipping patch.", "dim")
self.progress(1.0, "")
self.log("\n✓ Everything is up to date!", "ok")
client_ver = get_client_version(self.out_dir)
if client_ver:
self.log(f"Client version: {client_ver}", "dim")
self.log_q.put((f"__VERSION__{client_ver}", ""))
else:
self.log("Could not read client version from WoW.exe", "dim")
self.log_q.put(("__DONE__", ""))
except Exception as e:
self.log(f"\n✗ {e}", "err")
self.progress(0.0, "")
self.log_q.put(("__ERROR__", ""))
+17 -19
View File
@@ -15,8 +15,13 @@ import time
import math
import threading
import queue
import json
from functools import cache
from pathlib import Path
from .platform_ops import user_data_dir, user_config_dir
UPDATER_VERSION = "1.3.1"
SERVER = "https://octowow.st"
@@ -28,29 +33,23 @@ DOWNLOAD_TIMEOUT = 10 # seconds without any data before a transfer aborts
# next to this script — never the current working directory, which varies with
# how the app was launched. This anchors the default game folder.
if getattr(sys, "frozen", False):
APP_DIR = os.path.dirname(os.path.abspath(sys.executable))
APP_DIR = os.path.dirname(os.path.abspath(sys.executable)) # Directory where frozen .exe is executed from
#TODO : Compile into binary ? Make into AppImage ?
else:
APP_DIR = os.path.dirname(os.path.abspath(__file__))
APP_DIR = os.path.dirname(os.path.abspath(__file__)) # Directory where python3 script is executed from
def _default_app_data_dir() -> str:
base = os.environ.get("LOCALAPPDATA")
if base:
path = os.path.join(base, "OctoUpdater")
try:
os.makedirs(path, exist_ok=True)
return path
except OSError:
pass
return APP_DIR
# Set env vars
APP_DATA_DIR = user_data_dir()
APP_CONFIG_DIR = user_config_dir()
APP_DATA_DIR = _default_app_data_dir()
CONFIG_FILE = os.path.join(APP_CONFIG_DIR, "config.json") # TODO: Change this to point to ~/.config/OctoUpdater
CONFIG_FILE = os.path.join(APP_DATA_DIR, "config.json")
# TODO: Create the config directories on a linux host if they don't exist
# First-run default game folder, anchored to the app dir (not the CWD).
DEFAULT_GAME_DIR = os.path.join(APP_DIR, "OctoWoW")
DEFAULT_GAME_DIR = os.path.join(APP_DIR, "OctoWoW") # NOTE: May need to change later to point to other dir
def _relocate_legacy_data():
@@ -81,17 +80,18 @@ def load_config() -> dict:
return {}
# NOTE: writes the config , using tempfile to prevent data loss
def _atomic_write(path: str, text: str):
"""Write via a temp file + atomic rename so a crash mid-write can never
leave a truncated/corrupt file at `path`."""
tmp = path + ".tmp"
tmp = path + ".tmp"
with open(tmp, "w") as f:
f.write(text)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
# NOTE: executes write to config using thread safe lock ?
def save_config(data: dict):
with _CONFIG_LOCK:
try:
@@ -112,8 +112,6 @@ def update_config(mutator):
return cfg
def ensure_dir(path):
Path(path).mkdir(parents=True, exist_ok=True)
# ── logging ─────────────────────────────────────────────────────────────────
+4 -3
View File
@@ -21,9 +21,10 @@ from pathlib import Path
import urllib.error
import zipfile
import io
from app_log import log
from config import SERVER, UA, UPDATER_VERSION, load_config, update_config, ensure_dir
from net import ALLOWED_DOWNLOAD_HOSTS, secure_urlopen
from .app_log import log
from .config import SERVER, UA, UPDATER_VERSION, load_config, update_config
from .platform_ops import ensure_dir
from .net import ALLOWED_DOWNLOAD_HOSTS, secure_urlopen
MODS_REGISTRY = [
{
+76
View File
@@ -0,0 +1,76 @@
"""Optional MPQ content downloads using HTTPS and SHA-256 sidecars."""
import hashlib
import os
import urllib.request
from .config import UA, DOWNLOAD_TIMEOUT
from .platform_ops import ensure_dir
from .net import ALLOWED_DOWNLOAD_HOSTS, secure_urlopen
# Downloadable MPQ content patches (not shipped with the client). Each is
# a single HTTP file with a `<url>.sha256` sidecar; an update is available when
# that published sha differs from the on-disk file's.
MPQ_PATCHES = [
{
"file": "patch-O.mpq",
"name": "Octo Raid Visuals",
"description": "Adds ground markers and sounds for boss abilities in raids.",
"url": "https://dl.octowow.st/client/latest/Data/patch-O.mpq",
},
]
def mpq_patch_for(filename: str):
"""The registry entry whose file matches `filename` (case-insensitive)."""
lc = filename.lower()
return next((e for e in MPQ_PATCHES if e["file"].lower() == lc), None)
def sha256_file(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest().lower()
def fetch_mpq_sha256(url: str) -> str:
"""The published SHA256 of an MPQ patch, from its `<url>.sha256` sidecar."""
req = urllib.request.Request(url + ".sha256", headers={"User-Agent": UA})
with secure_urlopen(req, timeout=DOWNLOAD_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
return r.read().decode("ascii", "ignore").strip().split()[0].lower()
def download_mpq_patch(entry: dict, data_dir: str, on_progress=None):
"""Download the patch into <data_dir>/<file>, verifying its published
SHA256. Writes to a .part file and renames on success. on_progress(done,
total) is called as bytes arrive."""
url = entry["url"]
want = fetch_mpq_sha256(url)
ensure_dir(data_dir)
tmp = os.path.join(data_dir, entry["file"] + ".part")
h = hashlib.sha256()
req = urllib.request.Request(url, headers={"User-Agent": UA})
with secure_urlopen(req, timeout=DOWNLOAD_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
total = int(r.headers.get("Content-Length") or 0)
done = 0
with open(tmp, "wb") as f:
while True:
chunk = r.read(256 * 1024)
if not chunk:
break
f.write(chunk)
h.update(chunk)
done += len(chunk)
if on_progress:
on_progress(done, total)
if h.hexdigest().lower() != want:
try:
os.remove(tmp)
except OSError:
pass
raise RuntimeError("checksum verification failed")
os.replace(tmp, os.path.join(data_dir, entry["file"]))
-1
View File
@@ -18,7 +18,6 @@ import queue
from functools import cache
from pathlib import Path
from config import UA
+112 -1
View File
@@ -4,16 +4,67 @@ import shutil
import subprocess
import sys
from pathlib import Path
import hashlib
import platform
#from .config import APP_DATA_DIR, DOWNLOAD_TIMEOUT, ensure_dir
from .app_log import log
from .net import secure_urlopen
import urllib.request
# NOTE: Creates dir and verifies it was made
def ensure_dir(path):
Path(path).mkdir(parents=True, exist_ok=True)
# Windows: %LOCALAPPDATA%\octo-updater\
# Linux: $XDG_DATA_HOME/octo-updater or $HOME/.local/share/octo-updater
def user_data_dir() -> Path:
"""Alters which directory is used for application data. Returns Path object to directory"""
if sys.platform == "win32":
local_app_data = os.environ.get("LOCALAPPDATA")
if not local_app_data:
raise RuntimeError("LOCALAPPDATA is not set")
return Path(local_app_data)
# For linux , will first check for XDG_DATA_HOME"
local_app_data = os.environ.get("XDG_DATA_HOME")
if local_app_data:
return Path(local_app_data) / "OctoUpdater"
return Path.home() / ".local" / "share" / "OctoUpdater"
# Windows: %APPDATA%\octo-updater
# Linux: $XDG_CONFIG_HOME/octo-updater or $HOME/.config/octo-updater
def user_config_dir() -> Path:
"""Alters which directory is used for application data. Returns Path object to directory"""
if sys.platform == "win32":
local_config_data = os.environ.get("APPDATA")
if not local_config_data:
raise RuntimeError("APPDATA is not set")
return Path(local_config_data)
# For linux , will first check for XDG_CONFIG_HOME" , else set to ~/.config/octo-updater
local_config_data = os.environ.get("XDG_CONFIG_HOME")
if local_config_data:
return Path(local_config_data) / "OctoUpdater"
return Path.home() / ".config" / "OctoUpdater"
# NOTE: I think this is for opening links in the launcher ?
def open_directory(path: str) -> None:
if sys.platform == "win32":
subprocess.Popen(["explorer.exe", path])
else:
opener = shutil.which("xdg-open")
opener = shutil.which("xdg-open") # Point this to default linux browser ?
if not opener:
raise RuntimeError("xdg-open is required to open directories")
subprocess.Popen([opener, path])
# NOTE: If windows, creates ntfs junction , else , creates symlink
def create_directory_link(link: str, target: str) -> None:
if sys.platform == "win32":
flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
@@ -23,6 +74,7 @@ def create_directory_link(link: str, target: str) -> None:
return
os.symlink(target, link, target_is_directory=True)
# NOTE: allows dynamic scaling of UI window ?
def enable_dpi_awareness() -> None:
if sys.platform != "win32":
return
@@ -58,9 +110,11 @@ def request_defender_exclusion(path: str) -> bool:
return result > 32
# TODO: Find solution to launch client natively on linux , either using wine or bottles ?
def launch_executable(executable: str, cwd: str) -> None:
"""Launch the managed executable using native process semantics."""
if sys.platform != "win32":
# TODO: implement solution to run game on linux ....
raise RuntimeError("native Linux game launching is not configured")
flags = (getattr(subprocess, "DETACHED_PROCESS", 0)
| getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0))
@@ -95,3 +149,60 @@ def display_info() -> dict:
ctypes.windll.user32.EnumDisplaySettingsW(None, -1, ctypes.byref(dm))
return {"width": dm.dmPelsWidth, "height": dm.dmPelsHeight,
"refresh_rate": dm.dmDisplayFrequency}
# NOTE: Returns a path to an aria2c executable , otherwise returns none
def find_aria2c() -> str | None:
return shutil.which("aria2c")
# NOTE: Function that installs the exe for aria2c on windows devices, takes path as input , returns
# path to aria2c exe
def install_aria2c_windows( APP_DATA_DIR,
ALLOWED_DOWNLOAD_HOSTS,
log_fn) -> Path:
ARIA2C_PATH = os.path.join(APP_DATA_DIR, "aria2c.exe")
ARIA2_ZIP_SHA256 = "35f6514cc5dd7e98a87b3c4c2d25a0754b9b063dbe59bc0f22d483464f61e5b6"
ARIA2_ZIP_URL = (
"https://github.com/aria2/aria2/releases/download/"
"release-1.37.0/aria2-1.37.0-win-32bit-build1.zip"
)
#req = urllib.request.Request(ARIA2_ZIP_URL, headers={"User-Agent": UA})
req = urllib.request.Request(ARIA2_ZIP_URL)
log_fn("Fetching aria2c (one-time, ~2.5 MB)…", "acct")
with secure_urlopen(req, timeout=60, allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
data = r.read()
digest = hashlib.sha256(data).hexdigest()
if digest != ARIA2_ZIP_SHA256:
raise RuntimeError(
f"aria2 checksum mismatch (got {digest[:12]}…); refusing to run it")
import zipfile, io
with zipfile.ZipFile(io.BytesIO(data)) as zf:
name = next((n for n in zf.namelist()
if n.lower().endswith("aria2c.exe")), None)
if not name:
raise RuntimeError("aria2c.exe not found in the aria2 archive")
exe = zf.read(name)
ensure_dir(APP_DATA_DIR)
tmp = ARIA2C_PATH + ".part"
with open(tmp, "wb") as f:
f.write(exe)
os.replace(tmp, ARIA2C_PATH)
log_fn("aria2c ready.", "ok")
return ARIA2C_PATH
+8
View File
@@ -0,0 +1,8 @@
Run from the project root:
python3 -m pytest -q
These tests use pytest, tmp_path, and unittest.mock. Network access and aria2c
execution are mocked; no real torrent/MPQ downloads are performed.
Place these four files under your existing src/tests/ directory.
+26
View File
@@ -0,0 +1,26 @@
import queue
from src import app_log
def test_log_enqueues_message_and_tag(monkeypatch):
# Arrange
q = queue.Queue()
monkeypatch.setattr(app_log, "LOG_QUEUE", q)
# Act
app_log.log("ready", "ok")
# Assert
assert q.get_nowait() == ("ready", "ok")
def test_log_uses_empty_tag_by_default(monkeypatch):
# Arrange
q = queue.Queue()
monkeypatch.setattr(app_log, "LOG_QUEUE", q)
# Act
app_log.log("ready")
# Assert
assert q.get_nowait() == ("ready", "")
+152
View File
@@ -0,0 +1,152 @@
import math
from pathlib import Path
from unittest.mock import Mock
import pytest
from src import client
def test_remove_wdb_does_nothing_when_directory_is_absent(tmp_path, monkeypatch):
log = Mock()
monkeypatch.setattr(client, "log", log)
client.remove_wdb(str(tmp_path))
log.assert_not_called()
def test_remove_wdb_removes_existing_cache(tmp_path, monkeypatch):
wdb = tmp_path / "WDB"
wdb.mkdir()
(wdb / "cache.bin").write_bytes(b"x")
log = Mock()
monkeypatch.setattr(client, "log", log)
client.remove_wdb(str(tmp_path))
assert not wdb.exists()
log.assert_called_once_with("WDB cache cleared.", "dim")
def test_remove_wdb_logs_failure_without_raising(tmp_path, monkeypatch):
(tmp_path / "WDB").mkdir()
monkeypatch.setattr(client.shutil, "rmtree", Mock(side_effect=PermissionError("locked")))
log = Mock()
monkeypatch.setattr(client, "log", log)
client.remove_wdb(str(tmp_path))
assert "locked" in log.call_args.args[0]
assert log.call_args.args[1] == "err"
def test_get_client_version_returns_empty_when_executable_is_missing(tmp_path):
assert client.get_client_version(str(tmp_path)) == ""
def test_get_client_version_reads_valid_version_and_build(tmp_path):
exe = tmp_path / "WoW.exe"
data = bytearray(0x00437c0a)
data[0x00437bfc:0x00437c00] = b"5875"
data[0x00437c04:0x00437c0a] = b"1.12.1"
exe.write_bytes(data)
assert client.get_client_version(str(tmp_path)) == "1.12.1 (5875)"
def test_get_client_version_rejects_invalid_version_bytes(tmp_path):
exe = tmp_path / "WoW.exe"
data = bytearray(0x00437c0a)
data[0x00437bfc:0x00437c00] = b"5875"
data[0x00437c04:0x00437c0a] = b"broken"
exe.write_bytes(data)
assert client.get_client_version(str(tmp_path)) == ""
@pytest.mark.parametrize(("value", "expected"), [
(0, "0 KB"), (1023, "0 KB"), (1024, "1 KB"),
(1024**2, "1.0 MB"), (1024**3, "1.00 GB"),
])
def test_fmt_size_formats_boundaries(value, expected):
assert client.fmt_size(value) == expected
def test_fmt_size_truncates_gigabytes_instead_of_rounding_up():
assert client.fmt_size(8.7998 * 1024**3) == "8.79 GB"
@pytest.mark.parametrize(("value", "expected"), [
(0, "0 KB/s"), (1024, "1 KB/s"),
(1024**2 - 1, "1024 KB/s"), (1024**2, "1.0 MB/s"),
])
def test_fmt_speed_formats_boundaries(value, expected):
assert client.fmt_speed(value) == expected
def test_locale_patches_defaults_unknown_locale_to_english():
assert client.locale_patches("not-a-locale") == client.locale_patches("enUS")
def test_locale_patches_renames_reused_slot_for_russian():
patches = client.locale_patches("ruRU")
assert patches[-1] == (client._LOCALE_NAME_OFFSET(5), b"ruRU")
def test_update_config_wtf_creates_config_when_missing(tmp_path, monkeypatch):
writer = Mock()
monkeypatch.setattr(client, "write_config_wtf", writer)
tweaks = {"farClip": 500}
client.update_config_wtf(str(tmp_path), tweaks)
writer.assert_called_once_with(str(tmp_path), tweaks)
def test_update_config_wtf_updates_existing_values_and_preserves_unrelated_lines(tmp_path, monkeypatch):
cfg_dir = tmp_path / "WTF"
cfg_dir.mkdir()
cfg = cfg_dir / "Config.wtf"
cfg.write_text('SET farClip "100"\nSET untouched "yes"\n', encoding="utf-8")
monkeypatch.setattr(client, "log", Mock())
client.update_config_wtf(str(tmp_path), {"farClip": 777, "cameraDistance": 50,
"nameplateRange": 41, "fieldOfView": 110, "soundInBackground": True})
text = cfg.read_text(encoding="utf-8")
assert 'SET farClip "777"' in text
assert 'SET untouched "yes"' in text
assert 'SET CameraDistanceMax "50"' in text
def test_fov_default_for_display_uses_lowest_reference_for_narrow_display(monkeypatch):
monkeypatch.setattr(client, "_get_display_info_safe", lambda: {"width": 1024, "height": 1000})
assert client.fov_default_for_display() == 90
def test_fov_default_for_display_uses_highest_reference_for_ultrawide(monkeypatch):
monkeypatch.setattr(client, "_get_display_info_safe", lambda: {"width": 5120, "height": 1080})
assert client.fov_default_for_display() == 180
def test_fov_default_for_display_falls_back_to_16_9_when_display_lookup_fails(monkeypatch):
monkeypatch.setattr(client, "_get_display_info_safe", Mock(side_effect=RuntimeError("display unavailable")))
assert client.fov_default_for_display() == 110
def test_load_tweaks_config_merges_stored_values_with_defaults(monkeypatch):
monkeypatch.setattr(client, "load_config", lambda: {"tweaks": {"farClip": 999}})
monkeypatch.setattr(client, "fov_default_for_display", lambda: 150)
result = client.load_tweaks_config()
assert result["farClip"] == 999
assert result["fieldOfView"] == 150
assert result["cameraDistance"] == client.TWEAKS_DEFAULTS["cameraDistance"]
def test_save_tweaks_config_stores_values(monkeypatch):
cfg = {}
monkeypatch.setattr(client, "update_config", lambda fn: fn(cfg))
values = {"farClip": 900}
client.save_tweaks_config(values)
assert cfg["tweaks"] == values
+89
View File
@@ -0,0 +1,89 @@
import struct
from unittest.mock import patch
from src import client_patch
def test_read_pristine_prefers_cached_copy(tmp_path):
cache = tmp_path / "base-WoW.exe"
cache.write_bytes(b"cached")
client = tmp_path / "client"
client.mkdir()
(client / "WoW.exe").write_bytes(b"live")
with patch.object(client_patch, "PRISTINE_WOW_PATH", str(cache)):
assert client_patch.read_pristine_wow(str(client)) == b"cached"
def test_read_pristine_falls_back_to_live_exe(tmp_path):
client = tmp_path / "client"
client.mkdir()
(client / "WoW.exe").write_bytes(b"live")
with patch.object(client_patch, "PRISTINE_WOW_PATH",
str(tmp_path / "missing")):
assert client_patch.read_pristine_wow(str(client)) == b"live"
def test_refresh_pristine_wow_copies_when_assert_byte_matches(tmp_path):
client = tmp_path / "client"
client.mkdir()
exe = client / "WoW.exe"
exe.write_bytes(b"\xa1rest")
cache = tmp_path / "base.exe"
with patch.object(client_patch, "_LOCALE_ASSERT_OFFSET", 0), \
patch.object(client_patch, "PRISTINE_WOW_PATH", str(cache)):
client_patch.refresh_pristine_wow(str(client))
assert cache.read_bytes() == b"\xa1rest"
def test_build_tweaks_contains_expected_core_tweaks():
# Large enough for the uint16 read at 0x126.
buf = bytearray(0x200)
tweaks = {
"fieldOfView": 90,
"nameplateRange": 40,
"farClip": 500,
"frillDistance": 100,
"cameraDistance": 30,
"soundInBackground": True,
"alwaysAutoLoot": True,
"locale": "enUS",
}
with patch.object(client_patch, "locale_patches", return_value=[]):
result = client_patch.build_tweaks(buf, tweaks)
by_name = {item[0]: item for item in result}
assert "fieldOfView" in by_name
assert "largeAddress" in by_name
assert "alwaysAutoLoot" in by_name
assert by_name["largeAddress"][3] & 0x20
def test_patch_exe_writes_supported_patch_types(tmp_path):
client = tmp_path / "client"
client.mkdir()
exe = client / "WoW.exe"
exe.write_bytes(b"\x00" * 32)
fake_tweaks = [
("f", "float", 0, 1.5),
("i8", "int8", 4, 7),
("u16", "uint16", 6, 0x1234),
("raw", "bytes", None, [(8, b"AB")]),
]
with patch.object(client_patch, "read_pristine_wow",
return_value=b"\x00" * 32), \
patch.object(client_patch, "build_tweaks",
return_value=fake_tweaks):
client_patch.patch_exe(str(client), log_fn=lambda *a: None)
data = exe.read_bytes()
assert struct.unpack_from("<f", data, 0)[0] == 1.5
assert struct.unpack_from("<b", data, 4)[0] == 7
assert struct.unpack_from("<H", data, 6)[0] == 0x1234
assert data[8:10] == b"AB"
+79
View File
@@ -0,0 +1,79 @@
import os
import queue
from unittest.mock import Mock, patch
from src import client_sync
def test_fmt_size_and_speed():
assert client_sync.fmt_size(1024) == "1.0KiB"
assert client_sync.fmt_speed(1024) == "1.0KiB/s"
def test_prune_stale_client_files_removes_known_archive(tmp_path):
data = tmp_path / "Data"
data.mkdir()
p = data / "patch-6.mpq"
legacy_size = client_sync._LEGACY_ARCHIVES["patch-6.mpq"]
p.write_bytes(b"x")
# Avoid allocating a 451 MB fixture.
with patch.object(os.path, "getsize", return_value=legacy_size):
removed = client_sync.prune_stale_client_files(str(tmp_path), [])
assert "patch-6.mpq" in removed
assert not p.exists()
def test_prune_keeps_archive_when_current_torrent_contains_it(tmp_path):
data = tmp_path / "Data"
data.mkdir()
p = data / "patch-6.mpq"
p.write_bytes(b"x")
files = [(["Data", "patch-6.mpq"], 1)]
removed = client_sync.prune_stale_client_files(str(tmp_path), files)
assert removed == []
assert p.exists()
def test_verify_worker_reports_up_to_date(tmp_path):
(tmp_path / "WoW.exe").write_bytes(b"x")
log_q, prog_q = queue.Queue(), queue.Queue()
worker = client_sync.VerifyWorker(str(tmp_path), log_q, prog_q)
with patch.object(client_sync, "fetch_torrent",
return_value=(b"raw", [(["WoW.exe"], 1)])), \
patch.object(client_sync, "torrent_selection",
return_value=([], False)), \
patch.object(client_sync, "load_config", return_value={}):
worker.run()
messages = []
while not log_q.empty():
messages.append(log_q.get()[0])
assert "__UP_TO_DATE__" in messages
def test_verify_worker_reports_update_needed(tmp_path):
log_q, prog_q = queue.Queue(), queue.Queue()
worker = client_sync.VerifyWorker(str(tmp_path), log_q, prog_q)
with patch.object(client_sync, "fetch_torrent",
return_value=(b"raw", [(["WoW.exe"], 1)])), \
patch.object(client_sync, "torrent_selection",
return_value=([1], True)), \
patch.object(client_sync, "load_config", return_value={}):
worker.run()
messages = []
while not log_q.empty():
messages.append(log_q.get()[0])
assert "__UPDATE_NEEDED__" in messages
def test_update_worker_cancel_sets_flag(tmp_path):
worker = client_sync.UpdateWorker(
str(tmp_path), queue.Queue(), queue.Queue()
)
assert worker._cancel is False
worker.cancel()
assert worker._cancel is True
File diff suppressed because it is too large Load Diff
+221
View File
@@ -0,0 +1,221 @@
import errno
import urllib.error
import zipfile
from unittest.mock import Mock
import pytest
from src import mods
@pytest.mark.parametrize(("value", "expected"), [
("v1.2.0", (1, 2, 0)), ("V2.5-beta3", (2, 53)),
("", (0,)), ("release", (0,)),
])
def test_parse_version_extracts_numeric_components(value, expected):
assert mods._parse_version(value) == expected
def test_updater_update_available_returns_false_without_latest_tag():
assert mods.updater_update_available("") is False
def test_updater_update_available_zero_pads_equivalent_versions(monkeypatch):
monkeypatch.setattr(mods, "UPDATER_VERSION", "1.2.0")
assert mods.updater_update_available("v1.2") is False
def test_updater_update_available_detects_newer_version(monkeypatch):
monkeypatch.setattr(mods, "UPDATER_VERSION", "1.2.0")
assert mods.updater_update_available("v1.2.1") is True
def test_pick_asset_returns_first_matching_asset():
assets = [{"name": "mod.txt"}, {"name": "mod.zip"}]
assert mods._pick_asset(assets, "*.zip", None) == assets[1]
def test_pick_asset_prefers_name_without_excluded_text():
assets = [{"name": "mod-dxvk.zip"}, {"name": "mod.zip"}]
assert mods._pick_asset(assets, "*.zip", "-dxvk") == assets[1]
def test_pick_asset_returns_none_when_nothing_matches():
assert mods._pick_asset([{"name": "readme.txt"}], "*.zip", None) is None
def test_release_version_uses_tag_by_default():
mod = {"source": {"kind": "github_release"}}
assert mods._release_version(mod, {"tag_name": "v2.0"}) == "v2.0"
def test_release_version_can_extract_version_from_asset_name():
mod = {"source": {"version_from": "asset", "asset_pattern": "*.zip", "prefer_no": None}}
rel = {"assets": [{"name": "SuperWoW-1.4.2.zip"}], "tag_name": "static"}
assert mods._release_version(mod, rel) == "1.4.2"
def test_slim_release_discards_unused_release_fields():
rel = {"tag_name": "v1", "body": "large", "assets": [{"name": "x.zip", "size": 12,
"browser_download_url": "https://example/x", "extra": "drop"}]}
assert mods._slim_release(rel) == {"tag_name": "v1", "assets": [{"name": "x.zip", "size": 12,
"browser_download_url": "https://example/x"}]}
def test_add_dll_creates_dlls_file(tmp_path):
mods.add_dll(str(tmp_path), "Test.dll")
assert (tmp_path / "dlls.txt").read_text() == "Test.dll\n"
def test_add_dll_does_not_add_case_insensitive_duplicate(tmp_path):
path = tmp_path / "dlls.txt"
path.write_text("Test.dll\n")
mods.add_dll(str(tmp_path), "test.DLL")
assert path.read_text() == "Test.dll\n"
def test_remove_dll_removes_named_entry_case_insensitively(tmp_path):
path = tmp_path / "dlls.txt"
path.write_text("One.dll\nTwo.dll\n")
mods.remove_dll(str(tmp_path), "ONE.DLL")
assert path.read_text() == "Two.dll\n"
def test_remove_dll_deletes_empty_dlls_file(tmp_path):
path = tmp_path / "dlls.txt"
path.write_text("One.dll\n")
mods.remove_dll(str(tmp_path), "one.dll")
assert not path.exists()
def test_remove_dll_does_nothing_when_file_is_missing(tmp_path):
mods.remove_dll(str(tmp_path), "missing.dll")
assert not (tmp_path / "dlls.txt").exists()
def test_mod_installed_files_present_requires_nonempty_installed_file_list(tmp_path, monkeypatch):
monkeypatch.setattr(mods, "load_config", lambda: {"mods": {"m": {"installed_files": []}}})
assert mods.mod_installed_files_present({"id": "m"}, str(tmp_path)) is False
def test_mod_installed_files_present_returns_true_when_every_file_exists(tmp_path, monkeypatch):
(tmp_path / "a.dll").write_text("")
(tmp_path / "b.dll").write_text("")
monkeypatch.setattr(mods, "load_config", lambda: {"mods": {"m": {"installed_files": ["a.dll", "b.dll"]}}})
assert mods.mod_installed_files_present({"id": "m"}, str(tmp_path)) is True
@pytest.mark.parametrize("kind", ["direct_file", "direct_tar"])
def test_mod_supports_update_check_rejects_pinned_direct_sources(kind):
assert mods.mod_supports_update_check({"source": {"kind": kind}}) is False
def test_mod_update_available_requires_enabled_mod():
mod = {"source": {"kind": "github_release"}}
state = {"enabled": False, "installed_version": "1"}
assert mods.mod_update_available(mod, state, {"latest_version": "2"}) is False
def test_mod_update_available_respects_ignore_updates():
mod = {"source": {"kind": "github_release"}}
state = {"enabled": True, "ignore_updates": True, "installed_version": "1"}
assert mods.mod_update_available(mod, state, {"latest_version": "2"}) is False
def test_mod_update_available_detects_different_live_version():
mod = {"source": {"kind": "github_release"}}
state = {"enabled": True, "installed_version": "1"}
assert mods.mod_update_available(mod, state, {"latest_version": "2"}) is True
def test_same_git_repo_ignores_case_trailing_slash_and_dot_git():
assert mods._same_git_repo("https://github.com/Owner/Repo.git/", "https://github.com/owner/repo") is True
@pytest.mark.parametrize("url", [
"https://github.com/owner/repo", "https://sub.gitlab.com/owner/repo",
"https://codeberg.org/owner/repo",
])
def test_is_allowed_git_url_accepts_supported_https_hosts(url):
assert mods.is_allowed_git_url(url) is True
@pytest.mark.parametrize("url", ["http://github.com/a/b", "https://example.com/a/b", "not a url"])
def test_is_allowed_git_url_rejects_unsupported_urls(url):
assert mods.is_allowed_git_url(url) is False
def test_slim_addon_catalog_keeps_only_supported_fields():
catalog = [{"name": "A", "git": "g", "branch": "main", "ref": None, "description": "d",
"extra": 1, "toc": {"Title": "T", "Notes": "N", "Interface": "11200", "Other": "drop"}}]
result = mods._slim_addon_catalog(catalog)
assert result == [{"name": "A", "git": "g", "branch": "main", "ref": None, "description": "d",
"toc": {"Title": "T", "Notes": "N", "Interface": "11200"}}]
def test_read_toc_file_parses_metadata_and_ignores_other_lines(tmp_path):
path = tmp_path / "Addon.toc"
path.write_text("\ufeff## Title: My Addon\nfile.lua\n## Interface: 11200\n", encoding="utf-8")
assert mods.read_toc_file(str(path)) == {"Title": "My Addon", "Interface": "11200"}
def test_read_toc_file_returns_empty_dict_when_missing(tmp_path):
assert mods.read_toc_file(str(tmp_path / "missing.toc")) == {}
def test_parse_wow_colored_splits_colored_and_plain_segments():
text = "plain |cffff0000red|r normal"
assert mods.parse_wow_colored(text) == [("plain ", None), ("red", "#ff0000"), (" normal", None)]
def test_strip_wow_colors_removes_color_escape_sequences():
assert mods.strip_wow_colors("A |cff00ff00green|r B") == "A green B"
def test_git_parts_parses_github_dot_git_url():
kind, repo_url, owner, repo, api = mods._git_parts("https://github.com/Owner/Repo.git")
assert (kind, repo_url, owner, repo) == ("github", "https://github.com/Owner/Repo", "Owner", "Repo")
assert api == mods.GITHUB_API
def test_git_parts_parses_prefixed_gitea_url():
kind, repo_url, owner, repo, api = mods._git_parts("https://octowow.st/git/Microchip/octo-updater")
assert kind == "gitea"
assert repo_url == "https://octowow.st/git/Microchip/octo-updater"
assert api == "https://octowow.st/git/api/v1"
def test_git_parts_rejects_url_without_owner_and_repo():
with pytest.raises(ValueError, match="Unsupported git URL"):
mods._git_parts("https://github.com/onlyone")
def test_describe_net_error_maps_http_403_to_rate_limit_message():
e = urllib.error.HTTPError("https://api.github.com/x", 403, "Forbidden", {}, None)
assert "rate limit" in mods._describe_net_error(e)
def test_describe_net_error_maps_http_404_to_not_found_message():
e = urllib.error.HTTPError("https://api.github.com/x", 404, "Not Found", {}, None)
assert mods._describe_net_error(e) == "repository or branch not found"
def test_describe_install_error_maps_antivirus_like_os_errors():
e = OSError(errno.EACCES, "denied")
assert "antivirus" in mods.describe_install_error(e).lower()
def test_describe_install_error_maps_bad_zip_file():
assert "corrupted" in mods.describe_install_error(zipfile.BadZipFile()).lower()
def test_addon_cached_sha_returns_cached_value_without_network(monkeypatch):
key = "https://github.com/a/b#main"
monkeypatch.setattr(mods, "load_config", lambda: {"addon_sha_cache": {key: {"sha": "abc"}}})
assert mods.addon_cached_sha("https://github.com/a/b", branch="main") == "abc"
def test_addon_zip_url_builds_github_archive_url():
assert mods.addon_zip_url("https://github.com/a/b", "abc") == "https://github.com/a/b/archive/abc.zip"
def test_addon_zip_url_builds_gitlab_archive_url():
assert mods.addon_zip_url("https://gitlab.com/a/b", "abc") == "https://gitlab.com/a/b/-/archive/abc/b-abc.zip"
+68
View File
@@ -0,0 +1,68 @@
import hashlib
from unittest.mock import Mock, patch
import pytest
from src import mpq_patches
def response_cm(data: bytes, length=None):
response = Mock()
response.read.side_effect = [data, b""] if length is not None else None
if length is None:
response.read.return_value = data
response.headers = {"Content-Length": str(length)} if length is not None else {}
cm = Mock()
cm.__enter__ = Mock(return_value=response)
cm.__exit__ = Mock(return_value=False)
return cm
def test_mpq_patch_for_case_insensitive():
assert mpq_patches.mpq_patch_for("PATCH-o.MPQ")["file"] == "patch-O.mpq"
assert mpq_patches.mpq_patch_for("missing.mpq") is None
def test_sha256_file(tmp_path):
p = tmp_path / "x"
p.write_bytes(b"abc")
assert mpq_patches.sha256_file(str(p)) == hashlib.sha256(b"abc").hexdigest()
def test_fetch_mpq_sha256_parses_sidecar():
digest = "a" * 64
with patch.object(
mpq_patches, "secure_urlopen",
return_value=response_cm((digest + " patch.mpq\n").encode())
):
assert mpq_patches.fetch_mpq_sha256("https://example/x") == digest
def test_download_mpq_patch_success(tmp_path):
data = b"payload"
digest = hashlib.sha256(data).hexdigest()
entry = {"url": "https://example/x", "file": "patch.mpq"}
progress = Mock()
with patch.object(mpq_patches, "fetch_mpq_sha256", return_value=digest), \
patch.object(mpq_patches, "secure_urlopen",
return_value=response_cm(data, len(data))):
mpq_patches.download_mpq_patch(entry, str(tmp_path), progress)
assert (tmp_path / "patch.mpq").read_bytes() == data
assert not (tmp_path / "patch.mpq.part").exists()
progress.assert_called()
def test_download_mpq_patch_bad_checksum_removes_part(tmp_path):
data = b"payload"
entry = {"url": "https://example/x", "file": "patch.mpq"}
with patch.object(mpq_patches, "fetch_mpq_sha256", return_value="0" * 64), \
patch.object(mpq_patches, "secure_urlopen",
return_value=response_cm(data, len(data))):
with pytest.raises(RuntimeError, match="checksum"):
mpq_patches.download_mpq_patch(entry, str(tmp_path))
assert not (tmp_path / "patch.mpq").exists()
assert not (tmp_path / "patch.mpq.part").exists()
+62
View File
@@ -0,0 +1,62 @@
import urllib.request
from unittest.mock import Mock
import pytest
from src import net
def test_check_url_accepts_https_when_no_allowlist():
net._check_url("https://example.com/file", None)
def test_check_url_rejects_http():
with pytest.raises(RuntimeError, match="non-HTTPS"):
net._check_url("http://example.com/file", None)
def test_check_url_accepts_allowlisted_host_case_insensitively():
net._check_url("https://EXAMPLE.COM/file", {"example.com"})
def test_check_url_rejects_host_not_in_allowlist():
with pytest.raises(RuntimeError, match="unexpected host"):
net._check_url("https://evil.example/file", {"example.com"})
def test_check_url_does_not_treat_subdomain_as_exact_allowlist_match():
with pytest.raises(RuntimeError, match="unexpected host"):
net._check_url("https://cdn.example.com/file", {"example.com"})
def test_secure_urlopen_opens_valid_url_with_timeout(monkeypatch):
# Arrange
response = object()
opener = Mock()
opener.open.return_value = response
monkeypatch.setattr(net, "_SECURE_OPENER", opener)
# Act
result = net.secure_urlopen("https://example.com/file", 7, {"example.com"})
# Assert
assert result is response
opener.open.assert_called_once_with("https://example.com/file", timeout=7)
def test_secure_urlopen_validates_request_full_url_before_opening(monkeypatch):
# Arrange
opener = Mock()
monkeypatch.setattr(net, "_SECURE_OPENER", opener)
req = urllib.request.Request("http://example.com/file")
# Act / Assert
with pytest.raises(RuntimeError, match="non-HTTPS"):
net.secure_urlopen(req, 5)
opener.open.assert_not_called()
def test_redirect_handler_rejects_https_to_http_downgrade():
handler = net._HttpsOnlyRedirectHandler()
req = urllib.request.Request("https://example.com/start")
with pytest.raises(RuntimeError, match="non-HTTPS"):
handler.redirect_request(req, None, 302, "Found", {}, "http://example.com/end")
+15
View File
@@ -0,0 +1,15 @@
from unittest.mock import Mock
import octo_updater
def test_main_prepares_platform_and_data_before_starting_ui(monkeypatch):
calls = []
monkeypatch.setattr(octo_updater, "enable_dpi_awareness", lambda: calls.append("dpi"))
monkeypatch.setattr(octo_updater, "_relocate_legacy_data", lambda: calls.append("relocate"))
app = Mock()
monkeypatch.setattr(octo_updater, "OctoUpdaterApp", lambda: calls.append("construct") or app)
octo_updater.main()
assert calls == ["dpi", "relocate", "construct"]
app.mainloop.assert_called_once_with()
+783
View File
@@ -0,0 +1,783 @@
import io
import os
import zipfile
from pathlib import Path
from unittest.mock import Mock, call
import pytest
from src import platform_ops
EXPECTED_ARIA2_SHA256 = (
"35f6514cc5dd7e98a87b3c4c2d25a075"
"4b9b063dbe59bc0f22d483464f61e5b6"
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def make_aria2_zip(
exe_data=b"fake aria2c executable",
exe_name="aria2-1.37.0-win-32bit-build1/aria2c.exe",
):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as zf:
zf.writestr(exe_name, exe_data)
return buffer.getvalue()
def make_response(data):
response = Mock()
response.read.return_value = data
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
return response
def allow_fake_zip_checksum(monkeypatch):
digest = Mock()
digest.hexdigest.return_value = EXPECTED_ARIA2_SHA256
sha256 = Mock(return_value=digest)
monkeypatch.setattr(
platform_ops.hashlib,
"sha256",
sha256,
)
return sha256
# ---------------------------------------------------------------------------
# ensure_dir
# ---------------------------------------------------------------------------
def test_ensure_dir_creates_directory(tmp_path):
target = tmp_path / "test-directory"
platform_ops.ensure_dir(target)
assert target.is_dir()
def test_ensure_dir_creates_parent_directories(tmp_path):
target = tmp_path / "one" / "two" / "three"
platform_ops.ensure_dir(target)
assert target.is_dir()
def test_ensure_dir_existing_directory_does_not_fail(tmp_path):
target = tmp_path / "existing"
target.mkdir()
platform_ops.ensure_dir(target)
assert target.is_dir()
def test_ensure_dir_accepts_string_path(tmp_path):
target = tmp_path / "string-path"
platform_ops.ensure_dir(str(target))
assert target.is_dir()
# ---------------------------------------------------------------------------
# find_aria2c
# ---------------------------------------------------------------------------
def test_find_aria2c_returns_discovered_path(monkeypatch):
expected = "/usr/bin/aria2c"
which = Mock(return_value=expected)
monkeypatch.setattr(platform_ops.shutil, "which", which)
result = platform_ops.find_aria2c()
assert result == expected
which.assert_called_once_with("aria2c")
def test_find_aria2c_returns_none_when_missing(monkeypatch):
which = Mock(return_value=None)
monkeypatch.setattr(platform_ops.shutil, "which", which)
result = platform_ops.find_aria2c()
assert result is None
which.assert_called_once_with("aria2c")
def test_find_aria2c_returns_windows_path_unchanged(monkeypatch):
expected = r"C:\Program Files\aria2\aria2c.exe"
monkeypatch.setattr(
platform_ops.shutil,
"which",
Mock(return_value=expected),
)
assert platform_ops.find_aria2c() == expected
# ---------------------------------------------------------------------------
# install_aria2c_windows
# ---------------------------------------------------------------------------
def test_install_aria2c_windows_installs_executable(
tmp_path,
monkeypatch,
):
exe_data = b"fake aria2 executable"
archive = make_aria2_zip(exe_data)
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
result = platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
expected = tmp_path / "aria2c.exe"
assert Path(result) == expected
assert expected.read_bytes() == exe_data
def test_install_aria2c_windows_returns_aria2_path(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
result = platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert Path(result) == tmp_path / "aria2c.exe"
def test_install_aria2c_windows_uses_expected_url(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
captured = {}
def fake_urlopen(req, **kwargs):
captured["url"] = req.full_url
return make_response(archive)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
fake_urlopen,
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert captured["url"] == (
"https://github.com/aria2/aria2/releases/download/"
"release-1.37.0/aria2-1.37.0-win-32bit-build1.zip"
)
def test_install_aria2c_windows_uses_60_second_timeout(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
urlopen = Mock(return_value=make_response(archive))
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
urlopen,
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
_, kwargs = urlopen.call_args
assert kwargs["timeout"] == 60
def test_install_aria2c_windows_passes_allowed_hosts(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
urlopen = Mock(return_value=make_response(archive))
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
urlopen,
)
hosts = {
"github.com",
"objects.githubusercontent.com",
}
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS=hosts,
log_fn=Mock(),
)
_, kwargs = urlopen.call_args
assert kwargs["allowed_hosts"] is hosts
def test_install_aria2c_windows_reads_response(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
response = make_response(archive)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=response),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
response.read.assert_called_once_with()
def test_install_aria2c_windows_hashes_downloaded_archive(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
sha256 = allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
sha256.assert_called_once_with(archive)
def test_install_aria2c_windows_rejects_bad_checksum(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
digest = Mock()
digest.hexdigest.return_value = "0" * 64
monkeypatch.setattr(
platform_ops.hashlib,
"sha256",
Mock(return_value=digest),
)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
with pytest.raises(
RuntimeError,
match="aria2 checksum mismatch",
):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert not (tmp_path / "aria2c.exe").exists()
def test_install_aria2c_windows_checksum_error_contains_digest_prefix(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
bad_digest = "123456789abc" + ("0" * 52)
digest = Mock()
digest.hexdigest.return_value = bad_digest
monkeypatch.setattr(
platform_ops.hashlib,
"sha256",
Mock(return_value=digest),
)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
with pytest.raises(
RuntimeError,
match="123456789abc",
):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
def test_install_aria2c_windows_rejects_archive_without_executable(
tmp_path,
monkeypatch,
):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as zf:
zf.writestr("README.txt", b"no executable")
archive = buffer.getvalue()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
with pytest.raises(
RuntimeError,
match="aria2c.exe not found",
):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
def test_install_aria2c_windows_finds_nested_executable(
tmp_path,
monkeypatch,
):
exe_data = b"nested aria2 executable"
archive = make_aria2_zip(
exe_data=exe_data,
exe_name="release/bin/aria2c.exe",
)
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert (
tmp_path / "aria2c.exe"
).read_bytes() == exe_data
def test_install_aria2c_windows_executable_search_is_case_insensitive(
tmp_path,
monkeypatch,
):
exe_data = b"uppercase executable"
archive = make_aria2_zip(
exe_data=exe_data,
exe_name="release/ARIA2C.EXE",
)
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert (
tmp_path / "aria2c.exe"
).read_bytes() == exe_data
def test_install_aria2c_windows_calls_ensure_dir(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
real_ensure_dir = platform_ops.ensure_dir
ensure_dir = Mock(side_effect=real_ensure_dir)
monkeypatch.setattr(
platform_ops,
"ensure_dir",
ensure_dir,
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
ensure_dir.assert_called_once_with(tmp_path)
def test_install_aria2c_windows_ensure_dir_creates_destination(
tmp_path,
monkeypatch,
):
destination = tmp_path / "nested" / "app-data"
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=destination,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert destination.is_dir()
assert (destination / "aria2c.exe").is_file()
def test_install_aria2c_windows_writes_part_file_before_replace(
tmp_path,
monkeypatch,
):
exe_data = b"aria2 executable"
archive = make_aria2_zip(exe_data)
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
real_replace = os.replace
observed = {}
def inspect_replace(src, dst):
observed["src"] = src
observed["dst"] = dst
observed["part_data"] = Path(src).read_bytes()
return real_replace(src, dst)
monkeypatch.setattr(
platform_ops.os,
"replace",
inspect_replace,
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert Path(observed["src"]) == tmp_path / "aria2c.exe.part"
assert Path(observed["dst"]) == tmp_path / "aria2c.exe"
assert observed["part_data"] == exe_data
def test_install_aria2c_windows_removes_part_file_after_success(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert not (tmp_path / "aria2c.exe.part").exists()
def test_install_aria2c_windows_replaces_existing_executable(
tmp_path,
monkeypatch,
):
executable = tmp_path / "aria2c.exe"
executable.write_bytes(b"old")
archive = make_aria2_zip(
exe_data=b"new",
)
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
assert executable.read_bytes() == b"new"
def test_install_aria2c_windows_logs_download_and_success(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
logger = Mock()
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=logger,
)
assert logger.call_args_list == [
call(
"Fetching aria2c (one-time, ~2.5 MB)…",
"acct",
),
call(
"aria2c ready.",
"ok",
),
]
def test_install_aria2c_windows_does_not_log_ready_after_checksum_failure(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
digest = Mock()
digest.hexdigest.return_value = "0" * 64
monkeypatch.setattr(
platform_ops.hashlib,
"sha256",
Mock(return_value=digest),
)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
logger = Mock()
with pytest.raises(RuntimeError):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=logger,
)
assert call("aria2c ready.", "ok") not in logger.call_args_list
def test_install_aria2c_windows_propagates_network_failure(
tmp_path,
monkeypatch,
):
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(
side_effect=OSError(
"network unavailable"
)
),
)
with pytest.raises(
OSError,
match="network unavailable",
):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
def test_install_aria2c_windows_rejects_invalid_zip(
tmp_path,
monkeypatch,
):
invalid_zip = b"this is not a zip archive"
allow_fake_zip_checksum(monkeypatch)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(invalid_zip)),
)
with pytest.raises(zipfile.BadZipFile):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
def test_install_aria2c_windows_does_not_call_ensure_dir_on_bad_checksum(
tmp_path,
monkeypatch,
):
archive = make_aria2_zip()
digest = Mock()
digest.hexdigest.return_value = "0" * 64
monkeypatch.setattr(
platform_ops.hashlib,
"sha256",
Mock(return_value=digest),
)
monkeypatch.setattr(
platform_ops,
"secure_urlopen",
Mock(return_value=make_response(archive)),
)
ensure_dir = Mock()
monkeypatch.setattr(
platform_ops,
"ensure_dir",
ensure_dir,
)
with pytest.raises(RuntimeError):
platform_ops.install_aria2c_windows(
APP_DATA_DIR=tmp_path,
ALLOWED_DOWNLOAD_HOSTS={"github.com"},
log_fn=Mock(),
)
ensure_dir.assert_not_called()
+558
View File
@@ -0,0 +1,558 @@
import hashlib
import os
from unittest.mock import Mock
import pytest
from src import torrent
# ---------------------------------------------------------------------------
# ensure_aria2c
# ---------------------------------------------------------------------------
def test_ensure_aria2c_returns_existing_executable(monkeypatch):
expected = "/usr/bin/aria2c"
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=expected),
)
installer = Mock()
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == expected
installer.assert_not_called()
def test_ensure_aria2c_windows_returns_installed_binary(monkeypatch):
# Arrange
expected = r"C:\aria2\aria2c.exe"
monkeypatch.setattr(torrent.sys, "platform", "win32")
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=expected),
)
assert torrent.ensure_aria2c() == expected
def test_ensure_aria2c_windows_installs_when_binary_is_missing(monkeypatch):
# Arrange
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
monkeypatch.setattr(
torrent.sys,
"platform",
"win32",
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == "/fake/aria2c.exe"
installer.assert_called_once_with(
APP_DATA_DIR=torrent.APP_DATA_DIR,
ALLOWED_DOWNLOAD_HOSTS=torrent.ALLOWED_DOWNLOAD_HOSTS,
log_fn=torrent.log,
)
def test_ensure_aria2c_linux_returns_installed_binary(monkeypatch):
# Arrange
monkeypatch.setattr(
torrent.sys,
"platform",
"linux",
)
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value="/usr/bin/aria2c"),
)
installer = Mock()
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == "/usr/bin/aria2c"
installer.assert_not_called()
def test_ensure_aria2c_linux_raises_when_not_installed(monkeypatch):
# Arrange
monkeypatch.setattr(
torrent.sys,
"platform",
"linux",
)
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
installer = Mock()
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
with pytest.raises(
RuntimeError,
match="aria2c required but not installed",
):
torrent.ensure_aria2c()
installer.assert_not_called()
def test_ensure_aria2c_passes_custom_logger(monkeypatch):
# Arrange
monkeypatch.setattr(torrent.sys, "platform", "win32")
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
logger = Mock()
torrent.ensure_aria2c(log_fn=logger)
installer.assert_called_once_with(
APP_DATA_DIR=torrent.APP_DATA_DIR,
ALLOWED_DOWNLOAD_HOSTS=torrent.ALLOWED_DOWNLOAD_HOSTS,
log_fn=logger,
)
def test_ensure_aria2c_empty_path_triggers_install(monkeypatch):
# Arrange
monkeypatch.setattr(torrent.sys, "platform", "win32")
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=""),
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == "/fake/aria2c.exe"
installer.assert_called_once()
def test_ensure_aria2c_propagates_installer_failure(monkeypatch):
# Arrange
monkeypatch.setattr(torrent.sys, "platform", "win32")
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
monkeypatch.setattr(
torrent,
"install_aria2c_windows",
Mock(
side_effect=RuntimeError(
"installation failed"
)
),
)
with pytest.raises(
RuntimeError,
match="installation failed",
):
torrent.ensure_aria2c()
# ---------------------------------------------------------------------------
# torrent_version
# ---------------------------------------------------------------------------
def test_torrent_version_returns_sha1():
raw = b"torrent data"
assert torrent.torrent_version(raw) == hashlib.sha1(raw).hexdigest()
def test_torrent_version_changes_when_data_changes():
assert (
torrent.torrent_version(b"one")
!= torrent.torrent_version(b"two")
)
# ---------------------------------------------------------------------------
# torrent_selection
# ---------------------------------------------------------------------------
def test_torrent_selection_empty_when_files_match(tmp_path):
file = tmp_path / "file.bin"
file.write_bytes(b"1234")
files = [
(["file.bin"], 4),
]
need, missing = torrent.torrent_selection(
str(tmp_path),
files,
)
assert need == []
assert missing is False
def test_torrent_selection_marks_missing_file(tmp_path):
files = [
(["missing.bin"], 10),
]
need, missing = torrent.torrent_selection(
str(tmp_path),
files,
)
assert need == [1]
assert missing is True
def test_torrent_selection_marks_short_file(tmp_path):
file = tmp_path / "file.bin"
file.write_bytes(b"12")
files = [
(["file.bin"], 10),
]
need, missing = torrent.torrent_selection(
str(tmp_path),
files,
)
assert need == [1]
assert missing is False
assert file.exists()
def test_torrent_selection_removes_oversized_file(tmp_path):
file = tmp_path / "file.bin"
file.write_bytes(b"123456")
files = [
(["file.bin"], 4),
]
need, missing = torrent.torrent_selection(
str(tmp_path),
files,
)
assert need == [1]
assert missing is False
assert not file.exists()
def test_torrent_selection_drop_mismatched_removes_short_file(tmp_path):
file = tmp_path / "file.bin"
file.write_bytes(b"12")
files = [
(["file.bin"], 10),
]
need, missing = torrent.torrent_selection(
str(tmp_path),
files,
drop_mismatched=True,
)
assert need == [1]
assert missing is False
assert not file.exists()
# ---------------------------------------------------------------------------
# torrent_tree_intact
# ---------------------------------------------------------------------------
def test_torrent_tree_intact_returns_true_for_matching_files(tmp_path):
(tmp_path / "one.bin").write_bytes(b"123")
(tmp_path / "two.bin").write_bytes(b"12345")
files = [
(["one.bin"], 3),
(["two.bin"], 5),
]
assert torrent.torrent_tree_intact(
str(tmp_path),
files,
) is True
def test_torrent_tree_intact_returns_false_for_missing_file(tmp_path):
files = [
(["missing.bin"], 5),
]
assert torrent.torrent_tree_intact(
str(tmp_path),
files,
) is False
def test_torrent_tree_intact_returns_false_for_wrong_size(tmp_path):
(tmp_path / "file.bin").write_bytes(b"12")
files = [
(["file.bin"], 10),
]
assert torrent.torrent_tree_intact(
str(tmp_path),
files,
) is False
# ---------------------------------------------------------------------------
# shield / unshield
# ---------------------------------------------------------------------------
def test_unshield_protected_files_restores_backup(tmp_path):
original = tmp_path / "protected.dll"
backup = tmp_path / "protected.dll.octobak"
backup.write_bytes(b"original")
restored = torrent.unshield_protected_files([
(str(original), str(backup)),
])
assert original.read_bytes() == b"original"
assert not backup.exists()
assert restored == ["protected.dll"]
def test_unshield_removes_stub_for_previously_absent_file(tmp_path):
stub = tmp_path / "protected.dll"
stub.write_bytes(b"aria partial data")
restored = torrent.unshield_protected_files([
(str(stub), None),
])
assert not stub.exists()
assert restored == []
def test_recover_protected_files_restores_octobak(tmp_path):
backup = tmp_path / "file.bin.octobak"
backup.write_bytes(b"protected")
files = [
(["file.bin"], 9),
]
torrent.recover_protected_files(
str(tmp_path),
files,
)
assert (
tmp_path / "file.bin"
).read_bytes() == b"protected"
assert not backup.exists()
# ---------------------------------------------------------------------------
# clear_torrent_resume_state
# ---------------------------------------------------------------------------
def test_clear_torrent_resume_state_removes_aria2_files(
tmp_path,
monkeypatch,
):
monkeypatch.setattr(
torrent,
"TORRENT_STAGING_DIR",
str(tmp_path),
)
aria = tmp_path / "client.aria2"
metadata = tmp_path / "client.torrent"
unrelated = tmp_path / "keep.txt"
aria.write_bytes(b"a")
metadata.write_bytes(b"b")
unrelated.write_bytes(b"c")
torrent.clear_torrent_resume_state()
assert not aria.exists()
assert not metadata.exists()
assert unrelated.exists()
def test_clear_torrent_resume_state_handles_missing_directory(
tmp_path,
monkeypatch,
):
missing = tmp_path / "does-not-exist"
monkeypatch.setattr(
torrent,
"TORRENT_STAGING_DIR",
str(missing),
)
torrent.clear_torrent_resume_state()
# ---------------------------------------------------------------------------
# parse_aria_progress
# ---------------------------------------------------------------------------
def test_parse_aria_progress_download():
result = torrent.parse_aria_progress(
"1.2GiB/8.8GiB(13%) DL:5.0MiB"
)
assert result["progress"] == 0.13
assert result["bps"] == 5 * 1024 ** 2
assert result["checking"] is False
def test_parse_aria_progress_checksum():
result = torrent.parse_aria_progress(
"0B/1.5GiB(0%) "
"[Checksum:#f66a3e 236MiB/1.5GiB(15%)]"
)
assert result["progress"] == 0.15
assert result["done"] == 236 * 1024 ** 2
assert result["checking"] is True
def test_parse_aria_progress_without_download_speed():
result = torrent.parse_aria_progress(
"512MiB/1GiB(50%)"
)
assert result["progress"] == 0.5
assert result["bps"] == 0.0
def test_parse_aria_progress_returns_none_for_unrelated_line():
assert (
torrent.parse_aria_progress(
"Download complete"
)
is None
)
# ---------------------------------------------------------------------------
# stop_aria2c
# ---------------------------------------------------------------------------
def test_stop_aria2c_terminates_running_process(monkeypatch):
proc = Mock()
proc.poll.return_value = None
monkeypatch.setattr(
torrent,
"_active_aria2",
proc,
)
torrent.stop_aria2c()
proc.terminate.assert_called_once_with()
assert torrent._active_aria2 is None
def test_stop_aria2c_does_not_terminate_finished_process(monkeypatch):
proc = Mock()
proc.poll.return_value = 0
monkeypatch.setattr(
torrent,
"_active_aria2",
proc,
)
torrent.stop_aria2c()
proc.terminate.assert_not_called()
assert torrent._active_aria2 is None
def test_stop_aria2c_handles_no_process(monkeypatch):
monkeypatch.setattr(
torrent,
"_active_aria2",
None,
)
torrent.stop_aria2c()
assert torrent._active_aria2 is None
+58
View File
@@ -0,0 +1,58 @@
import io
import json
from unittest.mock import Mock
from src import ui
class JsonResponse(io.BytesIO):
def __enter__(self): return self
def __exit__(self, *args): return False
def test_strip_html_removes_tags_scripts_and_decodes_entities():
raw = '<p>Hello &amp; goodbye</p><script>bad()</script><br><b>World</b>'
assert ui._strip_html(raw) == "Hello & goodbye\n\nWorld"
def test_strip_html_formats_list_items_as_bullets():
assert ui._strip_html("<ul><li>One</li><li>Two</li></ul>") == "• One\n\n• Two"
def test_format_news_date_formats_valid_iso_date():
assert ui._format_news_date("2026-09-27T12:30:00+00:00") == "27 Sep 2026"
def test_format_news_date_returns_original_invalid_value():
assert ui._format_news_date("not-a-date") == "not-a-date"
def test_fetch_patch_notes_returns_items_newest_first(monkeypatch):
payload = {"items": [{"id": 1, "date": "2026-01-01"}, {"id": 2, "date": "2026-09-01"}]}
opener = Mock(return_value=JsonResponse(json.dumps(payload).encode()))
monkeypatch.setattr(ui, "secure_urlopen", opener)
result = ui.fetch_patch_notes()
assert [x["id"] for x in result] == [2, 1]
assert opener.call_args.kwargs["timeout"] == ui.NEWS_TIMEOUT
def test_fetch_patch_notes_returns_empty_list_when_items_missing(monkeypatch):
monkeypatch.setattr(ui, "secure_urlopen", Mock(return_value=JsonResponse(b"{}")))
assert ui.fetch_patch_notes() == []
def test_fetch_featured_post_returns_valid_post(monkeypatch):
payload = {"id": 7, "title": "News"}
monkeypatch.setattr(ui, "secure_urlopen", Mock(return_value=JsonResponse(json.dumps(payload).encode())))
assert ui.fetch_featured_post() == payload
def test_fetch_featured_post_returns_none_when_id_missing(monkeypatch):
monkeypatch.setattr(ui, "secure_urlopen", Mock(return_value=JsonResponse(b'{"title":"News"}')))
assert ui.fetch_featured_post() is None
def test_fetch_featured_post_returns_none_for_non_object_json(monkeypatch):
monkeypatch.setattr(ui, "secure_urlopen", Mock(return_value=JsonResponse(b'[]')))
assert ui.fetch_featured_post() is None
+419
View File
@@ -0,0 +1,419 @@
"""Torrent metadata, file selection, protected-file handling, and aria2c execution."""
import hashlib
import os
import re
import subprocess
import threading
import urllib.request
import sys
from functools import cache
from .app_log import log
from .config import APP_DATA_DIR, UA, DOWNLOAD_TIMEOUT
from .mods import MODS_REGISTRY
from .net import ALLOWED_DOWNLOAD_HOSTS, secure_urlopen
from .platform_ops import create_directory_link, ensure_dir, find_aria2c, install_aria2c_windows
CLIENT_TORRENT_URL = "https://dl.octowow.st/download/client.torrent"
# NOTE: Windows specific ZIP file
# NOTE: Windows sha hash for exe
#ARIA2C_PATH = os.path.join(APP_DATA_DIR, "aria2c.exe")
TORRENT_NAME = "client"
TORRENT_STAGING_DIR = os.path.join(APP_DATA_DIR, "torrent-root")
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
def ensure_aria2c(log_fn=log) -> str:
"""Return the path to aria2c.exe, downloading + checksum-verifying it into
APP_DATA_DIR on first use. Raises on failure."""
ARIA2C_PATH = find_aria2c() # NOTE: Returns a path to a aria2c executable , otherwise this var is None
if ARIA2C_PATH: # NOTE: If result from find_aria2c() isn't None , returns path
return ARIA2C_PATH
if sys.platform.startswith("win"): # NOTE: Checks if platform is windows
return install_aria2c_windows( APP_DATA_DIR=APP_DATA_DIR,
ALLOWED_DOWNLOAD_HOSTS=ALLOWED_DOWNLOAD_HOSTS,
log_fn=log_fn)
raise RuntimeError("aria2c required but not installed") # NOTE: If neither windows or linux path is returned , throws runtime error
def _bdecode(buf: bytes, pos: int = 0):
"""Minimal bencode decoder → (value, next_pos). Strings stay bytes."""
ch = buf[pos]
if ch == 0x69: # i<int>e
end = buf.index(b"e", pos)
return int(buf[pos + 1:end]), end + 1
if ch == 0x6c: # l<items>e
lst, p = [], pos + 1
while buf[p] != 0x65:
v, p = _bdecode(buf, p)
lst.append(v)
return lst, p + 1
if ch == 0x64: # d<pairs>e
d, p = {}, pos + 1
while buf[p] != 0x65:
k, p = _bdecode(buf, p)
v, p = _bdecode(buf, p)
d[k] = v
return d, p + 1
colon = buf.index(b":", pos) # <len>:<bytes>
n = int(buf[pos:colon])
start = colon + 1
return buf[start:start + n], start + n
def fetch_torrent(url: str = CLIENT_TORRENT_URL) -> tuple:
"""Download the .torrent → (raw_bytes, files). `files` is a list of
(path_parts, length). The raw bytes' SHA-1 is the client 'version'."""
req = urllib.request.Request(url, headers={"User-Agent": UA})
with secure_urlopen(req, timeout=DOWNLOAD_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
raw = r.read()
decoded, _ = _bdecode(raw)
info = decoded.get(b"info", {})
files = [([p.decode("latin1") for p in f.get(b"path", [])],
int(f.get(b"length", 0)))
for f in info.get(b"files", [])]
return raw, files
def torrent_version(raw: bytes) -> str:
"""SHA-1 of the whole .torrent file — the client 'version'. Changes whenever
the torrent is re-rolled, so it's what we compare to detect a new build and
decide whether aria2's resume state is stale. An occasional re-check when
only the announce/date changed (content unchanged) is harmless."""
return hashlib.sha1(raw).hexdigest()
@cache
def _torrent_excluded_files() -> frozenset:
"""Lower-cased basenames of every file the Mods tab installs, across all
mods — the client-root files the torrent sync leaves to the mod system.
The torrent ships several of them (VfPatcher.dll, nampower.dll, d3d9.dll,
…), but the Mods tab is the single source of truth for installing /
disabling / versioning them, so the sync must never fetch, re-add, or flag
them. Computed once and memoized (MODS_REGISTRY is static at import)."""
return frozenset(
os.path.basename(f).lower()
for mod in MODS_REGISTRY
for f in mod.get("installed_files", [])
)
def _is_torrent_excluded(parts) -> bool:
return len(parts) == 1 and parts[0].lower() in _torrent_excluded_files()
def _torrent_skip(parts, ignore_speech: bool, client_dir: str) -> bool:
"""Files the sync must leave alone: mod-owned client-root files, plus a
speech.MPQ the user opted to keep (Settings) — but only when it already
exists on disk. A missing speech.MPQ is still downloaded so a clean client
can launch; 'ignore' protects a custom copy, it doesn't skip a required file
that isn't there."""
if _is_torrent_excluded(parts):
return True
if ignore_speech and parts and parts[-1].lower() == "speech.mpq":
return os.path.exists(os.path.join(client_dir, *parts))
return False
# Files the sync must not rewrite (mod-owned client-root files, plus a kept
# custom speech.MPQ). Excluding them from --select-file stops aria2 fetching
# them on their own, but a torrent piece can straddle a file boundary, so
# repairing a selected neighbour re-downloads the shared piece and rewrites the
# excluded file's bytes too. We move them aside (rename) for the sync and put
# them back — instant and RAM-free, unlike copying a large speech.MPQ.
_SHIELD_SUFFIX = ".octobak"
def shield_protected_files(client_dir: str, files, ignore_speech: bool) -> list:
"""Prepare each sync-protected file (see _torrent_skip) for the sync. An
existing one is moved aside via a same-dir rename; an absent one is recorded
with backup=None, because the sync must never create it (only the Mods tab
installs these) yet aria2 may write a partial stub for it through a shared
piece. Returns [(orig, backup_or_None)] for unshielding."""
shielded = []
for parts, _length in files:
if not _torrent_skip(parts, ignore_speech, client_dir):
continue
p = os.path.join(client_dir, *parts)
if os.path.exists(p):
bak = p + _SHIELD_SUFFIX
try:
os.replace(p, bak) # same-fs, instant; aria2 sees p missing
shielded.append((p, bak))
except OSError:
pass
else:
shielded.append((p, None)) # must stay absent afterwards
return shielded
def unshield_protected_files(shielded) -> list:
"""Move each shielded file back, or delete the stub the sync created for a
file that was absent before. Returns the basenames restored."""
restored = []
for p, bak in shielded:
if bak is None:
try:
os.remove(p) # drop aria2's partial stub; stay absent
except OSError:
pass
continue
try:
os.replace(bak, p) # our version wins over aria2's partial
restored.append(os.path.basename(p))
except OSError:
pass
return restored
def recover_protected_files(client_dir: str, files):
"""Undo a shield interrupted by a crash: an orphaned '.octobak' beside a
torrent file is the real file — move it back into place."""
for parts, _length in files:
p = os.path.join(client_dir, *parts)
bak = p + _SHIELD_SUFFIX
if os.path.exists(bak):
try:
os.replace(bak, p)
except OSError:
pass
def torrent_selection(client_dir: str, files, drop_mismatched=False,
ignore_speech=False):
"""1-indexed list of torrent files that are missing or the wrong size on
disk (aria2 --select-file), plus whether any were entirely missing."""
need, missing = [], False
for i, (parts, length) in enumerate(files):
if _torrent_skip(parts, ignore_speech, client_dir):
continue
dest = os.path.join(client_dir, *parts)
try:
size = os.path.getsize(dest)
except OSError:
missing = True
need.append(i + 1)
continue
if size != length:
# oversized/corrupt file poisons resume — drop it; a short file is
# kept so aria2 can resume it
if drop_mismatched or size > length:
try:
os.remove(dest)
except OSError:
pass
need.append(i + 1)
return need, missing
def torrent_all_selection(client_dir: str, files, ignore_speech=False) -> list:
"""1-indexed list of every non-mod file in the torrent. Used for an
integrity pass: aria2 --check-integrity verifies each selected file's piece
hashes and re-downloads only the bad/missing pieces — catching same-size but
corrupted files that the size-based torrent_selection can't."""
return [i + 1 for i, (parts, _) in enumerate(files)
if not _torrent_skip(parts, ignore_speech, client_dir)]
def torrent_tree_intact(client_dir: str, files, ignore_speech=False) -> bool:
for parts, length in files:
if _torrent_skip(parts, ignore_speech, client_dir):
continue
try:
if os.path.getsize(os.path.join(client_dir, *parts)) != length:
return False
except OSError:
return False
return True
def clear_torrent_resume_state():
"""Delete aria2's saved control (.aria2) and metadata (.torrent) files in
the staging dir. They pin a specific torrent revision (info-hash) and record
which pieces are 'done', so a stale one causes an 'info hash mismatch' error
after the client torrent is re-rolled, or blocks re-downloading a file the
user deleted. Safe to call when nothing is there."""
try:
for name in os.listdir(TORRENT_STAGING_DIR):
if name.endswith((".aria2", ".torrent")):
try:
os.remove(os.path.join(TORRENT_STAGING_DIR, name))
except OSError:
pass
except OSError:
pass
def _ensure_torrent_link(client_dir: str) -> str:
"""Point <staging>/client at client_dir at client_dir via a platform-appropriate directory link so
aria2 writes the torrent's files straight into the real client dir. Returns
the staging dir to pass as aria2 --dir."""
staging = TORRENT_STAGING_DIR
ensure_dir(staging)
link = os.path.join(staging, TORRENT_NAME)
target = os.path.abspath(client_dir)
try:
if os.path.isdir(link) and \
os.path.abspath(os.path.realpath(link)) == target:
return staging
except OSError:
pass
# remove a stale junction/link (rmdir drops the reparse point, not its
# target's contents) then recreate it
try:
os.rmdir(link)
except OSError:
try:
os.remove(link)
except OSError:
pass
create_directory_link(link, target)
if not os.path.isdir(link):
raise RuntimeError("could not create torrent download link")
return staging
_SIZE_UNITS = {"B": 1, "KiB": 1024, "MiB": 1024 ** 2,
"GiB": 1024 ** 3, "TiB": 1024 ** 4}
_UNIT = "|".join(_SIZE_UNITS) # B|KiB|MiB|GiB|TiB
_SIZE = rf"[\d.]+(?:{_UNIT})" # e.g. 8.8GiB
_FRAC = rf"({_SIZE})/({_SIZE})\((\d+)%\)" # done/total(percent%)
_ARIA_FRAC = re.compile(_FRAC)
_ARIA_DL = re.compile(rf"DL:({_SIZE})")
# During --check-integrity aria2 prints the hash-check progress in a separate
# field, e.g. '… [Checksum:#f66a3e 236MiB/1.5GiB(15%)]', while the leading
# completed/total stays at 0%/0B (a complete client downloads nothing). Read the
# Checksum fraction so the bar tracks the check instead of freezing at 0.
_ARIA_CHK = re.compile(rf"Checksum:#\w+\s+{_FRAC}")
def _to_bytes(s: str) -> float:
m = re.match(rf"^([\d.]+)({_UNIT})$", s.strip())
return float(m.group(1)) * _SIZE_UNITS[m.group(2)] if m else 0.0
def parse_aria_progress(line: str):
"""Parse an aria2 summary line like '1.2GiB/8.8GiB(13%) … DL:5.0MiB' →
{progress, done, total, bps, checking}, or None for non-progress lines.
A checksum-check line is preferred over the (idle) download fraction."""
chk = _ARIA_CHK.search(line)
m = chk or _ARIA_FRAC.search(line)
if not m:
return None
dl = _ARIA_DL.search(line)
return {"progress": int(m.group(3)) / 100.0,
"done": _to_bytes(m.group(1)),
"total": _to_bytes(m.group(2)),
"bps": _to_bytes(dl.group(1)) if dl else 0.0,
"checking": chk is not None}
# The currently-running aria2c child, so it can be killed when the app quits
# (a daemon worker thread dying would otherwise orphan it, still downloading
# headless). --stop-with-process is aria2's own belt-and-suspenders for this,
# but it's unreliable on Windows — hence the explicit kill too.
_active_aria2: "subprocess.Popen | None" = None
_active_aria2_lock = threading.Lock()
def stop_aria2c():
"""Terminate the running aria2c child, if any. Safe to call from any thread
(e.g. the app's close handler)."""
global _active_aria2
with _active_aria2_lock:
proc, _active_aria2 = _active_aria2, None
if proc and proc.poll() is None:
try:
proc.terminate()
except Exception:
pass
def run_aria2c(client_dir, select_files=None, check_integrity=False,
on_progress=None, should_cancel=None, log_fn=log):
"""Sync the client torrent into client_dir with aria2c (leech-only). Blocks
until aria2c exits; raises on a non-zero exit or cancellation. Calls
on_progress(dict) per update and should_cancel()->bool to abort.
aria2 is handed the .torrent URL (not a local copy), so it always fetches
the server's current torrent at download time — no chance of running a stale
local .torrent if the user starts the update long after the verify."""
global _active_aria2
exe = ensure_aria2c(log_fn)
staging = _ensure_torrent_link(client_dir)
args = [
exe,
f"--dir={staging}",
# aria2 exits when this PID (the updater) does — stops an orphaned
# download if we're killed before the explicit stop_aria2c() runs.
f"--stop-with-process={os.getpid()}",
"--seed-time=0",
f"--check-integrity={'true' if check_integrity else 'false'}",
"--bt-remove-unselected-file=false",
"--continue=true",
"--allow-overwrite=true",
"--auto-file-renaming=false",
"--file-allocation=none",
"--disk-cache=128M",
"--stream-piece-selector=inorder",
"--max-tries=0",
"--retry-wait=5",
"--bt-stop-timeout=120",
"--auto-save-interval=15",
"--summary-interval=1",
"--human-readable=false",
"--truncate-console-readout=false",
"--console-log-level=warn",
"--enable-dht=true",
"--bt-enable-lpd=true",
"--max-connection-per-server=8",
"--split=16",
"--min-split-size=1M",
]
if select_files:
args.append("--select-file=" + ",".join(str(i) for i in select_files))
args.append(CLIENT_TORRENT_URL)
proc = subprocess.Popen(args, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True,
bufsize=1, creationflags=_NO_WINDOW)
with _active_aria2_lock:
_active_aria2 = proc
try:
for line in proc.stdout:
if should_cancel and should_cancel():
proc.terminate()
raise RuntimeError("Cancelled")
line = line.strip()
if not line:
continue
p = parse_aria_progress(line)
if p:
# aria2's console readout floods idle '0B/total(0%)' lines (no
# bytes, no checksum) between the once-a-second summary blocks
# that carry the real figure. Drop the idle ones so they can't
# stomp progress back to 0 under the UI's latest-wins draining.
informative = (p["done"] or p["progress"]
or p["bps"] or p["checking"])
if on_progress and informative:
on_progress(p)
else:
log_fn(f"[aria2] {line}", "dim")
finally:
try:
proc.stdout.close()
except Exception:
pass
with _active_aria2_lock:
if _active_aria2 is proc:
_active_aria2 = None
code = proc.wait()
if code != 0:
raise RuntimeError(f"aria2c exited with code {code}")
+28 -21
View File
@@ -21,18 +21,21 @@ from pathlib import Path
import tkinter as tk
from tkinter import filedialog
import urllib.error
from app_log import LOG_QUEUE as _LOG_Q, log
from config import *
from net import *
from client import *
from torrent import *
from mods import *
from platform_ops import open_directory, launch_executable, request_defender_exclusion
from .app_log import LOG_QUEUE as _LOG_Q, log
#from .config import _relocate_legacy_data
from .net import *
from .client import *
from .torrent import *
from .mods import *
from .platform_ops import open_directory, launch_executable, request_defender_exclusion
from . import config
# News tab: the latest announcement (forum 2, full post) fills the left panel,
# the patch-notes list (forum 4) fills the right.
NEWS_FEATURED_URL = f"{SERVER}/forum/octonews.php?forum=2&mode=full"
PATCHNOTES_URL = f"{SERVER}/forum/octonews.php?mode=list&forum=4&limit=8"
NEWS_FEATURED_URL = f"{config.SERVER}/forum/octonews.php?forum=2&mode=full"
PATCHNOTES_URL = f"{config.SERVER}/forum/octonews.php?mode=list&forum=4&limit=8"
NEWS_TIMEOUT = 8
NEWS_CACHE_TTL = 300
@@ -71,9 +74,11 @@ C_PARCH_LINK = "#a3561c"
C_PARCH_EDGE = "#b7a678"
FONT_BODY = ("Segoe UI", 9)
FONT_MONO = ("TkDefaultFont", 10)
FONT_VER = ("TkDefaultFont", 8)
NEWS_FEATURED_URL = f"{SERVER}/forum/octonews.php?forum=2&mode=full"
PATCHNOTES_URL = f"{SERVER}/forum/octonews.php?mode=list&forum=4&limit=8"
NEWS_FEATURED_URL = f"{config.SERVER}/forum/octonews.php?forum=2&mode=full"
PATCHNOTES_URL = f"{config.SERVER}/forum/octonews.php?mode=list&forum=4&limit=8"
NEWS_TIMEOUT = 8
NEWS_CACHE_TTL = 300
@@ -103,7 +108,8 @@ def _format_news_date(iso: str) -> str:
def fetch_patch_notes() -> list:
"""Patch-notes list → [{id, title, date, body, url?, author?}, …]"""
req = urllib.request.Request(PATCHNOTES_URL, headers={"User-Agent": UA})
#req = urllib.request.Request(PATCHNOTES_URL, headers={"User-Agent": config.UA})
req = urllib.request.Request(PATCHNOTES_URL)
with secure_urlopen(req, timeout=NEWS_TIMEOUT) as r:
data = json.load(r)
items = data.get("items", [])
@@ -115,7 +121,8 @@ def fetch_patch_notes() -> list:
def fetch_featured_post() -> dict | None:
"""Latest announcements-forum post → {id, title, author?, date, url, html}"""
req = urllib.request.Request(NEWS_FEATURED_URL, headers={"User-Agent": UA})
#req = urllib.request.Request(NEWS_FEATURED_URL, headers={"User-Agent": config.UA})
req = urllib.request.Request(NEWS_FEATURED_URL)
with secure_urlopen(req, timeout=NEWS_TIMEOUT) as r:
data = json.load(r)
return data if isinstance(data, dict) and data.get("id") else None
@@ -210,10 +217,10 @@ class OctoUpdaterApp(tk.Tk):
# Move a pre-1.3 config/cache from beside the .exe into the per-user
# data dir before anything reads them, so first-run detection and
# load_config() below see the relocated files (see _relocate_legacy_data).
_relocate_legacy_data()
#config._relocate_legacy_data()
# Detect first run before anything writes the config.
self._first_run = not os.path.exists(CONFIG_FILE)
self._first_run = not os.path.exists(config.CONFIG_FILE)
# Set when the user adds a Defender exclusion via Settings; checked at
# the next reconcile to skip the auto-prompt (so a manual add isn't
# double-prompted), then reset — so each folder change offers one unless
@@ -247,7 +254,7 @@ class OctoUpdaterApp(tk.Tk):
# effect only once Settings is closed (see _close_settings), so no
# live trace fires mid-edit.
self._game_path = tk.StringVar(
value=os.path.normpath(self._cfg.get("out_dir", DEFAULT_GAME_DIR)))
value=os.path.normpath(self._cfg.get("out_dir", config.DEFAULT_GAME_DIR)))
# Count of mods with an update available — shown as a badge on the
# MODS nav tab.
@@ -293,7 +300,7 @@ class OctoUpdaterApp(tk.Tk):
self._build()
out_dir = self._cfg.get("out_dir", DEFAULT_GAME_DIR)
out_dir = self._cfg.get("out_dir", config.DEFAULT_GAME_DIR)
if not os.path.exists(out_dir):
def _wipe(c):
c.pop("mods", None)
@@ -596,13 +603,13 @@ class OctoUpdaterApp(tk.Tk):
hdr.bind("<Leave>", lambda e: self._on_hdr_motion(None))
self._clear_wdb_var = tk.BooleanVar(
value=bool(self._cfg.get("clear_wdb_on_launch", False)))
value=bool(self._cfg.get("clear_wdb_on_launch", True)))
self._close_on_launch_var = tk.BooleanVar(
value=bool(self._cfg.get("close_on_launch", False)))
value=bool(self._cfg.get("close_on_launch", True)))
self._auto_mods_var = tk.BooleanVar(
value=bool(self._cfg.get("auto_install_mods", True)))
value=bool(self._cfg.get("auto_install_mods", False)))
self._auto_addons_var = tk.BooleanVar(
value=bool(self._cfg.get("auto_install_addons", True)))
value=bool(self._cfg.get("auto_install_addons", False)))
self._ignore_speech_var = tk.BooleanVar(
value=bool(self._cfg.get("ignore_speech", False)))
# Deferred "install missing" pending from turning an auto-install
-907
View File
@@ -1,907 +0,0 @@
"""Extracted from octo_updater.py. Keep this module focused on its named responsibility."""
import json
import hashlib
import os
import sys
import ssl
import re
import subprocess
import urllib.request
from urllib.parse import urlsplit
import shutil
import stat
import struct
import time
import math
import threading
import queue
from functools import cache
from pathlib import Path
from app_log import log
from config import APP_DATA_DIR, UA, DOWNLOAD_RETRY, DOWNLOAD_TIMEOUT, ensure_dir, load_config, update_config
from net import ALLOWED_DOWNLOAD_HOSTS, secure_urlopen
from client import locale_patches, write_config_wtf, update_config_wtf
from platform_ops import create_directory_link
# Torrent-based client sync (aria2c)
# ──────────────────────────────────────────────────────────────────────────────
# Client files are synced over BitTorrent with aria2c, fetched on first use.
# The download is differential: only files that are missing or the wrong size
# are pulled.
CLIENT_TORRENT_URL = "https://dl.octowow.st/download/client.torrent"
# Pinned aria2 Windows build. aria2 is GPLv2+, fetched and run unmodified; only
# aria2c.exe is used. The sha256 is of the release .zip (verified once).
ARIA2_ZIP_URL = ("https://github.com/aria2/aria2/releases/download/"
"release-1.37.0/aria2-1.37.0-win-32bit-build1.zip")
ARIA2_ZIP_SHA256 = "35f6514cc5dd7e98a87b3c4c2d25a0754b9b063dbe59bc0f22d483464f61e5b6"
ARIA2C_PATH = os.path.join(APP_DATA_DIR, "aria2c.exe")
# The torrent's top-level folder name: aria2 writes files under <dir>/<name>/…,
# so a junction <staging>/client → the real client dir lands them in place.
TORRENT_NAME = "client"
TORRENT_STAGING_DIR = os.path.join(APP_DATA_DIR, "torrent-root")
PRISTINE_WOW_PATH = os.path.join(APP_DATA_DIR, "base-WoW.exe")
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
def ensure_aria2c(log_fn=log) -> str:
"""Return the path to aria2c.exe, downloading + checksum-verifying it into
APP_DATA_DIR on first use. Raises on failure."""
if os.path.exists(ARIA2C_PATH):
return ARIA2C_PATH
log_fn("Fetching aria2c (one-time, ~2.5 MB)…", "acct")
req = urllib.request.Request(ARIA2_ZIP_URL, headers={"User-Agent": UA})
with secure_urlopen(req, timeout=60, allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
data = r.read()
digest = hashlib.sha256(data).hexdigest()
if digest != ARIA2_ZIP_SHA256:
raise RuntimeError(
f"aria2 checksum mismatch (got {digest[:12]}…); refusing to run it")
import zipfile, io
with zipfile.ZipFile(io.BytesIO(data)) as zf:
name = next((n for n in zf.namelist()
if n.lower().endswith("aria2c.exe")), None)
if not name:
raise RuntimeError("aria2c.exe not found in the aria2 archive")
exe = zf.read(name)
ensure_dir(APP_DATA_DIR)
tmp = ARIA2C_PATH + ".part"
with open(tmp, "wb") as f:
f.write(exe)
os.replace(tmp, ARIA2C_PATH)
log_fn("aria2c ready.", "ok")
return ARIA2C_PATH
def _bdecode(buf: bytes, pos: int = 0):
"""Minimal bencode decoder → (value, next_pos). Strings stay bytes."""
ch = buf[pos]
if ch == 0x69: # i<int>e
end = buf.index(b"e", pos)
return int(buf[pos + 1:end]), end + 1
if ch == 0x6c: # l<items>e
lst, p = [], pos + 1
while buf[p] != 0x65:
v, p = _bdecode(buf, p)
lst.append(v)
return lst, p + 1
if ch == 0x64: # d<pairs>e
d, p = {}, pos + 1
while buf[p] != 0x65:
k, p = _bdecode(buf, p)
v, p = _bdecode(buf, p)
d[k] = v
return d, p + 1
colon = buf.index(b":", pos) # <len>:<bytes>
n = int(buf[pos:colon])
start = colon + 1
return buf[start:start + n], start + n
def fetch_torrent(url: str = CLIENT_TORRENT_URL) -> tuple:
"""Download the .torrent → (raw_bytes, files). `files` is a list of
(path_parts, length). The raw bytes' SHA-1 is the client 'version'."""
req = urllib.request.Request(url, headers={"User-Agent": UA})
with secure_urlopen(req, timeout=DOWNLOAD_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
raw = r.read()
decoded, _ = _bdecode(raw)
info = decoded.get(b"info", {})
files = [([p.decode("latin1") for p in f.get(b"path", [])],
int(f.get(b"length", 0)))
for f in info.get(b"files", [])]
return raw, files
def torrent_version(raw: bytes) -> str:
"""SHA-1 of the whole .torrent file — the client 'version'. Changes whenever
the torrent is re-rolled, so it's what we compare to detect a new build and
decide whether aria2's resume state is stale. An occasional re-check when
only the announce/date changed (content unchanged) is harmless."""
return hashlib.sha1(raw).hexdigest()
@cache
def _torrent_excluded_files() -> frozenset:
"""Lower-cased basenames of every file the Mods tab installs, across all
mods — the client-root files the torrent sync leaves to the mod system.
The torrent ships several of them (VfPatcher.dll, nampower.dll, d3d9.dll,
…), but the Mods tab is the single source of truth for installing /
disabling / versioning them, so the sync must never fetch, re-add, or flag
them. Computed once and memoized (MODS_REGISTRY is static at import)."""
return frozenset(
os.path.basename(f).lower()
for mod in MODS_REGISTRY
for f in mod.get("installed_files", [])
)
def _is_torrent_excluded(parts) -> bool:
return len(parts) == 1 and parts[0].lower() in _torrent_excluded_files()
def _torrent_skip(parts, ignore_speech: bool, client_dir: str) -> bool:
"""Files the sync must leave alone: mod-owned client-root files, plus a
speech.MPQ the user opted to keep (Settings) — but only when it already
exists on disk. A missing speech.MPQ is still downloaded so a clean client
can launch; 'ignore' protects a custom copy, it doesn't skip a required file
that isn't there."""
if _is_torrent_excluded(parts):
return True
if ignore_speech and parts and parts[-1].lower() == "speech.mpq":
return os.path.exists(os.path.join(client_dir, *parts))
return False
# Files the sync must not rewrite (mod-owned client-root files, plus a kept
# custom speech.MPQ). Excluding them from --select-file stops aria2 fetching
# them on their own, but a torrent piece can straddle a file boundary, so
# repairing a selected neighbour re-downloads the shared piece and rewrites the
# excluded file's bytes too. We move them aside (rename) for the sync and put
# them back — instant and RAM-free, unlike copying a large speech.MPQ.
_SHIELD_SUFFIX = ".octobak"
def shield_protected_files(client_dir: str, files, ignore_speech: bool) -> list:
"""Prepare each sync-protected file (see _torrent_skip) for the sync. An
existing one is moved aside via a same-dir rename; an absent one is recorded
with backup=None, because the sync must never create it (only the Mods tab
installs these) yet aria2 may write a partial stub for it through a shared
piece. Returns [(orig, backup_or_None)] for unshielding."""
shielded = []
for parts, _length in files:
if not _torrent_skip(parts, ignore_speech, client_dir):
continue
p = os.path.join(client_dir, *parts)
if os.path.exists(p):
bak = p + _SHIELD_SUFFIX
try:
os.replace(p, bak) # same-fs, instant; aria2 sees p missing
shielded.append((p, bak))
except OSError:
pass
else:
shielded.append((p, None)) # must stay absent afterwards
return shielded
def unshield_protected_files(shielded) -> list:
"""Move each shielded file back, or delete the stub the sync created for a
file that was absent before. Returns the basenames restored."""
restored = []
for p, bak in shielded:
if bak is None:
try:
os.remove(p) # drop aria2's partial stub; stay absent
except OSError:
pass
continue
try:
os.replace(bak, p) # our version wins over aria2's partial
restored.append(os.path.basename(p))
except OSError:
pass
return restored
def recover_protected_files(client_dir: str, files):
"""Undo a shield interrupted by a crash: an orphaned '.octobak' beside a
torrent file is the real file — move it back into place."""
for parts, _length in files:
p = os.path.join(client_dir, *parts)
bak = p + _SHIELD_SUFFIX
if os.path.exists(bak):
try:
os.replace(bak, p)
except OSError:
pass
def torrent_selection(client_dir: str, files, drop_mismatched=False,
ignore_speech=False):
"""1-indexed list of torrent files that are missing or the wrong size on
disk (aria2 --select-file), plus whether any were entirely missing."""
need, missing = [], False
for i, (parts, length) in enumerate(files):
if _torrent_skip(parts, ignore_speech, client_dir):
continue
dest = os.path.join(client_dir, *parts)
try:
size = os.path.getsize(dest)
except OSError:
missing = True
need.append(i + 1)
continue
if size != length:
# oversized/corrupt file poisons resume — drop it; a short file is
# kept so aria2 can resume it
if drop_mismatched or size > length:
try:
os.remove(dest)
except OSError:
pass
need.append(i + 1)
return need, missing
def torrent_all_selection(client_dir: str, files, ignore_speech=False) -> list:
"""1-indexed list of every non-mod file in the torrent. Used for an
integrity pass: aria2 --check-integrity verifies each selected file's piece
hashes and re-downloads only the bad/missing pieces — catching same-size but
corrupted files that the size-based torrent_selection can't."""
return [i + 1 for i, (parts, _) in enumerate(files)
if not _torrent_skip(parts, ignore_speech, client_dir)]
def torrent_tree_intact(client_dir: str, files, ignore_speech=False) -> bool:
for parts, length in files:
if _torrent_skip(parts, ignore_speech, client_dir):
continue
try:
if os.path.getsize(os.path.join(client_dir, *parts)) != length:
return False
except OSError:
return False
return True
# Legacy leftovers the current client no longer ships. Locale data folders are
# matched by name; the old patch archives by name AND exact size, so a player
# mod that reused one of these names is never deleted.
_LOCALE_DATA_DIRS = {
"enus", "engb", "encn", "entw", "kokr", "frfr", "dede", "zhcn",
"zhtw", "eses", "esmx", "ruru", "ptbr", "ptpt", "itit",
}
_LEGACY_ARCHIVES = {
"patch-6.mpq": 451195806,
"patch-7.mpq": 175256564,
"patch-8.mpq": 484649870,
"patch-9.mpq": 506808141,
"patch-a.mpq": 241751337,
}
def prune_stale_client_files(client_dir: str, files) -> list:
"""Remove legacy Data/<locale>/ folders and known old-client patch MPQs the
current torrent no longer ships. Folders are removed by name (unless the torrent
still uses them); archives only when the name AND the exact size match a known
legacy one. Returns removed names."""
data_dir = os.path.join(client_dir, "Data")
if not os.path.isdir(data_dir):
return []
# what the current torrent puts directly in Data/ (.mpq files and subdirs)
expected, used_dirs = set(), set()
for parts, _length in files:
if not parts or parts[0] != "Data":
continue
if len(parts) == 2 and parts[1].lower().endswith(".mpq"):
expected.add(parts[1].lower())
elif len(parts) >= 3:
used_dirs.add(parts[1].lower())
removed = []
for name in os.listdir(data_dir):
lc, full = name.lower(), os.path.join(data_dir, name)
if os.path.isdir(full):
if lc in _LOCALE_DATA_DIRS and lc not in used_dirs:
try:
shutil.rmtree(full)
removed.append(name + "/")
except OSError:
pass
continue
if not lc.endswith(".mpq") or lc in expected:
continue
try:
size = os.path.getsize(full)
except OSError:
continue
if _LEGACY_ARCHIVES.get(lc) == size:
try:
os.remove(full)
removed.append(name)
except OSError:
pass
return removed
# Downloadable MPQ content patches (not shipped with the client). Each is
# a single HTTP file with a `<url>.sha256` sidecar; an update is available when
# that published sha differs from the on-disk file's.
MPQ_PATCHES = [
{
"file": "patch-O.mpq",
"name": "Octo Raid Visuals",
"description": "Adds ground markers and sounds for boss abilities in raids.",
"url": "https://dl.octowow.st/client/latest/Data/patch-O.mpq",
},
]
def mpq_patch_for(filename: str):
"""The registry entry whose file matches `filename` (case-insensitive)."""
lc = filename.lower()
return next((e for e in MPQ_PATCHES if e["file"].lower() == lc), None)
def sha256_file(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest().lower()
def fetch_mpq_sha256(url: str) -> str:
"""The published SHA256 of an MPQ patch, from its `<url>.sha256` sidecar."""
req = urllib.request.Request(url + ".sha256", headers={"User-Agent": UA})
with secure_urlopen(req, timeout=NEWS_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
return r.read().decode("ascii", "ignore").strip().split()[0].lower()
def download_mpq_patch(entry: dict, data_dir: str, on_progress=None):
"""Download the patch into <data_dir>/<file>, verifying its published
SHA256. Writes to a .part file and renames on success. on_progress(done,
total) is called as bytes arrive."""
url = entry["url"]
want = fetch_mpq_sha256(url)
ensure_dir(data_dir)
tmp = os.path.join(data_dir, entry["file"] + ".part")
h = hashlib.sha256()
req = urllib.request.Request(url, headers={"User-Agent": UA})
with secure_urlopen(req, timeout=DOWNLOAD_TIMEOUT,
allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
total = int(r.headers.get("Content-Length") or 0)
done = 0
with open(tmp, "wb") as f:
while True:
chunk = r.read(256 * 1024)
if not chunk:
break
f.write(chunk)
h.update(chunk)
done += len(chunk)
if on_progress:
on_progress(done, total)
if h.hexdigest().lower() != want:
try:
os.remove(tmp)
except OSError:
pass
raise RuntimeError("checksum verification failed")
os.replace(tmp, os.path.join(data_dir, entry["file"]))
def clear_torrent_resume_state():
"""Delete aria2's saved control (.aria2) and metadata (.torrent) files in
the staging dir. They pin a specific torrent revision (info-hash) and record
which pieces are 'done', so a stale one causes an 'info hash mismatch' error
after the client torrent is re-rolled, or blocks re-downloading a file the
user deleted. Safe to call when nothing is there."""
try:
for name in os.listdir(TORRENT_STAGING_DIR):
if name.endswith((".aria2", ".torrent")):
try:
os.remove(os.path.join(TORRENT_STAGING_DIR, name))
except OSError:
pass
except OSError:
pass
def refresh_pristine_wow(client_dir: str):
"""Cache the freshly-synced (unpatched) WoW.exe as the pristine base, so a
later re-patch always starts from clean bytes."""
exe = os.path.join(client_dir, "WoW.exe")
try:
with open(exe, "rb") as f:
f.seek(_LOCALE_ASSERT_OFFSET)
pristine = f.read(1) == b"\xa1"
if pristine:
shutil.copyfile(exe, PRISTINE_WOW_PATH)
log("Cached pristine WoW.exe base.", "dim")
except OSError:
pass
def read_pristine_wow(client_dir: str) -> bytes:
"""The clean base to patch from: the cached pristine exe if present, else
the on-disk WoW.exe."""
if os.path.exists(PRISTINE_WOW_PATH):
with open(PRISTINE_WOW_PATH, "rb") as f:
return f.read()
with open(os.path.join(client_dir, "WoW.exe"), "rb") as f:
return f.read()
def _ensure_torrent_link(client_dir: str) -> str:
"""Point <staging>/client at client_dir at client_dir via a platform-appropriate directory link so
aria2 writes the torrent's files straight into the real client dir. Returns
the staging dir to pass as aria2 --dir."""
staging = TORRENT_STAGING_DIR
ensure_dir(staging)
link = os.path.join(staging, TORRENT_NAME)
target = os.path.abspath(client_dir)
try:
if os.path.isdir(link) and \
os.path.abspath(os.path.realpath(link)) == target:
return staging
except OSError:
pass
# remove a stale junction/link (rmdir drops the reparse point, not its
# target's contents) then recreate it
try:
os.rmdir(link)
except OSError:
try:
os.remove(link)
except OSError:
pass
create_directory_link(link, target)
if not os.path.isdir(link):
raise RuntimeError("could not create torrent download link")
return staging
_SIZE_UNITS = {"B": 1, "KiB": 1024, "MiB": 1024 ** 2,
"GiB": 1024 ** 3, "TiB": 1024 ** 4}
_UNIT = "|".join(_SIZE_UNITS) # B|KiB|MiB|GiB|TiB
_SIZE = rf"[\d.]+(?:{_UNIT})" # e.g. 8.8GiB
_FRAC = rf"({_SIZE})/({_SIZE})\((\d+)%\)" # done/total(percent%)
_ARIA_FRAC = re.compile(_FRAC)
_ARIA_DL = re.compile(rf"DL:({_SIZE})")
# During --check-integrity aria2 prints the hash-check progress in a separate
# field, e.g. '… [Checksum:#f66a3e 236MiB/1.5GiB(15%)]', while the leading
# completed/total stays at 0%/0B (a complete client downloads nothing). Read the
# Checksum fraction so the bar tracks the check instead of freezing at 0.
_ARIA_CHK = re.compile(rf"Checksum:#\w+\s+{_FRAC}")
def _to_bytes(s: str) -> float:
m = re.match(rf"^([\d.]+)({_UNIT})$", s.strip())
return float(m.group(1)) * _SIZE_UNITS[m.group(2)] if m else 0.0
def parse_aria_progress(line: str):
"""Parse an aria2 summary line like '1.2GiB/8.8GiB(13%) … DL:5.0MiB' →
{progress, done, total, bps, checking}, or None for non-progress lines.
A checksum-check line is preferred over the (idle) download fraction."""
chk = _ARIA_CHK.search(line)
m = chk or _ARIA_FRAC.search(line)
if not m:
return None
dl = _ARIA_DL.search(line)
return {"progress": int(m.group(3)) / 100.0,
"done": _to_bytes(m.group(1)),
"total": _to_bytes(m.group(2)),
"bps": _to_bytes(dl.group(1)) if dl else 0.0,
"checking": chk is not None}
# The currently-running aria2c child, so it can be killed when the app quits
# (a daemon worker thread dying would otherwise orphan it, still downloading
# headless). --stop-with-process is aria2's own belt-and-suspenders for this,
# but it's unreliable on Windows — hence the explicit kill too.
_active_aria2: "subprocess.Popen | None" = None
_active_aria2_lock = threading.Lock()
def stop_aria2c():
"""Terminate the running aria2c child, if any. Safe to call from any thread
(e.g. the app's close handler)."""
global _active_aria2
with _active_aria2_lock:
proc, _active_aria2 = _active_aria2, None
if proc and proc.poll() is None:
try:
proc.terminate()
except Exception:
pass
def run_aria2c(client_dir, select_files=None, check_integrity=False,
on_progress=None, should_cancel=None, log_fn=log):
"""Sync the client torrent into client_dir with aria2c (leech-only). Blocks
until aria2c exits; raises on a non-zero exit or cancellation. Calls
on_progress(dict) per update and should_cancel()->bool to abort.
aria2 is handed the .torrent URL (not a local copy), so it always fetches
the server's current torrent at download time — no chance of running a stale
local .torrent if the user starts the update long after the verify."""
global _active_aria2
exe = ensure_aria2c(log_fn)
staging = _ensure_torrent_link(client_dir)
args = [
exe,
f"--dir={staging}",
# aria2 exits when this PID (the updater) does — stops an orphaned
# download if we're killed before the explicit stop_aria2c() runs.
f"--stop-with-process={os.getpid()}",
"--seed-time=0",
f"--check-integrity={'true' if check_integrity else 'false'}",
"--bt-remove-unselected-file=false",
"--continue=true",
"--allow-overwrite=true",
"--auto-file-renaming=false",
"--file-allocation=none",
"--disk-cache=128M",
"--stream-piece-selector=inorder",
"--max-tries=0",
"--retry-wait=5",
"--bt-stop-timeout=120",
"--auto-save-interval=15",
"--summary-interval=1",
"--human-readable=false",
"--truncate-console-readout=false",
"--console-log-level=warn",
"--enable-dht=true",
"--bt-enable-lpd=true",
"--max-connection-per-server=8",
"--split=16",
"--min-split-size=1M",
]
if select_files:
args.append("--select-file=" + ",".join(str(i) for i in select_files))
args.append(CLIENT_TORRENT_URL)
proc = subprocess.Popen(args, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True,
bufsize=1, creationflags=_NO_WINDOW)
with _active_aria2_lock:
_active_aria2 = proc
try:
for line in proc.stdout:
if should_cancel and should_cancel():
proc.terminate()
raise RuntimeError("Cancelled")
line = line.strip()
if not line:
continue
p = parse_aria_progress(line)
if p:
# aria2's console readout floods idle '0B/total(0%)' lines (no
# bytes, no checksum) between the once-a-second summary blocks
# that carry the real figure. Drop the idle ones so they can't
# stomp progress back to 0 under the UI's latest-wins draining.
informative = (p["done"] or p["progress"]
or p["bps"] or p["checking"])
if on_progress and informative:
on_progress(p)
else:
log_fn(f"[aria2] {line}", "dim")
finally:
try:
proc.stdout.close()
except Exception:
pass
with _active_aria2_lock:
if _active_aria2 is proc:
_active_aria2 = None
code = proc.wait()
if code != 0:
raise RuntimeError(f"aria2c exited with code {code}")
class VerifyWorker:
def __init__(self, out_dir: str, log_q: queue.Queue, prog_q: queue.Queue):
self.out_dir = out_dir
self.log_q = log_q
self.prog_q = prog_q
self._cancel = False
def cancel(self):
self._cancel = True
def log(self, msg, tag=""):
self.log_q.put((msg, tag))
def progress(self, value, label=""):
self.prog_q.put((value, label))
def run(self):
try:
self.log("Checking for updates…", "acct")
raw, files = fetch_torrent()
self.log("Checking game files…", "acct")
# aria2 selects by size (a patched WoW.exe keeps the torrent's size,
# so it's never flagged); need == files missing or wrong-sized.
ignore_speech = bool(load_config().get("ignore_speech", False))
need, _missing = torrent_selection(self.out_dir, files,
ignore_speech=ignore_speech)
have_exe = os.path.exists(os.path.join(self.out_dir, "WoW.exe"))
if have_exe and not need:
self.log("Everything is up to date!", "ok")
self.log_q.put(("__UP_TO_DATE__", ""))
else:
self.log("Update available.", "acct")
self.log_q.put(("__UPDATE_NEEDED__", ""))
except Exception as e:
self.log(f"Verification failed: {e}", "err")
self.log_q.put(("__UPDATE_NEEDED__", ""))
class UpdateWorker:
def __init__(self, out_dir: str, log_q: queue.Queue, prog_q: queue.Queue,
check_integrity: bool = False, overwrite_config: bool = False):
self.out_dir = out_dir
self.log_q = log_q
self.prog_q = prog_q
self._cancel = False
# Integrity mode: aria2 hash-checks every file's pieces and repairs
# them (catches same-size corruption size-based selection misses).
self.check_integrity = check_integrity
# Write a fresh Config.wtf on a reconcile.
self.overwrite_config = overwrite_config
def cancel(self):
self._cancel = True
def log(self, msg: str, tag: str = ""):
self.log_q.put((msg, tag))
def progress(self, value: float, label: str = "", status: str | None = None):
# status (when given) updates the big status line; None leaves it as-is.
self.prog_q.put((value, label, status))
def build_tweaks(self, buf, tweaks: dict | None = None):
if tweaks is None:
tweaks = load_tweaks_config()
fov_deg = tweaks.get("fieldOfView", TWEAKS_DEFAULTS["fieldOfView"])
fov = fov_deg * (math.pi / 180.0)
flags = struct.unpack_from("<H", buf, 0x126)[0] | 0x20
nameplate = float(tweaks.get("nameplateRange", TWEAKS_DEFAULTS["nameplateRange"]))
far_clip = float(tweaks.get("farClip", TWEAKS_DEFAULTS["farClip"]))
frill = float(tweaks.get("frillDistance", TWEAKS_DEFAULTS["frillDistance"]))
cam_dist = float(tweaks.get("cameraDistance", TWEAKS_DEFAULTS["cameraDistance"]))
snd_bg = 0x27 if tweaks.get("soundInBackground", TWEAKS_DEFAULTS["soundInBackground"]) else 0x14
always_loot = tweaks.get("alwaysAutoLoot", TWEAKS_DEFAULTS["alwaysAutoLoot"])
locale = tweaks.get("locale", TWEAKS_DEFAULTS["locale"])
# fmt: off
return [
("gameLanguage", "bytes", None, locale_patches(locale)),
("largeAddress", "uint16", 0x126, flags),
("fieldOfView", "float", 0x4089b4, fov),
("cameraDistance", "float", 0x4089a4, cam_dist),
("farClip", "float", 0x40fed8, far_clip),
("frillDistance", "float", 0x467958, frill),
("nameplateRange", "float", 0x40c448, nameplate),
("soundInBackground", "int8", 0x3a4869, snd_bg),
("alwaysAutoLoot", "bytes", None, [
(0x0c1ecf, bytes([0x75 if always_loot else 0x74])),
(0x0c2b25, bytes([0x75 if always_loot else 0x74])),
]),
# cameraSkipFix is baked into the torrent's WoW.exe, so we don't
# apply it. skillUiGateHijack and octowowUrlAllowlist below are
# baked in too, but the official launcher still applies these 2
# specific patches, so we mirror it in case the Octo devs drop them
# from WoW.exe again.
("octowowUrlAllowlist", "bytes", None, [
(0x45ccd8, bytes([
0x6f,0x63,0x74,0x6f,0x77,0x6f,0x77,0x2e,0x73,0x74,
0x00,0x00,0x00,0x00,0x00,0x00,
])),
]),
("skillUiGateHijack", "bytes", None, [
(0x002ddf90, bytes([
0x55,0x8b,0xec,0x83,0xec,0x08,0x53,0x56,0x57,0x8b,0x3d,0x60,0xab,0xce,0x00,0x83,
0xff,0xff,0x89,0x55,0xfc,0x89,0x4d,0xf8,0x74,0x79,0x8b,0x75,0x08,0x8b,0x15,0x58,
0xab,0xce,0x00,0x8b,0xc7,0x23,0xc6,0x8d,0x04,0x40,0x8b,0x4c,0x82,0x08,0xf6,0xc1,
0x01,0x8d,0x44,0x82,0x04,0x75,0x04,0x85,0xc9,0x75,0x05,0x33,0xc9,0x8d,0x49,0x00,
0xf6,0xc1,0x01,0x75,0x4e,0x85,0xc9,0x74,0x4a,0x39,0x31,0x74,0x13,0x8b,0xc7,0x23,
0xc6,0x8d,0x04,0x40,0x8d,0x04,0x82,0x8b,0x00,0x03,0xc1,0x8b,0x48,0x04,0xeb,0xe0,
0x8b,0x59,0x1c,0x8b,0x71,0x18,0x33,0xff,0x85,0xdb,0x7e,0x27,0x8d,0x64,0x24,0x00,
0x8b,0x4e,0x0c,0x8b,0x56,0x08,0x6a,0x00,0x6a,0x00,0x51,0x8b,0x4d,0xf8,0x52,0x8b,
0x55,0xfc,0xe8,0xb9,0xfd,0xff,0xff,0x84,0xc0,0x75,0x13,0x47,0x83,0xc6,0x20,0x3b,
0xfb,0x7c,0xdd,0x5f,0x5e,0x33,0xc0,0x5b,0x8b,0xe5,0x5d,0xc2,0x04,0x00,0x5f,0x8b,
0xc6,0x5e,0x5b,0x8b,0xe5,0x5d,0xc2,0x04,0x00,0x90,0x90,0x90,0x90,0x90,0x90,0x90,
])),
]),
]
# fmt: on
def patch_exe(self, tweaks: dict | None = None):
exe = os.path.join(self.out_dir, "WoW.exe")
if not os.path.exists(exe):
raise RuntimeError(f"WoW.exe not found in {self.out_dir}")
self.log("\nApplying binary tweaks to WoW.exe…")
# Patch the pristine (unpatched) base rather than the on-disk exe, so a
# re-patch (tweak or language change) never stacks on patched bytes.
buf = bytearray(read_pristine_wow(self.out_dir))
for label, kind, offset, value in self.build_tweaks(buf, tweaks):
self.log(f" {label}", "dim")
if kind == "float":
struct.pack_into("<f", buf, offset, value)
elif kind == "int8":
struct.pack_into("<b", buf, offset, value)
elif kind == "uint16":
struct.pack_into("<H", buf, offset, value)
elif kind == "bytes":
for off, data in value:
buf[off: off + len(data)] = data
with open(exe, "wb") as f:
f.write(buf)
self.log("WoW.exe patched.", "ok")
def run(self):
# The selection is recomputed here from the live torrent so an
# interrupted sync always resumes against the current file set.
try:
self.log("\nStarting client sync…\n", "acct")
self.progress(0.0, "Preparing…")
raw, files = fetch_torrent()
version = torrent_version(raw)
# aria2's saved control state pins a torrent revision and its
# completed pieces. When the torrent was re-rolled (new identity) or
# the folder changed, that state is stale — an 'info hash mismatch'
# error, or skipped re-downloads. Clear it and (for a new revision)
# drop wrong-sized files so they re-fetch clean.
cfg = load_config()
stale = (cfg.get("active_torrent_hash") != version or
cfg.get("active_client_dir") != os.path.abspath(self.out_dir))
if stale:
clear_torrent_resume_state()
update_config(lambda c: c.update({
"active_torrent_hash": version,
"active_client_dir": os.path.abspath(self.out_dir)}))
# Config.wtf is user game config, not in the torrent — (re)write it
# on a reconcile (overwrite_config), or when missing.
cfg_wtf = os.path.join(self.out_dir, "WTF", "Config.wtf")
if self.overwrite_config or not os.path.exists(cfg_wtf):
write_config_wtf(self.out_dir)
# WoW.exe on disk is patched, so an integrity check always flags it
# and re-fetches its pieces over the network. Restore the cached
# pristine base first: the check then passes with no re-download when
# the client is unchanged (aria2 still repairs it if the torrent's
# WoW.exe genuinely changed). It's re-patched after the check.
wow_path = os.path.join(self.out_dir, "WoW.exe")
if (self.check_integrity and os.path.exists(PRISTINE_WOW_PATH)
and os.path.exists(wow_path)):
shutil.copyfile(PRISTINE_WOW_PATH, wow_path)
# Put back any file a prior run shielded but couldn't restore (crash
# mid-sync), so its version isn't stranded as a .octobak.
recover_protected_files(self.out_dir, files)
# speech.MPQ is left unverified/un-updated when the user keeps a
# custom one (Settings → Ignore speech.mpq).
ignore_speech = bool(load_config().get("ignore_speech", False))
if self.check_integrity:
# Full piece-hash verify + repair of every non-mod file — aria2
# re-hashes them and re-fetches only the bad/missing pieces.
need, missing = torrent_all_selection(
self.out_dir, files, ignore_speech=ignore_speech), False
else:
need, missing = torrent_selection(
self.out_dir, files, drop_mismatched=stale,
ignore_speech=ignore_speech)
# A file the user deleted leaves its pieces marked done in the
# control file, so aria2 skips it forever — clear resume state.
if missing:
clear_torrent_resume_state()
wow_downloaded = any(files[i - 1][0] == ["WoW.exe"] for i in need)
if need:
self.log(
(f"Verifying {len(need)} file(s) via torrent…"
if self.check_integrity
else f"Syncing {len(need)} file(s) via torrent…"), "acct")
# A reconcile runs two aria2 phases: it hash-checks every file
# (p["checking"]), then fetches the bad/missing pieces. Flip the
# status from "Verifying" to "Updating" between them so it doesn't
# read "Verifying" while files are actually being updated. Only
# sent on a phase change.
_phase = {"status": None}
def _prog(p):
label = f"{fmt_size(p['done'])} / {fmt_size(p['total'])}"
if p["bps"]:
label += " • " + fmt_speed(p["bps"])
frac = p["done"] / p["total"] if p["total"] else 0.0
status = None
if self.check_integrity:
want = ("Verifying game files…" if p["checking"]
else "Updating game files…")
if want != _phase["status"]:
_phase["status"] = status = want
self.progress(min(frac, 1.0), label, status)
# Excluded files (mod-owned + kept speech.MPQ) can still be
# rewritten by aria2 through a shared torrent piece. Move them
# aside for the sync and put them back after, so the Mods tab /
# custom speech stays authoritative.
shielded = shield_protected_files(self.out_dir, files,
ignore_speech)
try:
run_aria2c(self.out_dir, select_files=need,
check_integrity=self.check_integrity,
on_progress=_prog,
should_cancel=lambda: self._cancel,
log_fn=self.log)
finally:
for name in unshield_protected_files(shielded):
self.log(f" kept mod file: {name}", "dim")
else:
self.log("All game files already present.", "dim")
if self._cancel:
self.log("\nUpdate cancelled.", "err")
self.progress(0.0, "Cancelled")
self.log_q.put(("__ERROR__", ""))
return
self.progress(1.0, "Verifying…")
if not torrent_tree_intact(self.out_dir, files,
ignore_speech=ignore_speech):
self.log("\n✗ Download incomplete — click Update to finish.", "err")
self.log_q.put(("__ERROR__", ""))
return
self.log("\nDownload complete.", "ok")
remove_wdb(self.out_dir)
# Drop legacy leftovers the current torrent no longer ships
for gone in prune_stale_client_files(self.out_dir, files):
self.log(f"Removed legacy file: {gone}", "dim")
# Cache the fresh pristine exe, then patch WoW.exe from that
# clean base — but only when the sync actually (re)downloaded it.
refresh_pristine_wow(self.out_dir)
if wow_downloaded:
self.progress(1.0, "Patching…")
self.patch_exe()
else:
self.log("\nWoW.exe unchanged — skipping patch.", "dim")
self.progress(1.0, "")
self.log("\n✓ Everything is up to date!", "ok")
client_ver = get_client_version(self.out_dir)
if client_ver:
self.log(f"Client version: {client_ver}", "dim")
self.log_q.put((f"__VERSION__{client_ver}", ""))
else:
self.log("Could not read client version from WoW.exe", "dim")
self.log_q.put(("__DONE__", ""))
except Exception as e:
self.log(f"\n✗ {e}", "err")
self.progress(0.0, "")
self.log_q.put(("__ERROR__", ""))