fix(lora): review findings for the LoraManager layout (#355)
- The example-images library subfolder goes through the folder guard like its root, so its canonical (normcased) form matches the media check on Windows for a library name with capitals. - A failed trigger-word load is retried by a running node only after TRIGGER_RETRY_SECONDS, not on every prompt run. - A review whose discovery cannot list a folder says so (folder-unreadable, the folder's name only) instead of a generic error.
This commit is contained in:
@@ -441,6 +441,19 @@ class LoraImportRouteTests(LoraImportAPITestCase):
|
||||
[os.path.basename(i["image_path"]) for i in images], ["image_1.png"]
|
||||
)
|
||||
|
||||
async def test_an_unlistable_folder_says_so_in_the_review(self):
|
||||
self._two_loras()
|
||||
error = PermissionError(13, "Permission denied", str(self.loras / "lost+found"))
|
||||
with patch.object(lora_import, "discover_sidecars", side_effect=error):
|
||||
resp, body = await self._post(PREVIEW)
|
||||
self.assertEqual((resp.status, body["code"]), (500, "folder-unreadable"))
|
||||
self.assertEqual(
|
||||
body["error"],
|
||||
"A LoRA folder could not be listed (Permission denied: lost+found). "
|
||||
"Nothing was changed.",
|
||||
)
|
||||
self.assertNotIn(str(self.tmp_root), json.dumps(body))
|
||||
|
||||
async def test_preview_requires_the_integration(self):
|
||||
IntegrationConfig.LORA_MANAGER_ENABLED = False
|
||||
resp, body = await self._post(PREVIEW)
|
||||
|
||||
@@ -1771,6 +1771,26 @@ class TestTriggerWordCacheLayout(unittest.TestCase):
|
||||
self.assertEqual(cache.get_trigger_words("old"), ["word"])
|
||||
self.assertIn("share gone", "\n".join(logs.output))
|
||||
|
||||
def test_a_failed_load_waits_before_a_node_retries(self):
|
||||
# Review finding: a node calls load while the cache is not loaded; a
|
||||
# failing discovery must not rescan on every prompt run.
|
||||
cache = TriggerWordCache()
|
||||
self.assertTrue(cache.retry_due())
|
||||
clock = [100.0]
|
||||
with (
|
||||
patch.object(lora_utils.time, "monotonic", lambda: clock[0]),
|
||||
patch.object(lora_utils, "discover_sidecars", side_effect=OSError("x")),
|
||||
self.assertLogs("prompt_manager.lora_utils", level="WARNING"),
|
||||
):
|
||||
cache.load(str(self.lm))
|
||||
self.assertFalse(cache.is_loaded)
|
||||
self.assertFalse(cache.retry_due())
|
||||
clock[0] += lora_utils.TRIGGER_RETRY_SECONDS
|
||||
self.assertTrue(cache.retry_due())
|
||||
cache.load(str(self.lm))
|
||||
self.assertTrue(cache.is_loaded)
|
||||
self.assertTrue(cache.retry_due())
|
||||
|
||||
def test_any_other_failure_never_raises(self):
|
||||
cache = TriggerWordCache()
|
||||
with (
|
||||
|
||||
@@ -334,6 +334,24 @@ class TestLoraTriggerWords(NodeTestCase):
|
||||
self.assertEqual(text, "a cat <lora:neon:1>")
|
||||
self.assertIn("share gone", "\n".join(logs.output))
|
||||
|
||||
def test_a_failed_load_is_not_retried_on_every_run(self):
|
||||
from py import lora_utils
|
||||
|
||||
cache = lora_utils.get_trigger_cache()
|
||||
cache.clear()
|
||||
self.addCleanup(cache.clear)
|
||||
with (
|
||||
mock.patch.dict(sys.modules, {"py.config": fake_integration_config(True)}),
|
||||
mock.patch.object(
|
||||
lora_utils, "discover_sidecars", side_effect=OSError("share gone")
|
||||
) as discover,
|
||||
mock.patch.object(lora_utils, "default_lora_roots", return_value=[]),
|
||||
self.assertLogs("prompt_manager.lora_utils", level="WARNING"),
|
||||
):
|
||||
for _ in range(3):
|
||||
self.node._inject_lora_trigger_words("a cat <lora:neon:1>")
|
||||
self.assertEqual(discover.call_count, 1)
|
||||
|
||||
|
||||
class TestPromptManagerEncoding(NodeTestCase):
|
||||
node_cls = PromptManager
|
||||
|
||||
Reference in New Issue
Block a user