Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 127 additions & 0 deletions Lib/test/test_capi/test_marshal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
import marshal
import os.path
import unittest

from test import support
from test.support import import_helper
from test.support import os_helper
from test.test_marshal import HelperMixin, omit_last_byte


# Skip this test if _testcapi is are not available.
_testcapi = import_helper.import_module('_testcapi')


@support.cpython_only
class CAPI_TestCase(unittest.TestCase, HelperMixin):

def test_read_from_file_error(self):
# A read error is reported as OSError, not EOFError.
# A directory cannot be read (on some platforms it cannot even
# be opened, which is reported as OSError as well).
os.mkdir(os_helper.TESTFN)
self.addCleanup(os_helper.rmdir, os_helper.TESTFN)
for func in (_testcapi.pymarshal_read_short_from_file,
_testcapi.pymarshal_read_long_from_file,
_testcapi.pymarshal_read_object_from_file,
_testcapi.pymarshal_read_last_object_from_file):
with self.subTest(func=func.__name__):
self.assertRaises(OSError, func, os_helper.TESTFN)

@unittest.skipUnless(os.path.exists('/dev/full'), 'requires /dev/full')
def test_write_to_file_error(self):
# A write error is reported as OSError.
# The data is large enough to not fit in the stdio buffer, so that
# the error is detected before the file is closed.
obj = b'x' * 100000
with self.assertRaises(OSError):
_testcapi.pymarshal_write_object_to_file(obj, '/dev/full',
marshal.version)

def test_write_unmarshallable_to_file(self):
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
with self.assertRaisesRegex(ValueError, 'unmarshallable object'):
_testcapi.pymarshal_write_object_to_file(object(), os_helper.TESTFN,
marshal.version)

def test_write_long_to_file(self):
for v in range(marshal.version + 1):
_testcapi.pymarshal_write_long_to_file(0x12345678, os_helper.TESTFN, v)
with open(os_helper.TESTFN, 'rb') as f:
data = f.read()
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(data, b'\x78\x56\x34\x12')

def test_write_object_to_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j, 'long line '*1000)
for v in range(marshal.version + 1):
_testcapi.pymarshal_write_object_to_file(obj, os_helper.TESTFN, v)
with open(os_helper.TESTFN, 'rb') as f:
data = f.read()
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(marshal.loads(data), obj)

def test_read_short_from_file(self):
with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x34\x12xxxx')
r, p = _testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, 0x1234)
self.assertEqual(p, 2)

with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x12')
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_long_from_file(self):
with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x78\x56\x34\x12xxxx')
r, p = _testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, 0x12345678)
self.assertEqual(p, 4)

with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x56\x34\x12')
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_last_object_from_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
for v in range(marshal.version + 1):
data = marshal.dumps(obj, v)
with open(os_helper.TESTFN, 'wb') as f:
f.write(data + b'xxxx')
r, p = _testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, obj)

with open(os_helper.TESTFN, 'wb') as f:
f.write(omit_last_byte(data))
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_object_from_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
for v in range(marshal.version + 1):
data = marshal.dumps(obj, v)
with open(os_helper.TESTFN, 'wb') as f:
f.write(data + b'xxxx')
r, p = _testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, obj)
self.assertEqual(p, len(data))

with open(os_helper.TESTFN, 'wb') as f:
f.write(omit_last_byte(data))
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)


if __name__ == "__main__":
unittest.main()
111 changes: 0 additions & 111 deletions Lib/test/test_marshal.py
Original file line number Diff line number Diff line change
Expand Up @@ -797,117 +797,6 @@ def test_slice(self):
with self.assertRaises(ValueError):
marshal.dumps(obj, version)

@support.cpython_only
@unittest.skipUnless(_testcapi, 'requires _testcapi')
class CAPI_TestCase(unittest.TestCase, HelperMixin):

def test_read_from_file_error(self):
# A read error is reported as OSError, not EOFError.
# A directory cannot be read (on some platforms it cannot even
# be opened, which is reported as OSError as well).
os.mkdir(os_helper.TESTFN)
self.addCleanup(os_helper.rmdir, os_helper.TESTFN)
for func in (_testcapi.pymarshal_read_short_from_file,
_testcapi.pymarshal_read_long_from_file,
_testcapi.pymarshal_read_object_from_file,
_testcapi.pymarshal_read_last_object_from_file):
with self.subTest(func=func.__name__):
self.assertRaises(OSError, func, os_helper.TESTFN)

@unittest.skipUnless(os.path.exists('/dev/full'), 'requires /dev/full')
def test_write_to_file_error(self):
# A write error is reported as OSError.
# The data is large enough to not fit in the stdio buffer, so that
# the error is detected before the file is closed.
obj = b'x' * 100000
with self.assertRaises(OSError):
_testcapi.pymarshal_write_object_to_file(obj, '/dev/full',
marshal.version)

def test_write_unmarshallable_to_file(self):
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
with self.assertRaisesRegex(ValueError, 'unmarshallable object'):
_testcapi.pymarshal_write_object_to_file(object(), os_helper.TESTFN,
marshal.version)

def test_write_long_to_file(self):
for v in range(marshal.version + 1):
_testcapi.pymarshal_write_long_to_file(0x12345678, os_helper.TESTFN, v)
with open(os_helper.TESTFN, 'rb') as f:
data = f.read()
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(data, b'\x78\x56\x34\x12')

def test_write_object_to_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j, 'long line '*1000)
for v in range(marshal.version + 1):
_testcapi.pymarshal_write_object_to_file(obj, os_helper.TESTFN, v)
with open(os_helper.TESTFN, 'rb') as f:
data = f.read()
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(marshal.loads(data), obj)

