Buckets:
| import os | |
| import shutil | |
| import sqlite3 | |
| from contextlib import closing | |
| 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_db_path() == str(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) | |
| 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") | |
| db.prepare_file_db_path("relative.db") | |
| assert os.getcwd() == str(tmp_path) | |
| assert not list(tmp_path.iterdir()) | |
| def test_legacy_copy_includes_rows_still_in_the_wal(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) | |
| writer = sqlite3.connect(legacy_db) | |
| writer.execute("PRAGMA journal_mode=WAL") | |
| writer.execute("PRAGMA wal_autocheckpoint=0") | |
| writer.execute("CREATE TABLE t (x)") | |
| writer.execute("INSERT INTO t VALUES (1)") | |
| writer.commit() | |
| # An unclean exit: the committed rows stay in comfyui.db-wal. | |
| shutil.copy(legacy_db, tmp_path / "main-only.db") | |
| shutil.copy(str(legacy_db) + "-wal", tmp_path / "saved-wal") | |
| writer.close() | |
| shutil.copy(tmp_path / "main-only.db", legacy_db) | |
| shutil.copy(tmp_path / "saved-wal", str(legacy_db) + "-wal") | |
| 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)) | |
| with closing(sqlite3.connect(user_db)) as copied: | |
| assert copied.execute("SELECT x FROM t").fetchall() == [(1,)] | |
| def test_legacy_copy_skipped_when_its_wal_cannot_be_checkpointed(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) | |
| writer = sqlite3.connect(legacy_db) | |
| writer.execute("PRAGMA journal_mode=WAL") | |
| writer.execute("PRAGMA wal_autocheckpoint=0") | |
| writer.execute("CREATE TABLE t (x)") | |
| writer.execute("INSERT INTO t VALUES (1)") | |
| writer.commit() | |
| reader = sqlite3.connect(legacy_db, isolation_level=None) | |
| reader.execute("BEGIN") | |
| reader.execute("SELECT x FROM t").fetchall() # an open read pins the WAL | |
| monkeypatch.setattr(db.args, "database_url", None) | |
| monkeypatch.setattr(db, "get_legacy_default_db_path", lambda: str(legacy_db)) | |
| try: | |
| db.copy_legacy_default_db(str(user_db)) | |
| finally: | |
| reader.close() | |
| writer.close() | |
| assert not user_db.exists() | |
| assert legacy_db.exists() | |
Xet Storage Details
- Size:
- 6.8 kB
- Xet hash:
- 7230b6b40db6bf0d1e39819ef6af1b37d7f4ebe23b92d5ba177d100279491e68
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.