Vito Sansevero
2fe9686ffc
With the database commits gone, throttled resets still stalled in rmtree: unlinking the last fixture's files waits for their pending writeback. Each reseed now moves the old output and input folders into <root>/reset-trash and makes empty ones; a background thread deletes the trash, and the server's cleanup stops it and removes the rest before the temp root goes.
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""Browser-test media folders are emptied by a rename, deleted later (#384)."""
|
|
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from tests.browser.folder_trash import TRASH_NAME, close_trash, replace_folder
|
|
|
|
|
|
class TestFolderTrash(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.addCleanup(close_trash, self.root)
|
|
|
|
def trash_threads(self):
|
|
return [t for t in threading.enumerate() if t.name == "pm-reset-trash"]
|
|
|
|
def test_replaces_the_folder_with_an_empty_one(self):
|
|
folder = self.root / "output"
|
|
(folder / "nested").mkdir(parents=True)
|
|
(folder / "nested" / "image.png").write_bytes(b"png")
|
|
replace_folder(folder)
|
|
self.assertTrue(folder.is_dir())
|
|
self.assertEqual(list(folder.iterdir()), [])
|
|
close_trash(self.root)
|
|
self.assertFalse((self.root / TRASH_NAME).exists())
|
|
self.assertEqual(self.trash_threads(), [])
|
|
|
|
def test_a_missing_folder_is_created_without_a_trash(self):
|
|
replace_folder(self.root / "input")
|
|
self.assertTrue((self.root / "input").is_dir())
|
|
self.assertFalse((self.root / TRASH_NAME).exists())
|
|
self.assertEqual(self.trash_threads(), [])
|
|
|
|
def test_repeated_replacements_share_one_thread_and_all_are_removed(self):
|
|
folder = self.root / "output"
|
|
for number in range(5):
|
|
folder.mkdir(exist_ok=True)
|
|
(folder / f"{number}.png").write_bytes(b"png")
|
|
replace_folder(folder)
|
|
self.assertEqual(len(self.trash_threads()), 1)
|
|
close_trash(self.root)
|
|
self.assertEqual(list(folder.iterdir()), [])
|
|
self.assertFalse((self.root / TRASH_NAME).exists())
|
|
close_trash(self.root) # idempotent
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|