56 lines
2.6 KiB
Python
56 lines
2.6 KiB
Python
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('cache', Path(__file__).with_name('gitea_archive_cache.py'))
|
|
cache = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(cache)
|
|
|
|
class CleanerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name).resolve()
|
|
self.now = 10000
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
def file(self, number, age, size=20):
|
|
sha = f'{number:040x}'
|
|
p = self.root / '1' / sha[:2] / (sha + '.zip')
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_bytes(b'x' * size)
|
|
os.utime(p, (self.now-age, self.now-age))
|
|
return p
|
|
def test_expiry_preserves_open_recent_unknown_and_symlink_targets(self):
|
|
old = self.file(1, 4000)
|
|
active = self.file(2, 4000)
|
|
fresh = self.file(3, 100)
|
|
unknown = self.root / 'repository.git'; unknown.write_bytes(b'protected')
|
|
(self.root/'outside').symlink_to('/tmp', target_is_directory=True)
|
|
s = active.stat()
|
|
r = cache.clean(self.root, self.now, {(s.st_dev, s.st_ino)}, budget=1000)
|
|
self.assertFalse(old.exists())
|
|
for p in (active, fresh, unknown): self.assertTrue(p.exists())
|
|
self.assertEqual(1, r['removed_files'])
|
|
def test_capacity_evicts_oldest_then_stops(self):
|
|
old = self.file(1, 1200); newer = self.file(2, 1000); newest = self.file(3, 700)
|
|
r = cache.clean(self.root, self.now, set(), budget=40)
|
|
self.assertFalse(old.exists()); self.assertTrue(newer.exists()); self.assertTrue(newest.exists())
|
|
self.assertEqual(40, r['after_bytes'])
|
|
def test_dry_run_does_not_delete_and_repeated_cleanup_is_idempotent(self):
|
|
p = self.file(1, 4000)
|
|
r = cache.clean(self.root, self.now, set(), dry_run=True)
|
|
self.assertTrue(p.exists()); self.assertEqual(1, r['removed_files'])
|
|
cache.clean(self.root, self.now, set())
|
|
self.assertEqual(0, cache.clean(self.root, self.now, set())['removed_files'])
|
|
def test_fresh_files_may_exceed_budget_without_deletion(self):
|
|
p = self.file(1, 100)
|
|
r = cache.clean(self.root, self.now, set(), budget=1)
|
|
self.assertTrue(p.exists()); self.assertTrue(r['over_budget'])
|
|
def test_symlink_root_is_rejected(self):
|
|
link = self.root/'link'; link.symlink_to(self.root, target_is_directory=True)
|
|
with self.assertRaises(RuntimeError): cache.clean(link, self.now, set())
|
|
|
|
if __name__ == '__main__': unittest.main()
|