mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-12 14:22:11 -04:00
localai-org-maint-botandEttore Di Giacinto
bf93008ef3
Backend processes shared the host temporary directory, so crashes could leave request images and audio behind until the filesystem filled. Give each process a locked LocalAI-owned runtime, remove scratch on exit, and sweep only marked abandoned runtimes at the next start. Also close known request error-path leaks in the Python media backends, CrispASR, LongCat Video, and stable-diffusion.cpp. Assisted-by: Codex:gpt-5 Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
42 lines
1.7 KiB
Python
42 lines
1.7 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from temp_utils import cleanup_paths, materialize_base64
|
|
|
|
|
|
class MaterializeBase64Test(unittest.TestCase):
|
|
def test_removes_materialized_file_after_success(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with mock.patch.object(tempfile, "tempdir", directory):
|
|
with materialize_base64("aGVsbG8=", suffix=".data") as path:
|
|
with open(path, "rb") as materialized:
|
|
self.assertEqual(materialized.read(), b"hello")
|
|
self.assertFalse(os.path.exists(path))
|
|
|
|
def test_removes_materialized_file_when_consumer_fails(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with mock.patch.object(tempfile, "tempdir", directory):
|
|
with self.assertRaisesRegex(RuntimeError, "decode failed"):
|
|
with materialize_base64("aGVsbG8="):
|
|
raise RuntimeError("decode failed")
|
|
self.assertEqual(os.listdir(directory), [])
|
|
|
|
|
|
class CleanupPathsTest(unittest.TestCase):
|
|
def test_removes_every_registered_path_after_failure(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
paths = [os.path.join(directory, name) for name in ("one.wav", "two.wav")]
|
|
with self.assertRaisesRegex(RuntimeError, "merge failed"):
|
|
with cleanup_paths() as registered:
|
|
for path in paths:
|
|
open(path, "wb").close()
|
|
registered.append(path)
|
|
raise RuntimeError("merge failed")
|
|
self.assertEqual(os.listdir(directory), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|