1 """Test program for the fcntl C module.
2 """
3 import multiprocessing
4 import platform
5 import os
6 import struct
7 import sys
8 import unittest
9 from test.support import verbose, cpython_only, get_pagesize
10 from test.support.import_helper import import_module
11 from test.support.os_helper import TESTFN, unlink
12
13
14 # Skip test if no fcntl module.
15 fcntl = import_module('fcntl')
16
17
18
19 class ESC[4;38;5;81mBadFile:
20 def __init__(self, fn):
21 self.fn = fn
22 def fileno(self):
23 return self.fn
24
25 def try_lockf_on_other_process_fail(fname, cmd):
26 f = open(fname, 'wb+')
27 try:
28 fcntl.lockf(f, cmd)
29 except BlockingIOError:
30 pass
31 finally:
32 f.close()
33
34 def try_lockf_on_other_process(fname, cmd):
35 f = open(fname, 'wb+')
36 fcntl.lockf(f, cmd)
37 fcntl.lockf(f, fcntl.LOCK_UN)
38 f.close()
39
40 class ESC[4;38;5;81mTestFcntl(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
41
42 def setUp(self):
43 self.f = None
44
45 def tearDown(self):
46 if self.f and not self.f.closed:
47 self.f.close()
48 unlink(TESTFN)
49
50 @staticmethod
51 def get_lockdata():
52 try:
53 os.O_LARGEFILE
54 except AttributeError:
55 start_len = "ll"
56 else:
57 start_len = "qq"
58
59 if (sys.platform.startswith(('netbsd', 'freebsd', 'openbsd'))
60 or sys.platform == 'darwin'):
61 if struct.calcsize('l') == 8:
62 off_t = 'l'
63 pid_t = 'i'
64 else:
65 off_t = 'lxxxx'
66 pid_t = 'l'
67 lockdata = struct.pack(off_t + off_t + pid_t + 'hh', 0, 0, 0,
68 fcntl.F_WRLCK, 0)
69 elif sys.platform.startswith('gnukfreebsd'):
70 lockdata = struct.pack('qqihhi', 0, 0, 0, fcntl.F_WRLCK, 0, 0)
71 elif sys.platform in ['hp-uxB', 'unixware7']:
72 lockdata = struct.pack('hhlllii', fcntl.F_WRLCK, 0, 0, 0, 0, 0, 0)
73 else:
74 lockdata = struct.pack('hh'+start_len+'hh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
75 if lockdata:
76 if verbose:
77 print('struct.pack: ', repr(lockdata))
78 return lockdata
79
80 def test_fcntl_fileno(self):
81 # the example from the library docs
82 self.f = open(TESTFN, 'wb')
83 rv = fcntl.fcntl(self.f.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
84 if verbose:
85 print('Status from fcntl with O_NONBLOCK: ', rv)
86 lockdata = self.get_lockdata()
87 rv = fcntl.fcntl(self.f.fileno(), fcntl.F_SETLKW, lockdata)
88 if verbose:
89 print('String from fcntl with F_SETLKW: ', repr(rv))
90 self.f.close()
91
92 def test_fcntl_file_descriptor(self):
93 # again, but pass the file rather than numeric descriptor
94 self.f = open(TESTFN, 'wb')
95 rv = fcntl.fcntl(self.f, fcntl.F_SETFL, os.O_NONBLOCK)
96 if verbose:
97 print('Status from fcntl with O_NONBLOCK: ', rv)
98 lockdata = self.get_lockdata()
99 rv = fcntl.fcntl(self.f, fcntl.F_SETLKW, lockdata)
100 if verbose:
101 print('String from fcntl with F_SETLKW: ', repr(rv))
102 self.f.close()
103
104 def test_fcntl_bad_file(self):
105 with self.assertRaises(ValueError):
106 fcntl.fcntl(-1, fcntl.F_SETFL, os.O_NONBLOCK)
107 with self.assertRaises(ValueError):
108 fcntl.fcntl(BadFile(-1), fcntl.F_SETFL, os.O_NONBLOCK)
109 with self.assertRaises(TypeError):
110 fcntl.fcntl('spam', fcntl.F_SETFL, os.O_NONBLOCK)
111 with self.assertRaises(TypeError):
112 fcntl.fcntl(BadFile('spam'), fcntl.F_SETFL, os.O_NONBLOCK)
113
114 @cpython_only
115 def test_fcntl_bad_file_overflow(self):
116 from _testcapi import INT_MAX, INT_MIN
117 # Issue 15989
118 with self.assertRaises(OverflowError):
119 fcntl.fcntl(INT_MAX + 1, fcntl.F_SETFL, os.O_NONBLOCK)
120 with self.assertRaises(OverflowError):
121 fcntl.fcntl(BadFile(INT_MAX + 1), fcntl.F_SETFL, os.O_NONBLOCK)
122 with self.assertRaises(OverflowError):
123 fcntl.fcntl(INT_MIN - 1, fcntl.F_SETFL, os.O_NONBLOCK)
124 with self.assertRaises(OverflowError):
125 fcntl.fcntl(BadFile(INT_MIN - 1), fcntl.F_SETFL, os.O_NONBLOCK)
126
127 @unittest.skipIf(
128 platform.machine().startswith('arm') and platform.system() == 'Linux',
129 "ARM Linux returns EINVAL for F_NOTIFY DN_MULTISHOT")
130 def test_fcntl_64_bit(self):
131 # Issue #1309352: fcntl shouldn't fail when the third arg fits in a
132 # C 'long' but not in a C 'int'.
133 try:
134 cmd = fcntl.F_NOTIFY
135 # This flag is larger than 2**31 in 64-bit builds
136 flags = fcntl.DN_MULTISHOT
137 except AttributeError:
138 self.skipTest("F_NOTIFY or DN_MULTISHOT unavailable")
139 fd = os.open(os.path.dirname(os.path.abspath(TESTFN)), os.O_RDONLY)
140 try:
141 fcntl.fcntl(fd, cmd, flags)
142 finally:
143 os.close(fd)
144
145 def test_flock(self):
146 # Solaris needs readable file for shared lock
147 self.f = open(TESTFN, 'wb+')
148 fileno = self.f.fileno()
149 fcntl.flock(fileno, fcntl.LOCK_SH)
150 fcntl.flock(fileno, fcntl.LOCK_UN)
151 fcntl.flock(self.f, fcntl.LOCK_SH | fcntl.LOCK_NB)
152 fcntl.flock(self.f, fcntl.LOCK_UN)
153 fcntl.flock(fileno, fcntl.LOCK_EX)
154 fcntl.flock(fileno, fcntl.LOCK_UN)
155
156 self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH)
157 self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH)
158
159 @unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError")
160 def test_lockf_exclusive(self):
161 self.f = open(TESTFN, 'wb+')
162 cmd = fcntl.LOCK_EX | fcntl.LOCK_NB
163 fcntl.lockf(self.f, cmd)
164 mp = multiprocessing.get_context('spawn')
165 p = mp.Process(target=try_lockf_on_other_process_fail, args=(TESTFN, cmd))
166 p.start()
167 p.join()
168 fcntl.lockf(self.f, fcntl.LOCK_UN)
169 self.assertEqual(p.exitcode, 0)
170
171 @unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError")
172 def test_lockf_share(self):
173 self.f = open(TESTFN, 'wb+')
174 cmd = fcntl.LOCK_SH | fcntl.LOCK_NB
175 fcntl.lockf(self.f, cmd)
176 mp = multiprocessing.get_context('spawn')
177 p = mp.Process(target=try_lockf_on_other_process, args=(TESTFN, cmd))
178 p.start()
179 p.join()
180 fcntl.lockf(self.f, fcntl.LOCK_UN)
181 self.assertEqual(p.exitcode, 0)
182
183 @cpython_only
184 def test_flock_overflow(self):
185 import _testcapi
186 self.assertRaises(OverflowError, fcntl.flock, _testcapi.INT_MAX+1,
187 fcntl.LOCK_SH)
188
189 @unittest.skipIf(sys.platform != 'darwin', "F_GETPATH is only available on macos")
190 def test_fcntl_f_getpath(self):
191 self.f = open(TESTFN, 'wb')
192 expected = os.path.abspath(TESTFN).encode('utf-8')
193 res = fcntl.fcntl(self.f.fileno(), fcntl.F_GETPATH, bytes(len(expected)))
194 self.assertEqual(expected, res)
195
196 @unittest.skipUnless(
197 hasattr(fcntl, "F_SETPIPE_SZ") and hasattr(fcntl, "F_GETPIPE_SZ"),
198 "F_SETPIPE_SZ and F_GETPIPE_SZ are not available on all platforms.")
199 def test_fcntl_f_pipesize(self):
200 test_pipe_r, test_pipe_w = os.pipe()
201 try:
202 # Get the default pipesize with F_GETPIPE_SZ
203 pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ)
204 pipesize = pipesize_default // 2 # A new value to detect change.
205 pagesize_default = get_pagesize()
206 if pipesize < pagesize_default: # the POSIX minimum
207 raise unittest.SkipTest(
208 'default pipesize too small to perform test.')
209 fcntl.fcntl(test_pipe_w, fcntl.F_SETPIPE_SZ, pipesize)
210 self.assertEqual(fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ),
211 pipesize)
212 finally:
213 os.close(test_pipe_r)
214 os.close(test_pipe_w)
215
216
217 if __name__ == '__main__':
218 unittest.main()