test: isolate the clock and log buffer from background threads of other tests

This commit is contained in:
Vito Sansevero
2026-10-08 18:01:34 -07:00
parent 3a67c4b952
commit 5110ff1aff
3 changed files with 28 additions and 6 deletions
+4 -1
View File
@@ -76,8 +76,11 @@ class IntegrationTestCase(unittest.TestCase):
def setUp(self):
ComfyUIMetadataIntegration._instance = None
comfyui_integration._integration_instance = None
# Patch the module's own ``time`` name rather than ``time.time`` globally:
# background threads from other tests (image monitor, watchdog) would
# otherwise tick this clock and make fresh registrations look stale.
self.clock = mock.patch.object(
comfyui_integration.time, "time", side_effect=self._now
comfyui_integration, "time", types.SimpleNamespace(time=self._now)
)
self.clock.start()
self.addCleanup(self.clock.stop)
+15 -4
View File
@@ -112,10 +112,12 @@ class TestAdminEndpointsUseConfigPath(unittest.TestCase):
with sqlite3.connect(db_file) as conn:
conn.execute(
"CREATE TABLE prompts (id INTEGER PRIMARY KEY, text TEXT, created_at TEXT)"
"CREATE TABLE prompts "
"(id INTEGER PRIMARY KEY, text TEXT, created_at TEXT)"
)
conn.execute(
"INSERT INTO prompts (text, created_at) VALUES ('test', '2024-01-01')"
"INSERT INTO prompts (text, created_at) "
"VALUES ('test', '2024-01-01')"
)
stub = self._make_api_stub(db_file)
@@ -155,20 +157,29 @@ class TestAdminEndpointsUseConfigPath(unittest.TestCase):
from py.api.admin import AdminRoutesMixin
from database.operations import PromptDatabase
with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
db_file = f.name
f.write(b"SQLite format 3\x00test data for backup")
# A WAL-safe backup copies pages through sqlite itself, so the
# configured path has to hold a real database, not arbitrary bytes.
real_db = PromptDatabase(db_file)
try:
stub = self._make_api_stub(db_file)
stub.db.model = real_db.model
result = self._run_async(
AdminRoutesMixin.backup_database(stub, MagicMock())
)
self.assertEqual(result.content_type, "application/octet-stream")
self.assertGreater(len(result.body), 0)
self.assertTrue(result.body.startswith(b"SQLite format 3\x00"))
finally:
os.unlink(db_file)
real_db.close()
for suffix in ("", "-wal", "-shm"):
if os.path.exists(db_file + suffix):
os.unlink(db_file + suffix)
def test_backup_reports_missing_custom_path(self):
"""backup_database should return 404 when configured path doesn't exist."""
+9 -1
View File
@@ -5,6 +5,7 @@ import os
import pathlib
import sys
import tempfile
import threading
import unittest
import unittest.mock as mock
@@ -32,7 +33,14 @@ class IsolatedLoggerTestCase(unittest.TestCase):
self.manager = object.__new__(PromptManagerLogger) # bypass singleton
self.manager.__init__()
self.manager.update_config({"console_logging": False})
self.manager._log_buffer.clear() # drop the "Updated logging configuration" entry
# Drop the "Updated logging configuration" entry.
self.manager._log_buffer.clear()
# Other tests leave daemon threads (image monitor, watchdog) logging under
# "prompt_manager.*"; only records from this test's thread may reach the buffer.
this_thread = threading.get_ident()
for handler in self.manager.logger.handlers:
if isinstance(handler, MemoryBufferHandler):
handler.addFilter(lambda record: record.thread == this_thread)
self.logger = logging.getLogger("prompt_manager.coverage_test")
self.addCleanup(self._restore_global_logging)