Compare commits

...

3 Commits

Author SHA1 Message Date
Will Miao
28e7c04b37 fix(settings): migrate all settings subdirectories on portable mode switch 2026-06-29 21:40:37 +08:00
Will Miao
28f99c46d3 fix(update): preserve user data dirs during Git-based update via git clean -e excludes
git clean -fd in _perform_git_update deleted untracked, non-ignored
directories (wildcards, stats, backups, civitai, caches, logs) during
portable-mode updates, since released tags do not list them in .gitignore.
Add -e excludes for all user-managed paths to both nightly and stable
update branches. Add regression tests for both paths.
2026-06-29 21:10:38 +08:00
Will Miao
205194f4e6 chore: add stats, wildcards, backups, and logs dirs to .gitignore 2026-06-29 19:46:04 +08:00
5 changed files with 319 additions and 62 deletions

4
.gitignore vendored
View File

@@ -7,6 +7,10 @@ py/run_test.py
.vscode/
cache/
civitai/
stats/
wildcards/
backups/
logs/
node_modules/
coverage/
.coverage

View File

@@ -16,6 +16,27 @@ logger = logging.getLogger(__name__)
NETWORK_EXCEPTIONS = (ClientError, OSError, asyncio.TimeoutError)
# User-managed directories that live inside the plugin folder (portable
# mode) and must survive a Git-based update. ``git clean -fd`` would
# otherwise delete them because they are untracked and, in released tags,
# not listed in ``.gitignore``. ``-e`` excludes a path from cleaning
# regardless of whether it is ignored.
_PRESERVE_DIRS = ('settings.json', 'civitai', 'wildcards', 'backups', 'stats', 'logs', 'cache', 'model_cache')
def _clean_excludes() -> List[str]:
"""Build the ``-e`` arguments for ``git clean`` from :data:`_PRESERVE_DIRS`."""
excludes: List[str] = []
for name in _PRESERVE_DIRS:
excludes.append('-e')
excludes.append(name)
# For directories, also exclude nested matches explicitly
# (``-e dir`` alone matches the dir entry; ``-e dir/**`` guards
# contents under all git versions as defense-in-depth).
excludes.append('-e')
excludes.append(f'{name}/**')
return excludes
class UpdateRoutes:
"""Routes for handling plugin update checks"""
@@ -365,6 +386,8 @@ class UpdateRoutes:
)
return False, ""
clean_excludes = _clean_excludes()
try:
# Open the Git repository
repo = git.Repo(plugin_root)
@@ -376,8 +399,9 @@ class UpdateRoutes:
if nightly:
# Reset to discard any local changes
repo.git.reset('--hard')
# Clean untracked files
repo.git.clean('-fd')
# Clean untracked files, but preserve user-managed directories
# (wildcards, backups, stats, civitai, caches, settings.json).
repo.git.clean('-fd', *clean_excludes)
# Switch to main branch and pull latest
main_branch = 'main'
@@ -394,8 +418,9 @@ class UpdateRoutes:
else:
# Reset to discard any local changes
repo.git.reset('--hard')
# Clean untracked files
repo.git.clean('-fd')
# Clean untracked files, but preserve user-managed directories
# (wildcards, backups, stats, civitai, caches, settings.json).
repo.git.clean('-fd', *clean_excludes)
# Get latest release tag
tags = sorted(repo.tags, key=lambda t: t.commit.committed_datetime, reverse=True)

View File

