aboutsummaryrefslogtreecommitdiffstats
path: root/tests
diff options
context:
space:
mode:
authorGravatar ionite34 <[email protected]>2022-11-28 14:26:11 +0800
committerGravatar ionite34 <[email protected]>2022-11-28 14:26:11 +0800
commit0de6825f3fa25afa4d0ccd0fabde91b07cc64d02 (patch)
treee4afaddb0f7ca3ada67898361f6e654409c6c2a1 /tests
parentMove rlimit_fsize to cfg (diff)
Refactor MemFS for implicit cleanup support
Diffstat (limited to 'tests')
-rw-r--r--tests/test_memfs.py55
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())