diff options
author | 2022-11-28 14:26:11 +0800 | |
---|---|---|
committer | 2022-11-28 14:26:11 +0800 | |
commit | 0de6825f3fa25afa4d0ccd0fabde91b07cc64d02 (patch) | |
tree | e4afaddb0f7ca3ada67898361f6e654409c6c2a1 /tests | |
parent | Move rlimit_fsize to cfg (diff) |
Refactor MemFS for implicit cleanup support
Diffstat (limited to 'tests')
-rw-r--r-- | tests/test_memfs.py | 55 |
1 files changed, 33 insertions, 22 deletions
diff --git a/tests/test_memfs.py b/tests/test_memfs.py index 8050562..53dee22 100644 --- a/tests/test_memfs.py +++ b/tests/test_memfs.py @@ -1,6 +1,6 @@ import logging +import warnings from concurrent.futures import ThreadPoolExecutor -from operator import attrgetter from unittest import TestCase, mock from uuid import uuid4 @@ -9,24 +9,26 @@ from snekbox.memfs import MemFS UUID_TEST = uuid4() -def get_memfs_with_context(): - return MemFS(10).__enter__() - - class MemFSTests(TestCase): def setUp(self): super().setUp() self.logger = logging.getLogger("snekbox.memfs") self.logger.setLevel(logging.WARNING) + warnings.filterwarnings( + "ignore", + ".*Implicitly cleaning up.*", + ResourceWarning, + "snekbox.memfs", + ) @mock.patch("snekbox.memfs.uuid4", lambda: UUID_TEST) def test_assignment_thread_safe(self): """Test concurrent mounting works in multi-thread environments.""" # Concurrently create MemFS in threads, check only 1 can be created # Others should result in RuntimeError - with ThreadPoolExecutor() as pool: + with ThreadPoolExecutor() as executor: memfs: MemFS | None = None - futures = [pool.submit(get_memfs_with_context) for _ in range(8)] + futures = [executor.submit(MemFS, 10) for _ in range(8)] for future in futures: # We should have exactly one result and all others RuntimeErrors if err := future.exception(): @@ -37,20 +39,29 @@ class MemFSTests(TestCase): # Original memfs should still exist afterwards self.assertIsInstance(memfs, MemFS) - self.assertTrue(memfs.path.exists()) - - def test_no_context_error(self): - """Accessing MemFS attributes before __enter__ raises RuntimeError.""" - cases = [ - attrgetter("path"), - attrgetter("name"), - attrgetter("home"), - attrgetter("output"), - lambda fs: fs.mkdir(""), - lambda fs: list(fs.files(1)), - ] + self.assertTrue(memfs.path.is_mount()) + + def test_cleanup(self): + """Test explicit cleanup.""" + memfs = MemFS(10) + path = memfs.path + self.assertTrue(path.is_mount()) + memfs.cleanup() + self.assertFalse(path.exists()) + + def test_context_cleanup(self): + """Context __exit__ should trigger cleanup.""" + with MemFS(10) as memfs: + path = memfs.path + self.assertTrue(path.is_mount()) + self.assertFalse(path.exists()) + def test_implicit_cleanup(self): + """Test implicit _cleanup triggered by GC.""" memfs = MemFS(10) - for case in cases: - with self.subTest(case=case), self.assertRaises(RuntimeError): - case(memfs) + path = memfs.path + self.assertTrue(path.is_mount()) + # Catch the warning about implicit cleanup + with self.assertWarns(ResourceWarning): + del memfs + self.assertFalse(path.exists()) |