58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
|
import tempfile
|
||
|
|
import time
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from backend.utils import cache as cache_utils
|
||
|
|
|
||
|
|
|
||
|
|
class CacheUtilsTests(unittest.TestCase):
|
||
|
|
def setUp(self) -> None:
|
||
|
|
self._tmp_dir = tempfile.TemporaryDirectory()
|
||
|
|
self._old_cache_dir = cache_utils.CACHE_DIR
|
||
|
|
cache_utils.CACHE_DIR = Path(self._tmp_dir.name) / "cache"
|
||
|
|
|
||
|
|
self._work_dir = Path(self._tmp_dir.name) / "work"
|
||
|
|
self._work_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
self._src_file = self._work_dir / "sample.txt"
|
||
|
|
self._src_file.write_text("hello", encoding="utf-8")
|
||
|
|
|
||
|
|
def tearDown(self) -> None:
|
||
|
|
cache_utils.CACHE_DIR = self._old_cache_dir
|
||
|
|
self._tmp_dir.cleanup()
|
||
|
|
|
||
|
|
def test_get_file_hash_returns_none_for_missing_file(self) -> None:
|
||
|
|
missing = self._work_dir / "missing.txt"
|
||
|
|
self.assertIsNone(cache_utils.get_file_hash(missing))
|
||
|
|
|
||
|
|
def test_save_and_load_round_trip(self) -> None:
|
||
|
|
payload = {"value": 123, "ok": True}
|
||
|
|
saved = cache_utils.save_to_cache(self._src_file, payload, model="m1", operation="transcribe")
|
||
|
|
self.assertTrue(saved)
|
||
|
|
|
||
|
|
loaded = cache_utils.load_from_cache(self._src_file, model="m1", operation="transcribe")
|
||
|
|
self.assertEqual(payload, loaded)
|
||
|
|
|
||
|
|
def test_load_from_cache_respects_max_age(self) -> None:
|
||
|
|
payload = {"value": 999}
|
||
|
|
self.assertTrue(cache_utils.save_to_cache(self._src_file, payload, operation="transcribe"))
|
||
|
|
|
||
|
|
time.sleep(0.02)
|
||
|
|
expired = cache_utils.load_from_cache(self._src_file, operation="transcribe", max_age=0.001)
|
||
|
|
self.assertIsNone(expired)
|
||
|
|
|
||
|
|
def test_clear_cache_deletes_files(self) -> None:
|
||
|
|
self.assertTrue(cache_utils.save_to_cache(self._src_file, {"a": 1}, operation="transcribe"))
|
||
|
|
self.assertTrue(cache_utils.save_to_cache(self._src_file, {"a": 2}, operation="summarize"))
|
||
|
|
|
||
|
|
deleted_count = cache_utils.clear_cache()
|
||
|
|
self.assertGreaterEqual(deleted_count, 1)
|
||
|
|
|
||
|
|
size_bytes, file_count = cache_utils.get_cache_size()
|
||
|
|
self.assertEqual(size_bytes, 0)
|
||
|
|
self.assertEqual(file_count, 0)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|