nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(history): repair Windows fixtures and review feedback

Browse Source 
Address #209 and PR #227 native path, cleanup, media and review-comment expectations. Document Windows portability preflight without changing production behavior.
This commit is contained in:
Vito Sansevero
2026-10-18 16:00:43 -07:00
parent b27842dc69
commit 86a3a44f75
6 changed files with 123 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+24
View File
@@ -110,6 +110,30 @@ not certify every existing feature, arbitrary host versions,
GPU/model changes, cross-pack installation or visual design changes. Those still
need targeted validation.
## Windows portability preflight
Before changing paths, SQLite, streams, restore or cleanup:
- Follow the boundary's existing native canonical/identity contract;
`Path.resolve()` alone does not fold Windows case. Keep lexical aliases
distinct from resolved targets. Cover mixed case, drive and UNC spellings,
plus foreign-namespace denial with an allowed native control.
See Python's [normcase documentation](https://docs.python.org/3.13/library/os.path.html#os.path.normcase).
- Close every owned database/model connection, worker and stream handle before
unlink or `TemporaryDirectory.cleanup()`. Await worker/stream retirement.
A SQLite [`with conn` manages transactions, not connection closure](https://docs.python.org/3.13/library/sqlite3.html#how-to-use-the-connection-context-manager),
and [`addCleanup` runs after teardown](https://docs.python.org/3.13/library/unittest.html#unittest.TestCase.addCleanup);
use explicit `finally` cleanup when teardown removes files first.
- Windows can reject unlink of an open file; Unix can remove its directory entry
while the handle remains usable. Preserve the shared behavior assertions and
exercise the actual native branch, then prove handle release and successful
cleanup. Do not hide failures with blanket skips or retries.
See Python's [remove/unlink semantics](https://docs.python.org/3.13/library/os.html#os.remove).
- Run native Windows CI early for these changes and include a Windows impact and
evidence note in the PR. All twelve OS/Python combinations and every applicable
exact-head check must pass before merge. Portable `ntpath`/`PureWindowsPath`
controls do not establish native filesystem, locking or cleanup success.
## Required merge checks
`.github/branch-protection.json` records the required checks: all twelve Python/OS
tests/test_api_history.py
+25 -17
View File
@@ -5,6 +5,7 @@ import base64
import copy
import importlib
import json
import os
import sqlite3
import tempfile
import threading
@@ -787,31 +788,38 @@ class HistoryAPITests(ImageAPITestCase):
source = str(self.comfy_dir / "source.db")
self.assertTrue(self.api.db.model.backup_database(source))
writer = PromptDatabase(self.api.db.model.db_path)
self.addCleanup(writer.close_all)
for method, read in (
("list_history_window", lambda: self.window(positive)),
("get_history_run", lambda: self.detail(start.run_id)),
):
original = getattr(self.api.db, method)
try:
for method, read in (
("list_history_window", lambda: self.window(positive)),
("get_history_run", lambda: self.detail(start.run_id)),
):
original = getattr(self.api.db, method)
def during_snapshot(*args, **kwargs):
result = original(*args, **kwargs)
with patch("database.execution_history.notify_history_restore"):
writer.model.restore_from_file(source)
return result
def during_snapshot(*args, **kwargs):
result = original(*args, **kwargs)
with patch("database.execution_history.notify_history_restore"):
writer.model.restore_from_file(source)
return result
with patch.object(self.api.db, method, side_effect=during_snapshot):
response, body = await read()
self.assert_error(
response, body, 409, "history-changed", {"action": "refresh-history"}
)
with patch.object(self.api.db, method, side_effect=during_snapshot):
response, body = await read()
self.assert_error(
response,
body,
409,
"history-changed",
{"action": "refresh-history"},
)
finally:
writer.close_all()
self.assertEqual(len(writer.model._connections), 0)
async def test_one_lazy_subscription_cleanup_and_no_reinitialization(self):
import database.execution_history as observers
positive = self._save_prompt("positive")
start = self.capture()
identity = str(Path(self.api.db.model.db_path).resolve())
identity = os.path.normcase(str(Path(self.api.db.model.db_path).resolve()))
self.assertNotIn(identity, observers._restore_observers)
executor = self.api._run_in_executor
tests/test_api_history_outputs.py
+29 -3
View File
@@ -901,7 +901,7 @@ class HistoryOutputAPITests(ImageAPITestCase):
await asyncio.wait_for(reached.wait(), 5)
pending.cancel()
with self.assertRaises(asyncio.CancelledError):
await pending
await asyncio.wait_for(pending, 5)
finally:
release.set()
await asyncio.wait_for(closed.wait(), 5)
@@ -945,16 +945,42 @@ class HistoryOutputAPITests(ImageAPITestCase):
source = self.restore_fixture()
fence = self.api._history_fence()
validate = fence.validate
mod = importlib.import_module("py.api.history_current")
close = mod._OpenedCurrentMedia.close
closed = asyncio.Event()
loop = asyncio.get_running_loop()
handles = []
blocked_unlinks = []
def track_close(opened):
close(opened)
handles.append(opened.handle)
loop.call_soon_threadsafe(closed.set)
def after(epoch, generation):
validate(epoch, generation)
self.api.db.model.restore_from_file(source)
path.unlink()
try:
path.unlink()
except PermissionError:
if os.name != "nt":
raise
self.assertTrue(path.exists())
blocked_unlinks.append(path)
with patch.object(fence, "validate", side_effect=after):
with (
patch.object(fence, "validate", side_effect=after),
patch.object(mod._OpenedCurrentMedia, "close", track_close),
):
response = await self.current(descriptor["url"])
self.assertEqual(response.status, 200)
self.assertEqual(await response.read(), expected)
await asyncio.wait_for(closed.wait(), 5)
self.assertEqual(len(handles), 1)
self.assertTrue(handles[0].closed)
for locked_path in blocked_unlinks:
locked_path.unlink()
self.assertFalse(path.exists())
response = await self.current(descriptor["url"])
self.assert_error(response, await response.json(), 409, "history-changed")
tests/test_history_current.py
+35 -28
View File
@@ -94,7 +94,8 @@ class CurrentLocationTests(unittest.TestCase):
Path, "stat", side_effect=AssertionError("locator performed IO")
):
self.assertEqual(
mod.candidate_locator(reference, scope), str(alias / path.name)
mod.candidate_locator(reference, scope),
os.path.normcase(str(alias / path.name)),
)
candidate = mod.resolve_current_candidate(reference, scope, {}).public
self.assertEqual(candidate["status"], "available")
@@ -221,28 +222,29 @@ class CurrentLocationTests(unittest.TestCase):
db = PromptDatabase(str(self.base / "gallery.db"))
self.addCleanup(db.close_all)
path = make_png(self.output / "image.png")
locator = os.path.normcase(str(path))
prompt = db.save_prompt("first", prompt_hash="first-hash")
other = db.save_prompt("second", prompt_hash="second-hash")
image = db.link_image_to_prompt(prompt, str(path))
db.link_image_to_prompt(other, str(self.temp / path.name))
conn = db.model.get_connection()
with self.assertRaises(Exception):
mod.read_history_gallery(conn, (str(path),))
mod.read_history_gallery(conn, (locator,))
with db.recorded_history_snapshot():
projected = mod.read_history_gallery(conn, (str(path),))
projected = mod.read_history_gallery(conn, (locator,))
self.assertTrue(conn.in_transaction)
self.assertEqual(projected[str(path)]["gallery_match"], "unique")
self.assertEqual(projected[str(path)]["gallery_image_id"], image)
self.assertEqual(projected[locator]["gallery_match"], "unique")
self.assertEqual(projected[locator]["gallery_image_id"], image)
db.link_image_to_prompt(other, str(path))
with db.recorded_history_snapshot():
projected = mod.read_history_gallery(conn, (str(path),))
self.assertEqual(projected[str(path)]["gallery_match"], "ambiguous")
self.assertIsNone(projected[str(path)]["gallery_image_id"])
self.assertIsNone(projected[str(path)]["saved"])
self.assertEqual(projected[str(path)]["library"]["status"], "ambiguous")
projected = mod.read_history_gallery(conn, (locator,))
self.assertEqual(projected[locator]["gallery_match"], "ambiguous")
self.assertIsNone(projected[locator]["gallery_image_id"])
self.assertIsNone(projected[locator]["saved"])
self.assertEqual(projected[locator]["library"]["status"], "ambiguous")
with self.assertRaises(ValueError):
mod.read_history_gallery(
conn, tuple(str(path) + str(i) for i in range(101))
conn, tuple(locator + str(i) for i in range(101))
)
def test_gallery_display_is_bounded_without_loading_workflow_or_prompt(self):
@@ -250,6 +252,7 @@ class CurrentLocationTests(unittest.TestCase):
db = PromptDatabase(str(self.base / "gallery.db"))
self.addCleanup(db.close_all)
path = make_png(self.output / "image.png")
locator = os.path.normcase(str(path))
prompt = db.save_prompt(
"secret prompt" * 10000, prompt_hash="large-hash", category="猫" * 12000
)
@@ -264,8 +267,8 @@ class CurrentLocationTests(unittest.TestCase):
conn.set_trace_callback(statements.append)
self.addCleanup(conn.set_trace_callback, None)
with db.recorded_history_snapshot():
result = mod.read_history_gallery(conn, (str(path),))
display = result[str(path)]["library"]
result = mod.read_history_gallery(conn, (locator,))
display = result[locator]["library"]
self.assertEqual(display["status"], "limited")
self.assertEqual(display["tags"], [])
self.assertFalse(
@@ -282,6 +285,8 @@ class CurrentLocationTests(unittest.TestCase):
self.addCleanup(db.close_all)
path = make_png(self.output / "image.png")
unique = make_png(self.output / "unique.png")
locator = os.path.normcase(str(path))
unique_locator = os.path.normcase(str(unique))
prompt = db.save_prompt("owner", prompt_hash="owner")
image = db.link_image_to_prompt(prompt, str(unique))
with db.model.get_connection() as conn:
@@ -305,7 +310,7 @@ class CurrentLocationTests(unittest.TestCase):
"INSERT INTO generated_images"
"(prompt_id,image_path,filename,file_path) "
"VALUES(?,?,'image.png',?)",
(owner, str(path), str(path)),
(owner, str(path), locator),
)
fetched = {"owners": 0, "pairs": 0}
@@ -319,12 +324,12 @@ class CurrentLocationTests(unittest.TestCase):
conn.row_factory = row_factory
with db.recorded_history_snapshot():
result = mod.read_history_gallery(conn, (str(path), str(unique)))
result = mod.read_history_gallery(conn, (locator, unique_locator))
self.assertEqual(fetched, {"owners": 3, "pairs": 2})
self.assertEqual(result[str(path)]["gallery_match"], "ambiguous")
self.assertEqual(result[str(unique)]["gallery_match"], "unique")
self.assertEqual(result[str(unique)]["library"]["status"], "limited")
self.assertIsNone(result[str(unique)]["library"].get("pair_id"))
self.assertEqual(result[locator]["gallery_match"], "ambiguous")
self.assertEqual(result[unique_locator]["gallery_match"], "unique")
self.assertEqual(result[unique_locator]["library"]["status"], "limited")
self.assertIsNone(result[unique_locator]["library"].get("pair_id"))
def test_host_root_failure_and_removal_do_not_search_another_root(self):
make_png(self.output / "image.png")
@@ -359,7 +364,7 @@ class CurrentLocationTests(unittest.TestCase):
mod = self.module("database.history_gallery")
db = PromptDatabase(str(self.base / "gallery.db"))
self.addCleanup(db.close_all)
path = str(self.output / "image.png")
path = os.path.normcase(str(self.output / "image.png"))
with db.model.get_connection() as conn:
for item in range(512):
owner = conn.execute(
@@ -393,6 +398,7 @@ class CurrentLocationTests(unittest.TestCase):
db = PromptDatabase(str(self.base / "gallery.db"))
self.addCleanup(db.close_all)
path = make_png(self.output / "image.png")
locator = os.path.normcase(str(path))
pair = db.save_prompt_pair("owner", "negative one")
other_pair = db.save_prompt_pair("owner", "negative two")
owner = db.get_prompt_pair(pair)["positive_prompt_id"]
@@ -409,7 +415,7 @@ class CurrentLocationTests(unittest.TestCase):
(image, pair_id, provenance),
)
with db.recorded_history_snapshot():
result = mod.read_history_gallery(conn, (str(path),))[str(path)]
result = mod.read_history_gallery(conn, (locator,))[locator]
self.assertEqual(result["gallery_match"], "unique")
self.assertEqual(result["gallery_image_id"], image)
self.assertEqual(result["library"]["status"], "limited")
@@ -446,14 +452,15 @@ class CurrentLocationTests(unittest.TestCase):
class HistoryImportTests(unittest.TestCase):
def test_api_import_does_not_require_a_running_host_server(self):
result = subprocess.run(
[
sys.executable,
"-c",
"import sys, types; "
"sys.modules['folder_paths'] = types.ModuleType('folder_paths'); "
script = "; ".join(
(
"import sys, types",
"sys.modules['folder_paths'] = types.ModuleType('folder_paths')",
"import py.api",
],
)
)
result = subprocess.run(
[sys.executable, "-c", script],
cwd=Path(__file__).resolve().parents[1],
capture_output=True,
text=True,
tests/test_history_read_fence.py
+3 -2
View File
@@ -128,17 +128,18 @@ class HistoryReadFenceTests(unittest.TestCase):
module = importlib.import_module("utils.history_read_fence")
original_connect = sqlite3.connect
epoch = db.get_execution_epoch()
database_path = self.path
for close in (False, True):
fence = self.fence(db)
generation = fence.begin()
class InterruptedConnection(sqlite3.Connection):
def close(connection):
def close(self):
super().close()
if close:
fence.close()
else:
notify_history_restore(self.path)
notify_history_restore(database_path)
def interrupted_connect(*args, **kwargs):
return original_connect(*args, factory=InterruptedConnection, **kwargs)
tests/test_history_reads.py
+7 -2
View File
@@ -369,14 +369,19 @@ class HistoryReadTests(unittest.TestCase):
conn.rollback()
trace = []
conn.set_trace_callback(trace.append)
with self.assertRaisesRegex(RuntimeError, "abort"):
def abort_snapshot() -> None:
with db.recorded_history_snapshot() as epoch:
self.assertEqual(epoch, db.get_execution_epoch())
db.get_execution_contexts("missing")
db.get_execution_outputs("missing")
self.assertTrue(conn.in_transaction)
raise RuntimeError("abort")
conn.set_trace_callback(None)
try:
self.assertRaisesRegex(RuntimeError, "abort", abort_snapshot)
finally:
conn.set_trace_callback(None)
self.assertIn("storage_epoch", next(q for q in trace if q.startswith("SELECT")))
self.assertFalse(conn.in_transaction)
with self.assertRaises(HistoryEpochChanged):