@@ -1568,7 +1568,7 @@ class SettingsManager:
previous_dir = os.path.dirname(previous_path) or target_dir
if os.path.abspath(previous_path) != os.path.abspath(target_path):
self._copy_model_cache_directory(previous_dir, target_dir)
self._migrate_settings_directory_content(previous_dir, target_dir)
logger.info("Switching settings file to: %s", target_path)
self._pending_portable_switch = {"other_path": other_path}
@@ -1603,46 +1603,52 @@ class SettingsManager:
finally:
self._pending_portable_switch = None
def _copy_model_cache_directory(self, source_dir: str, target_dir: str) -> None:
"""Copy model_cache artifacts when switching storage locations."""
def _migrate_settings_directory_content(
self, source_dir: str, target_dir: str
) -> None:
"""Migrate settings directory subdirectories when switching storage locations.
Copies the canonical subdirectories (cache, backups, logs, stats, wildcards)
from the old settings directory to the new one. Legacy cache artifacts
(model_cache, recipe_cache, etc.) are migrated lazily by
``resolve_cache_path_with_migration`` on first access.
Args:
source_dir: The previous settings directory path.
target_dir: The new settings directory path.
"""
if not source_dir or not target_dir:
return
source_cache_dir = os.path.join(source_dir, "model_cache")
target_cache_dir = os.path.join(target_dir, "model_cache")
if os.path.isdir(source_cache_dir) and os.path.abspath(
source_cache_dir
) != os.path.abspath(target_cache_dir):
try:
shutil.copytree(
source_cache_dir,
target_cache_dir,
dirs_exist_ok=True,
ignore=shutil.ignore_patterns("*.sqlite-shm", "*.sqlite-wal"),
)
except Exception as exc:
logger.warning(
"Failed to copy model_cache directory from %s to %s: %s",
source_cache_dir,
target_cache_dir,
exc,
)
def _copy_dir(name: str) -> None:
source = os.path.join(source_dir, name)
target = os.path.join(target_dir, name)
if os.path.isdir(source) and os.path.abspath(source) != os.path.abspath(
target
):
try:
shutil.copytree(
source,
target,
dirs_exist_ok=True,
ignore=shutil.ignore_patterns("*.sqlite-shm", "*.sqlite-wal"),
)
except Exception as exc:
logger.warning(
"Failed to copy directory %s from %s to %s: %s",
name,
source,
target,
exc,
)
source_cache_file = os.path.join(source_dir, "model_cache.sqlite")
target_cache_file = os.path.join(target_dir, "model_cache.sqlite")
if os.path.isfile(source_cache_file) and os.path.abspath(
source_cache_file
) != os.path.abspath(target_cache_file):
try:
shutil.copy2(source_cache_file, target_cache_file)
except Exception as exc:
logger.warning(
"Failed to copy model_cache.sqlite from %s to %s: %s",
source_cache_file,
target_cache_file,
exc,
)
# Managed subdirectories under settings_dir
_copy_dir("cache")
_copy_dir("backups")
_copy_dir("logs")
_copy_dir("stats")
_copy_dir("wildcards")
def _get_user_config_directory(self) -> str:
"""Return the user configuration directory, falling back to ~/.config."""

View File

