Add functions PySys_GetAttr(), PySys_GetAttrString(), PySys_GetOptionalAttr() and PySys_GetOptionalAttrString().
217 lines
8.4 KiB
Python
217 lines
8.4 KiB
Python
import unittest
|
|
import contextlib
|
|
import sys
|
|
from test import support
|
|
from test.support import import_helper
|
|
|
|
try:
|
|
import _testlimitedcapi
|
|
except ImportError:
|
|
_testlimitedcapi = None
|
|
|
|
NULL = None
|
|
|
|
class CAPITest(unittest.TestCase):
|
|
# TODO: Test the following functions:
|
|
#
|
|
# PySys_Audit()
|
|
# PySys_AuditTuple()
|
|
|
|
maxDiff = None
|
|
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getattr(self):
|
|
# Test PySys_GetAttr()
|
|
sys_getattr = _testlimitedcapi.sys_getattr
|
|
|
|
self.assertIs(sys_getattr('stdout'), sys.stdout)
|
|
with support.swap_attr(sys, '\U0001f40d', 42):
|
|
self.assertEqual(sys_getattr('\U0001f40d'), 42)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, r'lost sys\.nonexistent'):
|
|
sys_getattr('nonexistent')
|
|
with self.assertRaisesRegex(RuntimeError, r'lost sys\.\U0001f40d'):
|
|
sys_getattr('\U0001f40d')
|
|
self.assertRaises(TypeError, sys_getattr, 1)
|
|
self.assertRaises(TypeError, sys_getattr, [])
|
|
# CRASHES sys_getattr(NULL)
|
|
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getattrstring(self):
|
|
# Test PySys_GetAttrString()
|
|
getattrstring = _testlimitedcapi.sys_getattrstring
|
|
|
|
self.assertIs(getattrstring(b'stdout'), sys.stdout)
|
|
with support.swap_attr(sys, '\U0001f40d', 42):
|
|
self.assertEqual(getattrstring('\U0001f40d'.encode()), 42)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, r'lost sys\.nonexistent'):
|
|
getattrstring(b'nonexistent')
|
|
with self.assertRaisesRegex(RuntimeError, r'lost sys\.\U0001f40d'):
|
|
getattrstring('\U0001f40d'.encode())
|
|
self.assertRaises(UnicodeDecodeError, getattrstring, b'\xff')
|
|
# CRASHES getattrstring(NULL)
|
|
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getoptionalattr(self):
|
|
# Test PySys_GetOptionalAttr()
|
|
getoptionalattr = _testlimitedcapi.sys_getoptionalattr
|
|
|
|
self.assertIs(getoptionalattr('stdout'), sys.stdout)
|
|
with support.swap_attr(sys, '\U0001f40d', 42):
|
|
self.assertEqual(getoptionalattr('\U0001f40d'), 42)
|
|
|
|
self.assertIs(getoptionalattr('nonexistent'), AttributeError)
|
|
self.assertIs(getoptionalattr('\U0001f40d'), AttributeError)
|
|
self.assertRaises(TypeError, getoptionalattr, 1)
|
|
self.assertRaises(TypeError, getoptionalattr, [])
|
|
# CRASHES getoptionalattr(NULL)
|
|
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getoptionalattrstring(self):
|
|
# Test PySys_GetOptionalAttrString()
|
|
getoptionalattrstring = _testlimitedcapi.sys_getoptionalattrstring
|
|
|
|
self.assertIs(getoptionalattrstring(b'stdout'), sys.stdout)
|
|
with support.swap_attr(sys, '\U0001f40d', 42):
|
|
self.assertEqual(getoptionalattrstring('\U0001f40d'.encode()), 42)
|
|
|
|
self.assertIs(getoptionalattrstring(b'nonexistent'), AttributeError)
|
|
self.assertIs(getoptionalattrstring('\U0001f40d'.encode()), AttributeError)
|
|
self.assertRaises(UnicodeDecodeError, getoptionalattrstring, b'\xff')
|
|
# CRASHES getoptionalattrstring(NULL)
|
|
|
|
@support.cpython_only
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getobject(self):
|
|
# Test PySys_GetObject()
|
|
getobject = _testlimitedcapi.sys_getobject
|
|
|
|
self.assertIs(getobject(b'stdout'), sys.stdout)
|
|
with support.swap_attr(sys, '\U0001f40d', 42):
|
|
self.assertEqual(getobject('\U0001f40d'.encode()), 42)
|
|
|
|
self.assertIs(getobject(b'nonexistent'), AttributeError)
|
|
with support.catch_unraisable_exception() as cm:
|
|
self.assertIs(getobject(b'\xff'), AttributeError)
|
|
self.assertEqual(cm.unraisable.exc_type, UnicodeDecodeError)
|
|
self.assertRegex(str(cm.unraisable.exc_value),
|
|
"'utf-8' codec can't decode")
|
|
# CRASHES getobject(NULL)
|
|
|
|
@support.cpython_only
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_setobject(self):
|
|
# Test PySys_SetObject()
|
|
setobject = _testlimitedcapi.sys_setobject
|
|
|
|
value = ['value']
|
|
value2 = ['value2']
|
|
try:
|
|
self.assertEqual(setobject(b'newattr', value), 0)
|
|
self.assertIs(sys.newattr, value)
|
|
self.assertEqual(setobject(b'newattr', value2), 0)
|
|
self.assertIs(sys.newattr, value2)
|
|
self.assertEqual(setobject(b'newattr', NULL), 0)
|
|
self.assertNotHasAttr(sys, 'newattr')
|
|
self.assertEqual(setobject(b'newattr', NULL), 0)
|
|
finally:
|
|
with contextlib.suppress(AttributeError):
|
|
del sys.newattr
|
|
try:
|
|
self.assertEqual(setobject('\U0001f40d'.encode(), value), 0)
|
|
self.assertIs(getattr(sys, '\U0001f40d'), value)
|
|
self.assertEqual(setobject('\U0001f40d'.encode(), NULL), 0)
|
|
self.assertNotHasAttr(sys, '\U0001f40d')
|
|
finally:
|
|
with contextlib.suppress(AttributeError):
|
|
delattr(sys, '\U0001f40d')
|
|
|
|
with self.assertRaises(UnicodeDecodeError):
|
|
setobject(b'\xff', value)
|
|
# CRASHES setobject(NULL, value)
|
|
|
|
@support.cpython_only
|
|
@unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
|
|
def test_sys_getxoptions(self):
|
|
# Test PySys_GetXOptions()
|
|
getxoptions = _testlimitedcapi.sys_getxoptions
|
|
|
|
self.assertIs(getxoptions(), sys._xoptions)
|
|
|
|
xoptions = sys._xoptions
|
|
try:
|
|
sys._xoptions = 'non-dict'
|
|
self.assertEqual(getxoptions(), {})
|
|
self.assertIs(getxoptions(), sys._xoptions)
|
|
|
|
del sys._xoptions
|
|
self.assertEqual(getxoptions(), {})
|
|
self.assertIs(getxoptions(), sys._xoptions)
|
|
finally:
|
|
sys._xoptions = xoptions
|
|
self.assertIs(getxoptions(), sys._xoptions)
|
|
|
|
def _test_sys_formatstream(self, funname, streamname):
|
|
import_helper.import_module('ctypes')
|
|
from ctypes import pythonapi, c_char_p, py_object
|
|
func = getattr(pythonapi, funname)
|
|
func.argtypes = (c_char_p,)
|
|
|
|
# Supports plain C types.
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %s!', c_char_p(b'world'))
|
|
self.assertEqual(stream.getvalue(), 'Hello, world!')
|
|
|
|
# Supports Python objects.
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %R!', py_object('world'))
|
|
self.assertEqual(stream.getvalue(), "Hello, 'world'!")
|
|
|
|
# The total length is not limited.
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %s!', c_char_p(b'world'*200))
|
|
self.assertEqual(stream.getvalue(), 'Hello, ' + 'world'*200 + '!')
|
|
|
|
def test_sys_formatstdout(self):
|
|
# Test PySys_FormatStdout()
|
|
self._test_sys_formatstream('PySys_FormatStdout', 'stdout')
|
|
|
|
def test_sys_formatstderr(self):
|
|
# Test PySys_FormatStderr()
|
|
self._test_sys_formatstream('PySys_FormatStderr', 'stderr')
|
|
|
|
def _test_sys_writestream(self, funname, streamname):
|
|
import_helper.import_module('ctypes')
|
|
from ctypes import pythonapi, c_char_p
|
|
func = getattr(pythonapi, funname)
|
|
func.argtypes = (c_char_p,)
|
|
|
|
# Supports plain C types.
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %s!', c_char_p(b'world'))
|
|
self.assertEqual(stream.getvalue(), 'Hello, world!')
|
|
|
|
# There is a limit on the total length.
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %s!', c_char_p(b'world'*100))
|
|
self.assertEqual(stream.getvalue(), 'Hello, ' + 'world'*100 + '!')
|
|
with support.captured_output(streamname) as stream:
|
|
func(b'Hello, %s!', c_char_p(b'world'*200))
|
|
out = stream.getvalue()
|
|
self.assertEqual(out[:20], 'Hello, worldworldwor')
|
|
self.assertEqual(out[-13:], '... truncated')
|
|
self.assertGreater(len(out), 1000)
|
|
|
|
def test_sys_writestdout(self):
|
|
# Test PySys_WriteStdout()
|
|
self._test_sys_writestream('PySys_WriteStdout', 'stdout')
|
|
|
|
def test_sys_writestderr(self):
|
|
# Test PySys_WriteStderr()
|
|
self._test_sys_writestream('PySys_WriteStderr', 'stderr')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|