| LEFT | RIGHT |
| 1 "Test posix functions" | 1 "Test posix functions" |
| 2 | 2 |
| 3 from test import support | 3 from test import support |
| 4 | 4 |
| 5 # Skip these tests if there is no posix module. | 5 # Skip these tests if there is no posix module. |
| 6 posix = support.import_module('posix') | 6 posix = support.import_module('posix') |
| 7 | 7 |
| 8 import errno | 8 import errno |
| 9 import sys | 9 import sys |
| 10 import time | 10 import time |
| (...skipping 116 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 127 try: | 127 try: |
| 128 # we need to have some data to truncate | 128 # we need to have some data to truncate |
| 129 fp.write('test') | 129 fp.write('test') |
| 130 fp.flush() | 130 fp.flush() |
| 131 posix.ftruncate(fp.fileno(), 0) | 131 posix.ftruncate(fp.fileno(), 0) |
| 132 finally: | 132 finally: |
| 133 fp.close() | 133 fp.close() |
| 134 | 134 |
| 135 @unittest.skipUnless(hasattr(posix, 'truncate'), "test needs posix.truncate(
)") | 135 @unittest.skipUnless(hasattr(posix, 'truncate'), "test needs posix.truncate(
)") |
| 136 def test_truncate(self): | 136 def test_truncate(self): |
| 137 fp = open(support.TESTFN, 'w+') | 137 with open(support.TESTFN, 'w') as fp: |
| 138 try: | |
| 139 # we need to have some data to truncate | |
| 140 fp.write('test') | 138 fp.write('test') |
| 141 fp.flush() | 139 fp.flush() |
| 142 posix.truncate(support.TESTFN, 0) | 140 posix.truncate(support.TESTFN, 0) |
| 143 finally: | |
| 144 fp.close() | |
| 145 | 141 |
| 146 @unittest.skipUnless(hasattr(posix, 'fexecve'), "test needs posix.fexecve()"
) | 142 @unittest.skipUnless(hasattr(posix, 'fexecve'), "test needs posix.fexecve()"
) |
| 147 @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") | 143 @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") |
| 148 @unittest.skipUnless(hasattr(os, 'chdir'), "test needs os.chdir()") | 144 @unittest.skipUnless(hasattr(os, 'wait'), "test needs os.wait()") |
| 149 @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") | |
| 150 def test_fexecve(self): | 145 def test_fexecve(self): |
| 151 fp = os.open(sys.executable, os.O_RDONLY) | 146 fp = os.open(sys.executable, os.O_RDONLY) |
| 152 try: | 147 try: |
| 153 pid = os.fork() | 148 pid = os.fork() |
| 154 if pid == 0: | 149 if pid == 0: |
| 155 os.chdir(os.path.split(sys.executable)[0]) | 150 os.chdir(os.path.split(sys.executable)[0]) |
| 156 posix.fexecve(fp, [sys.executable, '-c', 'pass'], os.environ) | 151 posix.fexecve(fp, [sys.executable, '-c', 'pass'], os.environ) |
| 157 else: | 152 else: |
| 158 os.waitpid(pid, 0) | 153 self.assertEqual(os.wait(), (pid, 0)) |
| 159 finally: | 154 finally: |
| 160 os.close(fp) | 155 os.close(fp) |
| 161 | 156 |
| 162 @unittest.skipUnless(hasattr(posix, 'gethostid'), "test needs posix.gethosti
d()") | 157 @unittest.skipUnless(hasattr(posix, 'gethostid'), "test needs posix.gethosti
d()") |
| 163 def test_gethostid(self): | 158 def test_gethostid(self): |
| 164 self.assertTrue(isinstance(posix.gethostid(), int)) | 159 self.assertTrue(isinstance(posix.gethostid(), int)) |
| 165 | 160 |
| 166 @unittest.skipUnless(hasattr(posix, 'sethostname'), "test needs posix.sethos
tname()") | 161 @unittest.skipUnless(hasattr(posix, 'sethostid'), "test needs posix.sethosti
d()") |
| 167 def test_sethostname(self): | 162 def test_sethostid(self): |
| 168 import socket | 163 hid = os.gethostid() |
| 169 oldhn = socket.gethostname() | 164 try: |
| 170 try: | 165 posix.sethostid(2464787) |
| 171 posix.sethostname('new') | 166 except OSError as e: |
| 172 except OSError as inst: | 167 if sys.platform.startswith("freebsd"): |
| 173 # requires root permission | 168 self.assertEqual(e.errno, errno.EPERM) |
| 174 self.assertEqual(inst.errno, 1) | 169 else: |
| 170 self.assertEqual(e.errno, errno.EACCES) |
| 175 else: | 171 else: |
| 176 # running test as root! | 172 # running test as root! |
| 177 self.assertEqual(socket.gethostname(), 'new') | 173 self.assertEqual(os.gethostid(), 2464787) |
| 178 posix.sethostname(oldhn) | 174 posix.sethostid(hid) |
| 179 | 175 |
| 180 @unittest.skipUnless(hasattr(posix, 'waitid'), "test needs posix.waitid()") | 176 @unittest.skipUnless(hasattr(posix, 'waitid'), "test needs posix.waitid()") |
| 181 @unittest.skipUnless(hasattr(os, 'chdir'), "test needs os.chdir()") | |
| 182 @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") | 177 @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") |
| 183 def test_waitid(self): | 178 def test_waitid(self): |
| 184 pid = os.fork() | 179 pid = os.fork() |
| 185 if pid == 0: | 180 if pid == 0: |
| 186 os.chdir(os.path.split(sys.executable)[0]) | 181 os.chdir(os.path.split(sys.executable)[0]) |
| 187 posix.execve(sys.executable, [sys.executable, '-c', 'pass'], os.envi
ron) | 182 posix.execve(sys.executable, [sys.executable, '-c', 'pass'], os.envi
ron) |
| 188 else: | 183 else: |
| 189 res = posix.waitid(posix.P_PID, pid, posix.WEXITED) | 184 res = posix.waitid(posix.P_PID, pid, posix.WEXITED) |
| 190 self.assertEqual(pid, res[3]) | 185 self.assertEqual(pid, res.si_pid) |
| 191 | 186 |
| 192 @unittest.skipUnless(hasattr(posix, 'lockf'), "test needs posix.lockf()") | 187 @unittest.skipUnless(hasattr(posix, 'lockf'), "test needs posix.lockf()") |
| 193 def test_lockf(self): | 188 def test_lockf(self): |
| 194 fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) | 189 fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) |
| 195 try: | 190 try: |
| 196 os.write(fd, b'test') | 191 os.write(fd, b'test') |
| 197 os.lseek(fd, 0, os.SEEK_SET) | 192 os.lseek(fd, 0, os.SEEK_SET) |
| 198 posix.lockf(fd, posix.F_LOCK, 4) | 193 posix.lockf(fd, posix.F_LOCK, 4) |
| 199 # section is locked | 194 # section is locked |
| 200 posix.lockf(fd, posix.F_ULOCK, 4) | 195 posix.lockf(fd, posix.F_ULOCK, 4) |
| (...skipping 25 matching lines...) Expand all Loading... |
| 226 | 221 |
| 227 @unittest.skipUnless(hasattr(posix, 'posix_fallocate'), | 222 @unittest.skipUnless(hasattr(posix, 'posix_fallocate'), |
| 228 "test needs posix.posix_fallocate()") | 223 "test needs posix.posix_fallocate()") |
| 229 def test_posix_fallocate(self): | 224 def test_posix_fallocate(self): |
| 230 fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) | 225 fd = os.open(support.TESTFN, os.O_WRONLY | os.O_CREAT) |
| 231 try: | 226 try: |
| 232 posix.posix_fallocate(fd, 0, 10) | 227 posix.posix_fallocate(fd, 0, 10) |
| 233 except OSError as inst: | 228 except OSError as inst: |
| 234 # issue10812, ZFS doesn't appear to support posix_fallocate, | 229 # issue10812, ZFS doesn't appear to support posix_fallocate, |
| 235 # so skip Solaris-based since they are likely to have ZFS. | 230 # so skip Solaris-based since they are likely to have ZFS. |
| 236 if inst.errno != 22 or not sys.platform.startswith("sunos"): | 231 if inst.errno != errno.EINVAL or not sys.platform.startswith("sunos"
): |
| 237 raise | 232 raise |
| 238 finally: | 233 finally: |
| 239 os.close(fd) | 234 os.close(fd) |
| 240 | 235 |
| 241 @unittest.skipUnless(hasattr(posix, 'posix_fadvise'), | 236 @unittest.skipUnless(hasattr(posix, 'posix_fadvise'), |
| 242 "test needs posix.posix_fadvise()") | 237 "test needs posix.posix_fadvise()") |
| 243 def test_posix_fadvise(self): | 238 def test_posix_fadvise(self): |
| 244 fd = os.open(support.TESTFN, os.O_RDONLY) | 239 fd = os.open(support.TESTFN, os.O_RDONLY) |
| 245 try: | 240 try: |
| 246 posix.posix_fadvise(fd, 0, 0, posix.POSIX_FADV_WILLNEED) | 241 posix.posix_fadvise(fd, 0, 0, posix.POSIX_FADV_WILLNEED) |
| (...skipping 46 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 293 self.assertEqual(b'test1tt2t3', posix.read(fd, 10)) | 288 self.assertEqual(b'test1tt2t3', posix.read(fd, 10)) |
| 294 finally: | 289 finally: |
| 295 os.close(fd) | 290 os.close(fd) |
| 296 | 291 |
| 297 @unittest.skipUnless(hasattr(posix, 'readv'), "test needs posix.readv()") | 292 @unittest.skipUnless(hasattr(posix, 'readv'), "test needs posix.readv()") |
| 298 def test_readv(self): | 293 def test_readv(self): |
| 299 fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) | 294 fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT) |
| 300 try: | 295 try: |
| 301 os.write(fd, b'test1tt2t3') | 296 os.write(fd, b'test1tt2t3') |
| 302 os.lseek(fd, 0, os.SEEK_SET) | 297 os.lseek(fd, 0, os.SEEK_SET) |
| 303 a = bytearray(5) | 298 buf = [bytearray(i) for i in [5, 3, 2]] |
| 304 b = bytearray(3) | 299 self.assertEqual(posix.readv(fd, buf), 10) |
| 305 c = bytearray(2) | 300 self.assertEqual([b'test1', b'tt2', b't3'], [bytes(i) for i in buf]) |
| 306 self.assertEqual(posix.readv(fd, (a, b, c)), 10) | |
| 307 self.assertEqual((b'test1', b'tt2', b't3'), | |
| 308 (bytes(a), bytes(b), bytes(c))) | |
| 309 finally: | 301 finally: |
| 310 os.close(fd) | 302 os.close(fd) |
| 311 | 303 |
| 312 def test_dup(self): | 304 def test_dup(self): |
| 313 if hasattr(posix, 'dup'): | 305 if hasattr(posix, 'dup'): |
| 314 fp = open(support.TESTFN) | 306 fp = open(support.TESTFN) |
| 315 try: | 307 try: |
| 316 fd = posix.dup(fp.fileno()) | 308 fd = posix.dup(fp.fileno()) |
| 317 self.assertIsInstance(fd, int) | 309 self.assertIsInstance(fd, int) |
| 318 os.close(fd) | 310 os.close(fd) |
| (...skipping 278 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 597 for groups in [[0], list(range(16))]: | 589 for groups in [[0], list(range(16))]: |
| 598 posix.setgroups(groups) | 590 posix.setgroups(groups) |
| 599 self.assertListEqual(groups, posix.getgroups()) | 591 self.assertListEqual(groups, posix.getgroups()) |
| 600 | 592 |
| 601 | 593 |
| 602 def test_main(): | 594 def test_main(): |
| 603 support.run_unittest(PosixTester, PosixGroupsTester) | 595 support.run_unittest(PosixTester, PosixGroupsTester) |
| 604 | 596 |
| 605 if __name__ == '__main__': | 597 if __name__ == '__main__': |
| 606 test_main() | 598 test_main() |
| LEFT | RIGHT |