Microchip/octo-updater
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more unit tests

Browse Source
This commit is contained in:
Microchip
2026-09-26 14:11:59 -04:00
parent 8ed33d52e3
commit 82f9342918
8 changed files with 1149 additions and 1158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client.py
+1 -1
View File
@@ -20,7 +20,7 @@ from pathlib import Path
from .app_log import log
from .config import load_config, update_config
from .platform_ops import display_info
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)."""
src/config.py
-2
View File
@@ -112,8 +112,6 @@ def update_config(mutator):
return cfg
def ensure_dir(path):
Path(path).mkdir(parents=True, exist_ok=True)
# ── logging ─────────────────────────────────────────────────────────────────
src/mods.py
+2 -1
View File
@@ -22,7 +22,8 @@ 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 .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 = [
src/mpq_patches.py
+2 -1
View File
@@ -4,7 +4,8 @@ import hashlib
import os
import urllib.request
from .config import UA, DOWNLOAD_TIMEOUT, ensure_dir
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
src/platform_ops.py
+4
View File
@@ -10,6 +10,10 @@ import platform
from .app_log import log
# 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:
src/tests/test_platform_ops.py
+720 -866
View File
File diff suppressed because it is too large Load Diff
src/tests/test_torrent.py
+418 -285
View File
@@ -1,352 +1,485 @@
import hashlib
import os
from unittest.mock import Mock, patch
from unittest.mock import Mock
import pytest
from src import torrent
from pathlib import Path
def bencode(value):
if isinstance(value, int):
return b"i" + str(value).encode() + b"e"
if isinstance(value, bytes):
return str(len(value)).encode() + b":" + value
if isinstance(value, list):
return b"l" + b"".join(bencode(v) for v in value) + b"e"
if isinstance(value, dict):
return b"d" + b"".join(bencode(k) + bencode(v)
for k, v in sorted(value.items())) + b"e"
raise TypeError(type(value))
def test_bdecode_nested_values():
raw = bencode({b"a": 3, b"b": [b"x", 7]})
decoded, pos = torrent._bdecode(raw)
assert decoded == {b"a": 3, b"b": [b"x", 7]}
assert pos == len(raw)
# ---------------------------------------------------------------------------
# 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.platform_ops,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == expected
installer.assert_not_called()
def test_torrent_version_is_sha1_of_raw_bytes():
raw = b"torrent bytes"
def test_ensure_aria2c_returns_windows_path(monkeypatch):
expected = r"C:\aria2\aria2c.exe"
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=expected),
)
assert torrent.ensure_aria2c() == expected
def test_ensure_aria2c_installs_when_not_found(monkeypatch):
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent.platform_ops,
"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_passes_custom_logger(monkeypatch):
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent.platform_ops,
"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):
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=""),
)
installer = Mock(
return_value="/fake/aria2c.exe"
)
monkeypatch.setattr(
torrent.platform_ops,
"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):
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
monkeypatch.setattr(
torrent.platform_ops,
"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_fetch_torrent_extracts_paths_and_lengths():
raw = bencode({
b"info": {
b"files": [
{b"length": 10, b"path": [b"WoW.exe"]},
{b"length": 20, b"path": [b"Data", b"patch.MPQ"]},
]
}
})
response = Mock()
response.read.return_value = raw
cm = Mock()
cm.__enter__ = Mock(return_value=response)
cm.__exit__ = Mock(return_value=False)
with patch.object(torrent, "secure_urlopen", return_value=cm):
got_raw, files = torrent.fetch_torrent()
assert got_raw == raw
assert files == [(["WoW.exe"], 10), (["Data", "patch.MPQ"], 20)]
def test_torrent_version_changes_when_data_changes():
assert (
torrent.torrent_version(b"one")
!= torrent.torrent_version(b"two")
)
def test_torrent_selection_finds_missing_and_wrong_size(tmp_path):
files = [(["a.bin"], 3), (["b.bin"], 4), (["c.bin"], 5)]
(tmp_path / "a.bin").write_bytes(b"abc")
(tmp_path / "b.bin").write_bytes(b"xx")
# ---------------------------------------------------------------------------
# torrent_selection
# ---------------------------------------------------------------------------
with patch.object(torrent, "_torrent_skip", return_value=False):
need, missing = torrent.torrent_selection(str(tmp_path), files)
def test_torrent_selection_empty_when_files_match(tmp_path):
file = tmp_path / "file.bin"
file.write_bytes(b"1234")
assert need == [2, 3]
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_removes_oversized_file(tmp_path):
p = tmp_path / "a.bin"
p.write_bytes(b"too long")
with patch.object(torrent, "_torrent_skip", return_value=False):
need, missing = torrent.torrent_selection(
str(tmp_path), [(["a.bin"], 2)]
)
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 not p.exists()
assert file.exists()
def test_torrent_all_selection_skips_protected(tmp_path):
files = [(["a"], 1), (["b"], 1), (["c"], 1)]
with patch.object(torrent, "_torrent_skip",
side_effect=[False, True, False]):
assert torrent.torrent_all_selection(str(tmp_path), files) == [1, 3]
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_tree_intact(tmp_path):
(tmp_path / "a").write_bytes(b"123")
with patch.object(torrent, "_torrent_skip", return_value=False):
assert torrent.torrent_tree_intact(
str(tmp_path), [(["a"], 3)]
)
assert not torrent.torrent_tree_intact(
str(tmp_path), [(["a"], 4)]
)
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()
def test_shield_and_unshield_existing_file(tmp_path):
original = tmp_path / "mod.dll"
original.write_bytes(b"ours")
files = [(["mod.dll"], 4)]
# ---------------------------------------------------------------------------
# torrent_tree_intact
# ---------------------------------------------------------------------------
with patch.object(torrent, "_torrent_skip", return_value=True):
shielded = torrent.shield_protected_files(
str(tmp_path), files, False
)
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")
assert not original.exists()
assert (tmp_path / "mod.dll.octobak").exists()
files = [
(["one.bin"], 3),
(["two.bin"], 5),
]
# Simulate aria2 creating/replacing the original path.
original.write_bytes(b"aria")
restored = torrent.unshield_protected_files(shielded)
assert torrent.torrent_tree_intact(
str(tmp_path),
files,
) is True
assert original.read_bytes() == b"ours"
assert restored == ["mod.dll"]
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):
p = tmp_path / "mod.dll"
shielded = [(str(p), None)]
p.write_bytes(b"stub")
stub = tmp_path / "protected.dll"
stub.write_bytes(b"aria partial data")
assert torrent.unshield_protected_files(shielded) == []
assert not p.exists()
restored = torrent.unshield_protected_files([
(str(stub), None),
])
assert not stub.exists()
assert restored == []
def test_recover_protected_files(tmp_path):
p = tmp_path / "x.dll"
bak = tmp_path / "x.dll.octobak"
bak.write_bytes(b"saved")
torrent.recover_protected_files(str(tmp_path), [(["x.dll"], 5)])
assert p.read_bytes() == b"saved"
assert not bak.exists()
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()
@pytest.mark.parametrize(
("line", "checking", "progress", "bps"),
[
("1MiB/4MiB(25%) DL:2MiB", False, .25, 2 * 1024**2),
("0B/4MiB(0%) [Checksum:#abc 2MiB/4MiB(50%)]",
True, .50, 0),
],
)
def test_parse_aria_progress(line, checking, progress, bps):
got = torrent.parse_aria_progress(line)
assert got["checking"] is checking
assert got["progress"] == progress
assert got["bps"] == bps
# ---------------------------------------------------------------------------
# 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_parse_aria_progress_ignores_non_progress_line():
assert torrent.parse_aria_progress("download started") is None
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()
def test_clear_torrent_resume_state(tmp_path):
(tmp_path / "a.aria2").write_text("x")
(tmp_path / "b.torrent").write_text("x")
(tmp_path / "keep.txt").write_text("x")
# ---------------------------------------------------------------------------
# parse_aria_progress
# ---------------------------------------------------------------------------
with patch.object(torrent, "TORRENT_STAGING_DIR", str(tmp_path)):
torrent.clear_torrent_resume_state()
def test_parse_aria_progress_download():
result = torrent.parse_aria_progress(
"1.2GiB/8.8GiB(13%) DL:5.0MiB"
)
assert not (tmp_path / "a.aria2").exists()
assert not (tmp_path / "b.torrent").exists()
assert (tmp_path / "keep.txt").exists()
assert result["progress"] == 0.13
assert result["bps"] == 5 * 1024 ** 2
assert result["checking"] is False
def test_ensure_aria2c_returns_existing_binary(tmp_path):
exe = tmp_path / "aria2c.exe"
exe.write_bytes(b"x")
with patch.object(torrent, "ARIA2C_PATH", str(exe)):
assert torrent.ensure_aria2c() == str(exe)
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_ensure_aria2c_rejects_bad_checksum(tmp_path):
response = Mock()
response.read.return_value = b"not the expected archive"
cm = Mock()
cm.__enter__ = Mock(return_value=response)
cm.__exit__ = Mock(return_value=False)
def test_parse_aria_progress_without_download_speed():
result = torrent.parse_aria_progress(
"512MiB/1GiB(50%)"
)
with patch.object(torrent, "ARIA2C_PATH", str(tmp_path / "aria2c.exe")), \
patch.object(torrent, "secure_urlopen", return_value=cm):
with pytest.raises(RuntimeError, match="checksum mismatch"):
torrent.ensure_aria2c(log_fn=Mock())
assert result["progress"] == 0.5
assert result["bps"] == 0.0
class TestEnsureAria2c:
def test_returns_existing_aria2c(self, monkeypatch):
"""An aria2c already discoverable on PATH should be used directly."""
expected = Path("/usr/bin/aria2c")
find_mock = Mock(return_value=expected)
monkeypatch.setattr(torrent, "find_aria2c", find_mock)
result = torrent.ensure_aria2c()
assert result == expected
find_mock.assert_called_once_with()
def test_returns_existing_windows_aria2c(self, monkeypatch):
"""A Windows aria2c.exe discovered on PATH should also be returned."""
expected = Path(r"C:\Program Files\aria2\aria2c.exe")
find_mock = Mock(return_value=expected)
monkeypatch.setattr(torrent, "find_aria2c", find_mock)
result = torrent.ensure_aria2c()
assert result == expected
find_mock.assert_called_once_with()
def test_falls_back_to_windows_installer(self, monkeypatch):
"""Missing aria2c should invoke install_aria2c_windows()."""
expected = Path(r"C:\AppData\OctoUpdater\aria2c.exe")
find_mock = Mock(return_value=None)
monkeypatch.setattr(torrent, "find_aria2c", find_mock)
# ensure_aria2c imports install_aria2c_windows inside the function,
# so patch the source module rather than torrent.
from src import platform_ops
installer = Mock(return_value=expected)
monkeypatch.setattr(
platform_ops,
"install_aria2c_windows",
installer,
def test_parse_aria_progress_returns_none_for_unrelated_line():
assert (
torrent.parse_aria_progress(
"Download complete"
)
is None
)
result = torrent.ensure_aria2c()
assert result == expected
find_mock.assert_called_once_with()
installer.assert_called_once()
# ---------------------------------------------------------------------------
# stop_aria2c
# ---------------------------------------------------------------------------
def test_passes_custom_logger_to_installer(self, monkeypatch):
"""The caller's logger should be forwarded to the installer."""
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
def test_stop_aria2c_terminates_running_process(monkeypatch):
proc = Mock()
proc.poll.return_value = None
from src import platform_ops
monkeypatch.setattr(
torrent,
"_active_aria2",
proc,
)
logger = Mock()
installer = Mock(return_value=Path("aria2c.exe"))
torrent.stop_aria2c()
monkeypatch.setattr(
platform_ops,
"install_aria2c_windows",
installer,
)
proc.terminate.assert_called_once_with()
assert torrent._active_aria2 is None
torrent.ensure_aria2c(log_fn=logger)
installer.assert_called_once_with(log_fn=logger)
def test_stop_aria2c_does_not_terminate_finished_process(monkeypatch):
proc = Mock()
proc.poll.return_value = 0
def test_does_not_install_when_aria2c_exists(self, monkeypatch):
"""Finding aria2c should short-circuit before installation."""
expected = Path("/usr/bin/aria2c")
monkeypatch.setattr(
torrent,
"_active_aria2",
proc,
)
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=expected),
)
torrent.stop_aria2c()
from src import platform_ops
proc.terminate.assert_not_called()
assert torrent._active_aria2 is None
installer = Mock()
monkeypatch.setattr(
platform_ops,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
def test_stop_aria2c_handles_no_process(monkeypatch):
monkeypatch.setattr(
torrent,
"_active_aria2",
None,
)
assert result == expected
installer.assert_not_called()
torrent.stop_aria2c()
def test_propagates_find_aria2c_error(self, monkeypatch):
"""Unexpected errors from executable discovery should propagate."""
find_mock = Mock(
side_effect=OSError("PATH lookup failed")
)
monkeypatch.setattr(
torrent,
"find_aria2c",
find_mock,
)
with pytest.raises(
OSError,
match="PATH lookup failed",
):
torrent.ensure_aria2c()
def test_propagates_installer_error(self, monkeypatch):
"""Installation failures should propagate to the caller."""
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=None),
)
from src import platform_ops
installer = Mock(
side_effect=RuntimeError("download failed")
)
monkeypatch.setattr(
platform_ops,
"install_aria2c_windows",
installer,
)
with pytest.raises(
RuntimeError,
match="download failed",
):
torrent.ensure_aria2c()
def test_empty_string_from_find_triggers_install(self, monkeypatch):
"""Any false-y discovery result should trigger installation."""
monkeypatch.setattr(
torrent,
"find_aria2c",
Mock(return_value=""),
)
from src import platform_ops
expected = Path("aria2c.exe")
installer = Mock(return_value=expected)
monkeypatch.setattr(
platform_ops,
"install_aria2c_windows",
installer,
)
result = torrent.ensure_aria2c()
assert result == expected
installer.assert_called_once()
assert torrent._active_aria2 is None
src/torrent.py
+2 -2
View File
@@ -9,10 +9,10 @@ import urllib.request
from functools import cache
from .app_log import log
from .config import APP_DATA_DIR, UA, DOWNLOAD_TIMEOUT, ensure_dir
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
from .platform_ops import create_directory_link, ensure_dir
CLIENT_TORRENT_URL = "https://dl.octowow.st/download/client.torrent"