test_resource.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179
  1. import contextlib
  2. import sys
  3. import unittest
  4. from test import support
  5. from test.support import import_helper
  6. from test.support import os_helper
  7. import time
  8. resource = import_helper.import_module('resource')
  9. # This test is checking a few specific problem spots with the resource module.
  10. class ResourceTest(unittest.TestCase):
  11. def test_args(self):
  12. self.assertRaises(TypeError, resource.getrlimit)
  13. self.assertRaises(TypeError, resource.getrlimit, 42, 42)
  14. self.assertRaises(TypeError, resource.setrlimit)
  15. self.assertRaises(TypeError, resource.setrlimit, 42, 42, 42)
  16. @unittest.skipIf(sys.platform == "vxworks",
  17. "setting RLIMIT_FSIZE is not supported on VxWorks")
  18. def test_fsize_ismax(self):
  19. try:
  20. (cur, max) = resource.getrlimit(resource.RLIMIT_FSIZE)
  21. except AttributeError:
  22. pass
  23. else:
  24. # RLIMIT_FSIZE should be RLIM_INFINITY, which will be a really big
  25. # number on a platform with large file support. On these platforms,
  26. # we need to test that the get/setrlimit functions properly convert
  27. # the number to a C long long and that the conversion doesn't raise
  28. # an error.
  29. self.assertEqual(resource.RLIM_INFINITY, max)
  30. resource.setrlimit(resource.RLIMIT_FSIZE, (cur, max))
  31. def test_fsize_enforced(self):
  32. try:
  33. (cur, max) = resource.getrlimit(resource.RLIMIT_FSIZE)
  34. except AttributeError:
  35. pass
  36. else:
  37. # Check to see what happens when the RLIMIT_FSIZE is small. Some
  38. # versions of Python were terminated by an uncaught SIGXFSZ, but
  39. # pythonrun.c has been fixed to ignore that exception. If so, the
  40. # write() should return EFBIG when the limit is exceeded.
  41. # At least one platform has an unlimited RLIMIT_FSIZE and attempts
  42. # to change it raise ValueError instead.
  43. try:
  44. try:
  45. resource.setrlimit(resource.RLIMIT_FSIZE, (1024, max))
  46. limit_set = True
  47. except ValueError:
  48. limit_set = False
  49. f = open(os_helper.TESTFN, "wb")
  50. try:
  51. f.write(b"X" * 1024)
  52. try:
  53. f.write(b"Y")
  54. f.flush()
  55. # On some systems (e.g., Ubuntu on hppa) the flush()
  56. # doesn't always cause the exception, but the close()
  57. # does eventually. Try flushing several times in
  58. # an attempt to ensure the file is really synced and
  59. # the exception raised.
  60. for i in range(5):
  61. time.sleep(.1)
  62. f.flush()
  63. except OSError:
  64. if not limit_set:
  65. raise
  66. if limit_set:
  67. # Close will attempt to flush the byte we wrote
  68. # Restore limit first to avoid getting a spurious error
  69. resource.setrlimit(resource.RLIMIT_FSIZE, (cur, max))
  70. finally:
  71. f.close()
  72. finally:
  73. if limit_set:
  74. resource.setrlimit(resource.RLIMIT_FSIZE, (cur, max))
  75. os_helper.unlink(os_helper.TESTFN)
  76. def test_fsize_toobig(self):
  77. # Be sure that setrlimit is checking for really large values
  78. too_big = 10**50
  79. try:
  80. (cur, max) = resource.getrlimit(resource.RLIMIT_FSIZE)
  81. except AttributeError:
  82. pass
  83. else:
  84. try:
  85. resource.setrlimit(resource.RLIMIT_FSIZE, (too_big, max))
  86. except (OverflowError, ValueError):
  87. pass
  88. try:
  89. resource.setrlimit(resource.RLIMIT_FSIZE, (max, too_big))
  90. except (OverflowError, ValueError):
  91. pass
  92. @unittest.skipUnless(hasattr(resource, "getrusage"), "needs getrusage")
  93. def test_getrusage(self):
  94. self.assertRaises(TypeError, resource.getrusage)
  95. self.assertRaises(TypeError, resource.getrusage, 42, 42)
  96. usageself = resource.getrusage(resource.RUSAGE_SELF)
  97. usagechildren = resource.getrusage(resource.RUSAGE_CHILDREN)
  98. # May not be available on all systems.
  99. try:
  100. usageboth = resource.getrusage(resource.RUSAGE_BOTH)
  101. except (ValueError, AttributeError):
  102. pass
  103. try:
  104. usage_thread = resource.getrusage(resource.RUSAGE_THREAD)
  105. except (ValueError, AttributeError):
  106. pass
  107. # Issue 6083: Reference counting bug
  108. @unittest.skipIf(sys.platform == "vxworks",
  109. "setting RLIMIT_CPU is not supported on VxWorks")
  110. def test_setrusage_refcount(self):
  111. try:
  112. limits = resource.getrlimit(resource.RLIMIT_CPU)
  113. except AttributeError:
  114. pass
  115. else:
  116. class BadSequence:
  117. def __len__(self):
  118. return 2
  119. def __getitem__(self, key):
  120. if key in (0, 1):
  121. return len(tuple(range(1000000)))
  122. raise IndexError
  123. resource.setrlimit(resource.RLIMIT_CPU, BadSequence())
  124. def test_pagesize(self):
  125. pagesize = resource.getpagesize()
  126. self.assertIsInstance(pagesize, int)
  127. self.assertGreaterEqual(pagesize, 0)
  128. @unittest.skipUnless(sys.platform == 'linux', 'test requires Linux')
  129. def test_linux_constants(self):
  130. for attr in ['MSGQUEUE', 'NICE', 'RTPRIO', 'RTTIME', 'SIGPENDING']:
  131. with contextlib.suppress(AttributeError):
  132. self.assertIsInstance(getattr(resource, 'RLIMIT_' + attr), int)
  133. def test_freebsd_contants(self):
  134. for attr in ['SWAP', 'SBSIZE', 'NPTS']:
  135. with contextlib.suppress(AttributeError):
  136. self.assertIsInstance(getattr(resource, 'RLIMIT_' + attr), int)
  137. @unittest.skipUnless(hasattr(resource, 'prlimit'), 'no prlimit')
  138. @support.requires_linux_version(2, 6, 36)
  139. def test_prlimit(self):
  140. self.assertRaises(TypeError, resource.prlimit)
  141. self.assertRaises(ProcessLookupError, resource.prlimit,
  142. -1, resource.RLIMIT_AS)
  143. limit = resource.getrlimit(resource.RLIMIT_AS)
  144. self.assertEqual(resource.prlimit(0, resource.RLIMIT_AS), limit)
  145. self.assertEqual(resource.prlimit(0, resource.RLIMIT_AS, limit),
  146. limit)
  147. # Issue 20191: Reference counting bug
  148. @unittest.skipUnless(hasattr(resource, 'prlimit'), 'no prlimit')
  149. @support.requires_linux_version(2, 6, 36)
  150. def test_prlimit_refcount(self):
  151. class BadSeq:
  152. def __len__(self):
  153. return 2
  154. def __getitem__(self, key):
  155. return limits[key] - 1 # new reference
  156. limits = resource.getrlimit(resource.RLIMIT_AS)
  157. self.assertEqual(resource.prlimit(0, resource.RLIMIT_AS, BadSeq()),
  158. limits)
  159. if __name__ == "__main__":
  160. unittest.main()