@@ -59,3 +59,180 @@ async def test_get_nightly_version_network_error_logs_warning(monkeypatch, caplo
assert changelog == []
assert "Unable to reach GitHub for nightly version" in caplog.text
assert "Traceback" not in caplog.text
def test_clean_excludes_covers_user_data_dirs():
"""git clean must receive -e excludes for every user-managed dir."""
excludes = update_routes._clean_excludes()
assert "-e" in excludes # at least one exclude flag present
for name in update_routes._PRESERVE_DIRS:
assert name in excludes
assert f"{name}/**" in excludes
@pytest.mark.asyncio
async def test_perform_git_update_preserves_user_dirs(monkeypatch, tmp_path):
"""``git clean`` must be called with -e excludes for user data dirs.
Regression test for portable-mode updates wiping wildcards/, stats/,
backups/, etc. because ``git clean -fd`` removed untracked, non-ignored
directories.
"""
calls = []
class FakeGit:
def reset(self, *args, **kwargs):
calls.append(("reset", args))
def clean(self, *args, **kwargs):
calls.append(("clean", args))
def checkout(self, *args, **kwargs):
calls.append(("checkout", args))
class FakeRemote:
def fetch(self):
calls.append(("fetch", ()))
def pull(self, *args, **kwargs):
calls.append(("pull", args))
class FakeRemotes:
origin = FakeRemote()
class FakeCommit:
hexsha = "abcdef123456"
class FakeHeads:
def __getitem__(self, name):
class Head:
def checkout(self_inner):
calls.append(("head-checkout", (name,)))
return Head()
class FakeBranches:
names = ["main"]
def __iter__(self):
class B:
name = "main"
return iter([B()])
class FakeRepo:
def __init__(self, path):
calls.append(("repo", (path,)))
git = FakeGit()
remotes = FakeRemotes()
head = type("H", (), {"commit": FakeCommit()})()
branches = FakeBranches()
heads = FakeHeads()
def create_head(self, name, ref):
calls.append(("create_head", (name, ref)))
class FakeGitModule:
class Repo:
def __new__(cls, path):
return FakeRepo(path)
class exc:
class GitError(Exception):
pass
import builtins
real_import = builtins.__import__
def fake_import(name, *args, **kwargs):
if name == "git":
return FakeGitModule
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", fake_import)
success, version = await update_routes.UpdateRoutes._perform_git_update(
str(tmp_path), nightly=True
)
assert success is True
clean_calls = [c for c in calls if c[0] == "clean"]
assert len(clean_calls) == 1
clean_args = clean_calls[0][1]
# Every preserved dir must be excluded via -e
for name in update_routes._PRESERVE_DIRS:
assert name in clean_args, f"{name} missing from git clean excludes"
assert f"{name}/**" in clean_args, f"{name}/** missing from git clean excludes"
# Ensure there's an -e before each name occurrence
idx = clean_args.index(name)
assert clean_args[idx - 1] == "-e"
@pytest.mark.asyncio
async def test_perform_git_update_stable_preserves_user_dirs(monkeypatch, tmp_path):
"""Stable (tag) update path must also pass -e excludes to git clean."""
calls = []
class FakeGit:
def reset(self, *args, **kwargs):
calls.append(("reset", args))
def clean(self, *args, **kwargs):
calls.append(("clean", args))
def checkout(self, *args, **kwargs):
calls.append(("checkout", args))
class FakeRemote:
def fetch(self):
calls.append(("fetch", ()))
class FakeRemotes:
origin = FakeRemote()
class FakeCommit:
committed_datetime = "2026-01-01"
class FakeTag:
name = "v9.9.9"
commit = FakeCommit()
class FakeRepo:
def __init__(self, path):
calls.append(("repo", (path,)))
git = FakeGit()
remotes = FakeRemotes()
tags = [FakeTag()]
class FakeGitModule:
class Repo:
def __new__(cls, path):
return FakeRepo(path)
class exc:
class GitError(Exception):
pass
import builtins
real_import = builtins.__import__
def fake_import(name, *args, **kwargs):
if name == "git":
return FakeGitModule
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", fake_import)
success, version = await update_routes.UpdateRoutes._perform_git_update(
str(tmp_path), nightly=False
)
assert success is True
assert version == "v9.9.9"
clean_calls = [c for c in calls if c[0] == "clean"]
assert len(clean_calls) == 1
clean_args = clean_calls[0][1]
for name in update_routes._PRESERVE_DIRS:
assert name in clean_args, f"{name} missing from git clean excludes (stable)"

View File

@@ -200,52 +200,97 @@ def _setup_storage_paths(tmp_path, monkeypatch):
return project_root, user_dir, user_settings_path
def _populate_cache(root_dir, marker_name, db_text):
cache_dir = root_dir / "model_cache"
cache_dir.mkdir(exist_ok=True)
marker_file = cache_dir / marker_name
marker_file.write_text(marker_name, encoding="utf-8")
(root_dir / "model_cache.sqlite").write_text(db_text, encoding="utf-8")
def _populate_settings_dir(root_dir):
"""Create test data for all managed subdirectories under a settings directory."""
(root_dir / "cache" / "symlink").mkdir(parents=True, exist_ok=True)
(root_dir / "cache" / "symlink" / "symlink_map.json").write_text(
'{"migrated": true}', encoding="utf-8"
)
(root_dir / "backups").mkdir(parents=True, exist_ok=True)
(root_dir / "backups" / "backup_test.zip").write_text(
"backup", encoding="utf-8"
)
(root_dir / "logs").mkdir(parents=True, exist_ok=True)
(root_dir / "logs" / "session.log").write_text("log", encoding="utf-8")
(root_dir / "stats").mkdir(parents=True, exist_ok=True)
(root_dir / "stats" / "stats.json").write_text(
'{"stats": true}', encoding="utf-8"
)
(root_dir / "wildcards").mkdir(parents=True, exist_ok=True)
(root_dir / "wildcards" / "test.txt").write_text("wildcard", encoding="utf-8")
def test_switch_to_portable_mode_copies_cache(tmp_path, monkeypatch):
def test_switch_to_portable_mode_copies_subdirectories(tmp_path, monkeypatch):
project_root, user_dir, user_settings = _setup_storage_paths(tmp_path, monkeypatch)
_populate_cache(user_dir, "user_marker.txt", "user_db")
_populate_settings_dir(user_dir)
manager = SettingsManager()
manager.set("use_portable_settings", True)
assert manager.settings_file == str(project_root / "settings.json")
marker_copy = project_root / "model_cache" / "user_marker.txt"
assert marker_copy.read_text(encoding="utf-8") == "user_marker.txt"
assert (project_root / "model_cache.sqlite").read_text(
# Managed subdirectories should all be migrated
assert (
project_root / "cache" / "symlink" / "symlink_map.json"
).read_text(encoding="utf-8") == '{"migrated": true}'
assert (
project_root / "backups" / "backup_test.zip"
).read_text(encoding="utf-8") == "backup"
assert (project_root / "logs" / "session.log").read_text(
encoding="utf-8"
) == "user_db"
) == "log"
assert (project_root / "stats" / "stats.json").read_text(
encoding="utf-8"
) == '{"stats": true}'
assert (project_root / "wildcards" / "test.txt").read_text(
encoding="utf-8"
) == "wildcard"
assert user_settings.exists()
def test_switching_back_to_user_config_moves_cache(tmp_path, monkeypatch):
def test_switching_back_to_user_config_moves_subdirectories(tmp_path, monkeypatch):
project_root, user_dir, user_settings = _setup_storage_paths(tmp_path, monkeypatch)
_populate_cache(user_dir, "user_marker.txt", "user_db")
_populate_settings_dir(user_dir)
manager = SettingsManager()
manager.set("use_portable_settings", True)
project_cache_dir = project_root / "model_cache"
project_cache_dir.mkdir(exist_ok=True)
(project_cache_dir / "project_marker.txt").write_text(
"project_marker", encoding="utf-8"
# Populate project-root managed subdirectories
(project_root / "cache" / "model").mkdir(parents=True, exist_ok=True)
(project_root / "cache" / "model" / "default.sqlite").write_text(
"project_db", encoding="utf-8"
)
(project_root / "backups" / "project_backup.zip").write_text(
"project_backup", encoding="utf-8"
)
(project_root / "logs" / "project.log").write_text(
"project_log", encoding="utf-8"
)
(project_root / "stats" / "project_stats.json").write_text(
'{"project": true}', encoding="utf-8"
)
(project_root / "wildcards" / "project.txt").write_text(
"project_wildcard", encoding="utf-8"
)
(project_root / "model_cache.sqlite").write_text("project_db", encoding="utf-8")
manager.set("use_portable_settings", False)
assert manager.settings_file == str(user_settings)
assert (user_dir / "model_cache" / "project_marker.txt").read_text(
assert (user_dir / "cache" / "model" / "default.sqlite").read_text(
encoding="utf-8"
) == "project_marker"
assert (user_dir / "model_cache.sqlite").read_text(encoding="utf-8") == "project_db"
) == "project_db"
assert (user_dir / "backups" / "project_backup.zip").read_text(
encoding="utf-8"
) == "project_backup"
assert (user_dir / "logs" / "project.log").read_text(
encoding="utf-8"
) == "project_log"
assert (user_dir / "stats" / "project_stats.json").read_text(
encoding="utf-8"
) == '{"project": true}'
assert (user_dir / "wildcards" / "project.txt").read_text(
encoding="utf-8"
) == "project_wildcard"
def test_download_path_template_parses_json_string(manager):