Vito
65b15bbe75
* fix(logs): Clear all logs deletes and empties only what its confirmation showed (#389) - /logs/files rows carry an identity string (size:mtime_ns:ino); the workspace sends back the active and rotated rows its dialog was built from. Under the files and handler locks every listed copy is re-checked with _matches_review and the listed active log's inode; any difference refuses with nothing changed (409 logs-changed). A copy that appeared after the confirmation is kept. - A second writer of the active log (our flushed stream is behind the file's size, or our file is now a rotated copy) refuses first (409 logs-foreign-writer); each unlink is preceded by a fresh identity check and the active log is emptied last. - An active log with a hard link (st_nlink > 1) is never truncated (409 logs-hard-link), also on the file-logging-off path. - The legacy admin page's POST without a body keeps clearing what is there, with the foreign-writer and hard-link refusals. - The browser fixture grows the active log through the handler's stream. * fix(logs): accept pre-1970 identities, refuse duplicate rows, say the foreign-writer check is best effort (#389) Review follow-ups: a negative mtime_ns no longer makes every workspace clear 400; a name listed twice is a malformed list; the confirmed-row cap is 10000; the hard-link refusal says how to recover; the hard-link and foreign-append tests assert the kept bytes before the refusal code, so on the old code they fail on lost data.
1230 lines
52 KiB
Python
1230 lines
52 KiB
Python
"""PromptManagerLogger: memory buffer, file rotation, level changes, truncation."""
|
|
|
|
import logging
|
|
import ntpath
|
|
import os
|
|
import pathlib
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import unittest.mock as mock
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from utils import logging_config # noqa: E402
|
|
from utils.logging_config import ( # noqa: E402
|
|
MemoryBufferHandler,
|
|
PromptManagerLogger,
|
|
RotatedChanged,
|
|
get_logger,
|
|
get_logger_manager,
|
|
)
|
|
|
|
try:
|
|
from tests.open_handles import assert_closed # noqa: E402
|
|
except ImportError: # discovered with tests/ as the top-level directory
|
|
from open_handles import assert_closed # noqa: E402
|
|
|
|
|
|
class IsolatedLoggerTestCase(unittest.TestCase):
|
|
"""A logger manager whose log directory is a temporary folder."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
package = os.path.join(self.tmp.name, "pkg")
|
|
os.makedirs(os.path.join(package, "utils"))
|
|
self.log_dir = os.path.join(package, "logs")
|
|
fake_file = os.path.join(package, "utils", "logging_config.py")
|
|
with mock.patch.object(logging_config, "__file__", fake_file):
|
|
self.manager = object.__new__(PromptManagerLogger) # bypass singleton
|
|
self.manager.__init__()
|
|
self.allowed_thread_names = set()
|
|
self.manager.update_config({"console_logging": False})
|
|
# 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 or the rotating file (a stray line after a truncate
|
|
# would roll the file over and resurrect the backups). update_config
|
|
# rebuilds the handlers, so the filter is re-applied after every call.
|
|
self._filter_to_this_thread()
|
|
original_update = self.manager.update_config
|
|
|
|
def update_and_refilter(new_config):
|
|
original_update(new_config)
|
|
self._filter_to_this_thread()
|
|
|
|
self.manager.update_config = update_and_refilter
|
|
self.logger = logging.getLogger("prompt_manager.coverage_test")
|
|
self.addCleanup(self._restore_global_logging)
|
|
|
|
def _filter_to_this_thread(self):
|
|
this_thread = threading.get_ident()
|
|
# A test may let named helper threads of its own through too.
|
|
allowed_names = self.allowed_thread_names
|
|
|
|
def allowed(record):
|
|
return record.thread == this_thread or record.threadName in allowed_names
|
|
|
|
for handler in self.manager.logger.handlers:
|
|
handler.filters.clear()
|
|
handler.addFilter(allowed)
|
|
|
|
def _restore_global_logging(self):
|
|
for handler in logging.getLogger("prompt_manager").handlers[:]:
|
|
handler.close()
|
|
# Windows refuses to delete an open log: prove every handle is closed.
|
|
assert_closed(self, self.tmp.name)
|
|
get_logger_manager()._setup_loggers()
|
|
self.tmp.cleanup()
|
|
|
|
def _rotate(self, max_file_size=400, backup_count=2, messages=30):
|
|
self.manager.update_config(
|
|
{"max_file_size": max_file_size, "backup_count": backup_count}
|
|
)
|
|
for i in range(messages):
|
|
self.logger.info("message %03d %s", i, "x" * 80)
|
|
|
|
|
|
class TestSingleton(unittest.TestCase):
|
|
def test_manager_and_loggers_are_shared(self):
|
|
self.assertIs(PromptManagerLogger(), PromptManagerLogger())
|
|
self.assertIs(get_logger_manager(), get_logger_manager())
|
|
self.assertIs(
|
|
get_logger("prompt_manager.x"), logging.getLogger("prompt_manager.x")
|
|
)
|
|
self.assertEqual(get_logger().name, "prompt_manager")
|
|
|
|
def test_init_is_a_noop_on_the_existing_instance(self):
|
|
manager = get_logger_manager()
|
|
before = manager.config
|
|
manager.__init__()
|
|
self.assertIs(manager.config, before)
|
|
|
|
|
|
class TestMemoryBuffer(IsolatedLoggerTestCase):
|
|
def test_records_are_buffered_most_recent_first(self):
|
|
self.logger.info("first")
|
|
self.logger.warning("second")
|
|
self.logger.error("third")
|
|
recent = self.manager.get_recent_logs(limit=2)
|
|
self.assertEqual([r["message"] for r in recent], ["third", "second"])
|
|
entry = recent[0]
|
|
self.assertEqual(entry["level"], "ERROR")
|
|
self.assertEqual(entry["logger"], self.logger.name)
|
|
self.assertIn("third", entry["formatted"])
|
|
for key in ("timestamp", "module", "filename", "lineno", "thread", "process"):
|
|
self.assertIn(key, entry)
|
|
|
|
def test_level_filter(self):
|
|
self.logger.info("info")
|
|
self.logger.warning("warn")
|
|
self.logger.error("err")
|
|
self.assertEqual(
|
|
[r["message"] for r in self.manager.get_recent_logs(level="warning")],
|
|
["err", "warn"],
|
|
)
|
|
self.assertEqual(len(self.manager.get_recent_logs(level="NOPE")), 3)
|
|
|
|
def test_buffer_is_capped_at_the_configured_size(self):
|
|
for i in range(1005):
|
|
self.logger.info("entry %d", i)
|
|
stats = self.manager.get_log_stats()
|
|
self.assertEqual(stats["buffer_count"], 1000)
|
|
self.assertEqual(stats["level_counts"], {"INFO": 1000})
|
|
self.assertEqual(stats["current_level"], "INFO")
|
|
self.assertEqual(
|
|
os.path.normcase(os.path.realpath(stats["log_directory"])),
|
|
os.path.normcase(os.path.realpath(self.log_dir)),
|
|
)
|
|
self.assertEqual(
|
|
self.manager.get_recent_logs(limit=1)[0]["message"], "entry 1004"
|
|
)
|
|
|
|
def test_handler_swallows_buffer_errors(self):
|
|
manager = mock.Mock()
|
|
manager.add_to_buffer.side_effect = RuntimeError("boom")
|
|
handler = MemoryBufferHandler(manager)
|
|
record = logging.LogRecord("x", logging.INFO, __file__, 1, "msg", (), None)
|
|
handler.emit(record) # must not raise
|
|
manager.add_to_buffer.assert_called_once()
|
|
|
|
|
|
class TestFilesAndRotation(IsolatedLoggerTestCase):
|
|
def test_rotation_creates_backups_and_lists_them_newest_first(self):
|
|
self._rotate()
|
|
files = self.manager.get_log_files()
|
|
names = sorted(f["filename"] for f in files)
|
|
self.assertEqual(
|
|
names,
|
|
["prompt_manager.log", "prompt_manager.log.1", "prompt_manager.log.2"],
|
|
)
|
|
main = [f for f in files if f["is_main"]]
|
|
self.assertEqual(len(main), 1)
|
|
self.assertTrue(all(f["size"] > 0 for f in files if not f["is_main"]))
|
|
modified = [f["modified"] for f in files]
|
|
self.assertEqual(modified, sorted(modified, reverse=True))
|
|
self.assertEqual(self.manager.get_log_stats()["log_files_count"], 3)
|
|
|
|
def test_read_log_file_tail_and_whole(self):
|
|
self._rotate()
|
|
tail = self.manager.read_log_file("prompt_manager.log.1", lines=1)
|
|
self.assertEqual(len(tail), 1)
|
|
whole = self.manager.read_log_file("prompt_manager.log.1", lines=0)
|
|
self.assertGreater(len(whole), 1)
|
|
self.assertTrue(whole[-1].endswith(tail[0]))
|
|
|
|
def test_read_log_file_rejects_missing_and_outside_paths(self):
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.manager.read_log_file("missing.log")
|
|
outside = os.path.join(self.tmp.name, "pkg", "outside.log")
|
|
with open(outside, "w", encoding="utf-8") as f:
|
|
f.write("secret\n")
|
|
with self.assertRaises(ValueError):
|
|
self.manager.read_log_file(os.path.join("..", "outside.log"))
|
|
|
|
def test_read_log_file_reraises_read_errors(self):
|
|
self.logger.info("something")
|
|
with mock.patch("builtins.open", side_effect=OSError("disk")):
|
|
with self.assertRaises(OSError):
|
|
self.manager.read_log_file("prompt_manager.log")
|
|
|
|
def test_unreadable_files_are_skipped_in_the_listing(self):
|
|
self.logger.info("something")
|
|
original_lstat = os.lstat
|
|
|
|
def lstat_failing_for_logs(path, *args, **kwargs):
|
|
if str(path).endswith(".log"):
|
|
raise OSError("denied")
|
|
return original_lstat(path, *args, **kwargs)
|
|
|
|
with mock.patch.object(logging_config.os, "lstat", lstat_failing_for_logs):
|
|
self.assertEqual(self.manager.get_log_files(), [])
|
|
|
|
def test_truncate_does_not_roll_the_log_over_while_deleting_backups(self):
|
|
# With a file limit smaller than one record, any log line written
|
|
# while the backups are being deleted rolls the main log over and
|
|
# resurrects prompt_manager.log.1. The handler's own stream must also
|
|
# be reset, or its stale offset reports the old size, rolls over on the
|
|
# next record anyway and pads the file with zeros on POSIX.
|
|
self._rotate(max_file_size=400, backup_count=2, messages=10)
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(result["errors"], [])
|
|
self.logger.info("after truncate")
|
|
leftovers = sorted(p.name for p in pathlib.Path(self.log_dir).iterdir())
|
|
contents = {
|
|
name: pathlib.Path(self.log_dir, name).read_bytes() for name in leftovers
|
|
}
|
|
self.assertEqual(leftovers, ["prompt_manager.log"], contents)
|
|
main_log = pathlib.Path(self.log_dir, "prompt_manager.log").read_bytes()
|
|
self.assertNotIn(b"\x00", main_log)
|
|
self.assertIn(b"after truncate", main_log)
|
|
|
|
def test_truncate_clears_main_log_backups_and_buffer(self):
|
|
self._rotate()
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(result["errors"], [])
|
|
self.assertEqual(
|
|
sorted(result["truncated"]),
|
|
["prompt_manager.log", "prompt_manager.log.1", "prompt_manager.log.2"],
|
|
)
|
|
self.assertFalse(
|
|
os.path.exists(os.path.join(self.log_dir, "prompt_manager.log.1"))
|
|
)
|
|
# Only the "Truncated log file" lines written after clearing remain
|
|
self.assertEqual(
|
|
[
|
|
r["message"]
|
|
for r in self.manager.get_recent_logs()
|
|
if "message 0" in r["message"]
|
|
],
|
|
[],
|
|
)
|
|
|
|
def test_truncate_reports_files_it_cannot_remove(self):
|
|
self._rotate()
|
|
|
|
def busy(path, *args, **kwargs):
|
|
raise PermissionError(13, "busy", str(path))
|
|
|
|
with mock.patch.object(logging_config.os, "unlink", busy):
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(len(result["errors"]), 2)
|
|
self.assertEqual(result["truncated"], ["prompt_manager.log"])
|
|
for error in result["errors"]:
|
|
self.assertNotIn(self.tmp.name, error)
|
|
self.assertRegex(error, r"prompt_manager\.log\.[12]")
|
|
|
|
|
|
class TestRotatedCopyRule(IsolatedLoggerTestCase):
|
|
"""Rule R (#329): only prompt_manager.log.<n> regular files are deletable."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.logger.info("active log line") # the active log exists
|
|
self.outside = os.path.join(self.tmp.name, "outside")
|
|
os.makedirs(self.outside)
|
|
|
|
def _path(self, name):
|
|
return os.path.join(self.log_dir, name)
|
|
|
|
def _write(self, name, data=b"rotated\n"):
|
|
with open(self._path(name), "wb") as handle:
|
|
handle.write(data)
|
|
return self._path(name)
|
|
|
|
def _symlink(self, name, target_name="secret.txt", data=b"secret " * 1000):
|
|
"""A symlink in the log folder to a file outside it, or None."""
|
|
target = os.path.join(self.outside, target_name)
|
|
with open(target, "wb") as handle:
|
|
handle.write(data)
|
|
try:
|
|
os.symlink(target, self._path(name))
|
|
except (OSError, NotImplementedError): # pragma: no cover - no privilege
|
|
return None
|
|
return target
|
|
|
|
def _require_symlink(self, name, **kwargs):
|
|
target = self._symlink(name, **kwargs)
|
|
if target is None: # pragma: no cover - Windows without the privilege
|
|
self.skipTest("this account cannot create symlinks")
|
|
return target
|
|
|
|
def _kinds(self):
|
|
return {row["name"]: row["kind"] for row in self.manager.scan_log_folder()}
|
|
|
|
def _rotated_rows(self):
|
|
return [r for r in self.manager.scan_log_folder() if r["kind"] == "rotated"]
|
|
|
|
def _names(self):
|
|
return sorted(os.listdir(self.log_dir))
|
|
|
|
def _file_handler(self):
|
|
return next(
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
)
|
|
|
|
def test_scan_classes_rotated_active_link_and_other(self):
|
|
for name in ("prompt_manager.log.1", "prompt_manager.log.2"):
|
|
self._write(name)
|
|
self._write("prompt_manager.log.3", b"x" * 3000)
|
|
for name in (
|
|
"prompt_manager.log.0",
|
|
"prompt_manager.log.01",
|
|
"prompt_manager.log.bak",
|
|
"prompt_manager.log.1.gz",
|
|
"prompt_manager.log.-1",
|
|
"notes.txt",
|
|
):
|
|
self._write(name)
|
|
os.makedirs(self._path("prompt_manager.log.5"))
|
|
self._require_symlink("prompt_manager.log.4")
|
|
kinds = self._kinds()
|
|
self.assertEqual(
|
|
kinds,
|
|
{
|
|
"prompt_manager.log": "active",
|
|
"prompt_manager.log.1": "rotated",
|
|
"prompt_manager.log.2": "rotated",
|
|
"prompt_manager.log.3": "rotated",
|
|
"prompt_manager.log.0": "other",
|
|
"prompt_manager.log.01": "other",
|
|
"prompt_manager.log.bak": "other",
|
|
"prompt_manager.log.1.gz": "other",
|
|
"prompt_manager.log.-1": "other",
|
|
"notes.txt": "other",
|
|
"prompt_manager.log.5": "other",
|
|
"prompt_manager.log.4": "link",
|
|
},
|
|
)
|
|
row = next(
|
|
r
|
|
for r in self.manager.scan_log_folder()
|
|
if r["name"] == "prompt_manager.log.3"
|
|
)
|
|
self.assertEqual(
|
|
set(row), {"name", "kind", "size", "mtime_ns", "ino", "modified"}
|
|
)
|
|
stat = os.lstat(self._path("prompt_manager.log.3"))
|
|
self.assertEqual(row["size"], 3000)
|
|
self.assertEqual(row["mtime_ns"], stat.st_mtime_ns)
|
|
self.assertEqual(row["ino"], stat.st_ino)
|
|
|
|
def test_scan_matches_names_case_insensitively_on_windows(self):
|
|
self._write("PROMPT_MANAGER.LOG.2")
|
|
if os.name == "nt": # pragma: no cover - native case folding
|
|
self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
|
|
return
|
|
# POSIX names are case-sensitive: another file, never deleted.
|
|
self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "other")
|
|
with mock.patch.object(logging_config, "_name_key", ntpath.normcase):
|
|
self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
|
|
result = self.manager.delete_rotated(self._rotated_rows())
|
|
self.assertEqual(result["removed"], ["PROMPT_MANAGER.LOG.2"])
|
|
|
|
def test_a_junction_is_classed_as_a_link_and_kept(self):
|
|
# Portable control: whatever os.path.isjunction reports is a link.
|
|
self._write("prompt_manager.log.4")
|
|
self._write("prompt_manager.log.1")
|
|
junction = self._path("prompt_manager.log.4")
|
|
real_isjunction = os.path.isjunction
|
|
|
|
def isjunction(path):
|
|
return os.fspath(path) == junction or real_isjunction(path)
|
|
|
|
with mock.patch.object(logging_config.os.path, "isjunction", isjunction):
|
|
self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
|
|
self.manager.delete_rotated(self._rotated_rows())
|
|
self.manager.truncate_logs()
|
|
self.assertIn("prompt_manager.log.4", self._names())
|
|
self.assertNotIn("prompt_manager.log.1", self._names())
|
|
|
|
@unittest.skipUnless(os.name == "nt", "junctions exist only on Windows")
|
|
def test_junction_is_never_deleted_on_windows(self): # pragma: no cover
|
|
import _winapi
|
|
|
|
target = os.path.join(self.outside, "kept")
|
|
os.makedirs(target)
|
|
with open(os.path.join(target, "inside.txt"), "wb") as handle:
|
|
handle.write(b"keep me")
|
|
_winapi.CreateJunction(target, self._path("prompt_manager.log.4"))
|
|
self._write("prompt_manager.log.1")
|
|
self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
|
|
self.manager.delete_rotated(self._rotated_rows())
|
|
self.manager.truncate_logs()
|
|
self.assertIn("prompt_manager.log.4", self._names())
|
|
self.assertTrue(os.path.exists(os.path.join(target, "inside.txt")))
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.manager.log_file_entry("prompt_manager.log.4")
|
|
os.rmdir(self._path("prompt_manager.log.4")) # removes the junction only
|
|
|
|
def test_delete_rotated_removes_exactly_the_listed_files(self):
|
|
for n in (1, 2, 3):
|
|
self._write(f"prompt_manager.log.{n}", b"r" * n)
|
|
self._write("prompt_manager.log.bak")
|
|
os.makedirs(self._path("prompt_manager.log.7"))
|
|
target = self._symlink("prompt_manager.log.4")
|
|
listed = [
|
|
r for r in self._rotated_rows() if r["name"] != "prompt_manager.log.3"
|
|
]
|
|
result = self.manager.delete_rotated(listed)
|
|
self.assertEqual(
|
|
result,
|
|
{"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
|
|
)
|
|
expected = [
|
|
"prompt_manager.log",
|
|
"prompt_manager.log.3",
|
|
"prompt_manager.log.7",
|
|
"prompt_manager.log.bak",
|
|
]
|
|
if target is not None:
|
|
expected.append("prompt_manager.log.4")
|
|
self.assertTrue(os.path.exists(target))
|
|
self.assertEqual(self._names(), sorted(expected))
|
|
|
|
def test_delete_rotated_refuses_all_when_one_changed(self):
|
|
def change_size(path):
|
|
with open(path, "ab") as handle:
|
|
handle.write(b"more")
|
|
|
|
def change_mtime(path):
|
|
stat = os.lstat(path)
|
|
os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
|
|
|
|
def change_inode(path):
|
|
stat = os.lstat(path)
|
|
with open(path, "rb") as handle:
|
|
data = handle.read()
|
|
other = path + ".new"
|
|
with open(other, "wb") as handle:
|
|
handle.write(data)
|
|
os.utime(other, ns=(stat.st_atime_ns, stat.st_mtime_ns))
|
|
os.replace(other, path)
|
|
if os.lstat(path).st_ino in (0, stat.st_ino): # pragma: no cover
|
|
self.skipTest("this filesystem reports no inode numbers")
|
|
|
|
def remove(path):
|
|
os.unlink(path)
|
|
|
|
def replace_by_link(path):
|
|
os.unlink(path)
|
|
target = os.path.join(self.outside, "elsewhere.txt")
|
|
with open(target, "wb") as handle:
|
|
handle.write(b"rotated\n")
|
|
try:
|
|
os.symlink(target, path)
|
|
except (OSError, NotImplementedError): # pragma: no cover
|
|
self.skipTest("this account cannot create symlinks")
|
|
|
|
changes = {
|
|
"size": change_size,
|
|
"mtime": change_mtime,
|
|
"inode": change_inode,
|
|
"missing": remove,
|
|
"link": replace_by_link,
|
|
}
|
|
for label, change in changes.items():
|
|
with self.subTest(change=label):
|
|
for n in (1, 2, 3):
|
|
self._write(f"prompt_manager.log.{n}", b"rotated\n")
|
|
reviewed = self._rotated_rows()
|
|
change(self._path("prompt_manager.log.2"))
|
|
before = self._names()
|
|
with self.assertRaises(RotatedChanged):
|
|
self.manager.delete_rotated(reviewed)
|
|
self.assertEqual(self._names(), before)
|
|
self.assertIn("prompt_manager.log.1", before)
|
|
self.assertIn("prompt_manager.log.3", before)
|
|
for name in self._names():
|
|
if name.startswith("prompt_manager.log."):
|
|
os.unlink(self._path(name))
|
|
|
|
def test_rollover_waits_while_cleanup_holds_the_handler_lock(self):
|
|
# The file is already over its limit: the next record rolls it over,
|
|
# renaming .1 to .2 and the active log to .1.
|
|
self.manager.update_config({"max_file_size": 64, "backup_count": 3})
|
|
self.logger.info("x" * 100)
|
|
self._write("prompt_manager.log.1", b"first copy\n")
|
|
self._write("prompt_manager.log.2", b"second copy\n")
|
|
reviewed = self._rotated_rows()
|
|
holding = threading.Event()
|
|
release = threading.Event()
|
|
logged = threading.Event()
|
|
real_unlink = os.unlink
|
|
self.allowed_thread_names.add("pm-test-rollover")
|
|
|
|
def log_from_another_thread():
|
|
self.logger.info("rolls the log over")
|
|
logged.set()
|
|
|
|
worker = threading.Thread(
|
|
target=log_from_another_thread, name="pm-test-rollover", daemon=True
|
|
)
|
|
|
|
def slow_unlink(path, *args, **kwargs):
|
|
if not holding.is_set():
|
|
worker.start() # its record must wait for the cleanup's lock
|
|
holding.set()
|
|
if not release.wait(10):
|
|
raise RuntimeError("the test never released the cleanup")
|
|
return real_unlink(path, *args, **kwargs)
|
|
|
|
result = {}
|
|
|
|
def cleanup():
|
|
result.update(self.manager.delete_rotated(reviewed))
|
|
|
|
with mock.patch.object(logging_config.os, "unlink", slow_unlink):
|
|
cleaner = threading.Thread(target=cleanup, daemon=True)
|
|
cleaner.start()
|
|
self.assertTrue(holding.wait(10), "the cleanup never started deleting")
|
|
# The record waits on the handler lock the cleanup holds.
|
|
self.assertFalse(logged.wait(0.3), "a record was written mid-cleanup")
|
|
release.set()
|
|
cleaner.join(10)
|
|
worker.join(10)
|
|
self.assertFalse(cleaner.is_alive())
|
|
self.assertTrue(logged.is_set())
|
|
self.assertEqual(
|
|
result,
|
|
{"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
|
|
)
|
|
# The rollover ran after the cleanup: only the old active log moved.
|
|
self.assertEqual(self._names(), ["prompt_manager.log", "prompt_manager.log.1"])
|
|
with open(self._path("prompt_manager.log.1"), "rb") as handle:
|
|
self.assertIn(b"x" * 100, handle.read())
|
|
|
|
def test_open_rotated_copy_is_reported_not_deleted(self):
|
|
for n in (1, 2, 3):
|
|
self._write(f"prompt_manager.log.{n}")
|
|
real_unlink = os.unlink
|
|
held = self._path("prompt_manager.log.2")
|
|
|
|
def unlink(path, *args, **kwargs):
|
|
if os.fspath(path) == held:
|
|
error = PermissionError(
|
|
13,
|
|
"The process cannot access the file because it is being "
|
|
"used by another process",
|
|
held,
|
|
)
|
|
error.winerror = 32
|
|
raise error
|
|
return real_unlink(path, *args, **kwargs)
|
|
|
|
with mock.patch.object(logging_config.os, "unlink", unlink):
|
|
result = self.manager.delete_rotated(self._rotated_rows())
|
|
self.assertEqual(
|
|
result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
|
|
)
|
|
[(name, error)] = result["failed"]
|
|
self.assertEqual(name, "prompt_manager.log.2")
|
|
self.assertIsInstance(error, PermissionError)
|
|
self.assertIn("prompt_manager.log.2", self._names())
|
|
|
|
@unittest.skipUnless(os.name == "nt", "only Windows refuses to unlink an open file")
|
|
def test_open_rotated_copy_is_reported_not_deleted_on_windows(
|
|
self,
|
|
): # pragma: no cover
|
|
for n in (1, 2, 3):
|
|
self._write(f"prompt_manager.log.{n}")
|
|
reviewed = self._rotated_rows()
|
|
handle = open(self._path("prompt_manager.log.2"), "rb")
|
|
try:
|
|
result = self.manager.delete_rotated(reviewed)
|
|
finally:
|
|
handle.close()
|
|
self.assertEqual(
|
|
result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
|
|
)
|
|
[(name, error)] = result["failed"]
|
|
self.assertEqual(name, "prompt_manager.log.2")
|
|
self.assertIsInstance(error, PermissionError)
|
|
self.assertIn("prompt_manager.log.2", self._names())
|
|
|
|
def test_cleanup_never_logs_while_holding_the_handler_lock(self):
|
|
for n in (1, 2):
|
|
self._write(f"prompt_manager.log.{n}")
|
|
handler = self._file_handler()
|
|
depth = 0
|
|
calls = []
|
|
real_acquire, real_release = handler.acquire, handler.release
|
|
|
|
def acquire():
|
|
nonlocal depth
|
|
real_acquire()
|
|
depth += 1
|
|
|
|
def release():
|
|
nonlocal depth
|
|
depth -= 1
|
|
real_release()
|
|
|
|
real_call_handlers = logging.Logger.callHandlers
|
|
|
|
def call_handlers(logger_self, record):
|
|
calls.append((record.getMessage(), depth))
|
|
return real_call_handlers(logger_self, record)
|
|
|
|
with (
|
|
mock.patch.object(handler, "acquire", acquire),
|
|
mock.patch.object(handler, "release", release),
|
|
mock.patch.object(logging.Logger, "callHandlers", call_handlers),
|
|
):
|
|
self.manager.delete_rotated(self._rotated_rows())
|
|
self.assertEqual(depth, 0)
|
|
self._write("prompt_manager.log.1")
|
|
self.manager.truncate_logs()
|
|
self.assertEqual(depth, 0)
|
|
self.assertTrue(calls, "the cleanup logged no summary")
|
|
self.assertEqual([c for c in calls if c[1]], [])
|
|
|
|
def test_truncate_keeps_other_files_and_links(self):
|
|
# Corrected (#329): the legacy glob deleted every prompt_manager.log*.
|
|
self._rotate()
|
|
self._write("prompt_manager.log.bak", b"mine")
|
|
self._write("prompt_manager.log.1.gz", b"zip")
|
|
target = self._symlink("prompt_manager.log.4")
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(result["errors"], [])
|
|
self.assertEqual(
|
|
sorted(result["truncated"]),
|
|
["prompt_manager.log", "prompt_manager.log.1", "prompt_manager.log.2"],
|
|
)
|
|
names = self._names()
|
|
self.assertIn("prompt_manager.log.bak", names)
|
|
self.assertIn("prompt_manager.log.1.gz", names)
|
|
self.assertNotIn("prompt_manager.log.1", names)
|
|
if target is not None:
|
|
self.assertIn("prompt_manager.log.4", names)
|
|
self.assertTrue(os.path.islink(self._path("prompt_manager.log.4")))
|
|
with open(target, "rb") as handle:
|
|
self.assertEqual(handle.read(), b"secret " * 1000)
|
|
|
|
@unittest.skipIf(os.name == "nt", "Windows cannot rename a log that is open")
|
|
def test_truncate_never_empties_a_renamed_log_the_handler_still_writes(self):
|
|
# The user renamed the active log to keep it; the handler still has it
|
|
# open. Clear must empty the new prompt_manager.log, not the kept copy.
|
|
self.logger.info("worth keeping")
|
|
kept = os.path.join(self.log_dir, "saved_copy.txt")
|
|
os.rename(self._path("prompt_manager.log"), kept)
|
|
with open(kept, "rb") as handle:
|
|
before = handle.read()
|
|
self._write("prompt_manager.log", b"fresh active log\n")
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(result["errors"], [])
|
|
with open(kept, "rb") as handle:
|
|
self.assertTrue(handle.read().startswith(before))
|
|
with open(self._path("prompt_manager.log"), "rb") as handle:
|
|
self.assertEqual(handle.read(), b"")
|
|
|
|
def test_name_surrogate_reparse_points_are_links_and_placeholders_are_not(self):
|
|
regular = stat.S_IFREG | 0o644
|
|
path = self._write("prompt_manager.log.3")
|
|
for tag, is_link in (
|
|
(0xA000001D, True), # WSL symlink
|
|
(0xA0000003, True), # mount point (junction)
|
|
(0x9000001A, False), # OneDrive cloud placeholder: a regular file
|
|
(0, False),
|
|
):
|
|
with self.subTest(tag=hex(tag)):
|
|
st = mock.Mock(st_mode=regular, st_reparse_tag=tag)
|
|
self.assertIs(logging_config._is_link(path, st), is_link)
|
|
|
|
def test_truncate_with_file_logging_off_never_creates_the_active_log(self):
|
|
self.manager.update_config({"file_logging": False})
|
|
os.unlink(self._path("prompt_manager.log"))
|
|
self._write("prompt_manager.log.1")
|
|
result = self.manager.truncate_logs()
|
|
self.assertEqual(result, {"truncated": ["prompt_manager.log.1"], "errors": []})
|
|
self.assertEqual(self._names(), [])
|
|
|
|
def test_truncate_never_empties_a_linked_active_log(self):
|
|
self.manager.update_config({"file_logging": False})
|
|
os.unlink(self._path("prompt_manager.log"))
|
|
target = self._require_symlink("prompt_manager.log", data=b"keep")
|
|
self.manager.truncate_logs()
|
|
with open(target, "rb") as handle:
|
|
self.assertEqual(handle.read(), b"keep")
|
|
|
|
def test_log_files_mark_kind_and_lstat_links(self):
|
|
self._write("prompt_manager.log.1", b"r" * 10)
|
|
self._write("prompt_manager.log.bak", b"b")
|
|
target = self._require_symlink("prompt_manager.log.4")
|
|
rows = {row["filename"]: row for row in self.manager.get_log_files()}
|
|
self.assertEqual(rows["prompt_manager.log"]["kind"], "active")
|
|
self.assertTrue(rows["prompt_manager.log"]["is_main"])
|
|
self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
|
|
self.assertEqual(rows["prompt_manager.log.bak"]["kind"], "other")
|
|
link = rows["prompt_manager.log.4"]
|
|
self.assertEqual(link["kind"], "link")
|
|
self.assertEqual(
|
|
link["size"], os.lstat(self._path("prompt_manager.log.4")).st_size
|
|
)
|
|
self.assertNotEqual(link["size"], os.path.getsize(target))
|
|
for row in rows.values():
|
|
self.assertEqual(
|
|
set(row),
|
|
{"filename", "path", "size", "modified", "is_main", "kind", "identity"},
|
|
)
|
|
|
|
def test_an_unreadable_timestamp_does_not_break_the_scan(self):
|
|
self._write("prompt_manager.log.1")
|
|
real = logging_config.datetime
|
|
|
|
class Clock:
|
|
@staticmethod
|
|
def fromtimestamp(value):
|
|
raise OSError(22, "Invalid argument")
|
|
|
|
with mock.patch.object(logging_config, "datetime", Clock):
|
|
rows = {r["name"]: r for r in self.manager.scan_log_folder()}
|
|
self.assertIs(logging_config.datetime, real)
|
|
self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
|
|
self.assertEqual(rows["prompt_manager.log.1"]["modified"], "")
|
|
|
|
def test_stats_exclude_links(self):
|
|
self._write("prompt_manager.log.1", b"r" * 10)
|
|
before = self.manager.get_log_stats()
|
|
self._require_symlink("prompt_manager.log.4")
|
|
after = self.manager.get_log_stats()
|
|
self.assertEqual(after["log_files_count"], before["log_files_count"])
|
|
self.assertEqual(after["total_log_size"], before["total_log_size"])
|
|
|
|
def test_log_file_entry_accepts_only_listed_regular_files(self):
|
|
self._write("prompt_manager.log.1")
|
|
self._write("prompt_manager.log.bak")
|
|
os.makedirs(self._path("folder.log"))
|
|
self._symlink("prompt_manager.log.4")
|
|
self.assertEqual(
|
|
os.path.normcase(str(self.manager.log_file_entry("prompt_manager.log.1"))),
|
|
os.path.normcase(self._path("prompt_manager.log.1")),
|
|
)
|
|
self.manager.log_file_entry("prompt_manager.log")
|
|
self.manager.log_file_entry("prompt_manager.log.bak")
|
|
for name in (
|
|
"prompt_manager.log.4",
|
|
"folder.log",
|
|
"missing.log",
|
|
"",
|
|
".",
|
|
"..",
|
|
"../outside/secret.txt",
|
|
"..\\outside\\secret.txt",
|
|
"prompt_manager.log:stream",
|
|
"prompt_manager.log\x00",
|
|
):
|
|
with self.subTest(name=name):
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.manager.log_file_entry(name)
|
|
|
|
|
|
class TestClearOnlyWhatWasConfirmed(IsolatedLoggerTestCase):
|
|
"""#389: Clear all logs empties or deletes no more than its confirmation showed."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.logger.info("active log line") # the active log exists
|
|
self.outside = os.path.join(self.tmp.name, "outside")
|
|
os.makedirs(self.outside)
|
|
|
|
def _path(self, name):
|
|
return os.path.join(self.log_dir, name)
|
|
|
|
def _write(self, name, data=b"rotated\n"):
|
|
with open(self._path(name), "wb") as handle:
|
|
handle.write(data)
|
|
|
|
def _read(self, path):
|
|
with open(path, "rb") as handle:
|
|
return handle.read()
|
|
|
|
def _snapshot(self):
|
|
return {name: self._read(self._path(name)) for name in os.listdir(self.log_dir)}
|
|
|
|
def _assert_unchanged(self, before, growing="prompt_manager.log"):
|
|
"""Nothing deleted or emptied (the refusal itself logs to ``growing``)."""
|
|
after = self._snapshot()
|
|
self.assertEqual(sorted(after), sorted(before))
|
|
for name, data in before.items():
|
|
if name == growing:
|
|
self.assertTrue(after[name].startswith(data), name)
|
|
else:
|
|
self.assertEqual(after[name], data, name)
|
|
|
|
def _confirmed(self):
|
|
"""What the workspace dialog showed: the listed active and rotated rows."""
|
|
return [
|
|
{"name": row["filename"], "identity": row["identity"]}
|
|
for row in self.manager.get_log_files()
|
|
if row["kind"] in ("active", "rotated")
|
|
]
|
|
|
|
def _file_handler(self):
|
|
return next(
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
)
|
|
|
|
def test_listed_identity_is_a_string_that_survives_json_numbers(self):
|
|
# mtime_ns (~1.7e18) and many inodes exceed 2**53: as JSON numbers a
|
|
# browser would round them and every confirmed clear would be refused.
|
|
self._write("prompt_manager.log.1")
|
|
for row in self.manager.get_log_files():
|
|
with self.subTest(name=row["filename"]):
|
|
self.assertIsInstance(row["identity"], str)
|
|
self.assertRegex(row["identity"], r"^\d+:-?\d+:\d+$")
|
|
|
|
def test_a_copy_dated_before_1970_can_be_confirmed(self):
|
|
self._write("prompt_manager.log.1")
|
|
try:
|
|
os.utime(self._path("prompt_manager.log.1"), ns=(0, -(10**9)))
|
|
except (OSError, OverflowError, ValueError) as error:
|
|
self.skipTest(f"this filesystem cannot store a time before 1970: {error}")
|
|
confirmed = self._confirmed()
|
|
if not any(":-" in row["identity"] for row in confirmed):
|
|
self.skipTest("the filesystem did not keep the negative time")
|
|
result = self.manager.truncate_logs(confirmed)
|
|
self.assertIn("prompt_manager.log.1", result["truncated"])
|
|
|
|
def test_a_rollover_after_the_confirmation_deletes_nothing(self):
|
|
self._write("prompt_manager.log.1", b"one" * 10)
|
|
self._write("prompt_manager.log.2", b"two" * 10)
|
|
confirmed = self._confirmed()
|
|
self.logger.info("written after the confirmation")
|
|
self._file_handler().doRollover() # .2 -> .3, .1 -> .2, log -> .1
|
|
before = self._snapshot()
|
|
with self.assertRaises(logging_config.ClearRefused) as caught:
|
|
self.manager.truncate_logs(confirmed)
|
|
self.assertEqual(caught.exception.code, "logs-changed")
|
|
self._assert_unchanged(before)
|
|
|
|
def test_a_copy_that_appeared_after_the_confirmation_is_kept(self):
|
|
self._write("prompt_manager.log.1", b"one")
|
|
confirmed = self._confirmed()
|
|
self._write("prompt_manager.log.5", b"not shown")
|
|
result = self.manager.truncate_logs(confirmed)
|
|
self.assertEqual(result["errors"], [])
|
|
self.assertEqual(
|
|
sorted(result["truncated"]), ["prompt_manager.log", "prompt_manager.log.1"]
|
|
)
|
|
self.assertEqual(
|
|
sorted(os.listdir(self.log_dir)),
|
|
["prompt_manager.log", "prompt_manager.log.5"],
|
|
)
|
|
|
|
def _clear(self, confirmed=None):
|
|
"""Run a clear and return what it raised (None), so the files are
|
|
asserted first: on the old code the test fails on lost bytes."""
|
|
try:
|
|
if confirmed is None:
|
|
self.manager.truncate_logs() # the legacy page's call
|
|
else:
|
|
self.manager.truncate_logs(confirmed)
|
|
except Exception as error: # noqa: BLE001 - asserted below
|
|
return error
|
|
return None
|
|
|
|
def test_a_foreign_append_to_the_active_log_is_refused(self):
|
|
# A second ComfyUI sharing the folder writes the same active log.
|
|
self._write("prompt_manager.log.1")
|
|
listings = (None, "listed")
|
|
for listing in listings:
|
|
with self.subTest(listed=listing is not None):
|
|
confirmed = self._confirmed() if listing else None
|
|
with open(self._path("prompt_manager.log"), "ab") as other:
|
|
other.write(b"line from another process\n")
|
|
before = self._snapshot()
|
|
error = self._clear(confirmed)
|
|
self._assert_unchanged(before)
|
|
self.assertEqual(getattr(error, "code", None), "logs-foreign-writer")
|
|
self.logger.info("our own line") # our stream catches up again
|
|
|
|
@unittest.skipIf(os.name == "nt", "Windows cannot rename a log that is open")
|
|
def test_a_foreign_rotation_of_our_log_is_refused(self):
|
|
# Another process rolled the shared log over: our handler still writes
|
|
# the file now named .1, and a new active log is the other's.
|
|
os.rename(self._path("prompt_manager.log"), self._path("prompt_manager.log.1"))
|
|
self._write("prompt_manager.log", b"the other process's log\n")
|
|
before = self._snapshot()
|
|
with self.assertRaises(logging_config.ClearRefused) as caught:
|
|
self.manager.truncate_logs()
|
|
self.assertEqual(caught.exception.code, "logs-foreign-writer")
|
|
# Our handler still writes the renamed file, so the refusal lands there.
|
|
self._assert_unchanged(before, growing="prompt_manager.log.1")
|
|
|
|
def test_a_copy_that_changes_mid_clear_stops_the_clear(self):
|
|
# The identity is checked again just before each unlink: another
|
|
# process rotating while we delete stops the clear there.
|
|
for n in (1, 2):
|
|
self._write(f"prompt_manager.log.{n}", b"r" * n)
|
|
confirmed = self._confirmed()
|
|
real_unlink = os.unlink
|
|
calls = []
|
|
|
|
def unlink_then_rotate(path, *args, **kwargs):
|
|
real_unlink(path, *args, **kwargs)
|
|
if not calls:
|
|
with open(self._path("prompt_manager.log.2"), "ab") as handle:
|
|
handle.write(b"rotated in by another process")
|
|
calls.append(os.path.basename(path))
|
|
|
|
active_before = self._read(self._path("prompt_manager.log"))
|
|
with mock.patch.object(logging_config.os, "unlink", unlink_then_rotate):
|
|
result = self.manager.truncate_logs(confirmed)
|
|
self.assertEqual(calls, ["prompt_manager.log.1"])
|
|
self.assertEqual(result["truncated"], ["prompt_manager.log.1"])
|
|
self.assertEqual(len(result["errors"]), 1)
|
|
self.assertIn("prompt_manager.log.2", result["errors"][0])
|
|
self.assertTrue(os.path.exists(self._path("prompt_manager.log.2")))
|
|
# The active log is emptied last, so it is kept too.
|
|
self.assertTrue(
|
|
self._read(self._path("prompt_manager.log")).startswith(active_before)
|
|
)
|
|
|
|
def _require_hard_link(self, source, link):
|
|
try:
|
|
os.link(source, link)
|
|
except (OSError, NotImplementedError, AttributeError) as error:
|
|
self.skipTest(f"this filesystem cannot make hard links: {error}")
|
|
|
|
def test_a_hard_linked_active_log_is_never_emptied(self):
|
|
other_name = os.path.join(self.outside, "kept-by-the-user.log")
|
|
self._require_hard_link(self._path("prompt_manager.log"), other_name)
|
|
kept = self._read(other_name)
|
|
self.assertTrue(kept)
|
|
for listed in (False, True):
|
|
with self.subTest(listed=listed):
|
|
error = self._clear(self._confirmed() if listed else None)
|
|
self.assertTrue(self._read(other_name).startswith(kept))
|
|
self.assertEqual(getattr(error, "code", None), "logs-hard-link")
|
|
|
|
def test_a_hard_linked_active_log_with_file_logging_off_is_kept(self):
|
|
self.manager.update_config({"file_logging": False})
|
|
other_name = os.path.join(self.outside, "kept-by-the-user.log")
|
|
self._require_hard_link(self._path("prompt_manager.log"), other_name)
|
|
kept = self._read(other_name)
|
|
error = self._clear()
|
|
self.assertEqual(self._read(other_name), kept)
|
|
self.assertEqual(getattr(error, "code", None), "logs-hard-link")
|
|
|
|
def test_a_malformed_confirmation_is_refused_before_anything_changes(self):
|
|
self._write("prompt_manager.log.1")
|
|
before = self._snapshot()
|
|
for expected in (
|
|
"prompt_manager.log.1",
|
|
[{"name": "prompt_manager.log.1"}],
|
|
[{"name": "prompt_manager.log.1", "identity": 7}],
|
|
[{"name": "prompt_manager.log.1", "identity": "1:2"}],
|
|
[{"name": 1, "identity": "1:2:3"}],
|
|
[
|
|
{"name": "prompt_manager.log.1", "identity": "1:2:3"},
|
|
{"name": "prompt_manager.log.1", "identity": "1:2:3"},
|
|
],
|
|
["prompt_manager.log.1"],
|
|
):
|
|
with self.subTest(expected=expected):
|
|
with self.assertRaises(ValueError):
|
|
self.manager.truncate_logs(expected)
|
|
self._assert_unchanged(before)
|
|
|
|
|
|
class TestConfiguration(IsolatedLoggerTestCase):
|
|
def test_in_flight_record_cannot_reopen_retired_file_handler(self) -> None:
|
|
handler = next(
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
)
|
|
stream = handler.stream
|
|
selected = threading.Event()
|
|
resume = threading.Event()
|
|
errors = []
|
|
message = "in-flight record selected before file logging disabled"
|
|
|
|
def pause_before_handler_lock(record: logging.LogRecord) -> bool:
|
|
if record.getMessage() == message:
|
|
selected.set()
|
|
if not resume.wait(5):
|
|
raise RuntimeError("logging thread was not released")
|
|
return True
|
|
|
|
def log_from_background_thread() -> None:
|
|
try:
|
|
self.manager.logger.info(message)
|
|
except BaseException as error:
|
|
errors.append(error)
|
|
|
|
# Allow this record through the fixture's normal main-thread filter.
|
|
handler.filters.clear()
|
|
handler.addFilter(pause_before_handler_lock)
|
|
worker = threading.Thread(target=log_from_background_thread, daemon=True)
|
|
worker.start()
|
|
try:
|
|
self.assertTrue(selected.wait(5), "logger did not select the file handler")
|
|
self.assertFalse(stream.closed)
|
|
self.manager.update_config({"file_logging": False})
|
|
self.assertNotIn(handler, self.manager.logger.handlers)
|
|
self.assertTrue(stream.closed)
|
|
self.assertIsNone(handler.stream)
|
|
|
|
resume.set()
|
|
worker.join(5)
|
|
self.assertFalse(worker.is_alive(), "logging thread did not finish")
|
|
self.assertEqual(errors, [])
|
|
self.assertIsNone(handler.stream)
|
|
self.assertTrue(stream.closed)
|
|
self.assertFalse(self.manager.config["file_logging"])
|
|
self.assertFalse(
|
|
any(
|
|
isinstance(h, logging.FileHandler)
|
|
for h in self.manager.logger.handlers
|
|
)
|
|
)
|
|
self.assertNotIn(
|
|
message,
|
|
pathlib.Path(self.log_dir, "prompt_manager.log").read_text(),
|
|
)
|
|
finally:
|
|
resume.set()
|
|
worker.join(5)
|
|
handler.close()
|
|
|
|
def test_rebuild_closes_replaced_file_stream(self) -> None:
|
|
old_handler = next(
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
)
|
|
stream = old_handler.stream
|
|
self.addCleanup(old_handler.close)
|
|
self.assertFalse(stream.closed)
|
|
|
|
self.manager.update_config({"file_logging": False})
|
|
|
|
self.assertTrue(stream.closed)
|
|
self.assertIsNone(old_handler.stream)
|
|
self.assertNotIn(old_handler, self.manager.logger.handlers)
|
|
log_file = pathlib.Path(self.log_dir, "prompt_manager.log")
|
|
renamed = log_file.rename(log_file.with_suffix(".retired"))
|
|
renamed.unlink()
|
|
self.assertFalse(renamed.exists())
|
|
|
|
def test_repeated_rebuilds_retire_previous_handlers(self) -> None:
|
|
for change in (
|
|
{"max_file_size": 4096},
|
|
{"backup_count": 3},
|
|
{"console_logging": True},
|
|
{"console_logging": False},
|
|
):
|
|
with self.subTest(change=change):
|
|
previous = self.manager.logger.handlers[:]
|
|
for handler in previous:
|
|
self.addCleanup(handler.close)
|
|
old_file = next(
|
|
h for h in previous if isinstance(h, logging.FileHandler)
|
|
)
|
|
stream = old_file.stream
|
|
self.assertFalse(stream.closed)
|
|
console_streams = [
|
|
h.stream for h in previous if type(h) is logging.StreamHandler
|
|
]
|
|
|
|
self.manager.update_config(change)
|
|
|
|
self.assertTrue(stream.closed)
|
|
self.assertIsNone(old_file.stream)
|
|
self.assertTrue(all(h._closed for h in previous))
|
|
self.assertTrue(all(not s.closed for s in console_streams))
|
|
self.assertTrue(
|
|
all(h not in self.manager.logger.handlers for h in previous)
|
|
)
|
|
current_files = [
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
]
|
|
self.assertEqual(len(current_files), 1)
|
|
self.assertFalse(current_files[0].stream.closed)
|
|
message = f"usable replacement after {change}"
|
|
self.logger.info(message)
|
|
self.assertIn(
|
|
message,
|
|
pathlib.Path(self.log_dir, "prompt_manager.log").read_text(),
|
|
)
|
|
self.assertEqual(
|
|
self.manager.get_recent_logs(limit=1)[0]["message"], message
|
|
)
|
|
|
|
def test_rebuild_does_not_close_external_handler(self) -> None:
|
|
external_path = pathlib.Path(self.tmp.name, "external.log")
|
|
external_handler = logging.FileHandler(
|
|
external_path, mode="w", encoding="utf-8"
|
|
)
|
|
self.addCleanup(external_handler.close)
|
|
other_logger = logging.Logger("external_shared_sink", level=logging.INFO)
|
|
other_logger.addHandler(external_handler)
|
|
self.addCleanup(other_logger.removeHandler, external_handler)
|
|
self.manager.logger.addHandler(external_handler)
|
|
|
|
with mock.patch.object(
|
|
external_handler, "close", wraps=external_handler.close
|
|
) as close:
|
|
self.manager.update_config({"backup_count": 3})
|
|
close.assert_not_called()
|
|
|
|
self.assertNotIn(external_handler, self.manager.logger.handlers)
|
|
self.assertFalse(external_handler.stream.closed)
|
|
other_logger.info("external sink still usable")
|
|
self.assertIn("external sink still usable", external_path.read_text())
|
|
|
|
def test_level_only_change_keeps_current_file_handler(self) -> None:
|
|
handler = next(
|
|
h
|
|
for h in self.manager.logger.handlers
|
|
if isinstance(h, logging.FileHandler)
|
|
)
|
|
stream = handler.stream
|
|
|
|
self.manager.update_config({"level": "DEBUG"})
|
|
|
|
self.assertIn(handler, self.manager.logger.handlers)
|
|
self.assertIs(handler.stream, stream)
|
|
self.assertFalse(stream.closed)
|
|
self.logger.debug("same sink still usable")
|
|
self.assertIn(
|
|
"same sink still usable",
|
|
pathlib.Path(self.log_dir, "prompt_manager.log").read_text(),
|
|
)
|
|
|
|
def test_replacement_manager_closes_previous_owned_handlers(self) -> None:
|
|
previous = self.manager.logger.handlers[:]
|
|
for handler in previous:
|
|
self.addCleanup(handler.close)
|
|
old_file = next(h for h in previous if isinstance(h, logging.FileHandler))
|
|
stream = old_file.stream
|
|
self.assertFalse(stream.closed)
|
|
fake_file = pathlib.Path(self.tmp.name, "pkg", "utils", "logging_config.py")
|
|
|
|
with mock.patch.object(logging_config, "__file__", str(fake_file)):
|
|
replacement = object.__new__(PromptManagerLogger)
|
|
replacement.__init__()
|
|
|
|
self.assertIs(replacement.logger, self.manager.logger)
|
|
self.assertTrue(stream.closed)
|
|
self.assertIsNone(old_file.stream)
|
|
self.assertTrue(all(h._closed for h in previous))
|
|
self._filter_to_this_thread()
|
|
self.logger.info("replacement manager sink")
|
|
self.assertEqual(
|
|
replacement.get_recent_logs(limit=1)[0]["message"],
|
|
"replacement manager sink",
|
|
)
|
|
|
|
def test_level_change_applies_to_every_prompt_manager_logger(self):
|
|
self.manager.update_config({"level": "DEBUG"})
|
|
self.assertEqual(logging.getLogger("prompt_manager").level, logging.DEBUG)
|
|
self.assertEqual(
|
|
logging.getLogger("prompt_manager.database").level, logging.DEBUG
|
|
)
|
|
self.assertEqual(self.manager.get_config()["level"], "DEBUG")
|
|
self.logger.debug("now visible")
|
|
self.assertEqual(self.manager.get_recent_logs(limit=1)[0]["level"], "DEBUG")
|
|
self.manager.update_config({"level": "bogus"}) # falls back to INFO
|
|
self.assertEqual(logging.getLogger("prompt_manager").level, logging.INFO)
|
|
|
|
def test_get_config_returns_a_copy(self):
|
|
config = self.manager.get_config()
|
|
config["level"] = "CRITICAL"
|
|
self.assertEqual(self.manager.get_config()["level"], "INFO")
|
|
|
|
def test_handler_flags_rebuild_the_handlers(self):
|
|
root = logging.getLogger("prompt_manager")
|
|
self.assertFalse(any(type(h) is logging.StreamHandler for h in root.handlers))
|
|
self.assertTrue(any(isinstance(h, logging.FileHandler) for h in root.handlers))
|
|
self.manager.update_config({"console_logging": True, "file_logging": False})
|
|
self.assertTrue(any(type(h) is logging.StreamHandler for h in root.handlers))
|
|
self.assertFalse(any(isinstance(h, logging.FileHandler) for h in root.handlers))
|
|
self.assertTrue(any(isinstance(h, MemoryBufferHandler) for h in root.handlers))
|
|
self.manager.update_config({"console_logging": False})
|
|
|
|
|
|
class TestConsoleNoise(IsolatedLoggerTestCase):
|
|
"""ComfyUI's console only sees warnings unless the level is DEBUG."""
|
|
|
|
def _console_handler(self) -> logging.StreamHandler:
|
|
return next(
|
|
h
|
|
for h in logging.getLogger("prompt_manager").handlers
|
|
if type(h) is logging.StreamHandler
|
|
)
|
|
|
|
def test_records_do_not_propagate_to_comfyui_root_logger(self) -> None:
|
|
self.assertFalse(logging.getLogger("prompt_manager").propagate)
|
|
|
|
def test_console_hides_info_at_the_default_level(self) -> None:
|
|
self.manager.update_config({"console_logging": True})
|
|
self.assertEqual(self._console_handler().level, logging.WARNING)
|
|
self.logger.info("startup chatter")
|
|
self.assertEqual(self.manager.get_recent_logs(limit=1)[0]["level"], "INFO")
|
|
log_file = pathlib.Path(self.log_dir) / "prompt_manager.log"
|
|
self.assertIn("startup chatter", log_file.read_text())
|
|
self.manager.update_config({"console_logging": False})
|
|
|
|
def test_debug_level_opens_the_console(self) -> None:
|
|
self.manager.update_config({"console_logging": True, "level": "DEBUG"})
|
|
self.assertEqual(self._console_handler().level, logging.DEBUG)
|
|
self.manager.update_config({"level": "INFO"})
|
|
self.assertEqual(self._console_handler().level, logging.WARNING)
|
|
self.manager.update_config({"console_logging": False})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|