summaryrefslogtreecommitdiffstatshomepage
path: root/tests/extmod/vfs_userfs.py
blob: a7b87ec0b0d0c76160a764320572b7e23b62b942 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
# test VFS functionality with a user-defined filesystem
# also tests parts of io.IOBase implementation

import sys

try:
    import io, vfs

    io.IOBase
except (ImportError, AttributeError):
    print("SKIP")
    raise SystemExit


class UserFile(io.IOBase):
    buffer_size = 16

    def __init__(self, mode, data):
        assert isinstance(data, bytes)
        self.is_text = mode.find("b") == -1
        self.data = data
        self.pos = 0

    def read(self):
        if self.is_text:
            return str(self.data, "utf8")
        else:
            return self.data

    def readinto(self, buf):
        assert not self.is_text
        n = 0
        while n < len(buf) and self.pos < len(self.data):
            buf[n] = self.data[self.pos]
            n += 1
            self.pos += 1
        return n

    def ioctl(self, req, arg):
        print("ioctl", req, arg)
        if req == 4:  # MP_STREAM_CLOSE
            return 0
        if req == 11:  # MP_STREAM_GET_BUFFER_SIZE
            return UserFile.buffer_size
        return -1


class UserFS:
    def __init__(self, files):
        self.files = files

    def mount(self, readonly, mksfs):
        pass

    def umount(self):
        pass

    def stat(self, path):
        print("stat", path)
        if path in self.files:
            return (32768, 0, 0, 0, 0, 0, 0, 0, 0, 0)
        raise OSError

    def open(self, path, mode):
        print("open", path, mode)
        return UserFile(mode, self.files[path])


# create and mount a user filesystem
user_files = {
    "/data.txt": b"some data in a text file",
    "/usermod1.py": b"print('in usermod1')\nimport usermod2",
    "/usermod2.py": b"print('in usermod2')",
    "/usermod3.py": b"syntax error",
    "/usermod4.mpy": b"syntax error",
    "/usermod5.py": b"print('in usermod5')",
    "/usermod6.py": b"print('in usermod6')",
}
vfs.mount(UserFS(user_files), "/userfs")

# open and read a file
f = open("/userfs/data.txt")
print(f.read())

# import files from the user filesystem
sys.path.append("/userfs")
import usermod1

# import a .py file with a syntax error (file should be closed on error)
try:
    import usermod3
except SyntaxError:
    print("SyntaxError in usermod3")

# import a .mpy file with a syntax error (file should be closed on error)
try:
    import usermod4
except ValueError:
    print("ValueError in usermod4")

# Test an import with largest buffer size
UserFile.buffer_size = 255
import usermod5

# Test an import with over-size buffer size (should be safely limited internally)
UserFile.buffer_size = 1024
import usermod6

# unmount and undo path addition
vfs.umount("/userfs")
sys.path.pop()