test_marshal.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654
  1. from test import support
  2. from test.support import os_helper, requires_debug_ranges
  3. from test.support.script_helper import assert_python_ok
  4. import array
  5. import io
  6. import marshal
  7. import sys
  8. import unittest
  9. import os
  10. import types
  11. import textwrap
  12. try:
  13. import _testcapi
  14. except ImportError:
  15. _testcapi = None
  16. class HelperMixin:
  17. def helper(self, sample, *extra):
  18. new = marshal.loads(marshal.dumps(sample, *extra))
  19. self.assertEqual(sample, new)
  20. try:
  21. with open(os_helper.TESTFN, "wb") as f:
  22. marshal.dump(sample, f, *extra)
  23. with open(os_helper.TESTFN, "rb") as f:
  24. new = marshal.load(f)
  25. self.assertEqual(sample, new)
  26. finally:
  27. os_helper.unlink(os_helper.TESTFN)
  28. class IntTestCase(unittest.TestCase, HelperMixin):
  29. def test_ints(self):
  30. # Test a range of Python ints larger than the machine word size.
  31. n = sys.maxsize ** 2
  32. while n:
  33. for expected in (-n, n):
  34. self.helper(expected)
  35. n = n >> 1
  36. def test_int64(self):
  37. # Simulate int marshaling with TYPE_INT64.
  38. maxint64 = (1 << 63) - 1
  39. minint64 = -maxint64-1
  40. for base in maxint64, minint64, -maxint64, -(minint64 >> 1):
  41. while base:
  42. s = b'I' + int.to_bytes(base, 8, 'little', signed=True)
  43. got = marshal.loads(s)
  44. self.assertEqual(base, got)
  45. if base == -1: # a fixed-point for shifting right 1
  46. base = 0
  47. else:
  48. base >>= 1
  49. got = marshal.loads(b'I\xfe\xdc\xba\x98\x76\x54\x32\x10')
  50. self.assertEqual(got, 0x1032547698badcfe)
  51. got = marshal.loads(b'I\x01\x23\x45\x67\x89\xab\xcd\xef')
  52. self.assertEqual(got, -0x1032547698badcff)
  53. got = marshal.loads(b'I\x08\x19\x2a\x3b\x4c\x5d\x6e\x7f')
  54. self.assertEqual(got, 0x7f6e5d4c3b2a1908)
  55. got = marshal.loads(b'I\xf7\xe6\xd5\xc4\xb3\xa2\x91\x80')
  56. self.assertEqual(got, -0x7f6e5d4c3b2a1909)
  57. def test_bool(self):
  58. for b in (True, False):
  59. self.helper(b)
  60. class FloatTestCase(unittest.TestCase, HelperMixin):
  61. def test_floats(self):
  62. # Test a few floats
  63. small = 1e-25
  64. n = sys.maxsize * 3.7e250
  65. while n > small:
  66. for expected in (-n, n):
  67. self.helper(float(expected))
  68. n /= 123.4567
  69. f = 0.0
  70. s = marshal.dumps(f, 2)
  71. got = marshal.loads(s)
  72. self.assertEqual(f, got)
  73. # and with version <= 1 (floats marshalled differently then)
  74. s = marshal.dumps(f, 1)
  75. got = marshal.loads(s)
  76. self.assertEqual(f, got)
  77. n = sys.maxsize * 3.7e-250
  78. while n < small:
  79. for expected in (-n, n):
  80. f = float(expected)
  81. self.helper(f)
  82. self.helper(f, 1)
  83. n *= 123.4567
  84. class StringTestCase(unittest.TestCase, HelperMixin):
  85. def test_unicode(self):
  86. for s in ["", "Andr\xe8 Previn", "abc", " "*10000]:
  87. self.helper(marshal.loads(marshal.dumps(s)))
  88. def test_string(self):
  89. for s in ["", "Andr\xe8 Previn", "abc", " "*10000]:
  90. self.helper(s)
  91. def test_bytes(self):
  92. for s in [b"", b"Andr\xe8 Previn", b"abc", b" "*10000]:
  93. self.helper(s)
  94. class ExceptionTestCase(unittest.TestCase):
  95. def test_exceptions(self):
  96. new = marshal.loads(marshal.dumps(StopIteration))
  97. self.assertEqual(StopIteration, new)
  98. class CodeTestCase(unittest.TestCase):
  99. def test_code(self):
  100. co = ExceptionTestCase.test_exceptions.__code__
  101. new = marshal.loads(marshal.dumps(co))
  102. self.assertEqual(co, new)
  103. def test_many_codeobjects(self):
  104. # Issue2957: bad recursion count on code objects
  105. count = 5000 # more than MAX_MARSHAL_STACK_DEPTH
  106. codes = (ExceptionTestCase.test_exceptions.__code__,) * count
  107. marshal.loads(marshal.dumps(codes))
  108. def test_different_filenames(self):
  109. co1 = compile("x", "f1", "exec")
  110. co2 = compile("y", "f2", "exec")
  111. co1, co2 = marshal.loads(marshal.dumps((co1, co2)))
  112. self.assertEqual(co1.co_filename, "f1")
  113. self.assertEqual(co2.co_filename, "f2")
  114. @requires_debug_ranges()
  115. def test_minimal_linetable_with_no_debug_ranges(self):
  116. # Make sure when demarshalling objects with `-X no_debug_ranges`
  117. # that the columns are None.
  118. co = ExceptionTestCase.test_exceptions.__code__
  119. code = textwrap.dedent("""
  120. import sys
  121. import marshal
  122. with open(sys.argv[1], 'rb') as f:
  123. co = marshal.load(f)
  124. positions = list(co.co_positions())
  125. assert positions[0][2] is None
  126. assert positions[0][3] is None
  127. """)
  128. try:
  129. with open(os_helper.TESTFN, 'wb') as f:
  130. marshal.dump(co, f)
  131. assert_python_ok('-X', 'no_debug_ranges',
  132. '-c', code, os_helper.TESTFN)
  133. finally:
  134. os_helper.unlink(os_helper.TESTFN)
  135. @support.cpython_only
  136. def test_same_filename_used(self):
  137. s = """def f(): pass\ndef g(): pass"""
  138. co = compile(s, "myfile", "exec")
  139. co = marshal.loads(marshal.dumps(co))
  140. for obj in co.co_consts:
  141. if isinstance(obj, types.CodeType):
  142. self.assertIs(co.co_filename, obj.co_filename)
  143. class ContainerTestCase(unittest.TestCase, HelperMixin):
  144. d = {'astring': 'foo@bar.baz.spam',
  145. 'afloat': 7283.43,
  146. 'anint': 2**20,
  147. 'ashortlong': 2,
  148. 'alist': ['.zyx.41'],
  149. 'atuple': ('.zyx.41',)*10,
  150. 'aboolean': False,
  151. 'aunicode': "Andr\xe8 Previn"
  152. }
  153. def test_dict(self):
  154. self.helper(self.d)
  155. def test_list(self):
  156. self.helper(list(self.d.items()))
  157. def test_tuple(self):
  158. self.helper(tuple(self.d.keys()))
  159. def test_sets(self):
  160. for constructor in (set, frozenset):
  161. self.helper(constructor(self.d.keys()))
  162. class BufferTestCase(unittest.TestCase, HelperMixin):
  163. def test_bytearray(self):
  164. b = bytearray(b"abc")
  165. self.helper(b)
  166. new = marshal.loads(marshal.dumps(b))
  167. self.assertEqual(type(new), bytes)
  168. def test_memoryview(self):
  169. b = memoryview(b"abc")
  170. self.helper(b)
  171. new = marshal.loads(marshal.dumps(b))
  172. self.assertEqual(type(new), bytes)
  173. def test_array(self):
  174. a = array.array('B', b"abc")
  175. new = marshal.loads(marshal.dumps(a))
  176. self.assertEqual(new, b"abc")
  177. class BugsTestCase(unittest.TestCase):
  178. def test_bug_5888452(self):
  179. # Simple-minded check for SF 588452: Debug build crashes
  180. marshal.dumps([128] * 1000)
  181. def test_patch_873224(self):
  182. self.assertRaises(Exception, marshal.loads, b'0')
  183. self.assertRaises(Exception, marshal.loads, b'f')
  184. self.assertRaises(Exception, marshal.loads, marshal.dumps(2**65)[:-1])
  185. def test_version_argument(self):
  186. # Python 2.4.0 crashes for any call to marshal.dumps(x, y)
  187. self.assertEqual(marshal.loads(marshal.dumps(5, 0)), 5)
  188. self.assertEqual(marshal.loads(marshal.dumps(5, 1)), 5)
  189. def test_fuzz(self):
  190. # simple test that it's at least not *totally* trivial to
  191. # crash from bad marshal data
  192. for i in range(256):
  193. c = bytes([i])
  194. try:
  195. marshal.loads(c)
  196. except Exception:
  197. pass
  198. def test_loads_recursion(self):
  199. def run_tests(N, check):
  200. # (((...None...),),)
  201. check(b')\x01' * N + b'N')
  202. check(b'(\x01\x00\x00\x00' * N + b'N')
  203. # [[[...None...]]]
  204. check(b'[\x01\x00\x00\x00' * N + b'N')
  205. # {None: {None: {None: ...None...}}}
  206. check(b'{N' * N + b'N' + b'0' * N)
  207. # frozenset([frozenset([frozenset([...None...])])])
  208. check(b'>\x01\x00\x00\x00' * N + b'N')
  209. # Check that the generated marshal data is valid and marshal.loads()
  210. # works for moderately deep nesting
  211. run_tests(100, marshal.loads)
  212. # Very deeply nested structure shouldn't blow the stack
  213. def check(s):
  214. self.assertRaises(ValueError, marshal.loads, s)
  215. run_tests(2**20, check)
  216. def test_recursion_limit(self):
  217. # Create a deeply nested structure.
  218. head = last = []
  219. # The max stack depth should match the value in Python/marshal.c.
  220. # BUG: https://bugs.python.org/issue33720
  221. # Windows always limits the maximum depth on release and debug builds
  222. #if os.name == 'nt' and hasattr(sys, 'gettotalrefcount'):
  223. if os.name == 'nt':
  224. MAX_MARSHAL_STACK_DEPTH = 1000
  225. elif sys.platform == 'wasi':
  226. MAX_MARSHAL_STACK_DEPTH = 1500
  227. else:
  228. MAX_MARSHAL_STACK_DEPTH = 2000
  229. for i in range(MAX_MARSHAL_STACK_DEPTH - 2):
  230. last.append([0])
  231. last = last[-1]
  232. # Verify we don't blow out the stack with dumps/load.
  233. data = marshal.dumps(head)
  234. new_head = marshal.loads(data)
  235. # Don't use == to compare objects, it can exceed the recursion limit.
  236. self.assertEqual(len(new_head), len(head))
  237. self.assertEqual(len(new_head[0]), len(head[0]))
  238. self.assertEqual(len(new_head[-1]), len(head[-1]))
  239. last.append([0])
  240. self.assertRaises(ValueError, marshal.dumps, head)
  241. def test_exact_type_match(self):
  242. # Former bug:
  243. # >>> class Int(int): pass
  244. # >>> type(loads(dumps(Int())))
  245. # <type 'int'>
  246. for typ in (int, float, complex, tuple, list, dict, set, frozenset):
  247. # Note: str subclasses are not tested because they get handled
  248. # by marshal's routines for objects supporting the buffer API.
  249. subtyp = type('subtyp', (typ,), {})
  250. self.assertRaises(ValueError, marshal.dumps, subtyp())
  251. # Issue #1792 introduced a change in how marshal increases the size of its
  252. # internal buffer; this test ensures that the new code is exercised.
  253. def test_large_marshal(self):
  254. size = int(1e6)
  255. testString = 'abc' * size
  256. marshal.dumps(testString)
  257. def test_invalid_longs(self):
  258. # Issue #7019: marshal.loads shouldn't produce unnormalized PyLongs
  259. invalid_string = b'l\x02\x00\x00\x00\x00\x00\x00\x00'
  260. self.assertRaises(ValueError, marshal.loads, invalid_string)
  261. def test_multiple_dumps_and_loads(self):
  262. # Issue 12291: marshal.load() should be callable multiple times
  263. # with interleaved data written by non-marshal code
  264. # Adapted from a patch by Engelbert Gruber.
  265. data = (1, 'abc', b'def', 1.0, (2, 'a', ['b', b'c']))
  266. for interleaved in (b'', b'0123'):
  267. ilen = len(interleaved)
  268. positions = []
  269. try:
  270. with open(os_helper.TESTFN, 'wb') as f:
  271. for d in data:
  272. marshal.dump(d, f)
  273. if ilen:
  274. f.write(interleaved)
  275. positions.append(f.tell())
  276. with open(os_helper.TESTFN, 'rb') as f:
  277. for i, d in enumerate(data):
  278. self.assertEqual(d, marshal.load(f))
  279. if ilen:
  280. f.read(ilen)
  281. self.assertEqual(positions[i], f.tell())
  282. finally:
  283. os_helper.unlink(os_helper.TESTFN)
  284. def test_loads_reject_unicode_strings(self):
  285. # Issue #14177: marshal.loads() should not accept unicode strings
  286. unicode_string = 'T'
  287. self.assertRaises(TypeError, marshal.loads, unicode_string)
  288. def test_bad_reader(self):
  289. class BadReader(io.BytesIO):
  290. def readinto(self, buf):
  291. n = super().readinto(buf)
  292. if n is not None and n > 4:
  293. n += 10**6
  294. return n
  295. for value in (1.0, 1j, b'0123456789', '0123456789'):
  296. self.assertRaises(ValueError, marshal.load,
  297. BadReader(marshal.dumps(value)))
  298. def test_eof(self):
  299. data = marshal.dumps(("hello", "dolly", None))
  300. for i in range(len(data)):
  301. self.assertRaises(EOFError, marshal.loads, data[0: i])
  302. def test_deterministic_sets(self):
  303. # bpo-37596: To support reproducible builds, sets and frozensets need to
  304. # have their elements serialized in a consistent order (even when they
  305. # have been scrambled by hash randomization):
  306. for kind in ("set", "frozenset"):
  307. for elements in (
  308. "float('nan'), b'a', b'b', b'c', 'x', 'y', 'z'",
  309. # Also test for bad interactions with backreferencing:
  310. "('Spam', 0), ('Spam', 1), ('Spam', 2), ('Spam', 3), ('Spam', 4), ('Spam', 5)",
  311. ):
  312. s = f"{kind}([{elements}])"
  313. with self.subTest(s):
  314. # First, make sure that our test case still has different
  315. # orders under hash seeds 0 and 1. If this check fails, we
  316. # need to update this test with different elements. Skip
  317. # this part if we are configured to use any other hash
  318. # algorithm (for example, using Py_HASH_EXTERNAL):
  319. if sys.hash_info.algorithm in {"fnv", "siphash24"}:
  320. args = ["-c", f"print({s})"]
  321. _, repr_0, _ = assert_python_ok(*args, PYTHONHASHSEED="0")
  322. _, repr_1, _ = assert_python_ok(*args, PYTHONHASHSEED="1")
  323. self.assertNotEqual(repr_0, repr_1)
  324. # Then, perform the actual test:
  325. args = ["-c", f"import marshal; print(marshal.dumps({s}))"]
  326. _, dump_0, _ = assert_python_ok(*args, PYTHONHASHSEED="0")
  327. _, dump_1, _ = assert_python_ok(*args, PYTHONHASHSEED="1")
  328. self.assertEqual(dump_0, dump_1)
  329. LARGE_SIZE = 2**31
  330. pointer_size = 8 if sys.maxsize > 0xFFFFFFFF else 4
  331. class NullWriter:
  332. def write(self, s):
  333. pass
  334. @unittest.skipIf(LARGE_SIZE > sys.maxsize, "test cannot run on 32-bit systems")
  335. class LargeValuesTestCase(unittest.TestCase):
  336. def check_unmarshallable(self, data):
  337. self.assertRaises(ValueError, marshal.dump, data, NullWriter())
  338. @support.bigmemtest(size=LARGE_SIZE, memuse=2, dry_run=False)
  339. def test_bytes(self, size):
  340. self.check_unmarshallable(b'x' * size)
  341. @support.bigmemtest(size=LARGE_SIZE, memuse=2, dry_run=False)
  342. def test_str(self, size):
  343. self.check_unmarshallable('x' * size)
  344. @support.bigmemtest(size=LARGE_SIZE, memuse=pointer_size + 1, dry_run=False)
  345. def test_tuple(self, size):
  346. self.check_unmarshallable((None,) * size)
  347. @support.bigmemtest(size=LARGE_SIZE, memuse=pointer_size + 1, dry_run=False)
  348. def test_list(self, size):
  349. self.check_unmarshallable([None] * size)
  350. @support.bigmemtest(size=LARGE_SIZE,
  351. memuse=pointer_size*12 + sys.getsizeof(LARGE_SIZE-1),
  352. dry_run=False)
  353. def test_set(self, size):
  354. self.check_unmarshallable(set(range(size)))
  355. @support.bigmemtest(size=LARGE_SIZE,
  356. memuse=pointer_size*12 + sys.getsizeof(LARGE_SIZE-1),
  357. dry_run=False)
  358. def test_frozenset(self, size):
  359. self.check_unmarshallable(frozenset(range(size)))
  360. @support.bigmemtest(size=LARGE_SIZE, memuse=2, dry_run=False)
  361. def test_bytearray(self, size):
  362. self.check_unmarshallable(bytearray(size))
  363. def CollectObjectIDs(ids, obj):
  364. """Collect object ids seen in a structure"""
  365. if id(obj) in ids:
  366. return
  367. ids.add(id(obj))
  368. if isinstance(obj, (list, tuple, set, frozenset)):
  369. for e in obj:
  370. CollectObjectIDs(ids, e)
  371. elif isinstance(obj, dict):
  372. for k, v in obj.items():
  373. CollectObjectIDs(ids, k)
  374. CollectObjectIDs(ids, v)
  375. return len(ids)
  376. class InstancingTestCase(unittest.TestCase, HelperMixin):
  377. keys = (123, 1.2345, 'abc', (123, 'abc'), frozenset({123, 'abc'}))
  378. def helper3(self, rsample, recursive=False, simple=False):
  379. #we have two instances
  380. sample = (rsample, rsample)
  381. n0 = CollectObjectIDs(set(), sample)
  382. for v in range(3, marshal.version + 1):
  383. s3 = marshal.dumps(sample, v)
  384. n3 = CollectObjectIDs(set(), marshal.loads(s3))
  385. #same number of instances generated
  386. self.assertEqual(n3, n0)
  387. if not recursive:
  388. #can compare with version 2
  389. s2 = marshal.dumps(sample, 2)
  390. n2 = CollectObjectIDs(set(), marshal.loads(s2))
  391. #old format generated more instances
  392. self.assertGreater(n2, n0)
  393. #if complex objects are in there, old format is larger
  394. if not simple:
  395. self.assertGreater(len(s2), len(s3))
  396. else:
  397. self.assertGreaterEqual(len(s2), len(s3))
  398. def testInt(self):
  399. intobj = 123321
  400. self.helper(intobj)
  401. self.helper3(intobj, simple=True)
  402. def testFloat(self):
  403. floatobj = 1.2345
  404. self.helper(floatobj)
  405. self.helper3(floatobj)
  406. def testStr(self):
  407. strobj = "abcde"*3
  408. self.helper(strobj)
  409. self.helper3(strobj)
  410. def testBytes(self):
  411. bytesobj = b"abcde"*3
  412. self.helper(bytesobj)
  413. self.helper3(bytesobj)
  414. def testList(self):
  415. for obj in self.keys:
  416. listobj = [obj, obj]
  417. self.helper(listobj)
  418. self.helper3(listobj)
  419. def testTuple(self):
  420. for obj in self.keys:
  421. tupleobj = (obj, obj)
  422. self.helper(tupleobj)
  423. self.helper3(tupleobj)
  424. def testSet(self):
  425. for obj in self.keys:
  426. setobj = {(obj, 1), (obj, 2)}
  427. self.helper(setobj)
  428. self.helper3(setobj)
  429. def testFrozenSet(self):
  430. for obj in self.keys:
  431. frozensetobj = frozenset({(obj, 1), (obj, 2)})
  432. self.helper(frozensetobj)
  433. self.helper3(frozensetobj)
  434. def testDict(self):
  435. for obj in self.keys:
  436. dictobj = {"hello": obj, "goodbye": obj, obj: "hello"}
  437. self.helper(dictobj)
  438. self.helper3(dictobj)
  439. def testModule(self):
  440. with open(__file__, "rb") as f:
  441. code = f.read()
  442. if __file__.endswith(".py"):
  443. code = compile(code, __file__, "exec")
  444. self.helper(code)
  445. self.helper3(code)
  446. def testRecursion(self):
  447. obj = 1.2345
  448. d = {"hello": obj, "goodbye": obj, obj: "hello"}
  449. d["self"] = d
  450. self.helper3(d, recursive=True)
  451. l = [obj, obj]
  452. l.append(l)
  453. self.helper3(l, recursive=True)
  454. class CompatibilityTestCase(unittest.TestCase):
  455. def _test(self, version):
  456. with open(__file__, "rb") as f:
  457. code = f.read()
  458. if __file__.endswith(".py"):
  459. code = compile(code, __file__, "exec")
  460. data = marshal.dumps(code, version)
  461. marshal.loads(data)
  462. def test0To3(self):
  463. self._test(0)
  464. def test1To3(self):
  465. self._test(1)
  466. def test2To3(self):
  467. self._test(2)
  468. def test3To3(self):
  469. self._test(3)
  470. class InterningTestCase(unittest.TestCase, HelperMixin):
  471. strobj = "this is an interned string"
  472. strobj = sys.intern(strobj)
  473. def testIntern(self):
  474. s = marshal.loads(marshal.dumps(self.strobj))
  475. self.assertEqual(s, self.strobj)
  476. self.assertEqual(id(s), id(self.strobj))
  477. s2 = sys.intern(s)
  478. self.assertEqual(id(s2), id(s))
  479. def testNoIntern(self):
  480. s = marshal.loads(marshal.dumps(self.strobj, 2))
  481. self.assertEqual(s, self.strobj)
  482. self.assertNotEqual(id(s), id(self.strobj))
  483. s2 = sys.intern(s)
  484. self.assertNotEqual(id(s2), id(s))
  485. @support.cpython_only
  486. @unittest.skipUnless(_testcapi, 'requires _testcapi')
  487. class CAPI_TestCase(unittest.TestCase, HelperMixin):
  488. def test_write_long_to_file(self):
  489. for v in range(marshal.version + 1):
  490. _testcapi.pymarshal_write_long_to_file(0x12345678, os_helper.TESTFN, v)
  491. with open(os_helper.TESTFN, 'rb') as f:
  492. data = f.read()
  493. os_helper.unlink(os_helper.TESTFN)
  494. self.assertEqual(data, b'\x78\x56\x34\x12')
  495. def test_write_object_to_file(self):
  496. obj = ('\u20ac', b'abc', 123, 45.6, 7+8j, 'long line '*1000)
  497. for v in range(marshal.version + 1):
  498. _testcapi.pymarshal_write_object_to_file(obj, os_helper.TESTFN, v)
  499. with open(os_helper.TESTFN, 'rb') as f:
  500. data = f.read()
  501. os_helper.unlink(os_helper.TESTFN)
  502. self.assertEqual(marshal.loads(data), obj)
  503. def test_read_short_from_file(self):
  504. with open(os_helper.TESTFN, 'wb') as f:
  505. f.write(b'\x34\x12xxxx')
  506. r, p = _testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
  507. os_helper.unlink(os_helper.TESTFN)
  508. self.assertEqual(r, 0x1234)
  509. self.assertEqual(p, 2)
  510. with open(os_helper.TESTFN, 'wb') as f:
  511. f.write(b'\x12')
  512. with self.assertRaises(EOFError):
  513. _testcapi.pymarshal_read_short_from_file(os_helper.TESTFN)
  514. os_helper.unlink(os_helper.TESTFN)
  515. def test_read_long_from_file(self):
  516. with open(os_helper.TESTFN, 'wb') as f:
  517. f.write(b'\x78\x56\x34\x12xxxx')
  518. r, p = _testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
  519. os_helper.unlink(os_helper.TESTFN)
  520. self.assertEqual(r, 0x12345678)
  521. self.assertEqual(p, 4)
  522. with open(os_helper.TESTFN, 'wb') as f:
  523. f.write(b'\x56\x34\x12')
  524. with self.assertRaises(EOFError):
  525. _testcapi.pymarshal_read_long_from_file(os_helper.TESTFN)
  526. os_helper.unlink(os_helper.TESTFN)
  527. def test_read_last_object_from_file(self):
  528. obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
  529. for v in range(marshal.version + 1):
  530. data = marshal.dumps(obj, v)
  531. with open(os_helper.TESTFN, 'wb') as f:
  532. f.write(data + b'xxxx')
  533. r, p = _testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
  534. os_helper.unlink(os_helper.TESTFN)
  535. self.assertEqual(r, obj)
  536. with open(os_helper.TESTFN, 'wb') as f:
  537. f.write(data[:1])
  538. with self.assertRaises(EOFError):
  539. _testcapi.pymarshal_read_last_object_from_file(os_helper.TESTFN)
  540. os_helper.unlink(os_helper.TESTFN)
  541. def test_read_object_from_file(self):
  542. obj = ('\u20ac', b'abc', 123, 45.6, 7+8j)
  543. for v in range(marshal.version + 1):
  544. data = marshal.dumps(obj, v)
  545. with open(os_helper.TESTFN, 'wb') as f:
  546. f.write(data + b'xxxx')
  547. r, p = _testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
  548. os_helper.unlink(os_helper.TESTFN)
  549. self.assertEqual(r, obj)
  550. self.assertEqual(p, len(data))
  551. with open(os_helper.TESTFN, 'wb') as f:
  552. f.write(data[:1])
  553. with self.assertRaises(EOFError):
  554. _testcapi.pymarshal_read_object_from_file(os_helper.TESTFN)
  555. os_helper.unlink(os_helper.TESTFN)
  556. if __name__ == "__main__":
  557. unittest.main()