mirror of
https://github.com/Comfy-Org/ComfyUI.git
synced 2026-09-12 12:03:56 +08:00
Constantine
5f0c4e18cb
* fix: respect user directory for default database * refactor: use None default for --database-url instead of argv scan Per review: default --database-url to None and treat a non-None value as explicit at resolution time. Removes the sys.argv scan and the database_url_explicit attribute. database_default_path now serves directly as the legacy copy source. Adds regression tests for the unchanged no-flag default path and explicit URLs at the legacy location. * refactor: rename legacy database to .bak after copy Per review: after copying the legacy install-dir database to the effective user directory, rename the original to comfyui.db.bak so a later launch without --user-directory cannot silently fall back to a diverged copy, while keeping the file around for recovery. Also hoist the database_default_path import to module scope. * fix: guard legacy migration on existing .bak and rename before copy Per review: bail out of the legacy migration when comfyui.db.bak already exists, so only the first run migrates and later launches with a fresh --user-directory cannot grab a database another instance is using. Rename before copy so os.replace fails fast if the legacy DB is held open by a running instance. --------- Co-authored-by: guill <jacob.e.segal@gmail.com>
135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
import os
|
|
|
|
from app.database import db
|
|
from comfy.cli_args import database_default_path
|
|
|
|
|
|
def test_default_database_url_uses_effective_user_directory(monkeypatch, tmp_path):
|
|
user_dir = tmp_path / "custom_user"
|
|
user_dir.mkdir()
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr("folder_paths.get_user_directory", lambda: str(user_dir))
|
|
|
|
assert db.get_database_url() == f"sqlite:///{user_dir / 'comfyui.db'}"
|
|
|
|
|
|
def test_default_db_path_matches_legacy_default_without_custom_user_directory(monkeypatch):
|
|
import folder_paths
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr(
|
|
"folder_paths.get_user_directory",
|
|
lambda: os.path.join(folder_paths.base_path, "user"),
|
|
)
|
|
|
|
assert os.path.abspath(db.get_db_path()) == database_default_path
|
|
|
|
|
|
def test_explicit_database_url_is_preserved(monkeypatch):
|
|
database_url = "sqlite:///:memory:"
|
|
|
|
monkeypatch.setattr(db.args, "database_url", database_url)
|
|
|
|
assert db.get_database_url() == database_url
|
|
|
|
|
|
def test_explicit_url_at_legacy_default_location_is_honoured(monkeypatch):
|
|
url = f"sqlite:///{database_default_path}"
|
|
|
|
monkeypatch.setattr(db.args, "database_url", url)
|
|
|
|
assert db.get_database_url() == url
|
|
assert db.get_db_path() == database_default_path
|
|
|
|
|
|
def test_explicit_database_url_skips_legacy_copy(monkeypatch, tmp_path):
|
|
legacy_db = tmp_path / "install" / "user" / "comfyui.db"
|
|
target_db = tmp_path / "target" / "comfyui.db"
|
|
legacy_db.parent.mkdir(parents=True)
|
|
target_db.parent.mkdir(parents=True)
|
|
legacy_db.write_bytes(b"legacy db")
|
|
|
|
monkeypatch.setattr(db.args, "database_url", f"sqlite:///{target_db}")
|
|
monkeypatch.setattr(db, "get_legacy_default_db_path", lambda: str(legacy_db))
|
|
|
|
db.copy_legacy_default_db(str(target_db))
|
|
|
|
assert not target_db.exists()
|
|
|
|
|
|
def test_legacy_default_database_is_copied_to_effective_user_directory(monkeypatch, tmp_path):
|
|
legacy_db = tmp_path / "install" / "user" / "comfyui.db"
|
|
user_dir = tmp_path / "custom_user"
|
|
legacy_db.parent.mkdir(parents=True)
|
|
user_dir.mkdir()
|
|
legacy_db.write_bytes(b"legacy db")
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr("folder_paths.get_user_directory", lambda: str(user_dir))
|
|
monkeypatch.setattr(db, "get_legacy_default_db_path", lambda: str(legacy_db))
|
|
|
|
db.copy_legacy_default_db(str(user_dir / "comfyui.db"))
|
|
|
|
assert (user_dir / "comfyui.db").read_bytes() == b"legacy db"
|
|
assert not legacy_db.exists()
|
|
assert legacy_db.with_suffix(".db.bak").read_bytes() == b"legacy db"
|
|
|
|
|
|
def test_legacy_copy_skipped_when_backup_already_exists(monkeypatch, tmp_path):
|
|
legacy_db = tmp_path / "install" / "user" / "comfyui.db"
|
|
backup_db = tmp_path / "install" / "user" / "comfyui.db.bak"
|
|
user_db = tmp_path / "custom_user" / "comfyui.db"
|
|
legacy_db.parent.mkdir(parents=True)
|
|
user_db.parent.mkdir(parents=True)
|
|
legacy_db.write_bytes(b"live db")
|
|
backup_db.write_bytes(b"old backup")
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr(db, "get_legacy_default_db_path", lambda: str(legacy_db))
|
|
|
|
db.copy_legacy_default_db(str(user_db))
|
|
|
|
assert not user_db.exists()
|
|
assert legacy_db.read_bytes() == b"live db"
|
|
assert backup_db.read_bytes() == b"old backup"
|
|
|
|
|
|
def test_legacy_default_database_does_not_overwrite_existing_effective_db(monkeypatch, tmp_path):
|
|
legacy_db = tmp_path / "install" / "user" / "comfyui.db"
|
|
user_db = tmp_path / "custom_user" / "comfyui.db"
|
|
legacy_db.parent.mkdir(parents=True)
|
|
user_db.parent.mkdir(parents=True)
|
|
legacy_db.write_bytes(b"legacy db")
|
|
user_db.write_bytes(b"user db")
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr(db, "get_legacy_default_db_path", lambda: str(legacy_db))
|
|
|
|
db.copy_legacy_default_db(str(user_db))
|
|
|
|
assert user_db.read_bytes() == b"user db"
|
|
assert legacy_db.read_bytes() == b"legacy db"
|
|
|
|
|
|
def test_prepare_file_database_creates_parent_directory(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "nested" / "comfyui.db"
|
|
|
|
monkeypatch.setattr(db.args, "database_url", None)
|
|
monkeypatch.setattr(db, "copy_legacy_default_db", lambda path: None)
|
|
|
|
db.prepare_file_db_path(str(db_path))
|
|
|
|
assert db_path.parent.is_dir()
|
|
|
|
|
|
def test_prepare_file_database_accepts_relative_database_path(monkeypatch, tmp_path):
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setattr(db.args, "database_url", "sqlite:///relative.db")
|
|
monkeypatch.setattr(db, "copy_legacy_default_db", lambda path: None)
|
|
|
|
db.prepare_file_db_path("relative.db")
|
|
|
|
assert os.getcwd() == str(tmp_path)
|
|
assert not list(tmp_path.iterdir())
|