mirror of
https://github.com/licsber/micropython.git
synced 2024-09-20 09:00:23 +08:00
4e0964b59f
When iterating over filesystem/folders with os.iterdir(), an open file (directory) handle is used internally. Currently this file handle is only closed once the iterator is completely drained, eg. once all entries have been looped over / converted into list etc. If a program opens an iterdir but does not loop over it, or starts to loop over the iterator but breaks out of the loop, then the handle never gets closed. In this state, when the iter object is cleaned up by the garbage collector this open handle can cause corruption of the filesystem. Fixes issues #6568 and #8506.
71 lines
1.6 KiB
Python
71 lines
1.6 KiB
Python
# Test ilistdir __del__ for VfsPosix.
|
|
import gc
|
|
|
|
try:
|
|
import os
|
|
|
|
os.VfsPosix
|
|
except (ImportError, AttributeError):
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
|
|
def test(testdir):
|
|
vfs = os.VfsPosix(testdir)
|
|
vfs.mkdir("/test_d1")
|
|
vfs.mkdir("/test_d2")
|
|
vfs.mkdir("/test_d3")
|
|
|
|
for i in range(10):
|
|
print(i)
|
|
|
|
# We want to partially iterate the ilistdir iterator to leave it in an
|
|
# open state, which will then test the finaliser when it's garbage collected.
|
|
idir = vfs.ilistdir("/")
|
|
print(any(idir))
|
|
|
|
# Alternate way of partially iterating the ilistdir object, modifying the
|
|
# filesystem while it's open.
|
|
for dname, *_ in vfs.ilistdir("/"):
|
|
vfs.rmdir(dname)
|
|
break
|
|
vfs.mkdir(dname)
|
|
|
|
# Also create a fully drained iterator and ensure trying to re-use it
|
|
# throws the correct exception.
|
|
idir_emptied = vfs.ilistdir("/")
|
|
l = list(idir_emptied)
|
|
print(len(l))
|
|
try:
|
|
next(idir_emptied)
|
|
except StopIteration:
|
|
pass
|
|
|
|
gc.collect()
|
|
|
|
# Create and delete a file, try to flush out any filesystem
|
|
# corruption that may be caused over the loops.
|
|
vfs.open("/test", "w").close()
|
|
vfs.remove("/test")
|
|
|
|
|
|
# We need an empty directory for testing.
|
|
# Skip the test if it already exists.
|
|
temp_dir = "vfs_posix_ilistdir_del_test_dir"
|
|
try:
|
|
os.stat(temp_dir)
|
|
print("SKIP")
|
|
raise SystemExit
|
|
except OSError:
|
|
pass
|
|
|
|
os.mkdir(temp_dir)
|
|
|
|
test(temp_dir)
|
|
|
|
# Remove tempdir.
|
|
for td in os.listdir(temp_dir):
|
|
os.rmdir("/".join((temp_dir, td)))
|
|
|
|
os.rmdir(temp_dir)
|