def test_read_short_from_file(self):
with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x34\x12xxxx')
r, p = _testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, 0x1234)
self.assertEqual(p, 2)

with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x12')
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_long_from_file(self):
with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x78\x56\x34\x12xxxx')
r, p = _testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, 0x12345678)
self.assertEqual(p, 4)

with open(os_helper.TESTFN, 'wb') as f:
f.write(b'\x56\x34\x12')
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_last_object_from_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
for v in range(marshal.version + 1):
data = marshal.dumps(obj, v)
with open(os_helper.TESTFN, 'wb') as f:
f.write(data + b'xxxx')
r, p = _testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, obj)

with open(os_helper.TESTFN, 'wb') as f:
f.write(omit_last_byte(data))
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_read_object_from_file(self):
obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
for v in range(marshal.version + 1):
data = marshal.dumps(obj, v)
with open(os_helper.TESTFN, 'wb') as f:
f.write(data + b'xxxx')
r, p = _testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)
self.assertEqual(r, obj)
self.assertEqual(p, len(data))

with open(os_helper.TESTFN, 'wb') as f:
f.write(omit_last_byte(data))
with self.assertRaises(EOFError):
_testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
os_helper.unlink(os_helper.TESTFN)


if __name__ == "__main__":
unittest.main()
2 changes: 1 addition & 1 deletion Modules/Setup.stdlib.in
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,7 @@
@MODULE__XXTESTFUZZ_TRUE@_xxtestfuzz _xxtestfuzz/_xxtestfuzz.c _xxtestfuzz/fuzzer.c
@MODULE__TESTBUFFER_TRUE@_testbuffer _testbuffer.c
@MODULE__TESTINTERNALCAPI_TRUE@_testinternalcapi _testinternalcapi.c _testinternalcapi/test_lock.c _testinternalcapi/pytime.c _testinternalcapi/set.c _testinternalcapi/test_critical_sections.c _testinternalcapi/complex.c _testinternalcapi/interpreter.c _testinternalcapi/tokenizer.c _testinternalcapi/tuple.c _testinternalcapi/typecache.c
@MODULE__TESTCAPI_TRUE@_testcapi _testcapimodule.c _testcapi/vectorcall.c _testcapi/heaptype.c _testcapi/abstract.c _testcapi/unicode.c _testcapi/dict.c _testcapi/set.c _testcapi/list.c _testcapi/tuple.c _testcapi/getargs.c _testcapi/datetime.c _testcapi/docstring.c _testcapi/mem.c _testcapi/watchers.c _testcapi/long.c _testcapi/float.c _testcapi/complex.c _testcapi/numbers.c _testcapi/structmember.c _testcapi/exceptions.c _testcapi/code.c _testcapi/buffer.c _testcapi/pyatomic.c _testcapi/run.c _testcapi/file.c _testcapi/codec.c _testcapi/immortal.c _testcapi/gc.c _testcapi/hash.c _testcapi/time.c _testcapi/bytes.c _testcapi/object.c _testcapi/modsupport.c _testcapi/monitoring.c _testcapi/config.c _testcapi/import.c _testcapi/frame.c _testcapi/type.c _testcapi/function.c _testcapi/module.c _testcapi/weakref.c
@MODULE__TESTCAPI_TRUE@_testcapi _testcapimodule.c _testcapi/vectorcall.c _testcapi/heaptype.c _testcapi/abstract.c _testcapi/unicode.c _testcapi/dict.c _testcapi/set.c _testcapi/list.c _testcapi/tuple.c _testcapi/getargs.c _testcapi/datetime.c _testcapi/docstring.c _testcapi/mem.c _testcapi/watchers.c _testcapi/long.c _testcapi/float.c _testcapi/complex.c _testcapi/numbers.c _testcapi/structmember.c _testcapi/exceptions.c _testcapi/code.c _testcapi/buffer.c _testcapi/pyatomic.c _testcapi/run.c _testcapi/file.c _testcapi/codec.c _testcapi/immortal.c _testcapi/gc.c _testcapi/hash.c _testcapi/time.c _testcapi/bytes.c _testcapi/object.c _testcapi/modsupport.c _testcapi/monitoring.c _testcapi/config.c _testcapi/import.c _testcapi/frame.c _testcapi/type.c _testcapi/function.c _testcapi/module.c _testcapi/weakref.c _testcapi/marshal.c
@MODULE__TESTLIMITEDCAPI_TRUE@_testlimitedcapi _testlimitedcapi.c _testlimitedcapi/abstract.c _testlimitedcapi/bytearray.c _testlimitedcapi/bytes.c _testlimitedcapi/capsule.c _testlimitedcapi/codec.c _testlimitedcapi/complex.c _testlimitedcapi/dict.c _testlimitedcapi/eval.c _testlimitedcapi/float.c _testlimitedcapi/heaptype_relative.c _testlimitedcapi/import.c _testlimitedcapi/list.c _testlimitedcapi/long.c _testlimitedcapi/object.c _testlimitedcapi/pyos.c _testlimitedcapi/set.c _testlimitedcapi/slots.c _testlimitedcapi/sys.c _testlimitedcapi/threadstate.c _testlimitedcapi/tuple.c _testlimitedcapi/unicode.c _testlimitedcapi/vectorcall_limited.c _testlimitedcapi/version.c _testlimitedcapi/file.c _testlimitedcapi/weakref.c _testlimitedcapi/run.c _testlimitedcapi/type.c
@MODULE__TESTCLINIC_TRUE@_testclinic _testclinic.c
@MODULE__TESTCLINIC_LIMITED_TRUE@_testclinic_limited _testclinic_limited.c
Expand Down
Loading
Loading