Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(204665)

Delta Between Two Patch Sets: Lib/test/test_capi.py

Issue 16510: Using appropriate checks in tests
Left Patch Set: Created 6 years, 3 months ago
Right Patch Set: Created 5 years, 10 months ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
Left: Side by side diff | Download
Right: Side by side diff | Download
« no previous file with change/comment | « Lib/test/test_bz2.py ('k') | Lib/test/test_cmd_line.py » ('j') | no next file with change/comment »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
LEFTRIGHT
1 # Run the _testcapi module tests (tests for the Python/C API): by defn, 1 # Run the _testcapi module tests (tests for the Python/C API): by defn,
2 # these are all functions _testcapi exports whose name begins with 'test_'. 2 # these are all functions _testcapi exports whose name begins with 'test_'.
3 3
4 from __future__ import with_statement
5 import os 4 import os
6 import pickle 5 import pickle
7 import random 6 import random
8 import subprocess 7 import subprocess
9 import sys 8 import sys
10 import time 9 import time
11 import unittest 10 import unittest
12 from test import support 11 from test import support
12 from test.support import MISSING_C_DOCSTRINGS
13 try: 13 try:
14 import _posixsubprocess 14 import _posixsubprocess
15 except ImportError: 15 except ImportError:
16 _posixsubprocess = None 16 _posixsubprocess = None
17 try: 17 try:
18 import threading 18 import threading
19 except ImportError: 19 except ImportError:
20 threading = None 20 threading = None
21 import _testcapi 21 # Skip this test if the _testcapi module isn't available.
22 _testcapi = support.import_module('_testcapi')
22 23
23 24
24 def testfunction(self): 25 def testfunction(self):
25 """some doc""" 26 """some doc"""
26 return self 27 return self
27 28
28 class InstanceMethod: 29 class InstanceMethod:
29 id = _testcapi.instancemethod(id) 30 id = _testcapi.instancemethod(id)
30 testfunction = _testcapi.instancemethod(testfunction) 31 testfunction = _testcapi.instancemethod(testfunction)
31 32
32 class CAPITest(unittest.TestCase): 33 class CAPITest(unittest.TestCase):
33 34
34 def test_instancemethod(self): 35 def test_instancemethod(self):
35 inst = InstanceMethod() 36 inst = InstanceMethod()
36 self.assertEqual(id(inst), inst.id()) 37 self.assertEqual(id(inst), inst.id())
37 self.assertIs(inst.testfunction(), inst) 38 self.assertIs(inst.testfunction(), inst)
38 self.assertEqual(inst.testfunction.__doc__, testfunction.__doc__) 39 self.assertEqual(inst.testfunction.__doc__, testfunction.__doc__)
39 self.assertEqual(InstanceMethod.testfunction.__doc__, testfunction.__doc __) 40 self.assertEqual(InstanceMethod.testfunction.__doc__, testfunction.__doc __)
40 41
41 InstanceMethod.testfunction.attribute = "test" 42 InstanceMethod.testfunction.attribute = "test"
42 self.assertEqual(testfunction.attribute, "test") 43 self.assertEqual(testfunction.attribute, "test")
43 self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute ", "test") 44 self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute ", "test")
44 45
45 @unittest.skipUnless(threading, 'Threading required for this test.') 46 @unittest.skipUnless(threading, 'Threading required for this test.')
46 def test_no_FatalError_infinite_loop(self): 47 def test_no_FatalError_infinite_loop(self):
47 p = subprocess.Popen([sys.executable, "-c", 48 with support.SuppressCrashReport():
48 'import _testcapi;' 49 p = subprocess.Popen([sys.executable, "-c",
49 '_testcapi.crash_no_current_thread()'], 50 'import _testcapi;'
50 stdout=subprocess.PIPE, 51 '_testcapi.crash_no_current_thread()'],
51 stderr=subprocess.PIPE) 52 stdout=subprocess.PIPE,
53 stderr=subprocess.PIPE)
52 (out, err) = p.communicate() 54 (out, err) = p.communicate()
53 self.assertEqual(out, b'') 55 self.assertEqual(out, b'')
54 # This used to cause an infinite loop. 56 # This used to cause an infinite loop.
55 self.assertTrue(err.rstrip().startswith( 57 self.assertTrue(err.rstrip().startswith(
56 b'Fatal Python error:' 58 b'Fatal Python error:'
57 b' PyThreadState_Get: no current thread')) 59 b' PyThreadState_Get: no current thread'))
58 60
59 def test_memoryview_from_NULL_pointer(self): 61 def test_memoryview_from_NULL_pointer(self):
60 self.assertRaises(ValueError, _testcapi.make_memoryview_from_NULL_pointe r) 62 self.assertRaises(ValueError, _testcapi.make_memoryview_from_NULL_pointe r)
61 63
(...skipping 39 matching lines...) Expand 10 before | Expand all | Expand 10 after
101 103
102 @unittest.skipUnless(_posixsubprocess, '_posixsubprocess required for this t est.') 104 @unittest.skipUnless(_posixsubprocess, '_posixsubprocess required for this t est.')
103 def test_subprocess_fork_exec(self): 105 def test_subprocess_fork_exec(self):
104 class Z(object): 106 class Z(object):
105 def __len__(self): 107 def __len__(self):
106 return 1 108 return 1
107 109
108 # Issue #15738: crash in subprocess_fork_exec() 110 # Issue #15738: crash in subprocess_fork_exec()
109 self.assertRaises(TypeError, _posixsubprocess.fork_exec, 111 self.assertRaises(TypeError, _posixsubprocess.fork_exec,
110 Z(),[b'1'],3,[1, 2],5,6,7,8,9,10,11,12,13,14,15,16,17) 112 Z(),[b'1'],3,[1, 2],5,6,7,8,9,10,11,12,13,14,15,16,17)
113
114 @unittest.skipIf(MISSING_C_DOCSTRINGS,
115 "Signature information for builtins requires docstrings")
116 def test_docstring_signature_parsing(self):
117
118 self.assertEqual(_testcapi.no_docstring.__doc__, None)
119 self.assertEqual(_testcapi.no_docstring.__text_signature__, None)
120
121 self.assertEqual(_testcapi.docstring_empty.__doc__, "")
122 self.assertEqual(_testcapi.docstring_empty.__text_signature__, None)
123
124 self.assertEqual(_testcapi.docstring_no_signature.__doc__,
125 "This docstring has no signature.")
126 self.assertEqual(_testcapi.docstring_no_signature.__text_signature__, No ne)
127
128 self.assertEqual(_testcapi.docstring_with_invalid_signature.__doc__,
129 "sig= (module, boo)\n"
130 "\n"
131 "This docstring has an invalid signature."
132 )
133 self.assertEqual(_testcapi.docstring_with_invalid_signature.__text_signa ture__, None)
134
135 self.assertEqual(_testcapi.docstring_with_signature.__doc__,
136 "This docstring has a valid signature.")
137 self.assertEqual(_testcapi.docstring_with_signature.__text_signature__, "(module, sig)")
138
139 self.assertEqual(_testcapi.docstring_with_signature_and_extra_newlines._ _doc__,
140 "This docstring has a valid signature and some extra newlines.")
141 self.assertEqual(_testcapi.docstring_with_signature_and_extra_newlines._ _text_signature__,
142 "(module, parameter)")
143
111 144
112 @unittest.skipUnless(threading, 'Threading required for this test.') 145 @unittest.skipUnless(threading, 'Threading required for this test.')
113 class TestPendingCalls(unittest.TestCase): 146 class TestPendingCalls(unittest.TestCase):
114 147
115 def pendingcalls_submit(self, l, n): 148 def pendingcalls_submit(self, l, n):
116 def callback(): 149 def callback():
117 #this function can be interrupted by thread switching so let's 150 #this function can be interrupted by thread switching so let's
118 #use an atomic operation 151 #use an atomic operation
119 l.append(None) 152 l.append(None)
120 153
(...skipping 64 matching lines...) Expand 10 before | Expand all | Expand 10 after
185 def test_pendingcalls_non_threaded(self): 218 def test_pendingcalls_non_threaded(self):
186 #again, just using the main thread, likely they will all be dispatched a t 219 #again, just using the main thread, likely they will all be dispatched a t
187 #once. It is ok to ask for too many, because we loop until we find a sl ot. 220 #once. It is ok to ask for too many, because we loop until we find a sl ot.
188 #the loop can be interrupted to dispatch. 221 #the loop can be interrupted to dispatch.
189 #there are only 32 dispatch slots, so we go for twice that! 222 #there are only 32 dispatch slots, so we go for twice that!
190 l = [] 223 l = []
191 n = 64 224 n = 64
192 self.pendingcalls_submit(l, n) 225 self.pendingcalls_submit(l, n)
193 self.pendingcalls_wait(l, n) 226 self.pendingcalls_wait(l, n)
194 227
228
229 class SubinterpreterTest(unittest.TestCase):
230
195 def test_subinterps(self): 231 def test_subinterps(self):
196 # XXX this test leaks in refleak runs
197 import builtins 232 import builtins
198 r, w = os.pipe() 233 r, w = os.pipe()
199 code = """if 1: 234 code = """if 1:
200 import sys, builtins, pickle 235 import sys, builtins, pickle
201 with open({:d}, "wb") as f: 236 with open({:d}, "wb") as f:
202 pickle.dump(id(sys.modules), f) 237 pickle.dump(id(sys.modules), f)
203 pickle.dump(id(builtins), f) 238 pickle.dump(id(builtins), f)
204 """.format(w) 239 """.format(w)
205 with open(r, "rb") as f: 240 with open(r, "rb") as f:
206 ret = _testcapi.run_in_subinterp(code) 241 ret = support.run_in_subinterp(code)
207 self.assertEqual(ret, 0) 242 self.assertEqual(ret, 0)
208 self.assertNotEqual(pickle.load(f), id(sys.modules)) 243 self.assertNotEqual(pickle.load(f), id(sys.modules))
209 self.assertNotEqual(pickle.load(f), id(builtins)) 244 self.assertNotEqual(pickle.load(f), id(builtins))
245
210 246
211 # Bug #6012 247 # Bug #6012
212 class Test6012(unittest.TestCase): 248 class Test6012(unittest.TestCase):
213 def test(self): 249 def test(self):
214 self.assertEqual(_testcapi.argparsing("Hello", "World"), 1) 250 self.assertEqual(_testcapi.argparsing("Hello", "World"), 1)
215 251
216 252
217 class EmbeddingTest(unittest.TestCase): 253 class EmbeddingTests(unittest.TestCase):
218 254 def setUp(self):
219 @unittest.skipIf(
220 sys.platform.startswith('win'),
221 "test doesn't work under Windows")
222 def test_subinterps(self):
223 # XXX only tested under Unix checkouts
224 basepath = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) 255 basepath = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
225 oldcwd = os.getcwd() 256 exename = "_testembed"
257 if sys.platform.startswith("win"):
258 ext = ("_d" if "_d" in sys.executable else "") + ".exe"
259 exename += ext
260 exepath = os.path.dirname(sys.executable)
261 else:
262 exepath = os.path.join(basepath, "Modules")
263 self.test_exe = exe = os.path.join(exepath, exename)
264 if not os.path.exists(exe):
265 self.skipTest("%r doesn't exist" % exe)
226 # This is needed otherwise we get a fatal error: 266 # This is needed otherwise we get a fatal error:
227 # "Py_Initialize: Unable to get the locale encoding 267 # "Py_Initialize: Unable to get the locale encoding
228 # LookupError: no codec search functions registered: can't find encoding " 268 # LookupError: no codec search functions registered: can't find encoding "
269 self.oldcwd = os.getcwd()
229 os.chdir(basepath) 270 os.chdir(basepath)
271
272 def tearDown(self):
273 os.chdir(self.oldcwd)
274
275 def run_embedded_interpreter(self, *args):
276 """Runs a test in the embedded interpreter"""
277 cmd = [self.test_exe]
278 cmd.extend(args)
279 p = subprocess.Popen(cmd,
280 stdout=subprocess.PIPE,
281 stderr=subprocess.PIPE)
282 (out, err) = p.communicate()
283 self.assertEqual(p.returncode, 0,
284 "bad returncode %d, stderr is %r" %
285 (p.returncode, err))
286 return out.decode("latin1"), err.decode("latin1")
287
288 def test_subinterps(self):
289 # This is just a "don't crash" test
290 out, err = self.run_embedded_interpreter()
291 if support.verbose:
292 print()
293 print(out)
294 print(err)
295
296 @staticmethod
297 def _get_default_pipe_encoding():
298 rp, wp = os.pipe()
230 try: 299 try:
231 exe = os.path.join(basepath, "Modules", "_testembed") 300 with os.fdopen(wp, 'w') as w:
232 if not os.path.exists(exe): 301 default_pipe_encoding = w.encoding
233 self.skipTest("%r doesn't exist" % exe)
234 p = subprocess.Popen([exe],
235 stdout=subprocess.PIPE,
236 stderr=subprocess.PIPE)
237 (out, err) = p.communicate()
238 self.assertEqual(p.returncode, 0,
239 "bad returncode %d, stderr is %r" %
240 (p.returncode, err))
241 if support.verbose:
242 print()
243 print(out.decode('latin1'))
244 print(err.decode('latin1'))
245 finally: 302 finally:
246 os.chdir(oldcwd) 303 os.close(rp)
304 return default_pipe_encoding
305
306 def test_forced_io_encoding(self):
307 # Checks forced configuration of embedded interpreter IO streams
308 out, err = self.run_embedded_interpreter("forced_io_encoding")
309 if support.verbose:
310 print()
311 print(out)
312 print(err)
313 expected_stdin_encoding = sys.__stdin__.encoding
314 expected_pipe_encoding = self._get_default_pipe_encoding()
315 expected_output = os.linesep.join([
316 "--- Use defaults ---",
317 "Expected encoding: default",
318 "Expected errors: default",
319 "stdin: {0}:strict",
320 "stdout: {1}:strict",
321 "stderr: {1}:backslashreplace",
322 "--- Set errors only ---",
323 "Expected encoding: default",
324 "Expected errors: surrogateescape",
325 "stdin: {0}:surrogateescape",
326 "stdout: {1}:surrogateescape",
327 "stderr: {1}:backslashreplace",
328 "--- Set encoding only ---",
329 "Expected encoding: latin-1",
330 "Expected errors: default",
331 "stdin: latin-1:strict",
332 "stdout: latin-1:strict",
333 "stderr: latin-1:backslashreplace",
334 "--- Set encoding and errors ---",
335 "Expected encoding: latin-1",
336 "Expected errors: surrogateescape",
337 "stdin: latin-1:surrogateescape",
338 "stdout: latin-1:surrogateescape",
339 "stderr: latin-1:backslashreplace"]).format(expected_stdin_encoding,
340 expected_pipe_encoding)
341 # This is useful if we ever trip over odd platform behaviour
342 self.maxDiff = None
343 self.assertEqual(out.strip(), expected_output)
247 344
248 class SkipitemTest(unittest.TestCase): 345 class SkipitemTest(unittest.TestCase):
249 346
250 def test_skipitem(self): 347 def test_skipitem(self):
251 """ 348 """
252 If this test failed, you probably added a new "format unit" 349 If this test failed, you probably added a new "format unit"
253 in Python/getargs.c, but neglected to update our poor friend 350 in Python/getargs.c, but neglected to update our poor friend
254 skipitem() in the same file. (If so, shame on you!) 351 skipitem() in the same file. (If so, shame on you!)
255 352
256 With a few exceptions**, this function brute-force tests all 353 With a few exceptions**, this function brute-force tests all
(...skipping 63 matching lines...) Expand 10 before | Expand all | Expand 10 after
320 # parse_tuple_and_keywords error handling tests 417 # parse_tuple_and_keywords error handling tests
321 self.assertRaises(TypeError, _testcapi.parse_tuple_and_keywords, 418 self.assertRaises(TypeError, _testcapi.parse_tuple_and_keywords,
322 (), {}, 42, []) 419 (), {}, 42, [])
323 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords, 420 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
324 (), {}, b'', 42) 421 (), {}, b'', 42)
325 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords, 422 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
326 (), {}, b'', [''] * 42) 423 (), {}, b'', [''] * 42)
327 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords, 424 self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
328 (), {}, b'', [42]) 425 (), {}, b'', [42])
329 426
330 def test_main(): 427 @unittest.skipUnless(threading, 'Threading required for this test.')
331 support.run_unittest(CAPITest, TestPendingCalls, 428 class TestThreadState(unittest.TestCase):
332 Test6012, EmbeddingTest, SkipitemTest) 429
333 430 @support.reap_threads
334 for name in dir(_testcapi): 431 def test_thread_state(self):
335 if name.startswith('test_'): 432 # some extra thread-state tests driven via _testcapi
336 test = getattr(_testcapi, name) 433 def target():
337 if support.verbose: 434 idents = []
338 print("internal", name) 435
339 test() 436 def callback():
340 437 idents.append(threading.get_ident())
341 # some extra thread-state tests driven via _testcapi 438
342 def TestThreadState(): 439 _testcapi._test_thread_state(callback)
343 if support.verbose: 440 a = b = callback
344 print("auto-thread-state") 441 time.sleep(1)
345 442 # Check our main thread is in the list exactly 3 times.
346 idents = [] 443 self.assertEqual(idents.count(threading.get_ident()), 3,
347 444 "Couldn't find main thread correctly in the list")
348 def callback(): 445
349 idents.append(threading.get_ident()) 446 target()
350 447 t = threading.Thread(target=target)
351 _testcapi._test_thread_state(callback)
352 a = b = callback
353 time.sleep(1)
354 # Check our main thread is in the list exactly 3 times.
355 if idents.count(threading.get_ident()) != 3:
356 raise support.TestFailed(
357 "Couldn't find main thread correctly in the list")
358
359 if threading:
360 import time
361 TestThreadState()
362 t = threading.Thread(target=TestThreadState)
363 t.start() 448 t.start()
364 t.join() 449 t.join()
365 450
451 class Test_testcapi(unittest.TestCase):
452 def test__testcapi(self):
453 for name in dir(_testcapi):
454 if name.startswith('test_'):
455 with self.subTest("internal", name=name):
456 test = getattr(_testcapi, name)
457 test()
366 458
367 if __name__ == "__main__": 459 if __name__ == "__main__":
368 test_main() 460 unittest.main()
LEFTRIGHT

RSS Feeds Recent Issues | This issue
This is Rietveld 894c83f36cb7+