Files
olivv-cs-ComfyUI-FunPack/core/tests/test_log.py
T
digital-garbageandClaude Opus 5 78c43acd67 contract: a colour type, and which run this is
Colour is a preference, so it is a setting: hex in, hex out, one spelling,
refused if it is not one. A module handed "red" in one place and "#f00" in
another guesses wrong somewhere.

Image and mask are NOT settings and deliberately have no type. They are
data -- tensors the app uploads and the graph wires -- and ComfyUI already
has socket types for both. Putting a drawn mask in a JSON settings payload
would be inventing a second way to move something that already moves.

Runs now have an id. Previews, progress, results and log lines all have to
be attributable to one generation, and ComfyUI gives a node no way to ask
which run it is in. So FunPack marks its own, every log record carries it,
and the history can be filtered to a single run. Readable rather than a
uuid, because it is going to be read: "run 7" is findable in a wall of
output and a hex string is not.

The colour element was caught by the registry-driven rule test before it was
finished -- an element with no catalogue demo is neither shown nor attacked,
so coverage keeps up with the registry instead of lagging behind it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-29 17:29:38 +03:00

122 lines
3.6 KiB
Python

"""The log: a severity, a source, and a message.
The format is the contract -- the app will parse it and a person will scan it --
so it is pinned rather than left to drift line by line.
"""
import pytest
from core import log, run
@pytest.fixture(autouse=True)
def clean():
log._reset()
run._reset()
yield
log._reset()
run._reset()
def test_the_shape_is_label_severity_source_message():
assert log.line("Warning", "FunPack LoRA Loader", "style.safetensors is empty") == (
"[Log] Warning: FunPack LoRA Loader | style.safetensors is empty")
def test_a_message_with_no_source_does_not_print_an_empty_one():
assert log.line("Info", "", "ready") == "[Log] Info: ready"
@pytest.mark.parametrize("fn,level", [
(log.info, "Info"), (log.alert, "Alert"),
(log.warning, "Warning"), (log.critical, "Critical"),
])
def test_each_level_records_itself(fn, level):
fn("somewhere", "something")
entry = log.history()[-1]
assert entry["level"] == level
assert entry["source"] == "somewhere"
assert entry["message"] == "something"
def test_a_record_carries_the_run_it_happened_in():
"""So a report can say which generation a line belongs to, and two runs in
one session can be told apart."""
log.info("before", "no run yet")
assert log.history()[-1]["run"] is None
first = run.start()
log.info("during", "inside a run")
assert log.history()[-1]["run"] == first
second = run.start()
log.info("during", "inside the next one")
assert log.history()[-1]["run"] == second
assert first != second
def test_history_can_be_filtered_to_one_run():
first = run.start()
log.info("a", "one")
run.start()
log.info("b", "two")
assert [r["source"] for r in log.history(run=first)] == ["a"]
def test_starting_a_run_clears_what_was_said_once():
run.start()
log.once("k", log.ALERT, "src", "inert")
run.start()
log.once("k", log.ALERT, "src", "inert")
assert len(log.history()) == 2
def test_it_prints_as_well_as_records(capsys):
log.warning("loader", "a file went missing")
assert "[Log] Warning: loader | a file went missing" in capsys.readouterr().err
def test_history_can_be_filtered_by_level():
log.info("a", "one")
log.warning("b", "two")
log.info("c", "three")
assert [r["source"] for r in log.history(log.INFO)] == ["a", "c"]
def test_the_history_is_bounded():
"""A long rental session must not turn the log into a memory leak."""
for i in range(log.HISTORY + 50):
log.info("loop", str(i))
assert len(log.history()) == log.HISTORY
assert log.history()[-1]["message"] == str(log.HISTORY + 49)
def test_failed_is_a_warning_because_it_means_a_feature_is_absent():
log.failed("some.module", ValueError("bad"))
entry = log.history()[-1]
assert entry["level"] == log.WARNING
assert "ValueError" in entry["message"] and "bad" in entry["message"]
# --- once, and what "once" means -------------------------------------------
def test_once_says_it_a_single_time():
for _ in range(10):
log.once("k", log.ALERT, "src", "inert")
assert len(log.history()) == 1
def test_once_is_per_run_not_per_process():
"""v4 deduped for the life of the interpreter, so a session reported the
first generation that went inert and stayed quiet for every one after."""
log.once("k", log.ALERT, "src", "inert")
log.new_run()
log.once("k", log.ALERT, "src", "inert")
assert len(log.history()) == 2
def test_different_keys_are_independent():
log.once("a", log.ALERT, "src", "one")
log.once("b", log.ALERT, "src", "two")
assert len(log.history()) == 2