File size: 4,739 Bytes
6329e19 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | 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())
|