Files
ComfyAssets-ComfyUI_PromptM…/tests/test_config.py
T
Vito 8870496ea3 feat(monitor): store certain prompts from workflows without a PromptManager node, with a System switch (#247, PR 2) (#467)
* feat(inspector): use recorded run history when the graph alone can't name the prompt (#386)

When a file's own graph leaves the prompt open, both inspect routes ask
recorded run history about the file (one run reported it and it was written
during that run). The save node that run reported it from anchors the walk
like a file-name match, so dataset branches that share one saver prefix
show their own prompt, labelled "Recorded run". Still open at one sampler
whose PromptManager text was recorded at execution (a language model's
output), the recorded text shows as "Execution recorded". Results the file
resolves alone never change; a failed lookup judges the file alone.

* fix(inspector): recorded texts follow history's saver; conflicts anchor nothing (#386)

Review follow-ups: the texts PromptManager recorded are looked up at the
sampler history's saver leads to, never at a sampler the unanchored walk
named; a file name naming a different saver than history anchors nothing;
outputs and texts are read in one transaction; "Recorded run" ranks below
every workflow-inferred source; unexpected lookup errors log at error level.

* fix(history): look up recorded runs by history's own spelling of the file (#386)

The history route inspects a case-normalized target on Windows, while
history matches file names exactly, so the occurrence's reference names
the file.

* refactor(inspector): share the inspector's resolve as inspection_resolution

inspect_prompt_evidence now calls inspection_resolution, which returns the
resolution to use, the anchored result and whether the recorded saver chose
it. Ingestion can call the same helper for files the default profile cannot
decide (#247).

* feat(ingestion): one per-file prompt decision for the scan and the monitor

utils/file_prompt.py holds decide_file_prompt, the ladder the output scan
runs today plus the inspector's rule (inspection profile, then the recorded
saver) before the legacy parser, with a SCAN and a MONITOR policy. Junk
text (blank, numbers, LoRA tags only, model file names) is never stored.
graph_prompt_text and parse_comfyui_prompt move here; the parser accepts
parsed metadata (#247).

* feat(ingestion): store a file's prompt in one transaction, one prompt per file

database/scanned_prompts.py links a file to the prompt of one text, creating
it when new, inside one BEGIN IMMEDIATE transaction: a file linked to
another prompt is a conflict, a failed link leaves no prompt behind and two
writers store one prompt. pair_ingestion gains file_link_rows,
link_conflict and Marks/mark_created; ingest_resolved_pair takes optional
marks for the components it creates. recorded_runs and named_prompt in
run_attribution become public (#247).

* feat(scan): decide each scanned file with the shared ladder (#247)

_ingest_scanned_file calls decide_file_prompt (scan policy) and stores the
result: pairs through ingest_resolved_pair and positives through
store_scanned_prompt, both with the scan marks. Files the default profile
leaves undecided now get the inspector's resolve, with the saver recorded
history names, before the legacy parser, so each branch of a shared-prefix
dataset run links its own prompt. Junk text is never stored, a file linked
to another prompt is a conflict instead of a second link, and a failed link
leaves no prompt behind. _BatchHistory reads the stored prompt and the
recorded saver from one recorded_runs pass. _parse_comfyui_prompt calls the
moved parse_comfyui_prompt. TESTING.md documents the decision.

* fix(ingestion): keep the file-name saver anchor out of the ladder (#390)

The #390 ruling queues a file that only its saver's file name resolves for
Needs review, never linking it. inspection_resolution gains name_anchor;
the ladder passes False, so such a file falls to the legacy floor in the
scan (and to Needs review) and is never stored by the monitor, while the
recorded saver still decides. The inspector keeps the name anchor.

* feat(config): gallery.monitoring.save_foreign_runs, default on (#247)

* fix(ingestion): no joined PromptManager text from the legacy floor; junk keeps its step's status

The legacy parser returns prepend + text + append, so its answer is refused
when a PromptManager-family node has a prepend or append part (#343). Junk
found at the undecided-graph step keeps that step's unlinked status and
review, so an ambiguous file is still queued for Needs review. The monitor
policy's ambiguous file no longer asks for the recorded run, and the file
size may be read lazily.

* fix(ingestion): keep the legacy answer that is a PromptManager node's own text

The joined-text refusal also refused the #170 parity files whose
PromptManager prepend is a link the browser parser does not follow: its
answer is then the node's literal text, which is what the node stores.
Refuse only an answer that is not such a node's own text.

* feat(monitor): store the certain prompt of a foreign workflow's output (#247)

A foreign output (no positive PromptManager node) is decided by the shared
ladder with the MONITOR policy. Certain files are stored at settle with
RUN_MARKS through survive_switch; files only recorded history can decide
wait in their own bounded set and are re-checked once after the run's
outputs are sealed, with the sealed reference. The setting
GalleryConfig.SAVE_FOREIGN_RUNS is read for every file.

* fix(scan): one prompt per file for recorded-run links; writes survive a library move

The recorded run's stored prompt now links in one transaction with the file's
link rows, so a file linked to another prompt is a conflict instead of a
second link. The scan's stores run through survive_library_switch. A single
file asks history once (a one-file _BatchHistory), the scan passes
in_output=False and reads the file size only for files with prompt
metadata. A conflict row reads "Kept: already linked to a different
prompt". TESTING.md records the rules.

* feat(settings): save_foreign_runs through POST /settings, applied live (#247)

GET /settings reports settings.save_foreign_runs; a save takes a strict
bool, writes gallery.monitoring.save_foreign_runs and publishes it to
GalleryConfig in the commit, with no restart field. The diagnostics
report lists it. The legacy contract snapshot gains the added key (an
addition, no legacy key moves).

* feat(system): Save prompts from other workflows in Sources & scans (#247, D147)

A host setting row below Watch for new outputs drafts save_foreign_runs
and saves with the one Save row; the server applies it at once, so no
restart is named. While monitoring is drafted off the row is disabled and
says it applies while Watch for new outputs is on. The concept comparison
maps the inserted row and its mutation check covers it.

* test(comfyui): a workflow of ComfyUI's own nodes creates and links its prompt (#247)

The real-host attribution spec gains the foreign case (CLIPTextEncode,
KSampler, SaveImage): its prompt is created and linked at settle and a
second run links to the same prompt. The fixture route lists the waiting
foreign files. TESTING.md records the monitor's foreign path and D147.

* test(vendor): the execution matrix counts the #247 real-host case

* fix(monitor): read a foreign file's size only when the ladder needs it (#247)

Uses PR 1's lazy file size and pins that a foreign file linked to another
prompt keeps its one link on the pair and the positive store paths.

* fix(ingestion): match a file's link rows under its resolved spelling too

The scan stores realpath spellings and the monitor the watcher's raw path,
so a file reached through a symlinked or junction output folder could get a
second generated_images row. file_link_rows now queries the path and its
realpath (each normalized and raw) and dedupes by row id. TESTING.md notes
the reverse order is matched only when the raw spelling is the realpath.

* fix(monitor): wait for the run unless history answered; monitoring off stops foreign writes (#247)

Review fixes: at settle a certain answer ends the wait only when the
lookup found a record or the answer is the inspector's or the recorded
saver's; the legacy floor waits for the re-check. A record without a
certain prompt ends the wait at once. Foreign writes need both
save_foreign_runs and monitoring_enabled, read again just before the
store. Dropped foreign files are logged outside the lock (evictions as
evictions, due ones too), and a drained re-check never drops a newer
reference.

* fix(system): the foreign row can be set while the monitor still watches; its note describes it (#247, D147)

Review fixes: while monitoring is saved off but a source still reads
watching until restart (D128), the row stays enabled with its note; the
note is the switch's aria-describedby. The browser spec restores
save_foreign_runs on the shared server. TESTING.md moves D147 into order
and records the monitor's review rules.

* test(comfyui): each branch of a shared-prefix foreign run links to its own prompt (#247 CUJ 1)

* fix(ingestion): a switched-off PromptManager node behind a pass-through stores nothing (#90)

The #90 graph walk follows only prompt-like inputs, so a PromptManagerText
with Save to library off reaching the encoder through another input name
(easy showAnything.anything, a custom node's value) was invisible to it,
and the inspector's resolve stored its text. The ladder now checks the
positive every step names: a resolved or inspected positive whose path
passes a node that is off, or a legacy answer that is such a node's own
text, makes the file not_saved.

* fix(ingestion): any switched-off node upstream of a positive stores nothing (#90)

The ladder walks every input upstream of each sampler or guider positive
(not negative, model, clip, latent or image inputs) for PromptManager nodes
whose Save to library is off or wired from an unreadable node, so the
sampler-less step, an off node with a prepend part and a PromptManagerPair
behind a pass-through no longer store its text. The legacy-floor guard reads
a node's resolved text. Every guard reports not_saved for a switch that is
off and not_saved_wired for one it cannot read.

* fix(ingestion): the switch walk skips negative inputs by #90's own rule

The upstream walk treated only names containing "negative" as negative, so
an off node on a neg, neg_text or text_neg input blocked an enabled
positive. prompt_graph gains is_negative_input (the _NEGATIVE_KEY rule, also
used by follows_input) and the walk uses it. The guards accept a node with
null inputs and a side whose provenance is not an object.

* fix(ingestion): the switch walk also skips any input named with "negative"

#90's rule matches only whole underscore-separated words, so negative2,
negativeprompt and NegativePrompt counted as positive inputs and an off
node wired only to one of them blocked the file. The walk skips those too;
follows_input is unchanged.

* fix(monitor): a foreign not_saved decision logs its skip and never waits (#247, #90)

The foreign path stored nothing for a ladder not_saved decision, but logged
it as "No certain prompt" and, when only the legacy answer was a
switched-off node's text, kept the file waiting for a seal that could only
store nothing. Such a decision now ends the wait at settle, and both settle
and the re-check log the opted-out path's own line (a wired switch says so).

Tests cover an only-PromptManager node switched off (opted out, not
foreign), a switched-off PromptManagerText behind a pass-through into a
core CLIPTextEncode, a wired switch, a negative-only switched-off node,
recorded history holding the off text, and foreign saving off.

* fix(monitor): an undecided not_saved waits for its run's record (#247, #90)

A not_saved from the ladder's undecided step is not final before a record
exists: the recorded saver may still resolve another branch. The settle
lookup now keeps such a file waiting; with the record it stores nothing and
logs the opted-out line. An early not_saved (a switched-off node upstream of
any sampler's positive) never reaches the lookup and stays final.

Tests: the undecided case (literal and wired switch) waits, then skips; a
switched-off node feeding one branch of a shared-prefix run never waits and
stores nothing, not even the branch the recorded saver names.

* docs(system): renumber the foreign-runs switch deviation to D158 (#247)

#430 took D147 on main; the Save prompts from other workflows row is D158.

* test(monitor): a switched-off node in the recorded saver's branch is never foreign (#247, #90)

PR 3's repair skips a file whose recorded saver's branch reads a
switched-off PromptManager node. On the monitor the same shape never reaches
the foreign path: a literal switch is the opted-out path's and a wired one
the node's own run's. Nothing is stored at settle or after the run, even
from a recorded run holding the off text.
2026-10-29 12:00:32 +00:00

636 lines
27 KiB
Python

"""
Tests for GalleryConfig and PromptManagerConfig.
The config module imports `from server import PromptServer` at module level,
which is ComfyUI's server. We mock it before import to run tests standalone.
"""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import MagicMock
# Mock ComfyUI's server module before importing config
_mock_server = MagicMock()
_mock_server.PromptServer.instance.routes = MagicMock()
sys.modules.setdefault("server", _mock_server)
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from py import config as config_module
from py.config import GalleryConfig, IntegrationConfig, PromptManagerConfig
class TestGalleryConfig(unittest.TestCase):
"""Test GalleryConfig class."""
def setUp(self):
"""Save original values to restore after each test."""
self._orig = GalleryConfig.get_config()
def tearDown(self):
"""Restore original config values."""
GalleryConfig.update_config(self._orig)
def test_get_config_structure(self):
config = GalleryConfig.get_config()
self.assertIn("monitoring", config)
self.assertIn("tracking", config)
self.assertIn("database", config)
self.assertIn("web_interface", config)
self.assertIn("performance", config)
def test_default_values(self):
config = GalleryConfig.get_config()
self.assertTrue(config["monitoring"]["enabled"])
self.assertEqual(
config["monitoring"]["extensions"],
[".png", ".jpg", ".jpeg", ".webp", ".gif"],
)
self.assertEqual(config["tracking"]["prompt_timeout"], 600)
self.assertEqual(config["web_interface"]["images_per_page"], 20)
def test_update_monitoring(self):
GalleryConfig.update_config(
{"monitoring": {"enabled": False, "processing_delay": 5.0}}
)
self.assertFalse(GalleryConfig.MONITORING_ENABLED)
self.assertEqual(GalleryConfig.PROCESSING_DELAY, 5.0)
def test_save_foreign_runs_defaults_on_and_only_booleans_apply(self):
"""#247: gallery.monitoring.save_foreign_runs, default true."""
self.assertIs(
GalleryConfig.get_config()["monitoring"]["save_foreign_runs"], True
)
GalleryConfig.update_config({"monitoring": {"save_foreign_runs": False}})
self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, False)
for junk in ("true", 1, None):
with self.subTest(value=junk):
with self.assertLogs("prompt_manager.config", "WARNING") as logs:
GalleryConfig.update_config(
{"monitoring": {"save_foreign_runs": junk}}
)
self.assertIn(
"Ignoring gallery monitoring save_foreign_runs: expected true "
"or false",
"\n".join(logs.output),
)
self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, False)
GalleryConfig.update_config({"monitoring": {"save_foreign_runs": True}})
self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, True)
def test_update_tracking(self):
GalleryConfig.update_config({"tracking": {"prompt_timeout": 300}})
self.assertEqual(GalleryConfig.PROMPT_TIMEOUT, 300)
def test_update_database(self):
GalleryConfig.update_config(
{"database": {"auto_cleanup": False, "max_image_age_days": 30}}
)
self.assertFalse(GalleryConfig.AUTO_CLEANUP_MISSING_FILES)
self.assertEqual(GalleryConfig.MAX_IMAGE_AGE_DAYS, 30)
def test_update_web_interface(self):
GalleryConfig.update_config(
{"web_interface": {"images_per_page": 50, "thumbnail_size": 512}}
)
self.assertEqual(GalleryConfig.IMAGES_PER_PAGE, 50)
# #317 R1: the never-read thumbnail_size is ignored (thumbnails.edge).
self.assertFalse(hasattr(GalleryConfig, "THUMBNAIL_SIZE"))
def test_update_performance(self):
GalleryConfig.update_config({"performance": {"max_concurrent_processing": 5}})
self.assertEqual(GalleryConfig.MAX_CONCURRENT_PROCESSING, 5)
def test_partial_update_preserves_other_values(self):
original_timeout = GalleryConfig.PROMPT_TIMEOUT
GalleryConfig.update_config({"monitoring": {"enabled": False}})
self.assertEqual(GalleryConfig.PROMPT_TIMEOUT, original_timeout)
def test_empty_update_changes_nothing(self):
before = GalleryConfig.get_config()
GalleryConfig.update_config({})
after = GalleryConfig.get_config()
self.assertEqual(before, after)
class TestPromptManagerConfig(unittest.TestCase):
"""Test PromptManagerConfig class."""
def setUp(self):
self._orig = PromptManagerConfig.get_config()
def tearDown(self):
PromptManagerConfig.update_config(self._orig)
def test_get_config_structure(self):
config = PromptManagerConfig.get_config()
self.assertIn("database", config)
self.assertIn("web_ui", config)
self.assertIn("performance", config)
self.assertIn("gallery", config)
def test_gallery_config_nested(self):
config = PromptManagerConfig.get_config()
# Gallery section should match GalleryConfig output
self.assertIn("monitoring", config["gallery"])
self.assertIn("tracking", config["gallery"])
def test_default_database_values(self):
config = PromptManagerConfig.get_config()
self.assertEqual(config["database"]["default_path"], "prompts.db")
self.assertTrue(config["database"]["enable_duplicate_detection"])
self.assertTrue(config["database"]["enable_auto_save"])
def test_default_web_ui_values(self):
config = PromptManagerConfig.get_config()
self.assertEqual(config["web_ui"]["webui_display_mode"], "newtab")
def test_update_database(self):
PromptManagerConfig.update_config({"database": {"default_path": "custom.db"}})
self.assertEqual(PromptManagerConfig.DEFAULT_DB_PATH, "custom.db")
def test_update_web_ui(self):
PromptManagerConfig.update_config({"web_ui": {"webui_display_mode": "popup"}})
self.assertEqual(PromptManagerConfig.WEBUI_DISPLAY_MODE, "popup")
def test_update_performance(self):
PromptManagerConfig.update_config({"performance": {"max_search_results": 50}})
self.assertEqual(PromptManagerConfig.MAX_SEARCH_RESULTS, 50)
def test_infinite_scroll_default_off_and_only_booleans_apply(self):
self.assertIn("infinite_scroll", PromptManagerConfig.get_config()["web_ui"])
PromptManagerConfig.update_config({"web_ui": {"infinite_scroll": True}})
self.assertTrue(PromptManagerConfig.INFINITE_SCROLL)
PromptManagerConfig.update_config({"web_ui": {"infinite_scroll": "yes"}})
self.assertTrue(PromptManagerConfig.INFINITE_SCROLL)
PromptManagerConfig.update_config({"web_ui": {"infinite_scroll": False}})
self.assertFalse(PromptManagerConfig.INFINITE_SCROLL)
def test_worker_threads_default_is_half_the_cores_and_current_is_in_range(self):
# The current value may come from config.json (loaded at import), so the
# rule is checked on the constant and only the range on the live value.
cores = PromptManagerConfig.max_worker_threads()
self.assertGreaterEqual(cores, 1)
self.assertEqual(PromptManagerConfig.DEFAULT_WORKER_THREADS, max(1, cores // 2))
current = PromptManagerConfig.get_config()["performance"]["worker_threads"]
self.assertGreaterEqual(current, 1)
self.assertLessEqual(current, cores)
def test_update_worker_threads_is_clamped_to_the_machine(self):
cores = PromptManagerConfig.max_worker_threads()
PromptManagerConfig.update_config({"performance": {"worker_threads": 1}})
self.assertEqual(PromptManagerConfig.WORKER_THREADS, 1)
PromptManagerConfig.update_config(
{"performance": {"worker_threads": cores + 50}}
)
self.assertEqual(PromptManagerConfig.WORKER_THREADS, cores)
PromptManagerConfig.update_config({"performance": {"worker_threads": 0}})
self.assertEqual(PromptManagerConfig.WORKER_THREADS, 1)
def test_update_worker_threads_ignores_junk_from_a_hand_edited_file(self):
PromptManagerConfig.update_config({"performance": {"worker_threads": 1}})
PromptManagerConfig.update_config({"performance": {"worker_threads": "lots"}})
self.assertEqual(PromptManagerConfig.WORKER_THREADS, 1)
PromptManagerConfig.update_config({"performance": {"worker_threads": True}})
self.assertEqual(PromptManagerConfig.WORKER_THREADS, 1)
def test_update_propagates_to_gallery(self):
PromptManagerConfig.update_config(
{"gallery": {"monitoring": {"enabled": False}}}
)
self.assertFalse(GalleryConfig.MONITORING_ENABLED)
# Restore
GalleryConfig.MONITORING_ENABLED = True
def test_save_and_load_file(self):
tmp = tempfile.NamedTemporaryFile(
delete=False, suffix=".json", dir=tempfile.gettempdir()
)
tmp.close()
try:
# Modify a value
PromptManagerConfig.MAX_SEARCH_RESULTS = 42
PromptManagerConfig.save_to_file(tmp.name)
# Reset and reload
PromptManagerConfig.MAX_SEARCH_RESULTS = 100
PromptManagerConfig.load_from_file(tmp.name)
self.assertEqual(PromptManagerConfig.MAX_SEARCH_RESULTS, 42)
finally:
os.unlink(tmp.name)
def test_load_nonexistent_file_uses_defaults(self):
# Should not raise, just log and continue
original = PromptManagerConfig.MAX_SEARCH_RESULTS
PromptManagerConfig.load_from_file("/nonexistent/path/config.json")
self.assertEqual(PromptManagerConfig.MAX_SEARCH_RESULTS, original)
def test_load_invalid_json_file(self):
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".json", mode="w")
tmp.write("not valid json {{")
tmp.close()
try:
original = PromptManagerConfig.MAX_SEARCH_RESULTS
PromptManagerConfig.load_from_file(tmp.name)
# Should not crash, keeps existing values
self.assertEqual(PromptManagerConfig.MAX_SEARCH_RESULTS, original)
finally:
os.unlink(tmp.name)
def test_save_creates_directory(self):
tmp_dir = tempfile.mkdtemp()
nested_path = os.path.join(tmp_dir, "subdir", "config.json")
try:
PromptManagerConfig.save_to_file(nested_path)
self.assertTrue(os.path.exists(nested_path))
with open(nested_path) as f:
data = json.load(f)
self.assertIn("database", data)
finally:
import shutil
shutil.rmtree(tmp_dir)
def test_saved_file_is_private_to_the_user(self):
import stat
tmp_dir = tempfile.mkdtemp()
try:
path = os.path.join(tmp_dir, "config.json")
PromptManagerConfig.save_to_file(path)
mode = stat.S_IMODE(os.stat(path).st_mode)
if os.name == "posix":
self.assertEqual(mode, 0o600)
with open(path) as f:
self.assertIn("database", json.load(f))
# Overwriting an existing, more open file tightens it as well
os.chmod(path, 0o644)
PromptManagerConfig.save_to_file(path)
if os.name == "posix":
self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
finally:
import shutil
shutil.rmtree(tmp_dir, ignore_errors=True)
def test_saved_file_is_valid_json(self):
tmp = tempfile.NamedTemporaryFile(
delete=False, suffix=".json", dir=tempfile.gettempdir()
)
tmp.close()
try:
PromptManagerConfig.save_to_file(tmp.name)
with open(tmp.name) as f:
data = json.load(f)
self.assertIsInstance(data, dict)
finally:
os.unlink(tmp.name)
class TestIntegrationConfig(unittest.TestCase):
"""Test IntegrationConfig for LoRA Manager settings."""
def setUp(self):
"""Save original values to restore after each test."""
self._orig = IntegrationConfig.get_config()
def tearDown(self):
"""Restore original config values."""
IntegrationConfig.update_config(self._orig)
def test_reset_to_disabled(self):
"""Integrations can be fully disabled via update_config."""
IntegrationConfig.update_config(
{
"lora_manager": {
"enabled": False,
"path": "",
"trigger_words_enabled": False,
"civitai_api_key": "",
}
}
)
config = IntegrationConfig.get_config()
lora = config["lora_manager"]
self.assertFalse(lora["enabled"])
self.assertEqual(lora["path"], "")
self.assertFalse(lora["trigger_words_enabled"])
self.assertEqual(lora["civitai_api_key"], "")
def test_get_config_structure(self):
config = IntegrationConfig.get_config()
self.assertIn("lora_manager", config)
lora = config["lora_manager"]
self.assertIn("enabled", lora)
self.assertIn("path", lora)
self.assertIn("trigger_words_enabled", lora)
self.assertIn("civitai_api_key", lora)
def test_update_config_enables(self):
IntegrationConfig.update_config(
{
"lora_manager": {
"enabled": True,
"path": "/some/path",
"trigger_words_enabled": True,
"civitai_api_key": "test-key-123",
}
}
)
self.assertTrue(IntegrationConfig.LORA_MANAGER_ENABLED)
self.assertEqual(IntegrationConfig.LORA_MANAGER_PATH, "/some/path")
self.assertTrue(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED)
self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, "test-key-123")
def test_update_partial(self):
"""Updating one field shouldn't affect others."""
# Reset to known state first
IntegrationConfig.update_config(
{"lora_manager": {"enabled": False, "path": "/known"}}
)
# Now update only enabled
IntegrationConfig.update_config({"lora_manager": {"enabled": True}})
self.assertTrue(IntegrationConfig.LORA_MANAGER_ENABLED)
self.assertEqual(IntegrationConfig.LORA_MANAGER_PATH, "/known")
def test_update_empty_dict_noop(self):
"""Updating with empty dict preserves current state."""
before = IntegrationConfig.get_config()
IntegrationConfig.update_config({})
after = IntegrationConfig.get_config()
self.assertEqual(before, after)
def test_roundtrip(self):
"""get_config → update_config → get_config should be stable."""
IntegrationConfig.update_config(
{"lora_manager": {"enabled": True, "path": "/test"}}
)
config1 = IntegrationConfig.get_config()
IntegrationConfig.update_config(config1)
config2 = IntegrationConfig.get_config()
self.assertEqual(config1, config2)
class TestConfigPathOverride(unittest.TestCase):
"""PROMPT_MANAGER_CONFIG_PATH relocates the persisted config.json."""
ENV = "PROMPT_MANAGER_CONFIG_PATH"
def setUp(self):
self._orig_env = os.environ.pop(self.ENV, None)
self._orig = PromptManagerConfig.get_config()
def tearDown(self):
if self._orig_env is None:
os.environ.pop(self.ENV, None)
else:
os.environ[self.ENV] = self._orig_env
PromptManagerConfig.update_config(self._orig)
def test_default_config_path_is_repo_config_json(self):
repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
expected = os.path.join(repo_root, "config.json")
self.assertEqual(
os.path.normcase(PromptManagerConfig.get_config_path()),
os.path.normcase(expected),
)
def test_env_override_is_used(self):
with tempfile.TemporaryDirectory() as tmpdir:
custom = os.path.join(tmpdir, "custom.json")
os.environ[self.ENV] = custom
self.assertEqual(PromptManagerConfig.get_config_path(), custom)
def test_save_without_path_uses_env_override(self):
with tempfile.TemporaryDirectory() as tmpdir:
custom = os.path.join(tmpdir, "nested", "custom.json")
os.environ[self.ENV] = custom
PromptManagerConfig.RESULT_TIMEOUT = 42
PromptManagerConfig.save_to_file()
with open(custom) as f:
saved = json.load(f)
self.assertEqual(saved["web_ui"]["result_timeout"], 42)
def test_load_without_path_uses_env_override(self):
with tempfile.TemporaryDirectory() as tmpdir:
custom = os.path.join(tmpdir, "custom.json")
with open(custom, "w") as f:
json.dump({"web_ui": {"result_timeout": 77}}, f)
os.environ[self.ENV] = custom
PromptManagerConfig.load_from_file()
self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, 77)
class TestAutotagConfig(unittest.TestCase):
"""``autotag.defaults.suggestion_limit``: integer 1..100, default 10 (D68)."""
def setUp(self):
self.config_class = getattr(config_module, "AutotagConfig", None)
self.assertIsNotNone(self.config_class, "AutotagConfig is not implemented")
self._orig = PromptManagerConfig.get_config()
self._defaults = {
attr: getattr(self.config_class, attr, None)
for attr in (
"MODEL_TYPE",
"EXISTING",
"GENERAL_THRESHOLD",
"CHARACTER_THRESHOLD",
"TEMPLATE",
"SUGGESTION_LIMIT",
)
}
def tearDown(self):
for attr, value in self._defaults.items():
setattr(self.config_class, attr, value)
PromptManagerConfig.update_config(self._orig)
def _load(self, value):
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "config.json")
with open(path, "w") as f:
# json writes NaN for float("nan"), which json.load reads back.
json.dump({"autotag": {"defaults": {"suggestion_limit": value}}}, f)
with self.assertLogs("prompt_manager.config", level="WARNING") as logs:
PromptManagerConfig.load_from_file(path)
return logs
def test_every_copy_of_the_tag_set_cap_is_one_value(self):
# A suggestion request is bounded by the image tag set cap; the
# copies that cannot import it (config loads before the database,
# the inspection envelope stays storage-free) are pinned to it here.
from database.annotations import MAX_IMAGE_TAGS
from py import autotag
from utils import inspection_response
self.assertEqual(autotag.MAX_SUGGESTION_LIMIT, MAX_IMAGE_TAGS)
self.assertEqual(config_module.MAX_SUGGESTION_LIMIT, MAX_IMAGE_TAGS)
self.assertEqual(inspection_response.MAX_IMAGE_TAGS, MAX_IMAGE_TAGS)
def test_default_is_ten_and_persisted_under_autotag_defaults(self):
self.assertEqual(config_module.DEFAULT_SUGGESTION_LIMIT, 10)
self.assertEqual(config_module.MAX_SUGGESTION_LIMIT, 100)
self.assertEqual(self.config_class.suggestion_limit(), 10)
self.assertEqual(
PromptManagerConfig.get_config()["autotag"],
{
"defaults": {
"model_type": "wd14-swinv2",
"existing": "skip",
"general_threshold": 0.35,
"character_threshold": 0.85,
"suggestion_limit": 10,
}
},
)
# The built-in template is not written; a saved one is.
self.config_class.TEMPLATE = "Saved."
self.assertEqual(
PromptManagerConfig.get_config()["autotag"]["defaults"]["template"],
"Saved.",
)
self.config_class.TEMPLATE = config_module.DEFAULT_TAG_TEMPLATE
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "config.json")
PromptManagerConfig.update_config(
{"autotag": {"defaults": {"suggestion_limit": 25}}}
)
PromptManagerConfig.save_to_file(path)
self.config_class.SUGGESTION_LIMIT = 10
PromptManagerConfig.load_from_file(path)
self.assertEqual(self.config_class.suggestion_limit(), 25)
def test_valid_limits_are_kept(self):
for value in (1, 10, 100):
with self.subTest(value=value):
PromptManagerConfig.update_config(
{"autotag": {"defaults": {"suggestion_limit": value}}}
)
self.assertEqual(self.config_class.suggestion_limit(), value)
def test_autotag_defaults_fall_back_per_key_on_load(self):
"""#328: each saved tagging default is checked alone on load; a bad
hand-edited value falls back to its default with one warning and
leaves the other keys as saved."""
good = {
"model_type": "gguf",
"existing": "merge",
"general_threshold": 0.4,
"character_threshold": 0.7,
"template": "Tags only.",
"suggestion_limit": 15,
}
fallback = {
"model_type": "wd14-swinv2",
"existing": "skip",
"general_threshold": 0.35,
"character_threshold": 0.85,
"template": config_module.DEFAULT_TAG_TEMPLATE,
"suggestion_limit": 10,
}
bad_values = {
"model_type": ["wd14-convnext", 3, None],
"existing": ["replace", "", True],
# #369: a saved threshold off the 0.05 step no page can show.
"general_threshold": [1.5, -0.1, float("nan"), "0.4", True, 0.37],
"character_threshold": [2, float("inf"), None, 0.33],
# 4000 characters are at most 16,000 UTF-8 bytes, so the 16 KiB
# byte cap can only be met past the character cap.
"template": ["", " ", "x" * 4001, "\U0001f600" * 4097, "\ud800", 7],
}
for key, values in bad_values.items():
for value in values:
with self.subTest(key=key, value=value):
saved = dict(good, **{key: value})
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "config.json")
with open(path, "w") as f:
json.dump({"autotag": {"defaults": saved}}, f)
with self.assertLogs(
"prompt_manager.config", level="WARNING"
) as logs:
PromptManagerConfig.load_from_file(path)
self.assertEqual(len(logs.records), 1, logs.output)
self.assertIn(f"autotag.defaults.{key}", logs.output[0])
loaded = {
k: getattr(self.config_class, attr)
for k, (attr, _v, _d) in self.config_class._KEYS.items()
}
self.assertEqual(loaded, dict(good, **{key: fallback[key]}))
def test_tagging_validators_moved_with_their_messages(self):
validate_threshold = config_module.validate_threshold
validate_template = config_module.validate_template
self.assertEqual(validate_threshold("general_threshold", 1), 1.0)
for value, message in (
(float("nan"), "general_threshold must be a finite number"),
("0.4", "general_threshold must be a finite number"),
(1.01, "general_threshold must be between 0 and 1"),
):
with self.assertRaises(ValueError) as caught:
validate_threshold("general_threshold", value)
self.assertEqual(str(caught.exception), message)
self.assertEqual(validate_template("Tags."), "Tags.")
for value, message in (
(" ", "instructions must be non-empty text"),
("x" * 4001, "instructions must be at most 4000 characters"),
("\ud800", "instructions must be valid Unicode text"),
):
with self.assertRaises(ValueError) as caught:
validate_template(value)
self.assertEqual(str(caught.exception), message)
self.assertEqual(
config_module.MODEL_TYPES, {"gguf", "hf", "wd14-swinv2", "wd14-vit"}
)
def test_saved_threshold_is_a_multiple_of_the_step(self):
"""#369: every page shows the thresholds in steps of 0.05, so a saved
one must be on that grid; it is stored exactly on it."""
from py.tagging_defaults import validate_saved_threshold as validate
for value in (0, 0.05, 0.15, 0.35, 0.85, 1):
with self.subTest(value=value):
self.assertEqual(validate("x", value), value)
self.assertEqual(validate("x", 0.1 + 0.2), 0.3)
for value, message in (
(0.37, "x must be a multiple of 0.05"),
(0.351, "x must be a multiple of 0.05"),
(0.35000000001, "x must be a multiple of 0.05"),
(1.5, "x must be between 0 and 1"),
("0.4", "x must be a finite number"),
):
with self.subTest(value=value):
with self.assertRaises(ValueError) as caught:
validate("x", value)
self.assertEqual(str(caught.exception), message)
def test_suggestion_limit_setting_validated_on_load(self):
for value in (0, 101, "10", float("nan"), 2.5, True, None, [10]):
with self.subTest(value=value):
self.config_class.SUGGESTION_LIMIT = 42
logs = self._load(value)
self.assertEqual(self.config_class.suggestion_limit(), 10)
self.assertIn("suggestion_limit", "\n".join(logs.output))
def test_missing_or_malformed_sections_keep_the_current_value(self):
self.config_class.SUGGESTION_LIMIT = 7
for config in ({}, {"autotag": {}}, {"autotag": {"defaults": {}}}):
PromptManagerConfig.update_config(config)
self.assertEqual(self.config_class.suggestion_limit(), 7)
with self.assertLogs("prompt_manager.config", level="WARNING"):
PromptManagerConfig.update_config({"autotag": "nope"})
with self.assertLogs("prompt_manager.config", level="WARNING"):
PromptManagerConfig.update_config({"autotag": {"defaults": [1]}})
self.assertEqual(self.config_class.suggestion_limit(), 7)
if __name__ == "__main__":
unittest.main()