feat(recipes): reconnect suggestions, undo, and base-model family tolerance

Enhance the deleted-LoRA reconnect flow in the recipe modal:

- Suggest local reconnect candidates when the panel opens, ranked by
  identity (same hash / same CivitAI version) then filename/name
  similarity, with a hard filter on confident base-model mismatches;
  the input gets a Combobox backed by the same endpoint as you type.
- Snapshot the pre-reconnect entry and offer a permanent restore:
  reconnected entries show an undo icon at the right end of the info
  row, with the original filename in the tooltip.
- Relax the manual reconnect base-model guard to a three-tier check:
  exact/unknown labels pass silently, same-architecture families
  (e.g. Pony <-> Illustrious) pass with a warning toast, and only
  cross-architecture mismatches stay hard-rejected.
This commit is contained in:
Will Miao
2026-08-30 08:17:35 +08:00
parent 6e31da7a70
commit 838a374a56
23 changed files with 1893 additions and 45 deletions
@@ -539,4 +539,241 @@ describe('RecipeModal resource item interactions', () => {
checkpointItem.click();
expect(navigateSpy).not.toHaveBeenCalled();
});
describe('reconnect suggestions', () => {
const suggestionsPayload = {
success: true,
suggestions: [
{
file_name: 'deleted-lora-v1.safetensors',
file_path: '/models/loras/deleted-lora-v1.safetensors',
model_name: 'Deleted LoRA v1',
base_model: 'SD 1.5',
preview_url: '/preview/deleted.png',
hash: 'abc123',
score: 0.95,
match_reason: 'same_version',
target_name: 'deleted-lora-v1',
},
],
};
function mockSuggestionsFetch(payload) {
const requests = [];
global.fetch = vi.fn(async (url, options) => {
requests.push({ url: String(url), options });
if (String(url).includes('/reconnect-suggestions')) {
return { ok: true, json: async () => payload };
}
if (String(url).includes('/recipe/lora/reconnect')) {
return {
ok: true,
json: async () => ({
success: true,
updated_lora: { name: 'deleted-lora-v1', modelName: 'Deleted LoRA v1', inLibrary: true },
}),
};
}
return { ok: true, json: async () => ({}) };
});
return requests;
}
async function openReconnectPanel(recipeModal, loraIndex) {
recipeModal.showRecipeDetails(recipeWithResources);
await flushWiring();
const item = document.querySelector(`[data-lora-index="${loraIndex}"]`);
item.querySelector('.lora-reconnect').click();
return item.querySelector('.lora-reconnect-container');
}
it('fetches suggestions when the panel opens and renders them as rows', async () => {
const recipeModal = await createRecipeModal();
mockSuggestionsFetch(suggestionsPayload);
const container = await openReconnectPanel(recipeModal, 2);
// The loading state shows synchronously while the fetch is in flight
expect(container.querySelector('.reconnect-suggestions-loading')).not.toBeNull();
await vi.waitFor(() => {
expect(container.querySelectorAll('.reconnect-suggestion').length).toBe(1);
});
expect(global.fetch).toHaveBeenCalledWith(
'/api/lm/recipe/recipe-resources/lora/2/reconnect-suggestions'
);
const row = container.querySelector('.reconnect-suggestion');
// Primary label is the file stem (what the match scored on and what
// gets submitted); the secondary line shows only the base model — the
// model name is noise and intentionally omitted.
expect(row.querySelector('.reconnect-suggestion-name').textContent).toBe('deleted-lora-v1');
expect(row.querySelector('.reconnect-suggestion-secondary').textContent).toBe('SD 1.5');
expect(row.querySelector('.reconnect-suggestion-reason').textContent).toBe('Same model version');
expect(row.title).toBe('deleted-lora-v1');
const preview = row.querySelector('.reconnect-suggestion-preview');
expect(preview.getAttribute('src')).toBe('/preview/deleted.png');
});
it('reconnects with the suggestion target_name when a row is clicked', async () => {
const recipeModal = await createRecipeModal();
const requests = mockSuggestionsFetch(suggestionsPayload);
const container = await openReconnectPanel(recipeModal, 2);
await vi.waitFor(() => {
expect(container.querySelectorAll('.reconnect-suggestion').length).toBe(1);
});
container.querySelector('.reconnect-suggestion').click();
await vi.waitFor(() => {
expect(requests.some(r => r.url === '/api/lm/recipe/lora/reconnect')).toBe(true);
});
const reconnectRequest = requests.find(r => r.url === '/api/lm/recipe/lora/reconnect');
expect(reconnectRequest.options.method).toBe('POST');
// lora_index rides as the DOM attribute string, same as the manual form
expect(JSON.parse(reconnectRequest.options.body)).toEqual({
recipe_id: 'recipe-resources',
lora_index: '2',
target_name: 'deleted-lora-v1',
});
});
it('warns when the reconnect crossed base-model families', async () => {
const recipeModal = await createRecipeModal();
global.fetch = vi.fn(async (url) => {
if (String(url).includes('/reconnect-suggestions')) {
return { ok: true, json: async () => suggestionsPayload };
}
if (String(url).includes('/recipe/lora/reconnect')) {
return {
ok: true,
json: async () => ({
success: true,
updated_lora: { name: 'deleted-lora-v1', modelName: 'Deleted LoRA v1', inLibrary: true },
base_model_mismatch: { recipe_base_model: 'Illustrious', lora_base_model: 'Pony' },
}),
};
}
return { ok: true, json: async () => ({}) };
});
const container = await openReconnectPanel(recipeModal, 2);
await vi.waitFor(() => {
expect(container.querySelectorAll('.reconnect-suggestion').length).toBe(1);
});
container.querySelector('.reconnect-suggestion').click();
await vi.waitFor(() => {
expect(showToastMock).toHaveBeenCalledWith(
'toast.recipes.reconnectBaseModelMismatch',
{ recipe: 'Illustrious', lora: 'Pony' },
'warning'
);
});
});
it('shows an empty state when no suggestions are available', async () => {
const recipeModal = await createRecipeModal();
mockSuggestionsFetch({ success: true, suggestions: [] });
const container = await openReconnectPanel(recipeModal, 3);
await vi.waitFor(() => {
expect(container.querySelector('.reconnect-suggestions-empty')).not.toBeNull();
});
expect(container.querySelector('.reconnect-suggestions-empty').textContent)
.toBe('No matching LoRAs in your local library');
expect(container.querySelectorAll('.reconnect-suggestion').length).toBe(0);
});
it('submits free text via the combobox onCommit when Enter is pressed', async () => {
const recipeModal = await createRecipeModal();
const requests = mockSuggestionsFetch({ success: true, suggestions: [] });
const container = await openReconnectPanel(recipeModal, 2);
const input = container.querySelector('.reconnect-input');
input.value = 'typed-lora-name';
input.dispatchEvent(new KeyboardEvent('keydown', { key: 'Enter', bubbles: true }));
await vi.waitFor(() => {
expect(requests.some(r => r.url === '/api/lm/recipe/lora/reconnect')).toBe(true);
});
const reconnectRequest = requests.find(r => r.url === '/api/lm/recipe/lora/reconnect');
expect(JSON.parse(reconnectRequest.options.body)).toEqual({
recipe_id: 'recipe-resources',
lora_index: '2',
target_name: 'typed-lora-name',
});
});
it('keeps the panel open when the combobox dropdown is clicked', async () => {
const recipeModal = await createRecipeModal();
mockSuggestionsFetch({ success: true, suggestions: [] });
recipeModal.showRecipeDetails(recipeWithResources);
await flushWiring();
// Open the panel directly — button wiring races the hydration re-render,
// and this test is about the document click handler, not the button.
recipeModal.showReconnectInput('2');
const container = document.querySelector('.lora-reconnect-container[data-lora-index="2"]');
expect(container.classList.contains('active')).toBe(true);
// The dropdown panel lives on document.body; clicking an option there is
// part of the reconnect interaction, not an outside click.
const panel = document.createElement('div');
panel.className = 'lm-combobox-panel';
document.body.appendChild(panel);
panel.dispatchEvent(new MouseEvent('click', { bubbles: true }));
expect(container.classList.contains('active')).toBe(true);
panel.remove();
// A genuine outside click still closes the panel
document.body.dispatchEvent(new MouseEvent('click', { bubbles: true }));
expect(container.classList.contains('active')).toBe(false);
});
});
it('offers undo for reconnected entries and restores via the API', async () => {
const recipeModal = await createRecipeModal();
const isolatedRecipe = JSON.parse(JSON.stringify(recipeWithResources));
isolatedRecipe.loras[0].reconnectSnapshot = { file_name: 'gone', isDeleted: true };
fetchRecipeDetailsMock.mockResolvedValue(isolatedRecipe);
const requests = [];
global.fetch = vi.fn(async (url, options) => {
requests.push({ url: String(url), options });
if (String(url).includes('/recipe/lora/restore')) {
return {
ok: true,
json: async () => ({
success: true,
updated_lora: { name: 'gone', modelName: 'Gone', inLibrary: false, isDeleted: true },
}),
};
}
return { ok: true, json: async () => ({}) };
});
recipeModal.showRecipeDetails(isolatedRecipe);
await flushWiring();
const item = document.querySelector('[data-lora-index="0"]');
const undoButton = item.querySelector('.lora-undo-reconnect');
expect(undoButton).not.toBeNull();
undoButton.click();
// Wait for the whole restore chain (fetch -> json -> toast), not just the
// request itself.
await vi.waitFor(() => {
expect(showToastMock).toHaveBeenCalledWith('toast.recipes.loraRestored', {}, 'success');
});
const restoreRequest = requests.find(r => r.url === '/api/lm/recipe/lora/restore');
expect(restoreRequest.options.method).toBe('POST');
expect(JSON.parse(restoreRequest.options.body)).toEqual({
recipe_id: 'recipe-resources',
lora_index: '0',
});
});
});
+394
View File
@@ -16,6 +16,7 @@ from py.services.recipe_scanner import RecipeScanner
from py.services import settings_manager as settings_manager_module
from py.utils.models import BaseModelMetadata
from py.utils.utils import calculate_recipe_fingerprint
from py.services.recipes.errors import RecipeValidationError
async def _wait_for_resort(scanner: RecipeScanner) -> None:
@@ -164,6 +165,285 @@ async def test_local_lora_lookup_requires_unambiguous_name_and_matching_base_mod
assert await scanner.get_local_lora_by_hash("b" * 64) is models[1]
def _suggestion_item(**overrides):
item = {
"sha256": "ab" * 32,
"file_name": "style.safetensors",
"file_path": "/models/loras/style.safetensors",
"folder": "",
"model_name": "Style LoRA",
"base_model": "SD 1.5",
"preview_url": "/preview/style.png",
}
item.update(overrides)
return item
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_same_hash_ranks_first(recipe_scanner):
scanner, stub = recipe_scanner
stub.cache_version = 1
same_hash = _suggestion_item(
file_name="zzz-unrelated.safetensors",
file_path="/models/loras/zzz-unrelated.safetensors",
model_name="Unrelated",
)
similar = _suggestion_item(
sha256="cd" * 32,
file_name="anime-style-v2.safetensors",
file_path="/models/loras/anime-style-v2.safetensors",
model_name="Anime Style",
)
stub._cache.raw_data = [same_hash, similar]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"hash": "ab" * 32, "file_name": "anime-style-v2.safetensors"},
recipe_base_model="SD 1.5",
)
assert suggestions[0]["match_reason"] == "same_hash"
assert suggestions[0]["file_path"] == same_hash["file_path"]
assert suggestions[0]["score"] >= 1.0
assert any(s["match_reason"] == "similar_filename" for s in suggestions[1:])
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_same_version(recipe_scanner):
scanner, stub = recipe_scanner
item = _suggestion_item()
stub._cache.raw_data = [item]
stub._cache.version_index[456] = item
suggestions = await scanner.suggest_reconnect_candidates(
entry={"modelVersionId": 456},
recipe_base_model="SD 1.5",
)
assert len(suggestions) == 1
assert suggestions[0]["match_reason"] == "same_version"
assert suggestions[0]["score"] >= 0.95
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_base_model_mismatch_excluded(recipe_scanner):
scanner, stub = recipe_scanner
matching = _suggestion_item(
file_name="anime-style.safetensors",
file_path="/models/loras/anime-style.safetensors",
model_name="Anime Style",
base_model="SD 1.5",
)
mismatched = _suggestion_item(
sha256="cd" * 32,
file_name="anime-style.safetensors",
file_path="/models/loras/sdxl/anime-style.safetensors",
folder="sdxl",
model_name="Anime Style",
base_model="SDXL 1.0",
)
stub._cache.raw_data = [matching, mismatched]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "anime-style.safetensors"},
recipe_base_model="SD 1.5",
)
# A confident base-model mismatch is a hard rejection — reconnect itself
# enforces that rule, so suggesting the mismatch would guarantee failure.
assert [s["file_path"] for s in suggestions] == [matching["file_path"]]
assert suggestions[0]["target_name"] == "anime-style"
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_base_model_unknown_stays_eligible(recipe_scanner):
scanner, stub = recipe_scanner
unknown_item = _suggestion_item(
file_name="anime-style.safetensors",
file_path="/models/loras/anime-style.safetensors",
model_name="Anime Style",
base_model="",
)
stub._cache.raw_data = [unknown_item]
# Unknown base model on the item side must not be rejected — reconnect
# accepts it too (find_matching_models lenient guard).
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "anime-style.safetensors"},
recipe_base_model="SD 1.5",
)
assert [s["file_path"] for s in suggestions] == [unknown_item["file_path"]]
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_same_hash_mismatched_base_model_excluded(
recipe_scanner,
):
scanner, stub = recipe_scanner
stub.cache_version = 1
mismatched = _suggestion_item(
file_name="zzz-unrelated.safetensors",
file_path="/models/loras/zzz-unrelated.safetensors",
model_name="Unrelated",
base_model="SDXL 1.0",
)
stub._cache.raw_data = [mismatched]
# Even the strongest identity signal (same hash) must not surface a
# candidate that reconnect would reject on base-model grounds.
suggestions = await scanner.suggest_reconnect_candidates(
entry={"hash": "ab" * 32, "file_name": "other.safetensors"},
recipe_base_model="SD 1.5",
)
assert suggestions == []
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_basename_collision_uses_folder_path(recipe_scanner):
scanner, stub = recipe_scanner
first = _suggestion_item(
file_name="anime-style.safetensors",
file_path="/models/loras/anime-style.safetensors",
model_name="Anime Style",
base_model="SD 1.5",
)
second = _suggestion_item(
sha256="cd" * 32,
file_name="anime-style.safetensors",
file_path="/models/loras/sd15/anime-style.safetensors",
folder="sd15",
model_name="Anime Style v2",
base_model="SD 1.5",
)
stub._cache.raw_data = [first, second]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "anime-style.safetensors"},
recipe_base_model="SD 1.5",
)
# Duplicate basenames disambiguate target_name with the folder path.
assert {s["target_name"] for s in suggestions} == {"anime-style", "sd15/anime-style"}
scanner, stub = recipe_scanner
checkpoint = _suggestion_item(sub_type="checkpoint")
lora = _suggestion_item(
sha256="cd" * 32,
file_path="/models/loras/other/style.safetensors",
folder="other",
)
stub._cache.raw_data = [checkpoint, lora]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "style.safetensors"},
recipe_base_model=None,
)
assert all(s["file_path"] != checkpoint["file_path"] for s in suggestions)
assert any(s["file_path"] == lora["file_path"] for s in suggestions)
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_respects_limit(recipe_scanner):
scanner, stub = recipe_scanner
stub._cache.raw_data = [
_suggestion_item(
sha256=f"{i:064x}",
file_name=f"anime-style-{i}.safetensors",
file_path=f"/models/loras/anime-style-{i}.safetensors",
model_name=f"Anime Style {i}",
)
for i in range(10)
]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "anime-style.safetensors"},
recipe_base_model="SD 1.5",
limit=3,
)
assert len(suggestions) == 3
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_query_substring(recipe_scanner):
scanner, stub = recipe_scanner
item = _suggestion_item(
file_name="anime-style.safetensors",
file_path="/models/loras/anime-style.safetensors",
model_name="Anime Style",
)
stub._cache.raw_data = [item]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "unrelated.safetensors"},
recipe_base_model="SD 1.5",
query="anime",
)
assert len(suggestions) == 1
assert suggestions[0]["match_reason"] == "similar_filename"
# Substring hits floor the ratio at 0.8: 0.5 + 0.4 * 0.8 + 0.1 base boost.
assert suggestions[0]["score"] == 0.92
assert suggestions[0]["target_name"] == "anime-style"
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_skips_items_without_hash(recipe_scanner):
scanner, stub = recipe_scanner
no_hash = _suggestion_item(sha256="")
stub._cache.raw_data = [no_hash]
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "style.safetensors"},
recipe_base_model="SD 1.5",
)
assert suggestions == []
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_short_query_no_substring_floor(recipe_scanner):
scanner, stub = recipe_scanner
item = _suggestion_item(
file_name="anime-style.safetensors",
file_path="/models/loras/anime-style.safetensors",
model_name="Anime Style",
)
stub._cache.raw_data = [item]
# A 1-2 character query is a substring of nearly everything; it must NOT
# floor the ratio, otherwise every library item surfaces as a suggestion.
suggestions = await scanner.suggest_reconnect_candidates(
entry={"file_name": "unrelated.safetensors"},
recipe_base_model="SD 1.5",
query="a",
)
assert suggestions == []
@pytest.mark.asyncio
async def test_suggest_reconnect_candidates_name_threshold_filters_generic_overlap(recipe_scanner):
scanner, stub = recipe_scanner
item = _suggestion_item(
file_name="not-artists-styles-pony.safetensors",
file_path="/models/loras/not-artists-styles-pony.safetensors",
model_name="Not Artists Styles for Pony Diffusion V6 XL",
)
stub._cache.raw_data = [item]
# Long names sharing generic tokens ("style", "pony", "diffusion") score
# ~0.638 — below the name-similarity threshold, so unrelated models stay
# out of the suggestions.
suggestions = await scanner.suggest_reconnect_candidates(
entry={"modelName": "Concept Art Twilight Style SDXL_LoRA_Pony Diffusion"},
recipe_base_model="Pony",
)
assert suggestions == []
def test_recipes_dir_uses_custom_settings_path(tmp_path: Path, monkeypatch):
RecipeScanner._instance = None
settings_manager_module.reset_settings_manager()
@@ -331,6 +611,120 @@ async def test_update_lora_entry_updates_cache_and_file(tmp_path: Path, recipe_s
assert cached_recipe["fingerprint"] == expected_fingerprint
async def test_update_lora_entry_snapshots_previous_state(tmp_path: Path, recipe_scanner):
scanner, stub = recipe_scanner
recipes_dir = Path(config.loras_roots[0]) / "recipes"
recipes_dir.mkdir(parents=True, exist_ok=True)
recipe_id = "recipe-snapshot"
recipe_path = recipes_dir / f"{recipe_id}.recipe.json"
original_entry = {
"file_name": "old",
"strength": 1.0,
"hash": "",
"isDeleted": True,
"exclude": True,
}
recipe_data = {
"id": recipe_id,
"file_path": str(tmp_path / "image.png"),
"title": "Original",
"modified": 0.0,
"created_date": 0.0,
"loras": [dict(original_entry)],
}
recipe_path.write_text(json.dumps(recipe_data))
await scanner.add_recipe(dict(recipe_data))
target_info = {
"sha256": "abc123",
"file_path": str(tmp_path / "loras" / "target.safetensors"),
"preview_url": "preview.png",
"civitai": {"id": 42, "name": "v1", "model": {"name": "Target"}},
}
stub.register_model("target", target_info)
await scanner.update_lora_entry(
recipe_id, 0, target_name="target", target_lora=target_info
)
with recipe_path.open("r", encoding="utf-8") as file_obj:
persisted = json.load(file_obj)
snapshot = persisted["loras"][0]["reconnectSnapshot"]
assert snapshot == original_entry
# Snapshots never nest
assert "reconnectSnapshot" not in snapshot
async def test_restore_lora_entry_round_trip(tmp_path: Path, recipe_scanner):
scanner, stub = recipe_scanner
recipes_dir = Path(config.loras_roots[0]) / "recipes"
recipes_dir.mkdir(parents=True, exist_ok=True)
recipe_id = "recipe-restore"
recipe_path = recipes_dir / f"{recipe_id}.recipe.json"
original_entry = {
"file_name": "old",
"strength": 1.0,
"hash": "",
"isDeleted": True,
"exclude": True,
}
recipe_data = {
"id": recipe_id,
"file_path": str(tmp_path / "image.png"),
"title": "Original",
"modified": 0.0,
"created_date": 0.0,
"loras": [dict(original_entry)],
}
recipe_path.write_text(json.dumps(recipe_data))
await scanner.add_recipe(dict(recipe_data))
target_info = {
"sha256": "abc123",
"file_path": str(tmp_path / "loras" / "target.safetensors"),
"preview_url": "preview.png",
"civitai": {"id": 42, "name": "v1", "model": {"name": "Target"}},
}
stub.register_model("target", target_info)
await scanner.update_lora_entry(
recipe_id, 0, target_name="target", target_lora=target_info
)
restored_recipe, restored_lora = await scanner.restore_lora_entry(recipe_id, 0)
entry = restored_recipe["loras"][0]
assert entry == original_entry
assert "reconnectSnapshot" not in entry
assert restored_lora["isDeleted"] is True
assert restored_lora["inLibrary"] is False
assert restored_recipe["fingerprint"] == calculate_recipe_fingerprint([original_entry])
with recipe_path.open("r", encoding="utf-8") as file_obj:
persisted = json.load(file_obj)
assert persisted["loras"][0] == original_entry
assert persisted["fingerprint"] == restored_recipe["fingerprint"]
async def test_restore_lora_entry_without_snapshot_rejected(tmp_path: Path, recipe_scanner):
scanner, _ = recipe_scanner
recipes_dir = Path(config.loras_roots[0]) / "recipes"
recipes_dir.mkdir(parents=True, exist_ok=True)
recipe_id = "recipe-no-snapshot"
recipe_path = recipes_dir / f"{recipe_id}.recipe.json"
recipe_path.write_text(
json.dumps({"id": recipe_id, "loras": [{"file_name": "plain"}]})
)
with pytest.raises(RecipeValidationError):
await scanner.restore_lora_entry(recipe_id, 0)
async def test_set_lora_entry_hash_invalid_persists_flag(tmp_path: Path, recipe_scanner):
scanner, _ = recipe_scanner
recipes_dir = Path(config.loras_roots[0]) / "recipes"
+183
View File
@@ -1315,6 +1315,189 @@ async def test_reconnect_lora_distinguishes_ambiguous_mismatched_and_missing(tmp
)
@pytest.mark.asyncio
async def test_reconnect_lora_family_compatible_succeeds_with_warning(tmp_path):
service = RecipePersistenceService(
exif_utils=DummyExifUtils(),
card_preview_width=512,
logger=logging.getLogger("test"),
)
pony_item = {
"file_name": "style.safetensors",
"folder": "",
"file_path": "/models/loras/style.safetensors",
"base_model": "Pony",
"sha256": "ab" * 32,
}
recipe_path = tmp_path / "recipe.json"
recipe_path.write_text(
json.dumps({"id": "r1", "base_model": "Illustrious", "loras": [{}]})
)
class DummyScanner:
async def get_recipe_json_path(self, recipe_id):
return str(recipe_path)
async def find_local_loras_by_name(self, name, base_model=None):
return [pony_item]
async def update_lora_entry(self, recipe_id, lora_index, *, target_name, target_lora):
assert target_lora is pony_item
return ({"id": "r1"}, {"file_name": target_lora["file_name"]})
async def find_recipes_by_fingerprint(self, fingerprint):
return []
result = await service.reconnect_lora(
recipe_scanner=DummyScanner(), recipe_id="r1", lora_index=0, target_name="style"
)
assert result.payload["success"] is True
assert result.payload["base_model_mismatch"] == {
"recipe_base_model": "Illustrious",
"lora_base_model": "Pony",
}
@pytest.mark.asyncio
async def test_reconnect_lora_exact_base_model_has_no_warning(tmp_path):
service = RecipePersistenceService(
exif_utils=DummyExifUtils(),
card_preview_width=512,
logger=logging.getLogger("test"),
)
item = {
"file_name": "style.safetensors",
"folder": "",
"file_path": "/models/loras/style.safetensors",
"base_model": "SDXL 1.0",
"sha256": "ab" * 32,
}
recipe_path = tmp_path / "recipe.json"
recipe_path.write_text(
json.dumps({"id": "r1", "base_model": "SDXL 1.0", "loras": [{}]})
)
class DummyScanner:
async def get_recipe_json_path(self, recipe_id):
return str(recipe_path)
async def find_local_loras_by_name(self, name, base_model=None):
return [item]
async def update_lora_entry(self, recipe_id, lora_index, *, target_name, target_lora):
return ({"id": "r1"}, {"file_name": target_lora["file_name"]})
async def find_recipes_by_fingerprint(self, fingerprint):
return []
result = await service.reconnect_lora(
recipe_scanner=DummyScanner(), recipe_id="r1", lora_index=0, target_name="style"
)
assert result.payload["success"] is True
assert "base_model_mismatch" not in result.payload
@pytest.mark.asyncio
async def test_get_reconnect_suggestions_loads_entry_and_delegates(tmp_path):
service = RecipePersistenceService(
exif_utils=DummyExifUtils(),
card_preview_width=512,
logger=logging.getLogger("test"),
)
recipe_path = tmp_path / "recipe.json"
recipe_path.write_text(
json.dumps(
{
"id": "r1",
"base_model": "SD 1.5",
"loras": [
{"file_name": "a.safetensors", "hash": "aaa"},
{"file_name": "b.safetensors", "hash": "bbb", "isDeleted": True},
],
}
)
)
class DummyScanner:
def __init__(self):
self.calls = []
async def get_recipe_json_path(self, recipe_id):
assert recipe_id == "r1"
return str(recipe_path)
async def suggest_reconnect_candidates(
self, *, entry, recipe_base_model, query=None, limit=5
):
self.calls.append(
{
"entry": entry,
"recipe_base_model": recipe_base_model,
"query": query,
}
)
return [
{
"file_name": "b.safetensors",
"score": 1.0,
"match_reason": "same_hash",
"target_name": "b",
}
]
scanner = DummyScanner()
result = await service.get_reconnect_suggestions(
recipe_scanner=scanner, recipe_id="r1", lora_index=1, query="b"
)
assert result.payload["success"] is True
assert result.payload["suggestions"][0]["target_name"] == "b"
assert scanner.calls == [
{
"entry": {"file_name": "b.safetensors", "hash": "bbb", "isDeleted": True},
"recipe_base_model": "SD 1.5",
"query": "b",
}
]
@pytest.mark.asyncio
async def test_get_reconnect_suggestions_validates_recipe_and_index(tmp_path):
service = RecipePersistenceService(
exif_utils=DummyExifUtils(),
card_preview_width=512,
logger=logging.getLogger("test"),
)
class MissingScanner:
async def get_recipe_json_path(self, recipe_id):
return str(tmp_path / "missing.json")
with pytest.raises(RecipeNotFoundError):
await service.get_reconnect_suggestions(
recipe_scanner=MissingScanner(), recipe_id="nope", lora_index=0
)
recipe_path = tmp_path / "recipe.json"
recipe_path.write_text(json.dumps({"id": "r1", "loras": []}))
class EmptyScanner:
async def get_recipe_json_path(self, recipe_id):
return str(recipe_path)
with pytest.raises(RecipeValidationError, match="lora_index"):
await service.get_reconnect_suggestions(
recipe_scanner=EmptyScanner(), recipe_id="r1", lora_index=0
)
@pytest.mark.asyncio
async def test_mark_lora_hash_invalid_delegates_and_reports(tmp_path):
service = RecipePersistenceService(
+53
View File
@@ -0,0 +1,53 @@
"""Unit tests for base-model architecture-family relations."""
from py.utils.base_model import (
RELATION_COMPATIBLE,
RELATION_INCOMPATIBLE,
RELATION_SAME,
RELATION_UNKNOWN,
base_model_relation,
)
def test_identical_labels_are_same():
assert base_model_relation("SDXL 1.0", "sdxl 1.0") == RELATION_SAME
assert base_model_relation(" Pony ", "pony") == RELATION_SAME
def test_sdxl_lineage_is_compatible():
assert base_model_relation("Pony", "Illustrious") == RELATION_COMPATIBLE
assert base_model_relation("Illustrious", "SDXL 1.0") == RELATION_COMPATIBLE
assert base_model_relation("NoobAI", "SDXL Lightning") == RELATION_COMPATIBLE
def test_sd1_lineage_is_compatible():
assert base_model_relation("SD 1.5", "SD 1.4") == RELATION_COMPATIBLE
assert base_model_relation("SD 1.5 LCM", "SD 1.5") == RELATION_COMPATIBLE
def test_flux1_lineage_is_compatible():
assert base_model_relation("Flux.1 D", "Flux.1 S") == RELATION_COMPATIBLE
def test_cross_architecture_is_incompatible():
assert base_model_relation("SD 1.5", "SDXL 1.0") == RELATION_INCOMPATIBLE
assert base_model_relation("Pony", "Flux.1 D") == RELATION_INCOMPATIBLE
def test_pony_v7_is_not_sdxl_compatible():
# Pony V7 is AuraFlow-based; sharing a name prefix with Pony means nothing.
assert base_model_relation("Pony", "Pony V7") == RELATION_INCOMPATIBLE
def test_unknown_labels_stay_unknown():
assert base_model_relation("", "SDXL 1.0") == RELATION_UNKNOWN
assert base_model_relation("SDXL 1.0", "unknown") == RELATION_UNKNOWN
assert base_model_relation(None, None) == RELATION_UNKNOWN
def test_unlisted_labels_fall_back_to_strict():
# A label missing from the family table only matches itself exactly —
# unknown new CivitAI labels must never be wrongly waved through.
assert base_model_relation("Wan Video", "Wan Video") == RELATION_SAME
assert base_model_relation("Wan Video", "Hunyuan Video") == RELATION_INCOMPATIBLE
assert base_model_relation("Wan Video", "Pony") == RELATION_INCOMPATIBLE