702 lines
29 KiB
Python
702 lines
29 KiB
Python
# Run the _testcapi module tests (tests for the Python/C API): by defn,
|
|
# these are all functions _testcapi exports whose name begins with 'test_'.
|
|
|
|
from collections import OrderedDict
|
|
import os
|
|
import pickle
|
|
import random
|
|
import re
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
import sysconfig
|
|
import textwrap
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from test import support
|
|
from test.support import MISSING_C_DOCSTRINGS
|
|
from test.support.script_helper import assert_python_failure, assert_python_ok
|
|
try:
|
|
import _posixsubprocess
|
|
except ImportError:
|
|
_posixsubprocess = None
|
|
|
|
# Skip this test if the _testcapi module isn't available.
|
|
_testcapi = support.import_module('_testcapi')
|
|
|
|
# Were we compiled --with-pydebug or with #define Py_DEBUG?
|
|
Py_DEBUG = hasattr(sys, 'gettotalrefcount')
|
|
|
|
|
|
def testfunction(self):
|
|
"""some doc"""
|
|
return self
|
|
|
|
class InstanceMethod:
|
|
id = _testcapi.instancemethod(id)
|
|
testfunction = _testcapi.instancemethod(testfunction)
|
|
|
|
class CAPITest(unittest.TestCase):
|
|
|
|
def test_instancemethod(self):
|
|
inst = InstanceMethod()
|
|
self.assertEqual(id(inst), inst.id())
|
|
self.assertTrue(inst.testfunction() is inst)
|
|
self.assertEqual(inst.testfunction.__doc__, testfunction.__doc__)
|
|
self.assertEqual(InstanceMethod.testfunction.__doc__, testfunction.__doc__)
|
|
|
|
InstanceMethod.testfunction.attribute = "test"
|
|
self.assertEqual(testfunction.attribute, "test")
|
|
self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute", "test")
|
|
|
|
def test_no_FatalError_infinite_loop(self):
|
|
with support.SuppressCrashReport():
|
|
p = subprocess.Popen([sys.executable, "-c",
|
|
'import _testcapi;'
|
|
'_testcapi.crash_no_current_thread()'],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
(out, err) = p.communicate()
|
|
self.assertEqual(out, b'')
|
|
# This used to cause an infinite loop.
|
|
self.assertTrue(err.rstrip().startswith(
|
|
b'Fatal Python error:'
|
|
b' PyThreadState_Get: no current thread'))
|
|
|
|
def test_memoryview_from_NULL_pointer(self):
|
|
self.assertRaises(ValueError, _testcapi.make_memoryview_from_NULL_pointer)
|
|
|
|
def test_exc_info(self):
|
|
raised_exception = ValueError("5")
|
|
new_exc = TypeError("TEST")
|
|
try:
|
|
raise raised_exception
|
|
except ValueError as e:
|
|
tb = e.__traceback__
|
|
orig_sys_exc_info = sys.exc_info()
|
|
orig_exc_info = _testcapi.set_exc_info(new_exc.__class__, new_exc, None)
|
|
new_sys_exc_info = sys.exc_info()
|
|
new_exc_info = _testcapi.set_exc_info(*orig_exc_info)
|
|
reset_sys_exc_info = sys.exc_info()
|
|
|
|
self.assertEqual(orig_exc_info[1], e)
|
|
|
|
self.assertSequenceEqual(orig_exc_info, (raised_exception.__class__, raised_exception, tb))
|
|
self.assertSequenceEqual(orig_sys_exc_info, orig_exc_info)
|
|
self.assertSequenceEqual(reset_sys_exc_info, orig_exc_info)
|
|
self.assertSequenceEqual(new_exc_info, (new_exc.__class__, new_exc, None))
|
|
self.assertSequenceEqual(new_sys_exc_info, new_exc_info)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
@unittest.skipUnless(_posixsubprocess, '_posixsubprocess required for this test.')
|
|
def test_seq_bytes_to_charp_array(self):
|
|
# Issue #15732: crash in _PySequence_BytesToCharpArray()
|
|
class Z(object):
|
|
def __len__(self):
|
|
return 1
|
|
self.assertRaises(TypeError, _posixsubprocess.fork_exec,
|
|
1,Z(),3,(1, 2),5,6,7,8,9,10,11,12,13,14,15,16,17)
|
|
# Issue #15736: overflow in _PySequence_BytesToCharpArray()
|
|
class Z(object):
|
|
def __len__(self):
|
|
return sys.maxsize
|
|
def __getitem__(self, i):
|
|
return b'x'
|
|
self.assertRaises(MemoryError, _posixsubprocess.fork_exec,
|
|
1,Z(),3,(1, 2),5,6,7,8,9,10,11,12,13,14,15,16,17)
|
|
|
|
@unittest.skipUnless(_posixsubprocess, '_posixsubprocess required for this test.')
|
|
def test_subprocess_fork_exec(self):
|
|
class Z(object):
|
|
def __len__(self):
|
|
return 1
|
|
|
|
# Issue #15738: crash in subprocess_fork_exec()
|
|
self.assertRaises(TypeError, _posixsubprocess.fork_exec,
|
|
Z(),[b'1'],3,(1, 2),5,6,7,8,9,10,11,12,13,14,15,16,17)
|
|
|
|
@unittest.skipIf(MISSING_C_DOCSTRINGS,
|
|
"Signature information for builtins requires docstrings")
|
|
def test_docstring_signature_parsing(self):
|
|
|
|
self.assertEqual(_testcapi.no_docstring.__doc__, None)
|
|
self.assertEqual(_testcapi.no_docstring.__text_signature__, None)
|
|
|
|
self.assertEqual(_testcapi.docstring_empty.__doc__, None)
|
|
self.assertEqual(_testcapi.docstring_empty.__text_signature__, None)
|
|
|
|
self.assertEqual(_testcapi.docstring_no_signature.__doc__,
|
|
"This docstring has no signature.")
|
|
self.assertEqual(_testcapi.docstring_no_signature.__text_signature__, None)
|
|
|
|
self.assertEqual(_testcapi.docstring_with_invalid_signature.__doc__,
|
|
"docstring_with_invalid_signature($module, /, boo)\n"
|
|
"\n"
|
|
"This docstring has an invalid signature."
|
|
)
|
|
self.assertEqual(_testcapi.docstring_with_invalid_signature.__text_signature__, None)
|
|
|
|
self.assertEqual(_testcapi.docstring_with_invalid_signature2.__doc__,
|
|
"docstring_with_invalid_signature2($module, /, boo)\n"
|
|
"\n"
|
|
"--\n"
|
|
"\n"
|
|
"This docstring also has an invalid signature."
|
|
)
|
|
self.assertEqual(_testcapi.docstring_with_invalid_signature2.__text_signature__, None)
|
|
|
|
self.assertEqual(_testcapi.docstring_with_signature.__doc__,
|
|
"This docstring has a valid signature.")
|
|
self.assertEqual(_testcapi.docstring_with_signature.__text_signature__, "($module, /, sig)")
|
|
|
|
self.assertEqual(_testcapi.docstring_with_signature_but_no_doc.__doc__, None)
|
|
self.assertEqual(_testcapi.docstring_with_signature_but_no_doc.__text_signature__,
|
|
"($module, /, sig)")
|
|
|
|
self.assertEqual(_testcapi.docstring_with_signature_and_extra_newlines.__doc__,
|
|
"\nThis docstring has a valid signature and some extra newlines.")
|
|
self.assertEqual(_testcapi.docstring_with_signature_and_extra_newlines.__text_signature__,
|
|
"($module, /, parameter)")
|
|
|
|
def test_c_type_with_matrix_multiplication(self):
|
|
M = _testcapi.matmulType
|
|
m1 = M()
|
|
m2 = M()
|
|
self.assertEqual(m1 @ m2, ("matmul", m1, m2))
|
|
self.assertEqual(m1 @ 42, ("matmul", m1, 42))
|
|
self.assertEqual(42 @ m1, ("matmul", 42, m1))
|
|
o = m1
|
|
o @= m2
|
|
self.assertEqual(o, ("imatmul", m1, m2))
|
|
o = m1
|
|
o @= 42
|
|
self.assertEqual(o, ("imatmul", m1, 42))
|
|
o = 42
|
|
o @= m1
|
|
self.assertEqual(o, ("matmul", 42, m1))
|
|
|
|
def test_return_null_without_error(self):
|
|
# Issue #23571: A function must not return NULL without setting an
|
|
# error
|
|
if Py_DEBUG:
|
|
code = textwrap.dedent("""
|
|
import _testcapi
|
|
from test import support
|
|
|
|
with support.SuppressCrashReport():
|
|
_testcapi.return_null_without_error()
|
|
""")
|
|
rc, out, err = assert_python_failure('-c', code)
|
|
self.assertRegex(err.replace(b'\r', b''),
|
|
br'Fatal Python error: a function returned NULL '
|
|
br'without setting an error\n'
|
|
br'SystemError: <built-in function '
|
|
br'return_null_without_error> returned NULL '
|
|
br'without setting an error\n'
|
|
br'\n'
|
|
br'Current thread.*:\n'
|
|
br' File .*", line 6 in <module>')
|
|
else:
|
|
with self.assertRaises(SystemError) as cm:
|
|
_testcapi.return_null_without_error()
|
|
self.assertRegex(str(cm.exception),
|
|
'return_null_without_error.* '
|
|
'returned NULL without setting an error')
|
|
|
|
def test_return_result_with_error(self):
|
|
# Issue #23571: A function must not return a result with an error set
|
|
if Py_DEBUG:
|
|
code = textwrap.dedent("""
|
|
import _testcapi
|
|
from test import support
|
|
|
|
with support.SuppressCrashReport():
|
|
_testcapi.return_result_with_error()
|
|
""")
|
|
rc, out, err = assert_python_failure('-c', code)
|
|
self.assertRegex(err.replace(b'\r', b''),
|
|
br'Fatal Python error: a function returned a '
|
|
br'result with an error set\n'
|
|
br'ValueError\n'
|
|
br'\n'
|
|
br'The above exception was the direct cause '
|
|
br'of the following exception:\n'
|
|
br'\n'
|
|
br'SystemError: <built-in '
|
|
br'function return_result_with_error> '
|
|
br'returned a result with an error set\n'
|
|
br'\n'
|
|
br'Current thread.*:\n'
|
|
br' File .*, line 6 in <module>')
|
|
else:
|
|
with self.assertRaises(SystemError) as cm:
|
|
_testcapi.return_result_with_error()
|
|
self.assertRegex(str(cm.exception),
|
|
'return_result_with_error.* '
|
|
'returned a result with an error set')
|
|
|
|
def test_buildvalue_N(self):
|
|
_testcapi.test_buildvalue_N()
|
|
|
|
def test_set_nomemory(self):
|
|
code = """if 1:
|
|
import _testcapi
|
|
|
|
class C(): pass
|
|
|
|
# The first loop tests both functions and that remove_mem_hooks()
|
|
# can be called twice in a row. The second loop checks a call to
|
|
# set_nomemory() after a call to remove_mem_hooks(). The third
|
|
# loop checks the start and stop arguments of set_nomemory().
|
|
for outer_cnt in range(1, 4):
|
|
start = 10 * outer_cnt
|
|
for j in range(100):
|
|
if j == 0:
|
|
if outer_cnt != 3:
|
|
_testcapi.set_nomemory(start)
|
|
else:
|
|
_testcapi.set_nomemory(start, start + 1)
|
|
try:
|
|
C()
|
|
except MemoryError as e:
|
|
if outer_cnt != 3:
|
|
_testcapi.remove_mem_hooks()
|
|
print('MemoryError', outer_cnt, j)
|
|
_testcapi.remove_mem_hooks()
|
|
break
|
|
"""
|
|
rc, out, err = assert_python_ok('-c', code)
|
|
self.assertIn(b'MemoryError 1 10', out)
|
|
self.assertIn(b'MemoryError 2 20', out)
|
|
self.assertIn(b'MemoryError 3 30', out)
|
|
|
|
def test_mapping_keys_values_items(self):
|
|
class Mapping1(dict):
|
|
def keys(self):
|
|
return list(super().keys())
|
|
def values(self):
|
|
return list(super().values())
|
|
def items(self):
|
|
return list(super().items())
|
|
class Mapping2(dict):
|
|
def keys(self):
|
|
return tuple(super().keys())
|
|
def values(self):
|
|
return tuple(super().values())
|
|
def items(self):
|
|
return tuple(super().items())
|
|
dict_obj = {'foo': 1, 'bar': 2, 'spam': 3}
|
|
|
|
for mapping in [{}, OrderedDict(), Mapping1(), Mapping2(),
|
|
dict_obj, OrderedDict(dict_obj),
|
|
Mapping1(dict_obj), Mapping2(dict_obj)]:
|
|
self.assertListEqual(_testcapi.get_mapping_keys(mapping),
|
|
list(mapping.keys()))
|
|
self.assertListEqual(_testcapi.get_mapping_values(mapping),
|
|
list(mapping.values()))
|
|
self.assertListEqual(_testcapi.get_mapping_items(mapping),
|
|
list(mapping.items()))
|
|
|
|
def test_mapping_keys_values_items_bad_arg(self):
|
|
self.assertRaises(AttributeError, _testcapi.get_mapping_keys, None)
|
|
self.assertRaises(AttributeError, _testcapi.get_mapping_values, None)
|
|
self.assertRaises(AttributeError, _testcapi.get_mapping_items, None)
|
|
|
|
class BadMapping:
|
|
def keys(self):
|
|
return None
|
|
def values(self):
|
|
return None
|
|
def items(self):
|
|
return None
|
|
bad_mapping = BadMapping()
|
|
self.assertRaises(TypeError, _testcapi.get_mapping_keys, bad_mapping)
|
|
self.assertRaises(TypeError, _testcapi.get_mapping_values, bad_mapping)
|
|
self.assertRaises(TypeError, _testcapi.get_mapping_items, bad_mapping)
|
|
|
|
|
|
class TestPendingCalls(unittest.TestCase):
|
|
|
|
def pendingcalls_submit(self, l, n):
|
|
def callback():
|
|
#this function can be interrupted by thread switching so let's
|
|
#use an atomic operation
|
|
l.append(None)
|
|
|
|
for i in range(n):
|
|
time.sleep(random.random()*0.02) #0.01 secs on average
|
|
#try submitting callback until successful.
|
|
#rely on regular interrupt to flush queue if we are
|
|
#unsuccessful.
|
|
while True:
|
|
if _testcapi._pending_threadfunc(callback):
|
|
break;
|
|
|
|
def pendingcalls_wait(self, l, n, context = None):
|
|
#now, stick around until l[0] has grown to 10
|
|
count = 0;
|
|
while len(l) != n:
|
|
#this busy loop is where we expect to be interrupted to
|
|
#run our callbacks. Note that callbacks are only run on the
|
|
#main thread
|
|
if False and support.verbose:
|
|
print("(%i)"%(len(l),),)
|
|
for i in range(1000):
|
|
a = i*i
|
|
if context and not context.event.is_set():
|
|
continue
|
|
count += 1
|
|
self.assertTrue(count < 10000,
|
|
"timeout waiting for %i callbacks, got %i"%(n, len(l)))
|
|
if False and support.verbose:
|
|
print("(%i)"%(len(l),))
|
|
|
|
def test_pendingcalls_threaded(self):
|
|
|
|
#do every callback on a separate thread
|
|
n = 32 #total callbacks
|
|
threads = []
|
|
class foo(object):pass
|
|
context = foo()
|
|
context.l = []
|
|
context.n = 2 #submits per thread
|
|
context.nThreads = n // context.n
|
|
context.nFinished = 0
|
|
context.lock = threading.Lock()
|
|
context.event = threading.Event()
|
|
|
|
threads = [threading.Thread(target=self.pendingcalls_thread,
|
|
args=(context,))
|
|
for i in range(context.nThreads)]
|
|
with support.start_threads(threads):
|
|
self.pendingcalls_wait(context.l, n, context)
|
|
|
|
def pendingcalls_thread(self, context):
|
|
try:
|
|
self.pendingcalls_submit(context.l, context.n)
|
|
finally:
|
|
with context.lock:
|
|
context.nFinished += 1
|
|
nFinished = context.nFinished
|
|
if False and support.verbose:
|
|
print("finished threads: ", nFinished)
|
|
if nFinished == context.nThreads:
|
|
context.event.set()
|
|
|
|
def test_pendingcalls_non_threaded(self):
|
|
#again, just using the main thread, likely they will all be dispatched at
|
|
#once. It is ok to ask for too many, because we loop until we find a slot.
|
|
#the loop can be interrupted to dispatch.
|
|
#there are only 32 dispatch slots, so we go for twice that!
|
|
l = []
|
|
n = 64
|
|
self.pendingcalls_submit(l, n)
|
|
self.pendingcalls_wait(l, n)
|
|
|
|
|
|
class SubinterpreterTest(unittest.TestCase):
|
|
|
|
def test_subinterps(self):
|
|
import builtins
|
|
r, w = os.pipe()
|
|
code = """if 1:
|
|
import sys, builtins, pickle
|
|
with open({:d}, "wb") as f:
|
|
pickle.dump(id(sys.modules), f)
|
|
pickle.dump(id(builtins), f)
|
|
""".format(w)
|
|
with open(r, "rb") as f:
|
|
ret = support.run_in_subinterp(code)
|
|
self.assertEqual(ret, 0)
|
|
self.assertNotEqual(pickle.load(f), id(sys.modules))
|
|
self.assertNotEqual(pickle.load(f), id(builtins))
|
|
|
|
|
|
# Bug #6012
|
|
class Test6012(unittest.TestCase):
|
|
def test(self):
|
|
self.assertEqual(_testcapi.argparsing("Hello", "World"), 1)
|
|
|
|
|
|
class SkipitemTest(unittest.TestCase):
|
|
|
|
def test_skipitem(self):
|
|
"""
|
|
If this test failed, you probably added a new "format unit"
|
|
in Python/getargs.c, but neglected to update our poor friend
|
|
skipitem() in the same file. (If so, shame on you!)
|
|
|
|
With a few exceptions**, this function brute-force tests all
|
|
printable ASCII*** characters (32 to 126 inclusive) as format units,
|
|
checking to see that PyArg_ParseTupleAndKeywords() return consistent
|
|
errors both when the unit is attempted to be used and when it is
|
|
skipped. If the format unit doesn't exist, we'll get one of two
|
|
specific error messages (one for used, one for skipped); if it does
|
|
exist we *won't* get that error--we'll get either no error or some
|
|
other error. If we get the specific "does not exist" error for one
|
|
test and not for the other, there's a mismatch, and the test fails.
|
|
|
|
** Some format units have special funny semantics and it would
|
|
be difficult to accommodate them here. Since these are all
|
|
well-established and properly skipped in skipitem() we can
|
|
get away with not testing them--this test is really intended
|
|
to catch *new* format units.
|
|
|
|
*** Python C source files must be ASCII. Therefore it's impossible
|
|
to have non-ASCII format units.
|
|
|
|
"""
|
|
empty_tuple = ()
|
|
tuple_1 = (0,)
|
|
dict_b = {'b':1}
|
|
keywords = ["a", "b"]
|
|
|
|
for i in range(32, 127):
|
|
c = chr(i)
|
|
|
|
# skip parentheses, the error reporting is inconsistent about them
|
|
# skip 'e', it's always a two-character code
|
|
# skip '|' and '$', they don't represent arguments anyway
|
|
if c in '()e|$':
|
|
continue
|
|
|
|
# test the format unit when not skipped
|
|
format = c + "i"
|
|
try:
|
|
_testcapi.parse_tuple_and_keywords(tuple_1, dict_b,
|
|
format, keywords)
|
|
when_not_skipped = False
|
|
except SystemError as e:
|
|
s = "argument 1 (impossible<bad format char>)"
|
|
when_not_skipped = (str(e) == s)
|
|
except TypeError:
|
|
when_not_skipped = False
|
|
|
|
# test the format unit when skipped
|
|
optional_format = "|" + format
|
|
try:
|
|
_testcapi.parse_tuple_and_keywords(empty_tuple, dict_b,
|
|
optional_format, keywords)
|
|
when_skipped = False
|
|
except SystemError as e:
|
|
s = "impossible<bad format char>: '{}'".format(format)
|
|
when_skipped = (str(e) == s)
|
|
|
|
message = ("test_skipitem_parity: "
|
|
"detected mismatch between convertsimple and skipitem "
|
|
"for format unit '{}' ({}), not skipped {}, skipped {}".format(
|
|
c, i, when_skipped, when_not_skipped))
|
|
self.assertIs(when_skipped, when_not_skipped, message)
|
|
|
|
def test_skipitem_with_suffix(self):
|
|
parse = _testcapi.parse_tuple_and_keywords
|
|
empty_tuple = ()
|
|
tuple_1 = (0,)
|
|
dict_b = {'b':1}
|
|
keywords = ["a", "b"]
|
|
|
|
supported = ('s#', 's*', 'z#', 'z*', 'u#', 'Z#', 'y#', 'y*', 'w#', 'w*')
|
|
for c in string.ascii_letters:
|
|
for c2 in '#*':
|
|
f = c + c2
|
|
with self.subTest(format=f):
|
|
optional_format = "|" + f + "i"
|
|
if f in supported:
|
|
parse(empty_tuple, dict_b, optional_format, keywords)
|
|
else:
|
|
with self.assertRaisesRegex(SystemError,
|
|
'impossible<bad format char>'):
|
|
parse(empty_tuple, dict_b, optional_format, keywords)
|
|
|
|
for c in map(chr, range(32, 128)):
|
|
f = 'e' + c
|
|
optional_format = "|" + f + "i"
|
|
with self.subTest(format=f):
|
|
if c in 'st':
|
|
parse(empty_tuple, dict_b, optional_format, keywords)
|
|
else:
|
|
with self.assertRaisesRegex(SystemError,
|
|
'impossible<bad format char>'):
|
|
parse(empty_tuple, dict_b, optional_format, keywords)
|
|
|
|
def test_parse_tuple_and_keywords(self):
|
|
# Test handling errors in the parse_tuple_and_keywords helper itself
|
|
self.assertRaises(TypeError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, 42, [])
|
|
self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, '', 42)
|
|
self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, '', [''] * 42)
|
|
self.assertRaises(ValueError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, '', [42])
|
|
|
|
def test_bad_use(self):
|
|
# Test handling invalid format and keywords in
|
|
# PyArg_ParseTupleAndKeywords()
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(1,), {}, '||O', ['a'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(1, 2), {}, '|O|O', ['a', 'b'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {'a': 1}, '$$O', ['a'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {'a': 1, 'b': 2}, '$O$O', ['a', 'b'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {'a': 1}, '$|O', ['a'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {'a': 1, 'b': 2}, '$O|O', ['a', 'b'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(1,), {}, '|O', ['a', 'b'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(1,), {}, '|OO', ['a'])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, '|$O', [''])
|
|
self.assertRaises(SystemError, _testcapi.parse_tuple_and_keywords,
|
|
(), {}, '|OO', ['a', ''])
|
|
|
|
def test_positional_only(self):
|
|
parse = _testcapi.parse_tuple_and_keywords
|
|
|
|
parse((1, 2, 3), {}, 'OOO', ['', '', 'a'])
|
|
parse((1, 2), {'a': 3}, 'OOO', ['', '', 'a'])
|
|
with self.assertRaisesRegex(TypeError,
|
|
r'function takes at least 2 positional arguments \(1 given\)'):
|
|
parse((1,), {'a': 3}, 'OOO', ['', '', 'a'])
|
|
parse((1,), {}, 'O|OO', ['', '', 'a'])
|
|
with self.assertRaisesRegex(TypeError,
|
|
r'function takes at least 1 positional arguments \(0 given\)'):
|
|
parse((), {}, 'O|OO', ['', '', 'a'])
|
|
parse((1, 2), {'a': 3}, 'OO$O', ['', '', 'a'])
|
|
with self.assertRaisesRegex(TypeError,
|
|
r'function takes exactly 2 positional arguments \(1 given\)'):
|
|
parse((1,), {'a': 3}, 'OO$O', ['', '', 'a'])
|
|
parse((1,), {}, 'O|O$O', ['', '', 'a'])
|
|
with self.assertRaisesRegex(TypeError,
|
|
r'function takes at least 1 positional arguments \(0 given\)'):
|
|
parse((), {}, 'O|O$O', ['', '', 'a'])
|
|
with self.assertRaisesRegex(SystemError, r'Empty parameter name after \$'):
|
|
parse((1,), {}, 'O|$OO', ['', '', 'a'])
|
|
with self.assertRaisesRegex(SystemError, 'Empty keyword'):
|
|
parse((1,), {}, 'O|OO', ['', 'a', ''])
|
|
|
|
|
|
class TestThreadState(unittest.TestCase):
|
|
|
|
@support.reap_threads
|
|
def test_thread_state(self):
|
|
# some extra thread-state tests driven via _testcapi
|
|
def target():
|
|
idents = []
|
|
|
|
def callback():
|
|
idents.append(threading.get_ident())
|
|
|
|
_testcapi._test_thread_state(callback)
|
|
a = b = callback
|
|
time.sleep(1)
|
|
# Check our main thread is in the list exactly 3 times.
|
|
self.assertEqual(idents.count(threading.get_ident()), 3,
|
|
"Couldn't find main thread correctly in the list")
|
|
|
|
target()
|
|
t = threading.Thread(target=target)
|
|
t.start()
|
|
t.join()
|
|
|
|
|
|
class Test_testcapi(unittest.TestCase):
|
|
def test__testcapi(self):
|
|
if support.verbose:
|
|
print()
|
|
for name in dir(_testcapi):
|
|
if not name.startswith('test_'):
|
|
continue
|
|
with self.subTest("internal", name=name):
|
|
if support.verbose:
|
|
print(f" {name}", flush=True)
|
|
test = getattr(_testcapi, name)
|
|
test()
|
|
|
|
|
|
class PyMemDebugTests(unittest.TestCase):
|
|
PYTHONMALLOC = 'debug'
|
|
# '0x04c06e0' or '04C06E0'
|
|
PTR_REGEX = r'(?:0x)?[0-9a-fA-F]+'
|
|
|
|
def check(self, code):
|
|
with support.SuppressCrashReport():
|
|
out = assert_python_failure('-c', code,
|
|
PYTHONMALLOC=self.PYTHONMALLOC)
|
|
stderr = out.err
|
|
return stderr.decode('ascii', 'replace')
|
|
|
|
def test_buffer_overflow(self):
|
|
out = self.check('import _testcapi; _testcapi.pymem_buffer_overflow()')
|
|
regex = (r"Debug memory block at address p={ptr}: API 'm'\n"
|
|
r" 16 bytes originally requested\n"
|
|
r" The [0-9] pad bytes at p-[0-9] are FORBIDDENBYTE, as expected.\n"
|
|
r" The [0-9] pad bytes at tail={ptr} are not all FORBIDDENBYTE \(0x[0-9a-f]{{2}}\):\n"
|
|
r" at tail\+0: 0x78 \*\*\* OUCH\n"
|
|
r" at tail\+1: 0xfb\n"
|
|
r" at tail\+2: 0xfb\n"
|
|
r" .*\n"
|
|
r" The block was made by call #[0-9]+ to debug malloc/realloc.\n"
|
|
r" Data at p: cb cb cb .*\n"
|
|
r"\n"
|
|
r"Fatal Python error: bad trailing pad byte")
|
|
regex = regex.format(ptr=self.PTR_REGEX)
|
|
regex = re.compile(regex, flags=re.DOTALL)
|
|
self.assertRegex(out, regex)
|
|
|
|
def test_api_misuse(self):
|
|
out = self.check('import _testcapi; _testcapi.pymem_api_misuse()')
|
|
regex = (r"Debug memory block at address p={ptr}: API 'm'\n"
|
|
r" 16 bytes originally requested\n"
|
|
r" The [0-9] pad bytes at p-[0-9] are FORBIDDENBYTE, as expected.\n"
|
|
r" The [0-9] pad bytes at tail={ptr} are FORBIDDENBYTE, as expected.\n"
|
|
r" The block was made by call #[0-9]+ to debug malloc/realloc.\n"
|
|
r" Data at p: cb cb cb .*\n"
|
|
r"\n"
|
|
r"Fatal Python error: bad ID: Allocated using API 'm', verified using API 'r'\n")
|
|
regex = regex.format(ptr=self.PTR_REGEX)
|
|
self.assertRegex(out, regex)
|
|
|
|
def check_malloc_without_gil(self, code):
|
|
out = self.check(code)
|
|
expected = ('Fatal Python error: Python memory allocator called '
|
|
'without holding the GIL')
|
|
self.assertIn(expected, out)
|
|
|
|
def test_pymem_malloc_without_gil(self):
|
|
# Debug hooks must raise an error if PyMem_Malloc() is called
|
|
# without holding the GIL
|
|
code = 'import _testcapi; _testcapi.pymem_malloc_without_gil()'
|
|
self.check_malloc_without_gil(code)
|
|
|
|
def test_pyobject_malloc_without_gil(self):
|
|
# Debug hooks must raise an error if PyObject_Malloc() is called
|
|
# without holding the GIL
|
|
code = 'import _testcapi; _testcapi.pyobject_malloc_without_gil()'
|
|
self.check_malloc_without_gil(code)
|
|
|
|
|
|
class PyMemMallocDebugTests(PyMemDebugTests):
|
|
PYTHONMALLOC = 'malloc_debug'
|
|
|
|
|
|
@unittest.skipUnless(support.with_pymalloc(), 'need pymalloc')
|
|
class PyMemPymallocDebugTests(PyMemDebugTests):
|
|
PYTHONMALLOC = 'pymalloc_debug'
|
|
|
|
|
|
@unittest.skipUnless(Py_DEBUG, 'need Py_DEBUG')
|
|
class PyMemDefaultTests(PyMemDebugTests):
|
|
# test default allocator of Python compiled in debug mode
|
|
PYTHONMALLOC = ''
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|