| 1 | # As a test suite for the os module, this is woefully inadequate, but this
|
|---|
| 2 | # does add tests for a few functions which have been determined to be more
|
|---|
| 3 | # portable than they had been thought to be.
|
|---|
| 4 |
|
|---|
| 5 | import os
|
|---|
| 6 | import errno
|
|---|
| 7 | import unittest
|
|---|
| 8 | import warnings
|
|---|
| 9 | import sys
|
|---|
| 10 | import signal
|
|---|
| 11 | import subprocess
|
|---|
| 12 | import time
|
|---|
| 13 | try:
|
|---|
| 14 | import resource
|
|---|
| 15 | except ImportError:
|
|---|
| 16 | resource = None
|
|---|
| 17 |
|
|---|
| 18 | from test import test_support
|
|---|
| 19 | from test.script_helper import assert_python_ok
|
|---|
| 20 | import mmap
|
|---|
| 21 | import uuid
|
|---|
| 22 |
|
|---|
| 23 | warnings.filterwarnings("ignore", "tempnam", RuntimeWarning, __name__)
|
|---|
| 24 | warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, __name__)
|
|---|
| 25 |
|
|---|
| 26 | # Tests creating TESTFN
|
|---|
| 27 | class FileTests(unittest.TestCase):
|
|---|
| 28 | def setUp(self):
|
|---|
| 29 | if os.path.exists(test_support.TESTFN):
|
|---|
| 30 | os.unlink(test_support.TESTFN)
|
|---|
| 31 | tearDown = setUp
|
|---|
| 32 |
|
|---|
| 33 | def test_access(self):
|
|---|
| 34 | f = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
|
|---|
| 35 | os.close(f)
|
|---|
| 36 | self.assertTrue(os.access(test_support.TESTFN, os.W_OK))
|
|---|
| 37 |
|
|---|
| 38 | def test_closerange(self):
|
|---|
| 39 | first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
|
|---|
| 40 | # We must allocate two consecutive file descriptors, otherwise
|
|---|
| 41 | # it will mess up other file descriptors (perhaps even the three
|
|---|
| 42 | # standard ones).
|
|---|
| 43 | second = os.dup(first)
|
|---|
| 44 | try:
|
|---|
| 45 | retries = 0
|
|---|
| 46 | while second != first + 1:
|
|---|
| 47 | os.close(first)
|
|---|
| 48 | retries += 1
|
|---|
| 49 | if retries > 10:
|
|---|
| 50 | # XXX test skipped
|
|---|
| 51 | self.skipTest("couldn't allocate two consecutive fds")
|
|---|
| 52 | first, second = second, os.dup(second)
|
|---|
| 53 | finally:
|
|---|
| 54 | os.close(second)
|
|---|
| 55 | # close a fd that is open, and one that isn't
|
|---|
| 56 | os.closerange(first, first + 2)
|
|---|
| 57 | self.assertRaises(OSError, os.write, first, "a")
|
|---|
| 58 |
|
|---|
| 59 | @test_support.cpython_only
|
|---|
| 60 | def test_rename(self):
|
|---|
| 61 | path = unicode(test_support.TESTFN)
|
|---|
| 62 | old = sys.getrefcount(path)
|
|---|
| 63 | self.assertRaises(TypeError, os.rename, path, 0)
|
|---|
| 64 | new = sys.getrefcount(path)
|
|---|
| 65 | self.assertEqual(old, new)
|
|---|
| 66 |
|
|---|
| 67 |
|
|---|
| 68 | class TemporaryFileTests(unittest.TestCase):
|
|---|
| 69 | def setUp(self):
|
|---|
| 70 | self.files = []
|
|---|
| 71 | os.mkdir(test_support.TESTFN)
|
|---|
| 72 |
|
|---|
| 73 | def tearDown(self):
|
|---|
| 74 | for name in self.files:
|
|---|
| 75 | os.unlink(name)
|
|---|
| 76 | os.rmdir(test_support.TESTFN)
|
|---|
| 77 |
|
|---|
| 78 | def check_tempfile(self, name):
|
|---|
| 79 | # make sure it doesn't already exist:
|
|---|
| 80 | self.assertFalse(os.path.exists(name),
|
|---|
| 81 | "file already exists for temporary file")
|
|---|
| 82 | # make sure we can create the file
|
|---|
| 83 | open(name, "w")
|
|---|
| 84 | self.files.append(name)
|
|---|
| 85 |
|
|---|
| 86 | def test_tempnam(self):
|
|---|
| 87 | if not hasattr(os, "tempnam"):
|
|---|
| 88 | return
|
|---|
| 89 | with warnings.catch_warnings():
|
|---|
| 90 | warnings.filterwarnings("ignore", "tempnam", RuntimeWarning,
|
|---|
| 91 | r"test_os$")
|
|---|
| 92 | warnings.filterwarnings("ignore", "tempnam", DeprecationWarning)
|
|---|
| 93 | self.check_tempfile(os.tempnam())
|
|---|
| 94 |
|
|---|
| 95 | name = os.tempnam(test_support.TESTFN)
|
|---|
| 96 | self.check_tempfile(name)
|
|---|
| 97 |
|
|---|
| 98 | name = os.tempnam(test_support.TESTFN, "pfx")
|
|---|
| 99 | self.assertTrue(os.path.basename(name)[:3] == "pfx")
|
|---|
| 100 | self.check_tempfile(name)
|
|---|
| 101 |
|
|---|
| 102 | def test_tmpfile(self):
|
|---|
| 103 | if not hasattr(os, "tmpfile"):
|
|---|
| 104 | return
|
|---|
| 105 | # As with test_tmpnam() below, the Windows implementation of tmpfile()
|
|---|
| 106 | # attempts to create a file in the root directory of the current drive.
|
|---|
| 107 | # On Vista and Server 2008, this test will always fail for normal users
|
|---|
| 108 | # as writing to the root directory requires elevated privileges. With
|
|---|
| 109 | # XP and below, the semantics of tmpfile() are the same, but the user
|
|---|
| 110 | # running the test is more likely to have administrative privileges on
|
|---|
| 111 | # their account already. If that's the case, then os.tmpfile() should
|
|---|
| 112 | # work. In order to make this test as useful as possible, rather than
|
|---|
| 113 | # trying to detect Windows versions or whether or not the user has the
|
|---|
| 114 | # right permissions, just try and create a file in the root directory
|
|---|
| 115 | # and see if it raises a 'Permission denied' OSError. If it does, then
|
|---|
| 116 | # test that a subsequent call to os.tmpfile() raises the same error. If
|
|---|
| 117 | # it doesn't, assume we're on XP or below and the user running the test
|
|---|
| 118 | # has administrative privileges, and proceed with the test as normal.
|
|---|
| 119 | with warnings.catch_warnings():
|
|---|
| 120 | warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
|
|---|
| 121 |
|
|---|
| 122 | if sys.platform == 'win32':
|
|---|
| 123 | name = '\\python_test_os_test_tmpfile.txt'
|
|---|
| 124 | if os.path.exists(name):
|
|---|
| 125 | os.remove(name)
|
|---|
| 126 | try:
|
|---|
| 127 | fp = open(name, 'w')
|
|---|
| 128 | except IOError, first:
|
|---|
| 129 | # open() failed, assert tmpfile() fails in the same way.
|
|---|
| 130 | # Although open() raises an IOError and os.tmpfile() raises an
|
|---|
| 131 | # OSError(), 'args' will be (13, 'Permission denied') in both
|
|---|
| 132 | # cases.
|
|---|
| 133 | try:
|
|---|
| 134 | fp = os.tmpfile()
|
|---|
| 135 | except OSError, second:
|
|---|
| 136 | self.assertEqual(first.args, second.args)
|
|---|
| 137 | else:
|
|---|
| 138 | self.fail("expected os.tmpfile() to raise OSError")
|
|---|
| 139 | return
|
|---|
| 140 | else:
|
|---|
| 141 | # open() worked, therefore, tmpfile() should work. Close our
|
|---|
| 142 | # dummy file and proceed with the test as normal.
|
|---|
| 143 | fp.close()
|
|---|
| 144 | os.remove(name)
|
|---|
| 145 |
|
|---|
| 146 | fp = os.tmpfile()
|
|---|
| 147 | fp.write("foobar")
|
|---|
| 148 | fp.seek(0,0)
|
|---|
| 149 | s = fp.read()
|
|---|
| 150 | fp.close()
|
|---|
| 151 | self.assertTrue(s == "foobar")
|
|---|
| 152 |
|
|---|
| 153 | def test_tmpnam(self):
|
|---|
| 154 | if not hasattr(os, "tmpnam"):
|
|---|
| 155 | return
|
|---|
| 156 | with warnings.catch_warnings():
|
|---|
| 157 | warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning,
|
|---|
| 158 | r"test_os$")
|
|---|
| 159 | warnings.filterwarnings("ignore", "tmpnam", DeprecationWarning)
|
|---|
| 160 |
|
|---|
| 161 | name = os.tmpnam()
|
|---|
| 162 | if sys.platform in ("win32",):
|
|---|
| 163 | # The Windows tmpnam() seems useless. From the MS docs:
|
|---|
| 164 | #
|
|---|
| 165 | # The character string that tmpnam creates consists of
|
|---|
| 166 | # the path prefix, defined by the entry P_tmpdir in the
|
|---|
| 167 | # file STDIO.H, followed by a sequence consisting of the
|
|---|
| 168 | # digit characters '0' through '9'; the numerical value
|
|---|
| 169 | # of this string is in the range 1 - 65,535. Changing the
|
|---|
| 170 | # definitions of L_tmpnam or P_tmpdir in STDIO.H does not
|
|---|
| 171 | # change the operation of tmpnam.
|
|---|
| 172 | #
|
|---|
| 173 | # The really bizarre part is that, at least under MSVC6,
|
|---|
| 174 | # P_tmpdir is "\\". That is, the path returned refers to
|
|---|
| 175 | # the root of the current drive. That's a terrible place to
|
|---|
| 176 | # put temp files, and, depending on privileges, the user
|
|---|
| 177 | # may not even be able to open a file in the root directory.
|
|---|
| 178 | self.assertFalse(os.path.exists(name),
|
|---|
| 179 | "file already exists for temporary file")
|
|---|
| 180 | else:
|
|---|
| 181 | self.check_tempfile(name)
|
|---|
| 182 |
|
|---|
| 183 | # Test attributes on return values from os.*stat* family.
|
|---|
| 184 | class StatAttributeTests(unittest.TestCase):
|
|---|
| 185 | def setUp(self):
|
|---|
| 186 | os.mkdir(test_support.TESTFN)
|
|---|
| 187 | self.fname = os.path.join(test_support.TESTFN, "f1")
|
|---|
| 188 | f = open(self.fname, 'wb')
|
|---|
| 189 | f.write("ABC")
|
|---|
| 190 | f.close()
|
|---|
| 191 |
|
|---|
| 192 | def tearDown(self):
|
|---|
| 193 | os.unlink(self.fname)
|
|---|
| 194 | os.rmdir(test_support.TESTFN)
|
|---|
| 195 |
|
|---|
| 196 | def test_stat_attributes(self):
|
|---|
| 197 | if not hasattr(os, "stat"):
|
|---|
| 198 | return
|
|---|
| 199 |
|
|---|
| 200 | import stat
|
|---|
| 201 | result = os.stat(self.fname)
|
|---|
| 202 |
|
|---|
| 203 | # Make sure direct access works
|
|---|
| 204 | self.assertEqual(result[stat.ST_SIZE], 3)
|
|---|
| 205 | self.assertEqual(result.st_size, 3)
|
|---|
| 206 |
|
|---|
| 207 | # Make sure all the attributes are there
|
|---|
| 208 | members = dir(result)
|
|---|
| 209 | for name in dir(stat):
|
|---|
| 210 | if name[:3] == 'ST_':
|
|---|
| 211 | attr = name.lower()
|
|---|
| 212 | if name.endswith("TIME"):
|
|---|
| 213 | def trunc(x): return int(x)
|
|---|
| 214 | else:
|
|---|
| 215 | def trunc(x): return x
|
|---|
| 216 | self.assertEqual(trunc(getattr(result, attr)),
|
|---|
| 217 | result[getattr(stat, name)])
|
|---|
| 218 | self.assertIn(attr, members)
|
|---|
| 219 |
|
|---|
| 220 | try:
|
|---|
| 221 | result[200]
|
|---|
| 222 | self.fail("No exception raised")
|
|---|
| 223 | except IndexError:
|
|---|
| 224 | pass
|
|---|
| 225 |
|
|---|
| 226 | # Make sure that assignment fails
|
|---|
| 227 | try:
|
|---|
| 228 | result.st_mode = 1
|
|---|
| 229 | self.fail("No exception raised")
|
|---|
| 230 | except (AttributeError, TypeError):
|
|---|
| 231 | pass
|
|---|
| 232 |
|
|---|
| 233 | try:
|
|---|
| 234 | result.st_rdev = 1
|
|---|
| 235 | self.fail("No exception raised")
|
|---|
| 236 | except (AttributeError, TypeError):
|
|---|
| 237 | pass
|
|---|
| 238 |
|
|---|
| 239 | try:
|
|---|
| 240 | result.parrot = 1
|
|---|
| 241 | self.fail("No exception raised")
|
|---|
| 242 | except AttributeError:
|
|---|
| 243 | pass
|
|---|
| 244 |
|
|---|
| 245 | # Use the stat_result constructor with a too-short tuple.
|
|---|
| 246 | try:
|
|---|
| 247 | result2 = os.stat_result((10,))
|
|---|
| 248 | self.fail("No exception raised")
|
|---|
| 249 | except TypeError:
|
|---|
| 250 | pass
|
|---|
| 251 |
|
|---|
| 252 | # Use the constructor with a too-long tuple.
|
|---|
| 253 | try:
|
|---|
| 254 | result2 = os.stat_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
|
|---|
| 255 | except TypeError:
|
|---|
| 256 | pass
|
|---|
| 257 |
|
|---|
| 258 |
|
|---|
| 259 | def test_statvfs_attributes(self):
|
|---|
| 260 | if not hasattr(os, "statvfs"):
|
|---|
| 261 | return
|
|---|
| 262 |
|
|---|
| 263 | try:
|
|---|
| 264 | result = os.statvfs(self.fname)
|
|---|
| 265 | except OSError, e:
|
|---|
| 266 | # On AtheOS, glibc always returns ENOSYS
|
|---|
| 267 | if e.errno == errno.ENOSYS:
|
|---|
| 268 | return
|
|---|
| 269 |
|
|---|
| 270 | # Make sure direct access works
|
|---|
| 271 | self.assertEqual(result.f_bfree, result[3])
|
|---|
| 272 |
|
|---|
| 273 | # Make sure all the attributes are there.
|
|---|
| 274 | members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
|
|---|
| 275 | 'ffree', 'favail', 'flag', 'namemax')
|
|---|
| 276 | for value, member in enumerate(members):
|
|---|
| 277 | self.assertEqual(getattr(result, 'f_' + member), result[value])
|
|---|
| 278 |
|
|---|
| 279 | # Make sure that assignment really fails
|
|---|
| 280 | try:
|
|---|
| 281 | result.f_bfree = 1
|
|---|
| 282 | self.fail("No exception raised")
|
|---|
| 283 | except TypeError:
|
|---|
| 284 | pass
|
|---|
| 285 |
|
|---|
| 286 | try:
|
|---|
| 287 | result.parrot = 1
|
|---|
| 288 | self.fail("No exception raised")
|
|---|
| 289 | except AttributeError:
|
|---|
| 290 | pass
|
|---|
| 291 |
|
|---|
| 292 | # Use the constructor with a too-short tuple.
|
|---|
| 293 | try:
|
|---|
| 294 | result2 = os.statvfs_result((10,))
|
|---|
| 295 | self.fail("No exception raised")
|
|---|
| 296 | except TypeError:
|
|---|
| 297 | pass
|
|---|
| 298 |
|
|---|
| 299 | # Use the constructor with a too-long tuple.
|
|---|
| 300 | try:
|
|---|
| 301 | result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
|
|---|
| 302 | except TypeError:
|
|---|
| 303 | pass
|
|---|
| 304 |
|
|---|
| 305 | def test_utime_dir(self):
|
|---|
| 306 | delta = 1000000
|
|---|
| 307 | st = os.stat(test_support.TESTFN)
|
|---|
| 308 | # round to int, because some systems may support sub-second
|
|---|
| 309 | # time stamps in stat, but not in utime.
|
|---|
| 310 | os.utime(test_support.TESTFN, (st.st_atime, int(st.st_mtime-delta)))
|
|---|
| 311 | st2 = os.stat(test_support.TESTFN)
|
|---|
| 312 | self.assertEqual(st2.st_mtime, int(st.st_mtime-delta))
|
|---|
| 313 |
|
|---|
| 314 | # Restrict test to Win32, since there is no guarantee other
|
|---|
| 315 | # systems support centiseconds
|
|---|
| 316 | if sys.platform == 'win32':
|
|---|
| 317 | def get_file_system(path):
|
|---|
| 318 | root = os.path.splitdrive(os.path.abspath(path))[0] + '\\'
|
|---|
| 319 | import ctypes
|
|---|
| 320 | kernel32 = ctypes.windll.kernel32
|
|---|
| 321 | buf = ctypes.create_string_buffer("", 100)
|
|---|
| 322 | if kernel32.GetVolumeInformationA(root, None, 0, None, None, None, buf, len(buf)):
|
|---|
| 323 | return buf.value
|
|---|
| 324 |
|
|---|
| 325 | if get_file_system(test_support.TESTFN) == "NTFS":
|
|---|
| 326 | def test_1565150(self):
|
|---|
| 327 | t1 = 1159195039.25
|
|---|
| 328 | os.utime(self.fname, (t1, t1))
|
|---|
| 329 | self.assertEqual(os.stat(self.fname).st_mtime, t1)
|
|---|
| 330 |
|
|---|
| 331 | def test_large_time(self):
|
|---|
| 332 | t1 = 5000000000 # some day in 2128
|
|---|
| 333 | os.utime(self.fname, (t1, t1))
|
|---|
| 334 | self.assertEqual(os.stat(self.fname).st_mtime, t1)
|
|---|
| 335 |
|
|---|
| 336 | def test_1686475(self):
|
|---|
| 337 | # Verify that an open file can be stat'ed
|
|---|
| 338 | try:
|
|---|
| 339 | os.stat(r"c:\pagefile.sys")
|
|---|
| 340 | except WindowsError, e:
|
|---|
| 341 | if e.errno == 2: # file does not exist; cannot run test
|
|---|
| 342 | return
|
|---|
| 343 | self.fail("Could not stat pagefile.sys")
|
|---|
| 344 |
|
|---|
| 345 | from test import mapping_tests
|
|---|
| 346 |
|
|---|
| 347 | class EnvironTests(mapping_tests.BasicTestMappingProtocol):
|
|---|
| 348 | """check that os.environ object conform to mapping protocol"""
|
|---|
| 349 | type2test = None
|
|---|
| 350 | def _reference(self):
|
|---|
| 351 | return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"}
|
|---|
| 352 | def _empty_mapping(self):
|
|---|
| 353 | os.environ.clear()
|
|---|
| 354 | return os.environ
|
|---|
| 355 | def setUp(self):
|
|---|
| 356 | self.__save = dict(os.environ)
|
|---|
| 357 | os.environ.clear()
|
|---|
| 358 | def tearDown(self):
|
|---|
| 359 | os.environ.clear()
|
|---|
| 360 | os.environ.update(self.__save)
|
|---|
| 361 |
|
|---|
| 362 | # Bug 1110478
|
|---|
| 363 | def test_update2(self):
|
|---|
| 364 | if os.path.exists("/bin/sh"):
|
|---|
| 365 | os.environ.update(HELLO="World")
|
|---|
| 366 | with os.popen("/bin/sh -c 'echo $HELLO'") as popen:
|
|---|
| 367 | value = popen.read().strip()
|
|---|
| 368 | self.assertEqual(value, "World")
|
|---|
| 369 |
|
|---|
| 370 | # On FreeBSD < 7 and OS X < 10.6, unsetenv() doesn't return a value (issue
|
|---|
| 371 | # #13415).
|
|---|
| 372 | @unittest.skipIf(sys.platform.startswith(('freebsd', 'darwin')),
|
|---|
| 373 | "due to known OS bug: see issue #13415")
|
|---|
| 374 | def test_unset_error(self):
|
|---|
| 375 | if sys.platform == "win32":
|
|---|
| 376 | # an environment variable is limited to 32,767 characters
|
|---|
| 377 | key = 'x' * 50000
|
|---|
| 378 | self.assertRaises(ValueError, os.environ.__delitem__, key)
|
|---|
| 379 | else:
|
|---|
| 380 | # "=" is not allowed in a variable name
|
|---|
| 381 | key = 'key='
|
|---|
| 382 | self.assertRaises(OSError, os.environ.__delitem__, key)
|
|---|
| 383 |
|
|---|
| 384 | class WalkTests(unittest.TestCase):
|
|---|
| 385 | """Tests for os.walk()."""
|
|---|
| 386 |
|
|---|
| 387 | def test_traversal(self):
|
|---|
| 388 | import os
|
|---|
| 389 | from os.path import join
|
|---|
| 390 |
|
|---|
| 391 | # Build:
|
|---|
| 392 | # TESTFN/
|
|---|
| 393 | # TEST1/ a file kid and two directory kids
|
|---|
| 394 | # tmp1
|
|---|
| 395 | # SUB1/ a file kid and a directory kid
|
|---|
| 396 | # tmp2
|
|---|
| 397 | # SUB11/ no kids
|
|---|
| 398 | # SUB2/ a file kid and a dirsymlink kid
|
|---|
| 399 | # tmp3
|
|---|
| 400 | # link/ a symlink to TESTFN.2
|
|---|
| 401 | # TEST2/
|
|---|
| 402 | # tmp4 a lone file
|
|---|
| 403 | walk_path = join(test_support.TESTFN, "TEST1")
|
|---|
| 404 | sub1_path = join(walk_path, "SUB1")
|
|---|
| 405 | sub11_path = join(sub1_path, "SUB11")
|
|---|
| 406 | sub2_path = join(walk_path, "SUB2")
|
|---|
| 407 | tmp1_path = join(walk_path, "tmp1")
|
|---|
| 408 | tmp2_path = join(sub1_path, "tmp2")
|
|---|
| 409 | tmp3_path = join(sub2_path, "tmp3")
|
|---|
| 410 | link_path = join(sub2_path, "link")
|
|---|
| 411 | t2_path = join(test_support.TESTFN, "TEST2")
|
|---|
| 412 | tmp4_path = join(test_support.TESTFN, "TEST2", "tmp4")
|
|---|
| 413 |
|
|---|
| 414 | # Create stuff.
|
|---|
| 415 | os.makedirs(sub11_path)
|
|---|
| 416 | os.makedirs(sub2_path)
|
|---|
| 417 | os.makedirs(t2_path)
|
|---|
| 418 | for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path:
|
|---|
| 419 | f = file(path, "w")
|
|---|
| 420 | f.write("I'm " + path + " and proud of it. Blame test_os.\n")
|
|---|
| 421 | f.close()
|
|---|
| 422 | if hasattr(os, "symlink"):
|
|---|
| 423 | os.symlink(os.path.abspath(t2_path), link_path)
|
|---|
| 424 | sub2_tree = (sub2_path, ["link"], ["tmp3"])
|
|---|
| 425 | else:
|
|---|
| 426 | sub2_tree = (sub2_path, [], ["tmp3"])
|
|---|
| 427 |
|
|---|
| 428 | # Walk top-down.
|
|---|
| 429 | all = list(os.walk(walk_path))
|
|---|
| 430 | self.assertEqual(len(all), 4)
|
|---|
| 431 | # We can't know which order SUB1 and SUB2 will appear in.
|
|---|
| 432 | # Not flipped: TESTFN, SUB1, SUB11, SUB2
|
|---|
| 433 | # flipped: TESTFN, SUB2, SUB1, SUB11
|
|---|
| 434 | flipped = all[0][1][0] != "SUB1"
|
|---|
| 435 | all[0][1].sort()
|
|---|
| 436 | self.assertEqual(all[0], (walk_path, ["SUB1", "SUB2"], ["tmp1"]))
|
|---|
| 437 | self.assertEqual(all[1 + flipped], (sub1_path, ["SUB11"], ["tmp2"]))
|
|---|
| 438 | self.assertEqual(all[2 + flipped], (sub11_path, [], []))
|
|---|
| 439 | self.assertEqual(all[3 - 2 * flipped], sub2_tree)
|
|---|
| 440 |
|
|---|
| 441 | # Prune the search.
|
|---|
| 442 | all = []
|
|---|
| 443 | for root, dirs, files in os.walk(walk_path):
|
|---|
| 444 | all.append((root, dirs, files))
|
|---|
| 445 | # Don't descend into SUB1.
|
|---|
| 446 | if 'SUB1' in dirs:
|
|---|
| 447 | # Note that this also mutates the dirs we appended to all!
|
|---|
| 448 | dirs.remove('SUB1')
|
|---|
| 449 | self.assertEqual(len(all), 2)
|
|---|
| 450 | self.assertEqual(all[0], (walk_path, ["SUB2"], ["tmp1"]))
|
|---|
| 451 | self.assertEqual(all[1], sub2_tree)
|
|---|
| 452 |
|
|---|
| 453 | # Walk bottom-up.
|
|---|
| 454 | all = list(os.walk(walk_path, topdown=False))
|
|---|
| 455 | self.assertEqual(len(all), 4)
|
|---|
| 456 | # We can't know which order SUB1 and SUB2 will appear in.
|
|---|
| 457 | # Not flipped: SUB11, SUB1, SUB2, TESTFN
|
|---|
| 458 | # flipped: SUB2, SUB11, SUB1, TESTFN
|
|---|
| 459 | flipped = all[3][1][0] != "SUB1"
|
|---|
| 460 | all[3][1].sort()
|
|---|
| 461 | self.assertEqual(all[3], (walk_path, ["SUB1", "SUB2"], ["tmp1"]))
|
|---|
| 462 | self.assertEqual(all[flipped], (sub11_path, [], []))
|
|---|
| 463 | self.assertEqual(all[flipped + 1], (sub1_path, ["SUB11"], ["tmp2"]))
|
|---|
| 464 | self.assertEqual(all[2 - 2 * flipped], sub2_tree)
|
|---|
| 465 |
|
|---|
| 466 | if hasattr(os, "symlink"):
|
|---|
| 467 | # Walk, following symlinks.
|
|---|
| 468 | for root, dirs, files in os.walk(walk_path, followlinks=True):
|
|---|
| 469 | if root == link_path:
|
|---|
| 470 | self.assertEqual(dirs, [])
|
|---|
| 471 | self.assertEqual(files, ["tmp4"])
|
|---|
| 472 | break
|
|---|
| 473 | else:
|
|---|
| 474 | self.fail("Didn't follow symlink with followlinks=True")
|
|---|
| 475 |
|
|---|
| 476 | def tearDown(self):
|
|---|
| 477 | # Tear everything down. This is a decent use for bottom-up on
|
|---|
| 478 | # Windows, which doesn't have a recursive delete command. The
|
|---|
| 479 | # (not so) subtlety is that rmdir will fail unless the dir's
|
|---|
| 480 | # kids are removed first, so bottom up is essential.
|
|---|
| 481 | for root, dirs, files in os.walk(test_support.TESTFN, topdown=False):
|
|---|
| 482 | for name in files:
|
|---|
| 483 | os.remove(os.path.join(root, name))
|
|---|
| 484 | for name in dirs:
|
|---|
| 485 | dirname = os.path.join(root, name)
|
|---|
| 486 | if not os.path.islink(dirname):
|
|---|
| 487 | os.rmdir(dirname)
|
|---|
| 488 | else:
|
|---|
| 489 | os.remove(dirname)
|
|---|
| 490 | os.rmdir(test_support.TESTFN)
|
|---|
| 491 |
|
|---|
| 492 | class MakedirTests (unittest.TestCase):
|
|---|
| 493 | def setUp(self):
|
|---|
| 494 | os.mkdir(test_support.TESTFN)
|
|---|
| 495 |
|
|---|
| 496 | def test_makedir(self):
|
|---|
| 497 | base = test_support.TESTFN
|
|---|
| 498 | path = os.path.join(base, 'dir1', 'dir2', 'dir3')
|
|---|
| 499 | os.makedirs(path) # Should work
|
|---|
| 500 | path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4')
|
|---|
| 501 | os.makedirs(path)
|
|---|
| 502 |
|
|---|
| 503 | # Try paths with a '.' in them
|
|---|
| 504 | self.assertRaises(OSError, os.makedirs, os.curdir)
|
|---|
| 505 | path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', os.curdir)
|
|---|
| 506 | os.makedirs(path)
|
|---|
| 507 | path = os.path.join(base, 'dir1', os.curdir, 'dir2', 'dir3', 'dir4',
|
|---|
| 508 | 'dir5', 'dir6')
|
|---|
| 509 | os.makedirs(path)
|
|---|
| 510 |
|
|---|
| 511 |
|
|---|
| 512 |
|
|---|
| 513 |
|
|---|
| 514 | def tearDown(self):
|
|---|
| 515 | path = os.path.join(test_support.TESTFN, 'dir1', 'dir2', 'dir3',
|
|---|
| 516 | 'dir4', 'dir5', 'dir6')
|
|---|
| 517 | # If the tests failed, the bottom-most directory ('../dir6')
|
|---|
| 518 | # may not have been created, so we look for the outermost directory
|
|---|
| 519 | # that exists.
|
|---|
| 520 | while not os.path.exists(path) and path != test_support.TESTFN:
|
|---|
| 521 | path = os.path.dirname(path)
|
|---|
| 522 |
|
|---|
| 523 | os.removedirs(path)
|
|---|
| 524 |
|
|---|
| 525 | class DevNullTests (unittest.TestCase):
|
|---|
| 526 | def test_devnull(self):
|
|---|
| 527 | f = file(os.devnull, 'w')
|
|---|
| 528 | f.write('hello')
|
|---|
| 529 | f.close()
|
|---|
| 530 | f = file(os.devnull, 'r')
|
|---|
| 531 | self.assertEqual(f.read(), '')
|
|---|
| 532 | f.close()
|
|---|
| 533 |
|
|---|
| 534 | class URandomTests (unittest.TestCase):
|
|---|
| 535 |
|
|---|
| 536 | def test_urandom_length(self):
|
|---|
| 537 | self.assertEqual(len(os.urandom(0)), 0)
|
|---|
| 538 | self.assertEqual(len(os.urandom(1)), 1)
|
|---|
| 539 | self.assertEqual(len(os.urandom(10)), 10)
|
|---|
| 540 | self.assertEqual(len(os.urandom(100)), 100)
|
|---|
| 541 | self.assertEqual(len(os.urandom(1000)), 1000)
|
|---|
| 542 |
|
|---|
| 543 | def test_urandom_value(self):
|
|---|
| 544 | data1 = os.urandom(16)
|
|---|
| 545 | data2 = os.urandom(16)
|
|---|
| 546 | self.assertNotEqual(data1, data2)
|
|---|
| 547 |
|
|---|
| 548 | def get_urandom_subprocess(self, count):
|
|---|
| 549 | # We need to use repr() and eval() to avoid line ending conversions
|
|---|
| 550 | # under Windows.
|
|---|
| 551 | code = '\n'.join((
|
|---|
| 552 | 'import os, sys',
|
|---|
| 553 | 'data = os.urandom(%s)' % count,
|
|---|
| 554 | 'sys.stdout.write(repr(data))',
|
|---|
| 555 | 'sys.stdout.flush()',
|
|---|
| 556 | 'print >> sys.stderr, (len(data), data)'))
|
|---|
| 557 | cmd_line = [sys.executable, '-c', code]
|
|---|
| 558 | p = subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
|
|---|
| 559 | stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|---|
| 560 | out, err = p.communicate()
|
|---|
| 561 | self.assertEqual(p.wait(), 0, (p.wait(), err))
|
|---|
| 562 | out = eval(out)
|
|---|
| 563 | self.assertEqual(len(out), count, err)
|
|---|
| 564 | return out
|
|---|
| 565 |
|
|---|
| 566 | def test_urandom_subprocess(self):
|
|---|
| 567 | data1 = self.get_urandom_subprocess(16)
|
|---|
| 568 | data2 = self.get_urandom_subprocess(16)
|
|---|
| 569 | self.assertNotEqual(data1, data2)
|
|---|
| 570 |
|
|---|
| 571 | @unittest.skipUnless(resource, "test requires the resource module")
|
|---|
| 572 | def test_urandom_failure(self):
|
|---|
| 573 | # Check urandom() failing when it is not able to open /dev/random.
|
|---|
| 574 | # We spawn a new process to make the test more robust (if getrlimit()
|
|---|
| 575 | # failed to restore the file descriptor limit after this, the whole
|
|---|
| 576 | # test suite would crash; this actually happened on the OS X Tiger
|
|---|
| 577 | # buildbot).
|
|---|
| 578 | code = """if 1:
|
|---|
| 579 | import errno
|
|---|
| 580 | import os
|
|---|
| 581 | import resource
|
|---|
| 582 |
|
|---|
| 583 | soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
|
|---|
| 584 | resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
|
|---|
| 585 | try:
|
|---|
| 586 | os.urandom(16)
|
|---|
| 587 | except OSError as e:
|
|---|
| 588 | assert e.errno == errno.EMFILE, e.errno
|
|---|
| 589 | else:
|
|---|
| 590 | raise AssertionError("OSError not raised")
|
|---|
| 591 | """
|
|---|
| 592 | assert_python_ok('-c', code)
|
|---|
| 593 |
|
|---|
| 594 |
|
|---|
| 595 | class ExecvpeTests(unittest.TestCase):
|
|---|
| 596 |
|
|---|
| 597 | def test_execvpe_with_bad_arglist(self):
|
|---|
| 598 | self.assertRaises(ValueError, os.execvpe, 'notepad', [], None)
|
|---|
| 599 |
|
|---|
| 600 |
|
|---|
| 601 | class Win32ErrorTests(unittest.TestCase):
|
|---|
| 602 | def test_rename(self):
|
|---|
| 603 | self.assertRaises(WindowsError, os.rename, test_support.TESTFN, test_support.TESTFN+".bak")
|
|---|
| 604 |
|
|---|
| 605 | def test_remove(self):
|
|---|
| 606 | self.assertRaises(WindowsError, os.remove, test_support.TESTFN)
|
|---|
| 607 |
|
|---|
| 608 | def test_chdir(self):
|
|---|
| 609 | self.assertRaises(WindowsError, os.chdir, test_support.TESTFN)
|
|---|
| 610 |
|
|---|
| 611 | def test_mkdir(self):
|
|---|
| 612 | f = open(test_support.TESTFN, "w")
|
|---|
| 613 | try:
|
|---|
| 614 | self.assertRaises(WindowsError, os.mkdir, test_support.TESTFN)
|
|---|
| 615 | finally:
|
|---|
| 616 | f.close()
|
|---|
| 617 | os.unlink(test_support.TESTFN)
|
|---|
| 618 |
|
|---|
| 619 | def test_utime(self):
|
|---|
| 620 | self.assertRaises(WindowsError, os.utime, test_support.TESTFN, None)
|
|---|
| 621 |
|
|---|
| 622 | def test_chmod(self):
|
|---|
| 623 | self.assertRaises(WindowsError, os.chmod, test_support.TESTFN, 0)
|
|---|
| 624 |
|
|---|
| 625 | class TestInvalidFD(unittest.TestCase):
|
|---|
| 626 | singles = ["fchdir", "fdopen", "dup", "fdatasync", "fstat",
|
|---|
| 627 | "fstatvfs", "fsync", "tcgetpgrp", "ttyname"]
|
|---|
| 628 | #singles.append("close")
|
|---|
| 629 | #We omit close because it doesn'r raise an exception on some platforms
|
|---|
| 630 | def get_single(f):
|
|---|
| 631 | def helper(self):
|
|---|
| 632 | if hasattr(os, f):
|
|---|
| 633 | self.check(getattr(os, f))
|
|---|
| 634 | return helper
|
|---|
| 635 | for f in singles:
|
|---|
| 636 | locals()["test_"+f] = get_single(f)
|
|---|
| 637 |
|
|---|
| 638 | def check(self, f, *args):
|
|---|
| 639 | try:
|
|---|
| 640 | f(test_support.make_bad_fd(), *args)
|
|---|
| 641 | except OSError as e:
|
|---|
| 642 | self.assertEqual(e.errno, errno.EBADF)
|
|---|
| 643 | else:
|
|---|
| 644 | self.fail("%r didn't raise a OSError with a bad file descriptor"
|
|---|
| 645 | % f)
|
|---|
| 646 |
|
|---|
| 647 | def test_isatty(self):
|
|---|
| 648 | if hasattr(os, "isatty"):
|
|---|
| 649 | self.assertEqual(os.isatty(test_support.make_bad_fd()), False)
|
|---|
| 650 |
|
|---|
| 651 | def test_closerange(self):
|
|---|
| 652 | if hasattr(os, "closerange"):
|
|---|
| 653 | fd = test_support.make_bad_fd()
|
|---|
| 654 | # Make sure none of the descriptors we are about to close are
|
|---|
| 655 | # currently valid (issue 6542).
|
|---|
| 656 | for i in range(10):
|
|---|
| 657 | try: os.fstat(fd+i)
|
|---|
| 658 | except OSError:
|
|---|
| 659 | pass
|
|---|
| 660 | else:
|
|---|
| 661 | break
|
|---|
| 662 | if i < 2:
|
|---|
| 663 | raise unittest.SkipTest(
|
|---|
| 664 | "Unable to acquire a range of invalid file descriptors")
|
|---|
| 665 | self.assertEqual(os.closerange(fd, fd + i-1), None)
|
|---|
| 666 |
|
|---|
| 667 | def test_dup2(self):
|
|---|
| 668 | if hasattr(os, "dup2"):
|
|---|
| 669 | self.check(os.dup2, 20)
|
|---|
| 670 |
|
|---|
| 671 | def test_fchmod(self):
|
|---|
| 672 | if hasattr(os, "fchmod"):
|
|---|
| 673 | self.check(os.fchmod, 0)
|
|---|
| 674 |
|
|---|
| 675 | def test_fchown(self):
|
|---|
| 676 | if hasattr(os, "fchown"):
|
|---|
| 677 | self.check(os.fchown, -1, -1)
|
|---|
| 678 |
|
|---|
| 679 | def test_fpathconf(self):
|
|---|
| 680 | if hasattr(os, "fpathconf"):
|
|---|
| 681 | self.check(os.fpathconf, "PC_NAME_MAX")
|
|---|
| 682 |
|
|---|
| 683 | def test_ftruncate(self):
|
|---|
| 684 | if hasattr(os, "ftruncate"):
|
|---|
| 685 | self.check(os.ftruncate, 0)
|
|---|
| 686 |
|
|---|
| 687 | def test_lseek(self):
|
|---|
| 688 | if hasattr(os, "lseek"):
|
|---|
| 689 | self.check(os.lseek, 0, 0)
|
|---|
| 690 |
|
|---|
| 691 | def test_read(self):
|
|---|
| 692 | if hasattr(os, "read"):
|
|---|
| 693 | self.check(os.read, 1)
|
|---|
| 694 |
|
|---|
| 695 | def test_tcsetpgrpt(self):
|
|---|
| 696 | if hasattr(os, "tcsetpgrp"):
|
|---|
| 697 | self.check(os.tcsetpgrp, 0)
|
|---|
| 698 |
|
|---|
| 699 | def test_write(self):
|
|---|
| 700 | if hasattr(os, "write"):
|
|---|
| 701 | self.check(os.write, " ")
|
|---|
| 702 |
|
|---|
| 703 | if sys.platform != 'win32':
|
|---|
| 704 | class Win32ErrorTests(unittest.TestCase):
|
|---|
| 705 | pass
|
|---|
| 706 |
|
|---|
| 707 | class PosixUidGidTests(unittest.TestCase):
|
|---|
| 708 | if hasattr(os, 'setuid'):
|
|---|
| 709 | def test_setuid(self):
|
|---|
| 710 | if os.getuid() != 0:
|
|---|
| 711 | self.assertRaises(os.error, os.setuid, 0)
|
|---|
| 712 | self.assertRaises(OverflowError, os.setuid, 1<<32)
|
|---|
| 713 |
|
|---|
| 714 | if hasattr(os, 'setgid'):
|
|---|
| 715 | def test_setgid(self):
|
|---|
| 716 | if os.getuid() != 0:
|
|---|
| 717 | self.assertRaises(os.error, os.setgid, 0)
|
|---|
| 718 | self.assertRaises(OverflowError, os.setgid, 1<<32)
|
|---|
| 719 |
|
|---|
| 720 | if hasattr(os, 'seteuid'):
|
|---|
| 721 | def test_seteuid(self):
|
|---|
| 722 | if os.getuid() != 0:
|
|---|
| 723 | self.assertRaises(os.error, os.seteuid, 0)
|
|---|
| 724 | self.assertRaises(OverflowError, os.seteuid, 1<<32)
|
|---|
| 725 |
|
|---|
| 726 | if hasattr(os, 'setegid'):
|
|---|
| 727 | def test_setegid(self):
|
|---|
| 728 | if os.getuid() != 0:
|
|---|
| 729 | self.assertRaises(os.error, os.setegid, 0)
|
|---|
| 730 | self.assertRaises(OverflowError, os.setegid, 1<<32)
|
|---|
| 731 |
|
|---|
| 732 | if hasattr(os, 'setreuid'):
|
|---|
| 733 | def test_setreuid(self):
|
|---|
| 734 | if os.getuid() != 0:
|
|---|
| 735 | self.assertRaises(os.error, os.setreuid, 0, 0)
|
|---|
| 736 | self.assertRaises(OverflowError, os.setreuid, 1<<32, 0)
|
|---|
| 737 | self.assertRaises(OverflowError, os.setreuid, 0, 1<<32)
|
|---|
| 738 |
|
|---|
| 739 | def test_setreuid_neg1(self):
|
|---|
| 740 | # Needs to accept -1. We run this in a subprocess to avoid
|
|---|
| 741 | # altering the test runner's process state (issue8045).
|
|---|
| 742 | subprocess.check_call([
|
|---|
| 743 | sys.executable, '-c',
|
|---|
| 744 | 'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
|
|---|
| 745 |
|
|---|
| 746 | if hasattr(os, 'setregid'):
|
|---|
| 747 | def test_setregid(self):
|
|---|
| 748 | if os.getuid() != 0:
|
|---|
| 749 | self.assertRaises(os.error, os.setregid, 0, 0)
|
|---|
| 750 | self.assertRaises(OverflowError, os.setregid, 1<<32, 0)
|
|---|
| 751 | self.assertRaises(OverflowError, os.setregid, 0, 1<<32)
|
|---|
| 752 |
|
|---|
| 753 | def test_setregid_neg1(self):
|
|---|
| 754 | # Needs to accept -1. We run this in a subprocess to avoid
|
|---|
| 755 | # altering the test runner's process state (issue8045).
|
|---|
| 756 | subprocess.check_call([
|
|---|
| 757 | sys.executable, '-c',
|
|---|
| 758 | 'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
|
|---|
| 759 | else:
|
|---|
| 760 | class PosixUidGidTests(unittest.TestCase):
|
|---|
| 761 | pass
|
|---|
| 762 |
|
|---|
| 763 | @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
|
|---|
| 764 | class Win32KillTests(unittest.TestCase):
|
|---|
| 765 | def _kill(self, sig):
|
|---|
| 766 | # Start sys.executable as a subprocess and communicate from the
|
|---|
| 767 | # subprocess to the parent that the interpreter is ready. When it
|
|---|
| 768 | # becomes ready, send *sig* via os.kill to the subprocess and check
|
|---|
| 769 | # that the return code is equal to *sig*.
|
|---|
| 770 | import ctypes
|
|---|
| 771 | from ctypes import wintypes
|
|---|
| 772 | import msvcrt
|
|---|
| 773 |
|
|---|
| 774 | # Since we can't access the contents of the process' stdout until the
|
|---|
| 775 | # process has exited, use PeekNamedPipe to see what's inside stdout
|
|---|
| 776 | # without waiting. This is done so we can tell that the interpreter
|
|---|
| 777 | # is started and running at a point where it could handle a signal.
|
|---|
| 778 | PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
|
|---|
| 779 | PeekNamedPipe.restype = wintypes.BOOL
|
|---|
| 780 | PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
|
|---|
| 781 | ctypes.POINTER(ctypes.c_char), # stdout buf
|
|---|
| 782 | wintypes.DWORD, # Buffer size
|
|---|
| 783 | ctypes.POINTER(wintypes.DWORD), # bytes read
|
|---|
| 784 | ctypes.POINTER(wintypes.DWORD), # bytes avail
|
|---|
| 785 | ctypes.POINTER(wintypes.DWORD)) # bytes left
|
|---|
| 786 | msg = "running"
|
|---|
| 787 | proc = subprocess.Popen([sys.executable, "-c",
|
|---|
| 788 | "import sys;"
|
|---|
| 789 | "sys.stdout.write('{}');"
|
|---|
| 790 | "sys.stdout.flush();"
|
|---|
| 791 | "input()".format(msg)],
|
|---|
| 792 | stdout=subprocess.PIPE,
|
|---|
| 793 | stderr=subprocess.PIPE,
|
|---|
| 794 | stdin=subprocess.PIPE)
|
|---|
| 795 | self.addCleanup(proc.stdout.close)
|
|---|
| 796 | self.addCleanup(proc.stderr.close)
|
|---|
| 797 | self.addCleanup(proc.stdin.close)
|
|---|
| 798 |
|
|---|
| 799 | count, max = 0, 100
|
|---|
| 800 | while count < max and proc.poll() is None:
|
|---|
| 801 | # Create a string buffer to store the result of stdout from the pipe
|
|---|
| 802 | buf = ctypes.create_string_buffer(len(msg))
|
|---|
| 803 | # Obtain the text currently in proc.stdout
|
|---|
| 804 | # Bytes read/avail/left are left as NULL and unused
|
|---|
| 805 | rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
|
|---|
| 806 | buf, ctypes.sizeof(buf), None, None, None)
|
|---|
| 807 | self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
|
|---|
| 808 | if buf.value:
|
|---|
| 809 | self.assertEqual(msg, buf.value)
|
|---|
| 810 | break
|
|---|
| 811 | time.sleep(0.1)
|
|---|
| 812 | count += 1
|
|---|
| 813 | else:
|
|---|
| 814 | self.fail("Did not receive communication from the subprocess")
|
|---|
| 815 |
|
|---|
| 816 | os.kill(proc.pid, sig)
|
|---|
| 817 | self.assertEqual(proc.wait(), sig)
|
|---|
| 818 |
|
|---|
| 819 | def test_kill_sigterm(self):
|
|---|
| 820 | # SIGTERM doesn't mean anything special, but make sure it works
|
|---|
| 821 | self._kill(signal.SIGTERM)
|
|---|
| 822 |
|
|---|
| 823 | def test_kill_int(self):
|
|---|
| 824 | # os.kill on Windows can take an int which gets set as the exit code
|
|---|
| 825 | self._kill(100)
|
|---|
| 826 |
|
|---|
| 827 | def _kill_with_event(self, event, name):
|
|---|
| 828 | tagname = "test_os_%s" % uuid.uuid1()
|
|---|
| 829 | m = mmap.mmap(-1, 1, tagname)
|
|---|
| 830 | m[0] = '0'
|
|---|
| 831 | # Run a script which has console control handling enabled.
|
|---|
| 832 | proc = subprocess.Popen([sys.executable,
|
|---|
| 833 | os.path.join(os.path.dirname(__file__),
|
|---|
| 834 | "win_console_handler.py"), tagname],
|
|---|
| 835 | creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
|
|---|
| 836 | # Let the interpreter startup before we send signals. See #3137.
|
|---|
| 837 | count, max = 0, 20
|
|---|
| 838 | while count < max and proc.poll() is None:
|
|---|
| 839 | if m[0] == '1':
|
|---|
| 840 | break
|
|---|
| 841 | time.sleep(0.5)
|
|---|
| 842 | count += 1
|
|---|
| 843 | else:
|
|---|
| 844 | self.fail("Subprocess didn't finish initialization")
|
|---|
| 845 | os.kill(proc.pid, event)
|
|---|
| 846 | # proc.send_signal(event) could also be done here.
|
|---|
| 847 | # Allow time for the signal to be passed and the process to exit.
|
|---|
| 848 | time.sleep(0.5)
|
|---|
| 849 | if not proc.poll():
|
|---|
| 850 | # Forcefully kill the process if we weren't able to signal it.
|
|---|
| 851 | os.kill(proc.pid, signal.SIGINT)
|
|---|
| 852 | self.fail("subprocess did not stop on {}".format(name))
|
|---|
| 853 |
|
|---|
| 854 | @unittest.skip("subprocesses aren't inheriting CTRL+C property")
|
|---|
| 855 | def test_CTRL_C_EVENT(self):
|
|---|
| 856 | from ctypes import wintypes
|
|---|
| 857 | import ctypes
|
|---|
| 858 |
|
|---|
| 859 | # Make a NULL value by creating a pointer with no argument.
|
|---|
| 860 | NULL = ctypes.POINTER(ctypes.c_int)()
|
|---|
| 861 | SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
|
|---|
| 862 | SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
|
|---|
| 863 | wintypes.BOOL)
|
|---|
| 864 | SetConsoleCtrlHandler.restype = wintypes.BOOL
|
|---|
| 865 |
|
|---|
| 866 | # Calling this with NULL and FALSE causes the calling process to
|
|---|
| 867 | # handle CTRL+C, rather than ignore it. This property is inherited
|
|---|
| 868 | # by subprocesses.
|
|---|
| 869 | SetConsoleCtrlHandler(NULL, 0)
|
|---|
| 870 |
|
|---|
| 871 | self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
|
|---|
| 872 |
|
|---|
| 873 | def test_CTRL_BREAK_EVENT(self):
|
|---|
| 874 | self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
|
|---|
| 875 |
|
|---|
| 876 |
|
|---|
| 877 | def test_main():
|
|---|
| 878 | test_support.run_unittest(
|
|---|
| 879 | FileTests,
|
|---|
| 880 | TemporaryFileTests,
|
|---|
| 881 | StatAttributeTests,
|
|---|
| 882 | EnvironTests,
|
|---|
| 883 | WalkTests,
|
|---|
| 884 | MakedirTests,
|
|---|
| 885 | DevNullTests,
|
|---|
| 886 | URandomTests,
|
|---|
| 887 | ExecvpeTests,
|
|---|
| 888 | Win32ErrorTests,
|
|---|
| 889 | TestInvalidFD,
|
|---|
| 890 | PosixUidGidTests,
|
|---|
| 891 | Win32KillTests
|
|---|
| 892 | )
|
|---|
| 893 |
|
|---|
| 894 | if __name__ == "__main__":
|
|---|
| 895 | test_main()
|
|---|
