From bb530c47bbec7fc8e561d5bf4c2a86860d4903dc Mon Sep 17 00:00:00 2001 From: DemonGatanjieu <1027066412@qq.com> Date: Sun, 2 Aug 2026 17:32:30 +0800 Subject: [PATCH] Delete tests directory --- tests/test_backend_safety.py | 528 ----------------------------------- tests/test_graph_splice.mjs | 193 ------------- tests/test_model_picker.mjs | 65 ----- 3 files changed, 786 deletions(-) delete mode 100644 tests/test_backend_safety.py delete mode 100644 tests/test_graph_splice.mjs delete mode 100644 tests/test_model_picker.mjs diff --git a/tests/test_backend_safety.py b/tests/test_backend_safety.py deleted file mode 100644 index f9885b8..0000000 --- a/tests/test_backend_safety.py +++ /dev/null @@ -1,528 +0,0 @@ -import importlib.util -import json -import os -import sys -import tempfile -import types -import unittest -from unittest import mock -from pathlib import Path - - -folder_paths = types.ModuleType("folder_paths") -sys.modules.setdefault("folder_paths", folder_paths) - -API_PATH = Path(__file__).parents[1] / "api" -package = types.ModuleType("anomalous_test_api") -package.__path__ = [str(API_PATH)] -sys.modules[package.__name__] = package - - -def load_api_module(name): - full_name = f"{package.__name__}.{name}" - spec = importlib.util.spec_from_file_location(full_name, API_PATH / f"{name}.py") - module = importlib.util.module_from_spec(spec) - sys.modules[full_name] = module - spec.loader.exec_module(module) - return module - - -utils = load_api_module("utils") -metadata = load_api_module("metadata") -models = load_api_module("models") - - -class JsonRequest: - def __init__(self, payload): - self.payload = payload - - async def json(self): - return self.payload - - -class PathSafetyTests(unittest.TestCase): - def setUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.root = os.path.realpath(self.temp_dir.name) - folder_paths.get_folder_paths = lambda folder_type: [self.root] - - def tearDown(self): - self.temp_dir.cleanup() - - def test_normal_subfolder_stays_inside_root(self): - base, target = utils.resolve_folder_subdir("checkpoints", 0, "nested/models") - self.assertEqual(base, self.root) - self.assertEqual(target, os.path.join(self.root, "nested", "models")) - - def test_parent_traversal_is_rejected(self): - with self.assertRaises(ValueError): - utils.resolve_folder_subdir("checkpoints", 0, "../outside") - - def test_absolute_path_is_rejected(self): - absolute_target = os.path.abspath(os.path.join(self.root, os.pardir, "outside")) - with self.assertRaises(ValueError): - utils.resolve_folder_subdir("checkpoints", 0, absolute_target) - - def test_negative_folder_index_is_rejected(self): - with self.assertRaises(ValueError): - utils.resolve_folder_subdir("checkpoints", -1, "/") - - def test_filename_must_be_a_basename(self): - self.assertEqual(utils.require_filename("cover.png"), "cover.png") - with self.assertRaises(ValueError): - utils.require_filename(os.path.join("nested", "cover.png")) - - -class ImageEndpointSafetyTests(unittest.IsolatedAsyncioTestCase): - async def asyncSetUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.root = os.path.realpath(self.temp_dir.name) - folder_paths.get_folder_paths = lambda folder_type: [self.root] - folder_paths.get_temp_directory = lambda: os.path.join(self.root, "temp") - Path(self.root, "cover.png").write_bytes(b"not-a-real-png") - - async def asyncTearDown(self): - self.temp_dir.cleanup() - - async def test_image_endpoint_rejects_absolute_filename(self): - outside = os.path.abspath(os.path.join(self.root, os.pardir, "outside.txt")) - request = types.SimpleNamespace(query={"type": "checkpoints", "path_idx": "0", "filename": outside}) - response = await utils.api_serve_image(request) - self.assertEqual(response.status, 400) - - async def test_image_endpoint_accepts_supported_local_media(self): - request = types.SimpleNamespace(query={"type": "checkpoints", "path_idx": "0", "filename": "cover.png"}) - response = await utils.api_serve_image(request) - self.assertEqual(response.status, 200) - - async def test_image_endpoint_rejects_non_media_files(self): - Path(self.root, "metadata.json").write_text("{}", encoding="utf-8") - request = types.SimpleNamespace(query={"type": "checkpoints", "path_idx": "0", "filename": "metadata.json"}) - response = await utils.api_serve_image(request) - self.assertEqual(response.status, 415) - - async def test_card_variant_uses_bounded_cached_thumbnail_without_touching_source(self): - from PIL import Image - - source = Path(self.root, "large.png") - Image.new("RGB", (1200, 800), (40, 80, 120)).save(source) - original_bytes = source.read_bytes() - request = types.SimpleNamespace(query={ - "type": "checkpoints", - "path_idx": "0", - "filename": source.name, - "variant": "card", - "t": str(source.stat().st_mtime_ns), - }) - - response = await utils.api_serve_image(request) - - self.assertEqual(response.status, 200) - self.assertNotEqual(os.path.realpath(response._path), os.path.realpath(source)) - self.assertEqual(response.headers.get("Content-Type"), "image/webp") - self.assertIn("immutable", response.headers.get("Cache-Control", "")) - with Image.open(response._path) as thumbnail: - self.assertLessEqual(max(thumbnail.size), utils.CARD_THUMBNAIL_EDGE) - self.assertEqual(source.read_bytes(), original_bytes) - - async def test_original_variant_keeps_original_cover_path(self): - request = types.SimpleNamespace(query={ - "type": "checkpoints", - "path_idx": "0", - "filename": "cover.png", - }) - - response = await utils.api_serve_image(request) - - self.assertEqual(os.path.realpath(response._path), os.path.realpath(Path(self.root, "cover.png"))) - - -class MetadataHashSelectionTests(unittest.TestCase): - def test_multifile_info_hash_is_selected_by_physical_size(self): - with tempfile.TemporaryDirectory() as temp_dir: - model_path = Path(temp_dir, "renamed-model.safetensors") - model_path.write_bytes(b"the-local-model") - dependency_hash = "A" * 64 - model_hash = "B" * 64 - info = { - "files": [ - { - "name": "dependency-vae.safetensors", - "sizeKB": 5 / 1024, - "hashes": {"SHA256": dependency_hash}, - }, - { - "name": "original-model-name.safetensors", - "sizeKB": model_path.stat().st_size / 1024, - "hashes": {"SHA256": model_hash}, - }, - ] - } - model_path.with_suffix(".info").write_text(json.dumps(info), encoding="utf-8") - - result = metadata.get_metadata(str(model_path)) - self.assertEqual(result["hash"], model_hash) - - def test_equal_size_multifile_info_does_not_use_filename_or_first_hash(self): - with tempfile.TemporaryDirectory() as temp_dir: - model_path = Path(temp_dir, "renamed-model.safetensors") - model_path.write_bytes(b"size-not-listed") - size_kb = model_path.stat().st_size / 1024 - info = { - "files": [ - {"name": "renamed-model.safetensors", "sizeKB": size_kb, "hashes": {"SHA256": "A" * 64}}, - {"name": "b.safetensors", "sizeKB": size_kb, "hashes": {"SHA256": "B" * 64}}, - ] - } - model_path.with_suffix(".info").write_text(json.dumps(info), encoding="utf-8") - - self.assertEqual(metadata.get_metadata(str(model_path))["hash"], "") - - -class MetadataCacheTests(unittest.TestCase): - def setUp(self): - metadata.clear_metadata_cache() - self.temp_dir = tempfile.TemporaryDirectory() - self.model_path = Path(self.temp_dir.name, "cached.safetensors") - self.info_path = self.model_path.with_suffix(".info") - self.model_path.write_bytes(b"model") - self.info_path.write_text(json.dumps({"name": "First", "trainedWords": ["one"]}), encoding="utf-8") - - def tearDown(self): - metadata.clear_metadata_cache() - self.temp_dir.cleanup() - - def test_unchanged_metadata_uses_cache_and_returns_independent_objects(self): - first = metadata.get_metadata(str(self.model_path)) - first["trainedWords"].append("mutated") - second = metadata.get_metadata(str(self.model_path)) - cache_info = metadata.get_metadata_cache_info() - - self.assertEqual(second["name"], "First") - self.assertEqual(second["trainedWords"], ["one"]) - self.assertEqual(cache_info.misses, 1) - self.assertEqual(cache_info.hits, 1) - - def test_sidecar_signature_change_invalidates_cached_metadata(self): - self.assertEqual(metadata.get_metadata(str(self.model_path))["name"], "First") - self.info_path.write_text(json.dumps({"name": "Second and changed"}), encoding="utf-8") - - self.assertEqual(metadata.get_metadata(str(self.model_path))["name"], "Second and changed") - self.assertEqual(metadata.get_metadata_cache_info().misses, 2) - - def test_embedded_hash_reader_is_cached_by_model_signature(self): - with mock.patch.object(metadata, "_read_safetensors_hash", return_value="A" * 64) as reader: - first = metadata._extract_safetensors_hash(str(self.model_path)) - second = metadata._extract_safetensors_hash(str(self.model_path)) - - self.assertEqual(first, second) - self.assertEqual(reader.call_count, 1) - - -class ModelSidecarLifecycleTests(unittest.IsolatedAsyncioTestCase): - async def asyncSetUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.root = Path(self.temp_dir.name) - self.original_folder_names_and_paths = getattr(folder_paths, "folder_names_and_paths", None) - folder_paths.get_folder_paths = lambda folder_type: [str(self.root)] - - async def asyncTearDown(self): - if self.original_folder_names_and_paths is None: - try: - del folder_paths.folder_names_and_paths - except AttributeError: - pass - else: - folder_paths.folder_names_and_paths = self.original_folder_names_and_paths - self.temp_dir.cleanup() - - def request(self, filename, **overrides): - payload = { - "type": "checkpoints", - "path_idx": 0, - "subfolder": "/", - "filename": filename, - "custom_name": "", - "custom_notes": "", - "physical_rename": False, - } - payload.update(overrides) - return JsonRequest(payload) - - async def test_reset_restores_civitai_backup_and_keeps_the_backup(self): - Path(self.root, "model.safetensors").write_bytes(b"model") - Path(self.root, "model.preview.png").write_bytes(b"custom") - Path(self.root, "model.civitai_bak.jpg").write_bytes(b"civitai") - - response = await models.api_update_metadata(self.request("model.safetensors", reset_cover=True)) - result = json.loads(response.text) - - self.assertTrue(result["cover_reset"]) - self.assertEqual(result["cover_reset_source"], "civitai_backup") - self.assertFalse(Path(self.root, "model.preview.png").exists()) - self.assertEqual(Path(self.root, "model.preview.jpg").read_bytes(), b"civitai") - self.assertEqual(Path(self.root, "model.civitai_bak.jpg").read_bytes(), b"civitai") - - async def test_reset_falls_back_to_original_bare_cover(self): - Path(self.root, "model.safetensors").write_bytes(b"model") - Path(self.root, "model.preview.webp").write_bytes(b"custom") - Path(self.root, "model.png").write_bytes(b"original") - - response = await models.api_update_metadata(self.request("model.safetensors", reset_cover=True)) - result = json.loads(response.text) - - self.assertTrue(result["cover_reset"]) - self.assertEqual(result["cover_reset_source"], "original_cover") - self.assertFalse(Path(self.root, "model.preview.webp").exists()) - self.assertEqual(Path(self.root, "model.png").read_bytes(), b"original") - - async def test_reset_preserves_only_cover_when_no_restore_source_exists(self): - Path(self.root, "model.safetensors").write_bytes(b"model") - Path(self.root, "model.preview.png").write_bytes(b"only-cover") - - response = await models.api_update_metadata(self.request("model.safetensors", reset_cover=True)) - result = json.loads(response.text) - - self.assertFalse(result["cover_reset"]) - self.assertEqual(result["cover_reset_source"], "preserved_current") - self.assertEqual(Path(self.root, "model.preview.png").read_bytes(), b"only-cover") - - async def test_reset_and_physical_rename_keep_the_model_extension(self): - Path(self.root, "old.safetensors").write_bytes(b"model") - Path(self.root, "old.preview.png").write_bytes(b"custom") - Path(self.root, "old.civitai_bak.jpg").write_bytes(b"civitai") - - response = await models.api_update_metadata(self.request( - "old.safetensors", - custom_name="new", - reset_cover=True, - physical_rename=True, - )) - result = json.loads(response.text) - - self.assertEqual(result["new_filename"], "new.safetensors") - self.assertTrue(Path(self.root, "new.safetensors").is_file()) - self.assertFalse(Path(self.root, "new.avi").exists()) - self.assertEqual(Path(self.root, "new.civitai_bak.jpg").read_bytes(), b"civitai") - self.assertEqual(Path(self.root, "new.preview.jpg").read_bytes(), b"civitai") - - async def test_delete_does_not_remove_a_same_stem_sibling_model_or_shared_sidecars(self): - Path(self.root, "shared.ckpt").write_bytes(b"delete-me") - Path(self.root, "shared.safetensors").write_bytes(b"keep-me") - Path(self.root, "shared.civitai_bak.png").write_bytes(b"shared-cover") - - response = await models.api_delete_model(self.request("shared.ckpt")) - result = json.loads(response.text) - - self.assertEqual(result["status"], "success") - self.assertTrue(result["sidecars_preserved"]) - self.assertFalse(Path(self.root, "shared.ckpt").exists()) - self.assertEqual(Path(self.root, "shared.safetensors").read_bytes(), b"keep-me") - self.assertEqual(Path(self.root, "shared.civitai_bak.png").read_bytes(), b"shared-cover") - - -class ModelListingPerformanceTests(unittest.IsolatedAsyncioTestCase): - async def asyncSetUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.root = Path(self.temp_dir.name) - self.original_folder_names_and_paths = getattr(folder_paths, "folder_names_and_paths", None) - folder_paths.get_folder_paths = lambda folder_type: [str(self.root)] - - async def asyncTearDown(self): - if self.original_folder_names_and_paths is None: - try: - del folder_paths.folder_names_and_paths - except AttributeError: - pass - else: - folder_paths.folder_names_and_paths = self.original_folder_names_and_paths - self.temp_dir.cleanup() - - async def test_listing_keeps_sort_cover_priority_and_stable_cache_token(self): - Path(self.root, "B.ckpt").write_bytes(b"b") - Path(self.root, "a.safetensors").write_bytes(b"aaa") - Path(self.root, "a.png").write_bytes(b"bare") - preview = Path(self.root, "a.preview.webp") - preview.write_bytes(b"preview") - request = types.SimpleNamespace(query={ - "type": "checkpoints", - "path_idx": "0", - "subfolder": "/", - "page": "1", - "limit": "0", - }) - - response = await models.api_get_models(request) - result = json.loads(response.text) - - self.assertEqual([item["filename"] for item in result["models"]], ["a.safetensors", "B.ckpt"]) - listed = result["models"][0] - self.assertIn("filename=a.preview.webp", listed["preview_url"]) - self.assertIn(f"t={preview.stat().st_mtime_ns}", listed["preview_url"]) - self.assertEqual(listed["size_bytes"], 3) - - async def test_exact_preview_paths_bypass_full_library_walk(self): - nested = Path(self.root, "nested") - nested.mkdir() - Path(nested, "model.safetensors").write_bytes(b"model") - Path(nested, "model.preview.png").write_bytes(b"cover") - Path(nested, "model.info").write_text( - json.dumps({"baseModel": "SDXL 1.0"}), - encoding="utf-8", - ) - folder_paths.folder_names_and_paths = {"checkpoints": object()} - - with mock.patch.object(models.os, "walk", side_effect=AssertionError("full walk should not run")): - previews = models._resolve_paths_to_previews_sync(["nested/model.safetensors"]) - details = models._resolve_paths_to_model_info_sync( - ["nested/model.safetensors"], - ["checkpoints"], - ) - - self.assertIn("nested/model.safetensors", previews) - self.assertIn("filename=model.preview.png", previews["nested/model.safetensors"]) - model_info = details["nested/model.safetensors"] - self.assertEqual(model_info["type"], "checkpoints") - self.assertEqual(model_info["metadata"]["baseModel"], "SDXL 1.0") - self.assertIn("filename=model.preview.png", model_info["preview_url"]) - - -class HashResolutionTests(unittest.IsolatedAsyncioTestCase): - async def asyncSetUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - temp_root = Path(self.temp_dir.name) - self.checkpoint_root = temp_root / "checkpoints" - self.lora_root = temp_root / "loras" - checkpoint_file = self.checkpoint_root / "a" / "shared.safetensors" - lora_file = self.lora_root / "b" / "shared.safetensors" - checkpoint_file.parent.mkdir(parents=True) - lora_file.parent.mkdir(parents=True) - checkpoint_file.write_bytes(b"wrong-size-match") - lora_file.write_bytes(b"correct-hash-model-with-a-different-size") - self.checkpoint_hash = "1" * 64 - self.lora_hash = "2" * 64 - for file_path, file_hash in ((checkpoint_file, self.checkpoint_hash), (lora_file, self.lora_hash)): - info_path = file_path.with_suffix(".info") - info_path.write_text(json.dumps({"files": [{"hashes": {"SHA256": file_hash}}]}), encoding="utf-8") - self.checkpoint_size = checkpoint_file.stat().st_size - self.lora_size = lora_file.stat().st_size - mapping = {"checkpoints": [str(self.checkpoint_root)], "loras": [str(self.lora_root)]} - folder_paths.get_folder_paths = lambda folder_type: mapping.get(folder_type, []) - - async def asyncTearDown(self): - self.temp_dir.cleanup() - - async def test_ambiguous_basename_alias_is_omitted(self): - response = await models.api_get_all_hashes(types.SimpleNamespace()) - hashes = json.loads(response.text) - self.assertNotIn("shared.safetensors", hashes) - self.assertEqual(hashes["a/shared.safetensors"]["hash"], self.checkpoint_hash) - self.assertEqual(hashes["b/shared.safetensors"]["hash"], self.lora_hash) - - async def test_matching_hash_and_size_resolve_same_file(self): - request = types.SimpleNamespace(query={"hash": self.lora_hash, "size": str(self.lora_size)}) - response = await models.api_resolve_hash(request) - result = json.loads(response.text) - self.assertTrue(result["found"]) - self.assertEqual(result["type"], "loras") - self.assertEqual(result["filename"], "b/shared.safetensors") - self.assertTrue(result["matched_by_hash"]) - self.assertTrue(result["matched_by_size"]) - - async def test_conflicting_hash_and_size_are_rejected_without_type_context(self): - request = types.SimpleNamespace(query={"hash": self.lora_hash, "size": str(self.checkpoint_size)}) - response = await models.api_resolve_hash(request) - result = json.loads(response.text) - self.assertFalse(result["found"]) - self.assertTrue(result["identity_conflict"]) - - async def test_stale_hash_recovers_by_unique_size_in_expected_type(self): - request = types.SimpleNamespace(query={ - "hash": self.lora_hash, - "size": str(self.checkpoint_size), - "type": "checkpoints", - }) - response = await models.api_resolve_hash(request) - result = json.loads(response.text) - self.assertTrue(result["found"]) - self.assertEqual(result["type"], "checkpoints") - self.assertEqual(result["filename"], "a/shared.safetensors") - self.assertTrue(result["stale_hash"]) - - async def test_expected_type_prevents_cross_category_hash_match(self): - request = types.SimpleNamespace(query={"hash": self.checkpoint_hash, "size": "", "type": "loras"}) - response = await models.api_resolve_hash(request) - result = json.loads(response.text) - self.assertFalse(result["found"]) - - async def test_filename_does_not_disambiguate_equal_sizes_with_stale_hash(self): - original = self.checkpoint_root / "a" / "shared.safetensors" - original_info = original.with_suffix(".info") - original_info.write_text(json.dumps({"files": [{ - "name": "original-checkpoint.safetensors", - "sizeKB": self.checkpoint_size / 1024, - "hashes": {"SHA256": self.checkpoint_hash}, - }]}), encoding="utf-8") - - duplicate_size = self.checkpoint_root / "c" / "other.safetensors" - duplicate_size.parent.mkdir(parents=True) - duplicate_size.write_bytes(original.read_bytes()) - duplicate_size.with_suffix(".info").write_text(json.dumps({"files": [{ - "name": "different-checkpoint.safetensors", - "sizeKB": self.checkpoint_size / 1024, - "hashes": {"SHA256": "3" * 64}, - }]}), encoding="utf-8") - - request = types.SimpleNamespace(query={ - "hash": self.lora_hash, - "size": str(self.checkpoint_size), - "filename": "original-checkpoint.safetensors", - "type": "checkpoints", - }) - response = await models.api_resolve_hash(request) - result = json.loads(response.text) - self.assertFalse(result["found"]) - self.assertTrue(result["ambiguous"]) - - async def test_batch_resolution_scans_each_type_group_once(self): - request = JsonRequest({"items": [ - { - "key": "checkpoint", - "hash": self.checkpoint_hash, - "size": self.checkpoint_size, - "type": "checkpoints", - }, - { - "key": "stale", - "hash": self.lora_hash, - "size": self.checkpoint_size, - "type": "checkpoints", - }, - ]}) - - with mock.patch.object( - models, - "_collect_resolution_candidates", - wraps=models._collect_resolution_candidates, - ) as collect: - response = await models.api_resolve_hash_batch(request) - - result = json.loads(response.text) - by_key = {entry["key"]: entry["result"] for entry in result["results"]} - self.assertEqual(collect.call_count, 1) - self.assertTrue(by_key["checkpoint"]["matched_by_hash"]) - self.assertTrue(by_key["checkpoint"]["matched_by_size"]) - self.assertTrue(by_key["stale"]["stale_hash"]) - self.assertEqual(by_key["stale"]["filename"], "a/shared.safetensors") - - async def test_batch_resolution_rejects_invalid_model_type(self): - response = await models.api_resolve_hash_batch(JsonRequest({"items": [{ - "key": "bad", - "hash": self.checkpoint_hash, - "type": "not-a-real-category", - }]})) - self.assertEqual(response.status, 400) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_graph_splice.mjs b/tests/test_graph_splice.mjs deleted file mode 100644 index 8d5bcbe..0000000 --- a/tests/test_graph_splice.mjs +++ /dev/null @@ -1,193 +0,0 @@ -import assert from 'node:assert/strict'; -import { readFile } from 'node:fs/promises'; - -const graphSpliceSource = await readFile(new URL('../web/modules/graph_splice.js', import.meta.url), 'utf8'); -const graphSpliceModule = await import(`data:text/javascript;base64,${Buffer.from(graphSpliceSource).toString('base64')}`); -const { - analyzeModelChainInsertion, - getModelChainInsertionCapabilities, - spliceModelChainNode, -} = graphSpliceModule; - -class MockNode { - constructor(id, { inputs = [], outputs = [], x = 0 } = {}) { - this.id = id; - this.inputs = inputs.map(type => ({ type, link: null })); - this.outputs = outputs.map(type => ({ type, links: [] })); - this.pos = [x, 0]; - this.size = [220, 120]; - this.graph = null; - } - - connect(originSlot, targetNode, targetSlot) { - return this.graph.connect(this, originSlot, targetNode, targetSlot); - } -} - -class MockGraph { - constructor(nodes = []) { - this._nodes = []; - this.links = {}; - this.nextNodeId = 100; - this.nextLinkId = 1; - this.beforeCount = 0; - this.afterCount = 0; - this.changed = 0; - nodes.forEach(node => this.add(node)); - } - - add(node) { - if (node.id === null || node.id === undefined) node.id = this.nextNodeId++; - if (!this._nodes.includes(node)) this._nodes.push(node); - node.graph = this; - } - - remove(node) { - for (const input of node.inputs || []) { - if (input.link !== null) this.removeLink(input.link); - } - for (const output of node.outputs || []) { - for (const linkId of [...(output.links || [])]) this.removeLink(linkId); - } - this._nodes = this._nodes.filter(item => item !== node); - node.graph = null; - } - - getNodeById(id) { - return this._nodes.find(node => node.id === id); - } - - removeLink(linkId) { - const link = this.links[linkId]; - if (!link) return; - const origin = this.getNodeById(link.origin_id); - const target = this.getNodeById(link.target_id); - if (origin?.outputs?.[link.origin_slot]) { - origin.outputs[link.origin_slot].links = origin.outputs[link.origin_slot].links.filter(id => id !== linkId); - } - if (target?.inputs?.[link.target_slot]?.link === linkId) { - target.inputs[link.target_slot].link = null; - } - delete this.links[linkId]; - } - - connect(origin, originSlot, target, targetSlot) { - if (!origin.outputs[originSlot] || !target.inputs[targetSlot]) return null; - const type = origin.outputs[originSlot].type; - if (type !== target.inputs[targetSlot].type) return null; - const previous = target.inputs[targetSlot].link; - if (previous !== null) this.removeLink(previous); - const id = this.nextLinkId++; - const link = { - id, - type, - origin_id: origin.id, - origin_slot: originSlot, - target_id: target.id, - target_slot: targetSlot, - }; - this.links[id] = link; - origin.outputs[originSlot].links.push(id); - target.inputs[targetSlot].link = id; - return link; - } - - beforeChange() { this.beforeCount += 1; } - afterChange() { this.afterCount += 1; } - change() { this.changed += 1; } - setDirtyCanvas() {} -} - -function chainNode(id, x = 0) { - return new MockNode(id, { inputs: ['MODEL', 'CLIP'], outputs: ['MODEL', 'CLIP'], x }); -} - -function loaderNode(id, x = 0) { - return new MockNode(id, { outputs: ['MODEL', 'CLIP'], x }); -} - -function connectBoth(graph, origin, target) { - origin.connect(0, target, 0); - origin.connect(1, target, 1); -} - -function assertDirectLink(graph, origin, originSlot, target, targetSlot) { - const linkId = target.inputs[targetSlot].link; - const link = graph.links[linkId]; - assert.ok(link); - assert.equal(link.origin_id, origin.id); - assert.equal(link.origin_slot, originSlot); -} - -{ - const source = loaderNode(1, 0); - const anchor = chainNode(2, 500); - const graph = new MockGraph([source, anchor]); - connectBoth(graph, source, anchor); - const inserted = chainNode(null); - - assert.equal(getModelChainInsertionCapabilities(graph, anchor).before.supported, true); - spliceModelChainNode({ graph, anchorNode: anchor, insertedNode: inserted, direction: 'before' }); - - assertDirectLink(graph, source, 0, inserted, 0); - assertDirectLink(graph, source, 1, inserted, 1); - assertDirectLink(graph, inserted, 0, anchor, 0); - assertDirectLink(graph, inserted, 1, anchor, 1); - assert.equal(graph.beforeCount, 1); - assert.equal(graph.afterCount, 1); -} - -{ - const anchor = loaderNode(1, 0); - const downstream = chainNode(2, 500); - const graph = new MockGraph([anchor, downstream]); - connectBoth(graph, anchor, downstream); - const inserted = chainNode(null); - - assert.equal(analyzeModelChainInsertion(graph, anchor, 'after').supported, true); - spliceModelChainNode({ graph, anchorNode: anchor, insertedNode: inserted, direction: 'after' }); - - assertDirectLink(graph, anchor, 0, inserted, 0); - assertDirectLink(graph, anchor, 1, inserted, 1); - assertDirectLink(graph, inserted, 0, downstream, 0); - assertDirectLink(graph, inserted, 1, downstream, 1); -} - -{ - const anchor = loaderNode(1); - const a = chainNode(2); - const b = chainNode(3); - const graph = new MockGraph([anchor, a, b]); - anchor.connect(0, a, 0); - anchor.connect(0, b, 0); - anchor.connect(1, a, 1); - - const analysis = analyzeModelChainInsertion(graph, anchor, 'after'); - assert.equal(analysis.supported, false); - assert.equal(analysis.code, 'ambiguous_downstream_branches'); -} - -{ - const source = loaderNode(1); - const anchor = chainNode(2); - const graph = new MockGraph([source, anchor]); - connectBoth(graph, source, anchor); - const inserted = chainNode(null); - const originalConnect = inserted.connect.bind(inserted); - inserted.connect = (originSlot, target, targetSlot) => { - if (inserted.outputs[originSlot]?.type === 'CLIP') return null; - return originalConnect(originSlot, target, targetSlot); - }; - - assert.throws( - () => spliceModelChainNode({ graph, anchorNode: anchor, insertedNode: inserted, direction: 'before' }), - error => error.code === 'connection_failed', - ); - assert.equal(graph._nodes.includes(inserted), false); - assertDirectLink(graph, source, 0, anchor, 0); - assertDirectLink(graph, source, 1, anchor, 1); - assert.equal(graph.beforeCount, 1); - assert.equal(graph.afterCount, 1); -} - -console.log('graph_splice tests passed'); diff --git a/tests/test_model_picker.mjs b/tests/test_model_picker.mjs deleted file mode 100644 index 7f6fb72..0000000 --- a/tests/test_model_picker.mjs +++ /dev/null @@ -1,65 +0,0 @@ -import assert from 'node:assert/strict'; -import { - collectMainModelContextRequests, - formatModelTypeLabel, - getBaseModelFamily, - inferPickerModelType, -} from '../web/modules/model_picker.js'; - -assert.deepEqual( - inferPickerModelType({ type: 'LoraLoader' }, { name: 'lora_name' }), - { label: 'LoRA', folderTypes: ['loras'], isLora: true }, -); -assert.deepEqual( - inferPickerModelType({ type: 'UNETLoader' }, { name: 'unet_name' }), - { label: 'UNET', folderTypes: ['diffusion_models', 'unet'], isLora: false }, -); -assert.equal(formatModelTypeLabel('checkpoints'), 'Checkpoint'); -assert.equal(formatModelTypeLabel('custom_type', 'Custom'), 'Custom'); - -assert.equal(getBaseModelFamily('SDXL 1.0'), 'sdxl'); -assert.equal(getBaseModelFamily('Stable Diffusion 1.5'), 'sd15'); -assert.equal(getBaseModelFamily('Pony Diffusion V6 XL'), 'pony'); -assert.equal(getBaseModelFamily('Illustrious XL v1.1'), 'illustrious'); -assert.equal(getBaseModelFamily('FLUX.1 D'), 'flux'); - -const checkpoint = { - id: 1, - type: 'CheckpointLoaderSimple', - inputs: [], - widgets: [{ name: 'ckpt_name', value: 'SDXL/main.safetensors' }], -}; -const lora = { - id: 2, - type: 'LoraLoader', - inputs: [ - { type: 'MODEL', link: 10 }, - { type: 'CLIP', link: 11 }, - ], - widgets: [{ name: 'lora_name', value: 'styles/detail.safetensors' }], -}; -const sampler = { - id: 3, - type: 'KSampler', - inputs: [{ type: 'MODEL', link: 12 }], - widgets: [], -}; -const graph = { - links: { - 10: { origin_id: 1 }, - 11: { origin_id: 1 }, - 12: { origin_id: 2 }, - }, - _nodes: [checkpoint, lora, sampler], -}; - -assert.deepEqual(collectMainModelContextRequests(graph, lora), [{ - path: 'SDXL/main.safetensors', - folder_types: ['checkpoints'], -}]); -assert.deepEqual(collectMainModelContextRequests(graph, sampler), [{ - path: 'SDXL/main.safetensors', - folder_types: ['checkpoints'], -}]); - -console.log('model_picker tests passed');