Files
comfyui_o1key/tests/test_updater.py
T

124 lines
6.0 KiB
Python

"""Offline Git checks for the sidebar updater."""
import importlib.util
import subprocess
import tempfile
import unittest
from pathlib import Path
UPDATER_PATH = Path(__file__).resolve().parents[1] / "utils" / "updater.py"
spec = importlib.util.spec_from_file_location("o1key_updater_under_test", UPDATER_PATH)
updater = importlib.util.module_from_spec(spec)
spec.loader.exec_module(updater)
class UpdaterTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
root = Path(self.temp.name)
self.remote = root / "remote.git"
self.author = root / "author"
self.install = root / "install"
self.git(root, "init", "--bare", str(self.remote))
self.git(root, "clone", str(self.remote), str(self.author))
self.git(self.author, "config", "user.email", "[email protected]")
self.git(self.author, "config", "user.name", "Updater Test")
self.git(self.author, "switch", "-c", "main")
(self.author / "requirements.txt").write_text("requests>=2\n", encoding="utf-8")
(self.author / "version.txt").write_text("1\n", encoding="utf-8")
self.commit_and_push()
self.git(root, "clone", "--branch", "main", str(self.remote), str(self.install))
self.previous_dir = updater.PLUGIN_DIR
self.previous_url = updater.RELEASE_REPOSITORY_URL
updater.PLUGIN_DIR = self.install
updater.RELEASE_REPOSITORY_URL = str(self.remote)
self.addCleanup(setattr, updater, "PLUGIN_DIR", self.previous_dir)
self.addCleanup(setattr, updater, "RELEASE_REPOSITORY_URL", self.previous_url)
def git(self, cwd, *args):
return subprocess.run(
["git", *args], cwd=cwd, check=True, capture_output=True, text=True,
).stdout.strip()
def commit_and_push(self):
self.git(self.author, "add", ".")
self.git(self.author, "commit", "-m", "test update")
self.git(self.author, "push", "origin", "main")
def test_fast_forward_and_requirements_change(self):
self.assertFalse(updater.check_for_update()["update_available"])
self.assertFalse(updater.update_package()["updated"])
# The release URL, not the user's origin, determines the update source.
self.git(self.install, "remote", "set-url", "origin", str(self.install / "unused-origin"))
(self.author / "requirements.txt").write_text("requests>=3\n", encoding="utf-8")
self.commit_and_push()
self.assertTrue(updater.check_for_update()["update_available"])
self.assertEqual((self.install / "requirements.txt").read_text(encoding="utf-8"), "requests>=2\n")
result = updater.update_package()
self.assertTrue(result["updated"])
self.assertTrue(result["requirements_changed"])
self.assertEqual((self.install / "requirements.txt").read_text(encoding="utf-8"), "requests>=3\n")
def test_local_modification_is_preserved(self):
(self.install / "version.txt").write_text("local work\n", encoding="utf-8")
self.assertFalse(updater.check_for_update()["update_available"])
self.assertFalse(updater.update_package()["updated"])
(self.author / "version.txt").write_text("new release\n", encoding="utf-8")
self.commit_and_push()
with self.assertRaises(updater.UpdateError) as caught:
updater.check_for_update()
self.assertEqual(caught.exception.code, "local_changes")
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "local_changes")
self.assertEqual((self.install / "version.txt").read_text(encoding="utf-8"), "local work\n")
def test_diverged_branch_is_preserved(self):
self.git(self.install, "config", "user.email", "[email protected]")
self.git(self.install, "config", "user.name", "Updater Test")
(self.install / "version.txt").write_text("local commit\n", encoding="utf-8")
self.git(self.install, "add", ".")
self.git(self.install, "commit", "-m", "local")
(self.author / "version.txt").write_text("remote commit\n", encoding="utf-8")
self.commit_and_push()
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "diverged")
self.assertEqual((self.install / "version.txt").read_text(encoding="utf-8"), "local commit\n")
def test_wrong_branch_is_reported(self):
self.git(self.install, "switch", "-c", "feature")
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "wrong_branch")
def test_zip_style_install_is_reported(self):
updater.PLUGIN_DIR = Path(self.temp.name) / "unpacked"
updater.PLUGIN_DIR.mkdir()
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "not_git")
self.assertIn("suggestion", caught.exception.as_dict())
def test_missing_release_repo_is_reported(self):
updater.RELEASE_REPOSITORY_URL = str(self.install / "missing-release.git")
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "fetch_failed")
self.assertEqual(caught.exception.status, 503)
def test_untracked_collision_is_preserved(self):
(self.install / "collision.txt").write_text("local file\n", encoding="utf-8")
(self.author / "collision.txt").write_text("release file\n", encoding="utf-8")
self.commit_and_push()
with self.assertRaises(updater.UpdateError) as caught:
updater.update_package()
self.assertEqual(caught.exception.code, "merge_blocked")
self.assertEqual((self.install / "collision.txt").read_text(encoding="utf-8"), "local file\n")
if __name__ == "__main__":
unittest.main()