Files
bc0edb1827 feat(nodes)!: fold Knowledge Base Loader into the Configuration node (#149)
The Knowledge Base directory was entered in two places — the Configuration node's
knowledge_base path and the Knowledge Base Loader node's path — a duplication with
no benefit. Remove the Knowledge Base Loader node and have the Configuration node
load the Knowledge Base itself, emitting it on a new knowledge_base output wired
into the Resolver.

- New nodes/kb_loading.py holds the shared, process-wide load-once cache and
  package-relative path resolution (extracted from the deleted loader node).
- Configuration node: gains a knowledge_base_reload input and a KNOWLEDGE_BASE
  output; RETURN is now (config, knowledge_base, warnings, errors). A KB load
  failure surfaces on errors while the config is still emitted (the Analyzer and
  Validator do not need the KB).
- Remove SceneCompilerKnowledgeBaseLoader from the node registrations.
- Rebuild the example workflow to wire Configuration.knowledge_base -> Resolver.
- Docs (README, docs/nodes.md) updated; tests moved to test_kb_loading.py.

BREAKING CHANGE: the Knowledge Base Loader node is removed. Workflows that used it
must load the Knowledge Base via the Configuration node's knowledge_base output.

Co-authored-by: kunail0804 <kunail0804@gmail.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 17:05:53 +02:00

92 lines
3.5 KiB
Python

"""Tests that the shipped example scenes compile to their documented outputs (#30)."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from tests.regression.golden_scenes import (
compile_prompt,
load_reference_knowledge_base,
)
EXAMPLES_DIR = Path(__file__).resolve().parent.parent / "examples"
SCENES_DIR = EXAMPLES_DIR / "scenes"
WORKFLOW_FILE = EXAMPLES_DIR / "workflows" / "scene_compiler_pipeline.json"
SCENE_FILES = sorted(SCENES_DIR.glob("*.scene.json"))
@pytest.fixture(scope="module")
def knowledge_base():
return load_reference_knowledge_base()
def test_example_scenes_exist() -> None:
assert SCENE_FILES, "no example scenes found"
for scene_file in SCENE_FILES:
expected = scene_file.with_name(scene_file.name.replace(".scene.json", ".prompts.json"))
assert expected.is_file(), f"missing expected outputs for {scene_file.name}"
@pytest.mark.parametrize("scene_file", SCENE_FILES, ids=lambda p: p.stem)
def test_example_scene_compiles_to_documented_outputs(scene_file: Path, knowledge_base) -> None:
scene = json.loads(scene_file.read_text(encoding="utf-8"))
expected_file = scene_file.with_name(scene_file.name.replace(".scene.json", ".prompts.json"))
expected = json.loads(expected_file.read_text(encoding="utf-8"))
assert compile_prompt(scene, knowledge_base) == expected
def test_workflow_is_valid_json_referencing_registered_nodes() -> None:
import nodes
workflow = json.loads(WORKFLOW_FILE.read_text(encoding="utf-8"))
node_types = {node["type"] for node in workflow["nodes"]}
# Every node type in the example workflow is a registered Scene Compiler node.
assert node_types <= set(nodes.NODE_CLASS_MAPPINGS)
# The full pipeline is present.
for required in (
"SceneCompilerConfiguration",
"SceneCompilerAnalyzer",
"SceneCompilerValidator",
"SceneCompilerResolver",
):
assert required in node_types
# ComfyUI stores widget values positionally, so a node whose widget list changes
# silently misaligns older saved workflows (a boolean landing where a string is
# expected, etc.). This guards the shipped example against that drift.
_WIDGET_PRIMITIVES = {"STRING", "INT", "FLOAT", "BOOLEAN"}
def _widget_count(input_types: dict) -> int:
"""Count the positional widgets a node exposes (excluding socket-only inputs)."""
count = 0
for section in ("required", "optional"):
for spec in input_types.get(section, {}).values():
type_name = spec[0]
options = spec[1] if len(spec) > 1 else {}
is_combo = isinstance(type_name, (list, tuple))
is_primitive = isinstance(type_name, str) and type_name in _WIDGET_PRIMITIVES
# ``forceInput`` primitives render as sockets, not widgets.
if is_combo or (is_primitive and not options.get("forceInput")):
count += 1
return count
def test_workflow_widget_values_align_with_current_nodes() -> None:
import nodes
workflow = json.loads(WORKFLOW_FILE.read_text(encoding="utf-8"))
for node in workflow["nodes"]:
node_cls = nodes.NODE_CLASS_MAPPINGS[node["type"]]
expected = _widget_count(node_cls.INPUT_TYPES())
actual = len(node.get("widgets_values", []))
assert actual == expected, (
f"{node['type']} has {actual} saved widget values but the node now "
f"defines {expected} widgets — the workflow is stale and would misalign."
)