400 lines
15 KiB
Python
400 lines
15 KiB
Python
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import PropertyMock, patch
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
if str(ROOT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(ROOT_DIR))
|
|
|
|
import app.updater.update_manager as updater_manager
|
|
from app.updater.updater_gui import ConfigError as UpdaterConfigError
|
|
from app.updater.updater_gui import UpdateConfig, UpdaterGUI
|
|
|
|
ConfigError = updater_manager.ConfigError
|
|
UpdateManager = updater_manager.UpdateManager
|
|
VersionInfo = updater_manager.VersionInfo
|
|
compare_versions = updater_manager.compare_versions
|
|
create_update_manager_from_settings = updater_manager.create_update_manager_from_settings
|
|
|
|
|
|
class TestUpdateManager(unittest.TestCase):
|
|
def test_compare_versions_basic_cases(self):
|
|
self.assertEqual(compare_versions("3.2.0", "3.1.9"), 1)
|
|
self.assertEqual(compare_versions("3.2", "3.2.0"), 0)
|
|
self.assertEqual(compare_versions("3.2.0", "3.2.1"), -1)
|
|
|
|
def test_check_for_updates_raises_when_supabase_config_missing(self):
|
|
manager = UpdateManager(supabase_url="", supabase_key="")
|
|
with self.assertRaises(ConfigError):
|
|
manager.check_for_updates()
|
|
|
|
def test_load_updater_connection_config_uses_fallback_when_remote_fails(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = Path(td) / "config.json"
|
|
cfg.write_text(
|
|
json.dumps(
|
|
{
|
|
"config_url": "https://invalid.example/config",
|
|
"default_environment": "main",
|
|
"fallback": {
|
|
"supabase_url": "https://kong.m1tcloud.cc",
|
|
"anon_key": "fallback-key",
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with patch("app.updater.update_manager.requests.get", side_effect=Exception("network down")):
|
|
conn = getattr(updater_manager, "load_updater_connection_config")(config_path=cfg, environment="main")
|
|
|
|
self.assertEqual(conn["source"], "fallback")
|
|
self.assertEqual(conn["supabase_url"], "https://kong.m1tcloud.cc")
|
|
self.assertEqual(conn["supabase_key"], "fallback-key")
|
|
|
|
def test_load_updater_connection_config_uses_remote_when_available(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = Path(td) / "config.json"
|
|
cfg.write_text(
|
|
json.dumps(
|
|
{
|
|
"config_url": "https://jwt.m1tcloud.cc/config",
|
|
"default_environment": "main",
|
|
"fallback": {
|
|
"supabase_url": "https://fallback.example",
|
|
"anon_key": "fallback-key",
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
remote_payload = {
|
|
"defaultEnvironment": "main",
|
|
"environments": {
|
|
"main": {
|
|
"enabled": True,
|
|
"priority": 1,
|
|
"supabaseUrl": "https://kong.m1tcloud.cc",
|
|
"anonKey": "remote-key",
|
|
}
|
|
},
|
|
}
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
@property
|
|
def text(self):
|
|
return json.dumps(remote_payload, ensure_ascii=False)
|
|
|
|
with patch("app.updater.update_manager.requests.get", return_value=_Resp()):
|
|
conn = getattr(updater_manager, "load_updater_connection_config")(config_path=cfg, environment="main")
|
|
|
|
self.assertEqual(conn["source"], "remote")
|
|
self.assertEqual(conn["supabase_url"], "https://kong.m1tcloud.cc")
|
|
self.assertEqual(conn["supabase_key"], "remote-key")
|
|
|
|
def test_create_update_manager_from_settings_reads_connection_config(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = Path(td) / "config.json"
|
|
cfg.write_text(
|
|
json.dumps(
|
|
{
|
|
"default_environment": "main",
|
|
"fallback": {
|
|
"supabase_url": "https://kong.m1tcloud.cc",
|
|
"anon_key": "fallback-key",
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
settings = {
|
|
"update": {
|
|
"connection_config_path": str(cfg),
|
|
"environment": "main",
|
|
"program_id": "voc_monitor",
|
|
"check_interval_hours": 2,
|
|
}
|
|
}
|
|
|
|
manager = create_update_manager_from_settings(settings)
|
|
self.assertEqual(manager.supabase_url, "https://kong.m1tcloud.cc")
|
|
self.assertEqual(manager.supabase_key, "fallback-key")
|
|
self.assertEqual(manager.check_interval, 2)
|
|
self.assertEqual(getattr(manager, "version_table"), "program_version")
|
|
|
|
def test_check_for_updates_fallbacks_table_name_on_404(self):
|
|
manager = UpdateManager(
|
|
supabase_url="https://kong.m1tcloud.cc",
|
|
supabase_key="dummy-key",
|
|
current_version="3.0.0",
|
|
)
|
|
setattr(manager, "version_table", "program_version")
|
|
|
|
class _Resp404:
|
|
status_code = 404
|
|
|
|
def raise_for_status(self):
|
|
import requests
|
|
|
|
raise requests.HTTPError("404 not found")
|
|
|
|
def json(self):
|
|
return {"message": "not found"}
|
|
|
|
class _Resp200:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return [
|
|
{
|
|
"version": "3.5.5",
|
|
"is_stable": True,
|
|
"release_note": "test",
|
|
"download_url": "https://example.com/update.zip",
|
|
"min_required_version": "3.0.0",
|
|
}
|
|
]
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def fake_get(*_args, **_kwargs):
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
return _Resp404()
|
|
return _Resp200()
|
|
|
|
with patch("app.updater.update_manager.requests.get", side_effect=fake_get):
|
|
info = manager.check_for_updates()
|
|
|
|
self.assertIsNotNone(info)
|
|
self.assertEqual(getattr(info, "version"), "3.5.5")
|
|
self.assertEqual(getattr(manager, "version_table"), "program_versions")
|
|
|
|
def test_prepare_update_fails_when_updater_exe_missing(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
temp_dir = Path(td)
|
|
manager = UpdateManager(
|
|
supabase_url="https://example.supabase.co",
|
|
supabase_key="dummy-key",
|
|
)
|
|
version_info = VersionInfo(
|
|
version="9.9.9",
|
|
is_stable=True,
|
|
download_url="https://example.com/update.zip",
|
|
)
|
|
|
|
with patch.object(UpdateManager, "install_dir", new_callable=PropertyMock, return_value=temp_dir):
|
|
ok, message = manager.prepare_update(version_info)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("updater.exe", message)
|
|
|
|
def test_prepare_update_writes_config_and_copies_updater(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
temp_dir = Path(td)
|
|
manager = UpdateManager(
|
|
supabase_url="https://example.supabase.co",
|
|
supabase_key="dummy-key",
|
|
)
|
|
version_info = VersionInfo(
|
|
version="9.9.9",
|
|
is_stable=True,
|
|
download_url="https://example.com/update.zip",
|
|
)
|
|
|
|
fake_updater = temp_dir / "updater.exe"
|
|
fake_updater.write_bytes(b"fake-updater")
|
|
config_path = temp_dir / "voc_updater_config.json"
|
|
temp_updater_path = temp_dir / "updater_temp.exe"
|
|
|
|
with patch.object(UpdateManager, "install_dir", new_callable=PropertyMock, return_value=temp_dir), patch.object(
|
|
UpdateManager,
|
|
"config_path",
|
|
new_callable=PropertyMock,
|
|
return_value=config_path,
|
|
), patch.object(
|
|
UpdateManager,
|
|
"temp_updater_path",
|
|
new_callable=PropertyMock,
|
|
return_value=temp_updater_path,
|
|
):
|
|
ok, message = manager.prepare_update(version_info)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertIn("완료", message)
|
|
self.assertTrue(config_path.exists())
|
|
self.assertTrue(temp_updater_path.exists())
|
|
|
|
config_data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(config_data["version"], "9.9.9")
|
|
self.assertEqual(config_data["download_url"], "https://example.com/update.zip")
|
|
|
|
def test_prepare_update_returns_permission_error_when_copy_fails(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
temp_dir = Path(td)
|
|
manager = UpdateManager(
|
|
supabase_url="https://example.supabase.co",
|
|
supabase_key="dummy-key",
|
|
)
|
|
version_info = VersionInfo(
|
|
version="9.9.9",
|
|
is_stable=True,
|
|
download_url="https://example.com/update.zip",
|
|
)
|
|
|
|
fake_updater = temp_dir / "updater.exe"
|
|
fake_updater.write_bytes(b"fake-updater")
|
|
config_path = temp_dir / "voc_updater_config.json"
|
|
temp_updater_path = temp_dir / "updater_temp.exe"
|
|
|
|
with patch.object(UpdateManager, "install_dir", new_callable=PropertyMock, return_value=temp_dir), patch.object(
|
|
UpdateManager,
|
|
"config_path",
|
|
new_callable=PropertyMock,
|
|
return_value=config_path,
|
|
), patch.object(
|
|
UpdateManager,
|
|
"temp_updater_path",
|
|
new_callable=PropertyMock,
|
|
return_value=temp_updater_path,
|
|
), patch("app.updater.update_manager.shutil.copy2", side_effect=PermissionError("denied")):
|
|
ok, message = manager.prepare_update(version_info)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("권한", message)
|
|
|
|
|
|
def build_updater_gui_stub() -> UpdaterGUI:
|
|
gui = object.__new__(UpdaterGUI)
|
|
gui.CONFIG_FILENAME = "voc_updater_config.json"
|
|
gui._update_status = lambda *_args, **_kwargs: None
|
|
gui._update_progress = lambda *_args, **_kwargs: None
|
|
gui.config = None
|
|
gui.download_path = None
|
|
return gui
|
|
|
|
|
|
class TestUpdaterGUI(unittest.TestCase):
|
|
def test_load_config_validates_required_fields(self):
|
|
gui = build_updater_gui_stub()
|
|
config_path = Path(tempfile.gettempdir()) / gui.CONFIG_FILENAME
|
|
config_path.write_text(
|
|
json.dumps({"download_url": "", "target_path": "", "version": ""}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
try:
|
|
with self.assertRaises(UpdaterConfigError):
|
|
gui._load_config()
|
|
finally:
|
|
if config_path.exists():
|
|
config_path.unlink()
|
|
|
|
def test_extract_and_replace_success(self):
|
|
gui = build_updater_gui_stub()
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
target_dir = root / "app_dir"
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
(target_dir / "old.txt").write_text("old", encoding="utf-8")
|
|
|
|
zip_file = root / "update.zip"
|
|
source_dir = root / "zip_source"
|
|
source_dir.mkdir(parents=True, exist_ok=True)
|
|
(source_dir / "new.txt").write_text("new", encoding="utf-8")
|
|
|
|
import zipfile
|
|
|
|
with zipfile.ZipFile(zip_file, "w") as zf:
|
|
zf.write(source_dir / "new.txt", arcname="new.txt")
|
|
|
|
gui.config = UpdateConfig(
|
|
download_url="https://example.com/update.zip",
|
|
target_path=str(target_dir),
|
|
version="9.9.9",
|
|
restart_exe="voc_noti.exe",
|
|
)
|
|
|
|
ok, _msg = gui._extract_and_replace(zip_file)
|
|
self.assertTrue(ok)
|
|
self.assertTrue((target_dir / "new.txt").exists())
|
|
|
|
def test_extract_and_replace_fails_on_bad_zip(self):
|
|
gui = build_updater_gui_stub()
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
target_dir = root / "app_dir"
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
bad_zip = root / "bad.zip"
|
|
bad_zip.write_text("not-a-zip", encoding="utf-8")
|
|
|
|
gui.config = UpdateConfig(
|
|
download_url="https://example.com/update.zip",
|
|
target_path=str(target_dir),
|
|
version="9.9.9",
|
|
restart_exe="voc_noti.exe",
|
|
)
|
|
|
|
ok, message = gui._extract_and_replace(bad_zip)
|
|
self.assertFalse(ok)
|
|
self.assertIn("손상된 zip", message)
|
|
|
|
def test_extract_and_replace_rolls_back_on_copy_error(self):
|
|
gui = build_updater_gui_stub()
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
target_dir = root / "app_dir"
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
(target_dir / "old.txt").write_text("old", encoding="utf-8")
|
|
|
|
zip_file = root / "update.zip"
|
|
source_dir = root / "zip_source"
|
|
source_dir.mkdir(parents=True, exist_ok=True)
|
|
(source_dir / "new.txt").write_text("new", encoding="utf-8")
|
|
|
|
import zipfile
|
|
|
|
with zipfile.ZipFile(zip_file, "w") as zf:
|
|
zf.write(source_dir / "new.txt", arcname="new.txt")
|
|
|
|
gui.config = UpdateConfig(
|
|
download_url="https://example.com/update.zip",
|
|
target_path=str(target_dir),
|
|
version="9.9.9",
|
|
restart_exe="voc_noti.exe",
|
|
)
|
|
|
|
with patch("app.updater.updater_gui.shutil.copy2", side_effect=PermissionError("denied")):
|
|
ok, message = gui._extract_and_replace(zip_file)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("denied", message)
|
|
self.assertTrue((target_dir / "old.txt").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|