mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-28 10:44:36 +08:00
Code Activity
Files
7aedaf919e0debde1a8366f1c382add828af8fc9
RustPython/Lib/test/test_struct.py
T
Jeong, YunWon a78d0eb577 Fix crashes found hunting the last open fuzzing record (#8524) 
* specialize: check the member descriptor's type before caching its slot offset

The LOAD_ATTR/STORE_ATTR specializations cached the slot offset of any member
descriptor found on the owner's type and then guarded the specialized
instruction on the type version alone, while descr_get()/descr_set() check on
every access that the instance belongs to the type the descriptor was defined
for. A descriptor taken from a wider class and bound to a narrower one read
past the instance's slot array once the cache warmed up:

    class Big:
        __slots__ = ("a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7")
    class Narrow:
        __slots__ = ("z",)
    Narrow.x = Big.__dict__["a7"]
    o = Narrow()
    for _ in range(1000):
        try: o.x
        except TypeError: pass
    # index out of bounds: the len is 1 but the index is 7 (object/core.rs)

A class with no slots at all reached the ext_ref().unwrap() on the same line.

Assisted-by: Claude

* socket: reserve recv()'s buffer fallibly

recv() and recvfrom() handed the caller's bufsize straight to
Vec::with_capacity, so an unreachable size aborted the process through
handle_alloc_error before any syscall was made:

    socket.socket().recv(2**62)
    # memory allocation of 4611686018427387904 bytes failed -> SIGABRT

try_reserve_exact reports MemoryError instead, which is what CPython raises.

Assisted-by: Claude

* types: count the __call__ and __get__ slot dispatches as recursion

Both wrappers re-enter Python without pushing a frame, so nothing counted the
nesting when the special method named the object it was looked up on:

    class C: pass
    c = C(); C.__call__ = c
    c()                                    # native stack overflow, SIGSEGV

    class D: pass
    d = D(); D.__get__ = d; D.x = d
    d.x                                    # the same, through descr_get

with_recursion around the two dispatches raises RecursionError instead, the
way Py_EnterRecursiveCall bounds a tp_call dispatch. It costs about 5% on a
__call__ dispatch and 3% on a __get__ dispatch through these wrappers.

Assisted-by: Claude

* typevar: show a ParamSpecArgs origin by its repr

ParamSpecArgs and ParamSpecKwargs fell back to a Rust `{:?}` of __origin__ when
it had no __name__. That walks the object graph natively through Debug for
PyInner, where no recursion guard sits, so a single repr() of a deeply nested
chain overflowed the native stack:

    a = object()
    for _ in range(30000):
        a = typing.ParamSpecArgs(a)
    repr(a)                                # SIGSEGV

The origin is formatted with its repr now, which is guarded, and a ParamSpec
origin is recognized by its type rather than by carrying a __name__.

Assisted-by: Claude

* Do not hold a lock across a call back into Python

Three places kept a lock while running code that can reach the same object, so
a callback that touched it wedged the process:

    _asyncio.future_add_to_awaited_by(fut, waiter)   # waiter.__hash__ adds again
    select.select(elements, [], [], 0)               # fileno() clears `elements`
    select.poll().poll(1000)                         # SIGALRM handler registers

The future's awaited-by field is read and written under its lock but the set is
built outside it, the list extraction re-reads the list on each step the way
map_iterable_object() does, and poll() waits on a copy of its descriptors. All
three ran forever before and now finish the way they do on CPython.

Assisted-by: Claude

* Validate memoryview.cast() arguments and export negative strides correctly

cast() accepted any struct format and any shape element. A zero-size
format ('0s') and a 0 in the shape both reached a division by zero;
cast() now takes only a native single character format, optionally
'@'-prefixed, and shape elements that are ints greater than zero.

A view with a negative stride starts at its last item, so the bytes it
exported began there and its own offsets walked off the front of them.
Such a view now exports the whole underlying buffer with `start` folded
into the descriptor's offsets, and zip_eq() hands over a whole run only
when both sides are contiguous in the last dimension.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Charge native recursion to the stack, not to the frame limit

with_recursion() checked the limit sys.setrecursionlimit() sets and
incremented the same counter that pushing a frame does, so a guard on a
native dispatch spent what Python code had left to call with, and did so
where sys._getframe() cannot see it: test.support.get_recursion_available()
reported frames that were no longer there. Py_EnterRecursiveCall bounds
the native stack instead, which is a separate budget, and the C stack
check with_recursion already performs is that bound.

The snippets pinning the guarded paths nest deep enough to reach the
stack rather than the frame limit.

Assisted-by: Claude

* Report a size that cannot be allocated instead of aborting on it

A size taken from Python went straight into an infallible allocation in
several places, so the process aborted through handle_alloc_error before
any exception could be raised:

- str/bytes/bytearray center(), ljust(), rjust() and zfill() reserved
  the padded result for the caller's width
- expandtabs() built its runs of spaces from a tabsize of any width;
  the argument is a C int, and a wider one does not fit
- Buffered{Reader,Writer,Random} allocated buffer_size, and read(),
  read1() and FileIO.read() their read size
- bytes(n) and bytearray(n) allocated n
- pbkdf2_hmac() allocated the derived key length, which is a C int

Each of these now reports MemoryError, or OverflowError where the
argument does not fit the type it is declared with.

new_zeroed_bytes() leaves the zeroing to the allocator, so a large
request costs the pages that are written to rather than all of them.

Assisted-by: Claude

* marshal: answer allow_code where a code object is written or read

allow_code was answered by walking the whole result a second time, with
no depth counter and no record of what it had already seen, so a value
that referred back to itself or nested deeply enough ran off the native
stack. w_object() and r_object() answer it where the code object is,
inside the walk that already bounds its depth and resolves references.

A container length is read the way r_long() reads one: it is signed, so
a length with the top bit set is out of range rather than four billion
items to reserve room for.

load() no longer holds a borrow of the buffer read() returned across the
seek() it makes afterwards.

Assisted-by: Claude

* Do not lock an object while running code that can reach it

Several places held a lock or a borrow of an object across a call back
into Python, so a callback that touched the same object waited on a lock
its own caller was holding:

- memoryview slice assignment read a source overlapping the destination,
  and __setitem__ converted the value while holding the write borrow
- BytesIO.readinto() read into a buffer viewing the same BytesIO
- array.__setitem__ converted the value under the array's write lock,
  and mmap.write() read a source viewing the same map
- bytearray.join() and bytearray.__mod__ drove Python with the
  bytearray borrowed
- array and bytearray answered "is this resizable" after taking the
  write lock, though an export is exactly a borrow someone else holds

A TextIOWrapper cookie now has to name a position inside what was
decoded in characters as well as in bytes; only the byte offset was
checked, and the character count is what read() and tell() index with.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Stop asserting a pbkdf2 message that depends on the width of a C long

The snippet asserted "key length is too great.", which pbkdf2_hmac()
only reaches once the length has been converted; where a C long is
narrower than the length asked for, the conversion fails first and says
so instead. Both are OverflowError, which is what the case is about.

test_support.test_get_recursion_depth passes now that a native recursion
guard no longer spends frames get_recursion_depth() cannot see.

Assisted-by: Claude

* Publish and read the same pointer for a thread's top frame

set_current_frame() casts the `Py<FrameObject>` it publishes straight to
`*mut FrameObject`, so ThreadSlot::top_frame holds the object's base.
sys._current_frames() read it back through Py::from_payload_ptr(), which
subtracts the payload offset from what it is given. The reference it took
therefore incremented, and later decremented, a word 48 bytes ahead of
the frame -- inside the object allocated before it, whose OnceLock state
word sits exactly there for two frames adjacent in the size class. The
neighbour then read an initialized-looking cold pointer that had never
been written and locked whatever the uninitialized word addressed, so
the thread that owned it crashed rather than the one that read.

The slot now holds `*mut Py<FrameObject>`, which is what both sides mean.

A thread parked in a call has no FrameObject for its topmost frame, so
top_frame is null there and the reader takes the materialize path
instead: test_sys.test_current_frames never reaches the branch. The
snippet takes _current_frames() against threads that are running.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Decide stop-the-world parking under the thread registry lock

do_suspend() published SUSPENDED first and only then re-read `requested`,
restoring itself to ATTACHED if the stop had ended in the meantime. That
made a thread the second writer able to leave SUSPENDED, so a stop whose
completion check had already observed the thread parked could be undone
behind the requester's back:

  worker    CAS ATTACHED -> SUSPENDED
  requester all_non_requester_suspended() -> true, world_stopped = true
  requester start_the_world(): requested = false, then walks the registry
  worker    reads requested == false, stores ATTACHED

With the store landing inside that walk the debug assertion in
start_the_world fires; with the walk already past the slot, a following
stop force-parks the thread DETACHED -> SUSPENDED, counts it as stopped,
and the store then puts it back to ATTACHED with the world declared
stopped and the thread running bytecode.

`requested` is set in init_thread_countdown() and cleared in
start_the_world() with the registry held, and start_the_world() keeps
holding it while releasing every SUSPENDED thread. Taking the registry
around the check and the transition therefore makes the two orders the
only ones possible: park before that release pass and be woken by it, or
find the request already withdrawn and stay ATTACHED. The requester is
left as the only writer that takes a thread out of SUSPENDED, and the
self-restore is gone.

suspend_if_needed() takes the VirtualMachine to reach the registry.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Keep an atexit callback alive while it is being compared

atexit.unregister() releases the callback list around each __eq__ call and
identified the entry it had compared by the address of its Box. __eq__ can
call atexit._clear(), which drops that Box, and atexit.register(), whose
new Box lands on the freed allocation; the identity search then matched
the freshly registered callback and removed it.

  atexit.register(a); atexit.register(b); atexit.register(c)
  # __eq__ runs _clear() then register(d), returns True
  atexit.unregister(probe)

left no callbacks registered where CPython leaves d.

Entries are Arc-shared now, so unregister() holds the one it is comparing
and matches it with Arc::ptr_eq: an address cannot be reused while the
comparison that named it is still running.

Assisted-by: Claude

* Hold atexit entries in PyRc rather than Arc

PyObjectRef is Send and Sync only under the threading feature, so an Arc
over a callback entry trips clippy::arc_with_non_send_sync in builds
without it, such as the wasm package. PyRc is Arc there and Rc otherwise.

Assisted-by: Claude

* Do not hold a buffer's storage while waiting for a peer

FileIO.readinto() and socket.recv_into()/recvfrom_into() took the target
buffer's write borrow and kept it for the whole call, including the wait
for data a pipe, socket or terminal may never deliver. What CPython holds
across that wait is the export, which only forbids resizing; the borrow is
a lock every other thread touching the same object waits on, so

    threading.Thread(target=lambda: sock.recv_into(buf)).start()
    len(buf)

did not answer until the peer sent. A thread parked on that lock is
ATTACHED and never reaches a safepoint, so gc.collect() in a third thread
waited for the peer as well: one incidental read of the buffer stopped the
world from being stopped at all.

The wait now runs against storage of its own and the bytes are copied over
once they arrive, with the export held throughout so the target still
cannot be resized meanwhile. A seekable file answers from itself rather
than from a peer, so FileIO.readinto() writes into the target directly
there and the buffered read path is unchanged.

Assisted-by: Claude

* Do not hold a buffer's storage while waiting to hand it over

socket.send()/sendall()/sendto()/sendmsg() and FileIO.write() kept the
source buffer's read borrow for the whole call, including the wait for a
peer that may never make room. That borrow is a lock every other thread
writing to the same object waits on, so

    threading.Thread(target=lambda: sock.sendall(buf)).start()
    buf[0] = 1

did not return until the peer read; and a thread parked there is ATTACHED
and never reaches a safepoint, so gc.collect() in a third thread waited
for the peer too -- the same wedge readinto() had on the receiving side.

ArgBytesLike::borrow_buf_unlocked() answers with bytes that survive the
borrow being dropped. An immutable object hands out a plain reference and
locks nothing, so those are sent where they lie and bytes and memoryviews
over them cost nothing; only bytes reached through a lock are copied out
first. The export is held throughout either way, so the source still
cannot be resized while it is being sent.

The regression snippet covers both directions now and is renamed for it.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Check select()'s descriptor limit while the sequence is walked

seq2set() collected the whole sequence and compared the result's length
against FD_SETSIZE afterwards. Selectable::try_from_object() calls
fileno(), which runs Python and can append to the list being walked, and
the walk re-reads the list on every step, so the collection had no end to
reach and the comparison was never made. seq2set in Modules/selectmodule.c
checks the count per element instead.

stdlib_select.py gains a fileno() that appends to its own list, and
releases its sockets with close() rather than by dropping the name.

Assisted-by: Claude

* Bound native recursion where the C stack cannot be measured

check_c_stack_overflow() answers no unconditionally under miri and on
musl, where the stack pointer is not read. Since 9c9905aff that check is
all with_recursion() does, so every guard placed on native recursion --
__call__ and __get__ dispatch among them -- was a no-op on those targets
and the nesting ran until the stack ran out.

with_recursion() now counts its own depth on those targets and refuses
past NATIVE_RECURSION_LIMIT_UNMEASURED. The count is separate from the
frame limit sys.setrecursionlimit() sets, and compiles away where the
stack pointer can be read.

Assisted-by: Claude

* Report the failure to allocate pbkdf2's key and Take.readinto's scratch

Both buffers are sized from an argument -- pbkdf2_hmac's dklen accepts up
to i32::MAX, and readinto's from the length of the destination -- and were
built with vec![0u8; n], which aborts the process on allocation failure.
new_zeroed_bytes() raises MemoryError instead.

Assisted-by: Claude

* Read the frozen-code tuple length as a signed length

The '(' branches in read_marshal_str_vec() and read_marshal_const_tuple()
took the length with read_u32() as usize, so a value with the top bit set
read as four billion items rather than as out of range. read_len() is what
every other length in this file goes through, and it reinterprets as i32.

Both readers serve deserialize_code(), which reads only the frozen modules
baked in at build time, so this changes no reachable behavior;
marshal.loads() already went through read_len().

Assisted-by: Claude

* Make the snippets from the fuzzer sweep assert what they check

builtin_str.py expanded a tab to 2**31-1 columns, allocating 2 GiB to
observe that the width is accepted; a string with no tab observes the same
acceptance without laying anything out.

stdlib_array.py caught the refusal of frombytes() on its own exported
buffer and passed silently when nothing was raised. stdlib_hashlib.py
used a bare `assert False` as its failure branch. Both now say so through
the same shapes the other snippets use.

stdlib_socket.py also accepts OverflowError from recv() with a size that
does not fit the platform's C int.

stdlib_threading_current_frames.py indexed the frame chain for "f123"
without first asserting it is there.

Assisted-by: Claude

* Ask for a marshal container's room instead of assuming it

A flagged tuple or list is published in the reference table before its
children are read, and the placeholder was built with vec![none; len].
The length is the input's to choose and read_len() lets it reach
i32::MAX, so marshal.loads(b"\xa8\xff\xff\xff\x7f") -- five bytes -- asks
for 17 GB of element slots and aborts the process where the allocator
cannot serve it. r_object() allocates the container up front too, but
PyTuple_New() reports what it cannot get.

The elements are now reserved with try_reserve_exact() and a refusal is
raised as MemoryError through the decoder's pending-error channel.
PyTuple::new_marshal_placeholder() held nothing but that allocation and
is gone; the caller builds the elements and uses new_ref().

Assisted-by: Claude

* Compare the blocking-buffer snippet against something

measure() asserted len(buf) == len(buf), which holds whatever the length
is; it now takes the length the caller expects. The pipe case compared
the drained total against the whole source, while an unbuffered write()
reports only what it transferred and a signal can cut that short; it now
compares against what write() returned.

Assisted-by: Claude

* Refuse array.frombytes() a source whose items are not bytes

frombytes() reads its argument as bytes, but accepted any contiguous
buffer, so array("i").frombytes(memoryview(array("d", [1.0]))) appended a
double's bytes read as ints instead of raising. array_array_frombytes_impl
requires an itemsize of 1; ArgBytesLike now reports the itemsize so the
same check can be made here.

The BufferError that a resize meets while the array is exported also names
the array rather than repeating bytearray's wording.

stdlib_array.py's frombytes-of-itself case used typecode "i", where the
new check answers before the resize guard is reached; it uses "b" so the
guard is what refuses, and asserts the wider case separately.

Assisted-by: Claude

* Tell apart the ways marshal data can be bad

A type byte no reader knows, a back reference that names nothing, and
TYPE_NULL all came out as a bare "bad marshal data" ValueError. r_object()
answers the first two with "bad marshal data (unknown type code)" and
"bad marshal data (invalid reference)", and read_object() answers the
third with a TypeError, "NULL object in marshal data for object", since
TYPE_NULL stands for no object rather than for a value.

MarshalError gains the three cases and deserialize_value() maps them.
The container-specific wording r_object() uses for a NULL read inside a
tuple or list is not reproduced; the exception type is.

Assisted-by: Claude

* Hash the collector's tables by address rather than by SipHash

A collection keys three sets and two maps by object address, and it visits
every tracked object and every edge between them, so the hashing is a
per-edge cost. Those tables used the default RandomState, whose SipHash
buys resistance against a caller choosing colliding keys -- and nothing
chooses these keys: they are addresses this process handed out into tables
that live and die inside one collection. A profile of gc.collect() over a
423k-object heap spent 45% of its samples in SipHash.

They now hash with a splitmix64 finalizer. The shifts matter: a table
picks its bucket from the low bits and an address arrives with those bits
zeroed by alignment, so a plain multiply leaves every object in a handful
of buckets and is slower than SipHash was.

The reachability walk also copied each object's referent vector out of the
map it was cached in, a second pass over every edge; it reads them in
place, and reference subtraction hands its vector to the map instead of
cloning it.

Measured over 423k live objects: 0.93s to 0.15s. Over 843k dead ones:
3.00s to 0.79s. extra_tests/snippets/stdlib_threading_gc_import.py, whose
collector thread calls gc.collect() in a loop, ran anywhere from 2.7s to
28s and now runs in 3.1-3.5s: a collection that takes longer leaves more
garbage for the next one to walk, so the cost fed back on itself.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* Keep the collection's candidates and their counts in one table

A collection built a set of candidates and, beside it, a map from the same
addresses to their reference counts. Both were probed for every edge in
the heap -- membership from the set, the count from the map -- so each
edge paid to hash the same address twice, and each candidate paid to be
inserted twice. The map alone answers both questions.

The candidates also keep a walkable order now, which the reference
subtraction pass needs since it writes the counts while reading the
candidates, and which the unreachable set is built from instead of a
set difference.

Over the 423k-object heap measured in the previous commit: 0.15s to
0.13s live, and 0.79s to 0.49s dead.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* gc: collect referents into one buffer instead of a vector per object

Step 3 allocated a `Vec` for every tracked object to hold its referents
and kept them all in a map until step 4 read them back. The referents now
go into a single growing buffer, with the map holding each object's range
into it.

Adds `PyObject::gc_extend_referent_ptrs`, which appends to a caller's
buffer; `gc_get_referent_ptrs` calls it with a fresh one.

Assisted-by: Claude

* memoryview and struct: match the checks and errors of the reference

memoryview:
- `cast()` accepted a source and destination that are both item types,
  which reinterprets the items rather than re-dividing the bytes; one side
  now has to be a byte format.
- A cast to `shape=()` returned without checking that the buffer holds
  exactly the one item that shape describes.
- `hash()` hashes the bytes, so it now raises ValueError for a view whose
  items are not bytes, rather than returning a hash that disagrees with the
  value the view compares equal to.
- `tobytes()` takes the `order` argument, with 'F' walking a
  multidimensional view down its columns; `BufferDescriptor` gained
  `for_each_segment_fortran` for that walk.

struct:
- A value the format has no room for reported "argument out of range"
  instead of naming the format and its range. The format character is now
  passed to the packing functions to report it.
- `Struct.__new__` no longer reads the format; `__init__` does, so
  `__init__` can be called again and a subclass can pass the format up.
  Methods raise RuntimeError until it has run, and `Struct` is a base type.

Removes the expectedFailure from test_Struct_reinitialization and
test_struct_subclass_instantiation.

Assisted-by: Claude
Assisted-by: Codex:GPT-5

* io: decide the readinto path by file type, not by seekability

FileIO.readinto wrote straight into the caller's buffer, holding its write
borrow, when the fd was seekable; otherwise it read aside into scratch and
copied. Seekability stood in for "this read answers without waiting on a
peer", which a pipe on Windows breaks: lseek on one succeeds, so the pipe
took the borrow-holding path and every other thread touching that bytearray
waited for the peer.

host_io::reads_without_waiting answers it directly -- seekability elsewhere,
GetFileType() == FILE_TYPE_DISK on Windows.

The regression snippet times each operation separately, so a failure names
the one that waited; it asserts the transfer is still in flight before
checking the export; and the socket case fills the connection until it
refuses rather than assuming a size that outruns it, which SO_SNDBUF on an
already-connected pair does not settle.

Assisted-by: Claude
2026-08-17 00:33:27 +09:00

1089 lines
45 KiB
Python
Vendored
Raw Blame History

from collections import abc
from itertools import combinations
import array
import gc
import math
import operator
import unittest
import struct
import sys
import weakref
from test import support
from test.support import import_helper
from test.support.script_helper import assert_python_ok
from test.support.testcase import ComplexesAreIdenticalMixin
ISBIGENDIAN = sys.byteorder == "big"
integer_codes = 'b', 'B', 'h', 'H', 'i', 'I', 'l', 'L', 'q', 'Q', 'n', 'N'
byteorders = '', '@', '=', '<', '>', '!'
INF = float('inf')
NAN = float('nan')
def iter_integer_formats(byteorders=byteorders):
for code in integer_codes:
for byteorder in byteorders:
if (byteorder not in ('', '@') and code in ('n', 'N')):
continue
yield code, byteorder
def string_reverse(s):
return s[::-1]
def bigendian_to_native(value):
if ISBIGENDIAN:
return value
else:
return string_reverse(value)
class StructTest(ComplexesAreIdenticalMixin, unittest.TestCase):
def test_isbigendian(self):
self.assertEqual((struct.pack('=i', 1)[0] == 0), ISBIGENDIAN)
def test_consistence(self):
self.assertRaises(struct.error, struct.calcsize, 'Z')
sz = struct.calcsize('i')
self.assertEqual(sz * 3, struct.calcsize('iii'))
fmt = 'cbxxxxxxhhhhiillffd?'
fmt3 = '3c3b18x12h6i6l6f3d3?'
sz = struct.calcsize(fmt)
sz3 = struct.calcsize(fmt3)
self.assertEqual(sz * 3, sz3)
self.assertRaises(struct.error, struct.pack, 'iii', 3)
self.assertRaises(struct.error, struct.pack, 'i', 3, 3, 3)
self.assertRaises((TypeError, struct.error), struct.pack, 'i', 'foo')
self.assertRaises((TypeError, struct.error), struct.pack, 'P', 'foo')
self.assertRaises(struct.error, struct.unpack, 'd', b'flap')
s = struct.pack('ii', 1, 2)
self.assertRaises(struct.error, struct.unpack, 'iii', s)
self.assertRaises(struct.error, struct.unpack, 'i', s)
def test_transitiveness(self):
c = b'a'
b = 1
h = 255
i = 65535
l = 65536
f = 3.1415
d = 3.1415
t = True
for prefix in ('', '@', '<', '>', '=', '!'):
for format in ('xcbhilfd?', 'xcBHILfd?'):
format = prefix + format
s = struct.pack(format, c, b, h, i, l, f, d, t)
cp, bp, hp, ip, lp, fp, dp, tp = struct.unpack(format, s)
self.assertEqual(cp, c)
self.assertEqual(bp, b)
self.assertEqual(hp, h)
self.assertEqual(ip, i)
self.assertEqual(lp, l)
self.assertEqual(int(100 * fp), int(100 * f))
self.assertEqual(int(100 * dp), int(100 * d))
self.assertEqual(tp, t)
def test_new_features(self):
# Test some of the new features in detail
# (format, argument, big-endian result, little-endian result, asymmetric)
tests = [
('c', b'a', b'a', b'a', 0),
('xc', b'a', b'\0a', b'\0a', 0),
('cx', b'a', b'a\0', b'a\0', 0),
('s', b'a', b'a', b'a', 0),
('0s', b'helloworld', b'', b'', 1),
('1s', b'helloworld', b'h', b'h', 1),
('9s', b'helloworld', b'helloworl', b'helloworl', 1),
('10s', b'helloworld', b'helloworld', b'helloworld', 0),
('11s', b'helloworld', b'helloworld\0', b'helloworld\0', 1),
('20s', b'helloworld', b'helloworld'+10*b'\0', b'helloworld'+10*b'\0', 1),
('0p', b'helloworld', b'', b'', 1),
('1p', b'helloworld', b'\x00', b'\x00', 1),
('2p', b'helloworld', b'\x01h', b'\x01h', 1),
('10p', b'helloworld', b'\x09helloworl', b'\x09helloworl', 1),
('11p', b'helloworld', b'\x0Ahelloworld', b'\x0Ahelloworld', 0),
('12p', b'helloworld', b'\x0Ahelloworld\0', b'\x0Ahelloworld\0', 1),
('20p', b'helloworld', b'\x0Ahelloworld'+9*b'\0', b'\x0Ahelloworld'+9*b'\0', 1),
('b', 7, b'\7', b'\7', 0),
('b', -7, b'\371', b'\371', 0),
('B', 7, b'\7', b'\7', 0),
('B', 249, b'\371', b'\371', 0),
('h', 700, b'\002\274', b'\274\002', 0),
('h', -700, b'\375D', b'D\375', 0),
('H', 700, b'\002\274', b'\274\002', 0),
('H', 0x10000-700, b'\375D', b'D\375', 0),
('i', 70000000, b'\004,\035\200', b'\200\035,\004', 0),
('i', -70000000, b'\373\323\342\200', b'\200\342\323\373', 0),
('I', 70000000, b'\004,\035\200', b'\200\035,\004', 0),
('I', 0x100000000-70000000, b'\373\323\342\200', b'\200\342\323\373', 0),
('l', 70000000, b'\004,\035\200', b'\200\035,\004', 0),
('l', -70000000, b'\373\323\342\200', b'\200\342\323\373', 0),
('L', 70000000, b'\004,\035\200', b'\200\035,\004', 0),
('L', 0x100000000-70000000, b'\373\323\342\200', b'\200\342\323\373', 0),
('f', 2.0, b'@\000\000\000', b'\000\000\000@', 0),
('d', 2.0, b'@\000\000\000\000\000\000\000',
b'\000\000\000\000\000\000\000@', 0),
('f', -2.0, b'\300\000\000\000', b'\000\000\000\300', 0),
('d', -2.0, b'\300\000\000\000\000\000\000\000',
b'\000\000\000\000\000\000\000\300', 0),
('?', 0, b'\0', b'\0', 0),
('?', 3, b'\1', b'\1', 1),
('?', True, b'\1', b'\1', 0),
('?', [], b'\0', b'\0', 1),
('?', (1,), b'\1', b'\1', 1),
]
for fmt, arg, big, lil, asy in tests:
for (xfmt, exp) in [('>'+fmt, big), ('!'+fmt, big), ('<'+fmt, lil),
('='+fmt, ISBIGENDIAN and big or lil)]:
res = struct.pack(xfmt, arg)
self.assertEqual(res, exp)
self.assertEqual(struct.calcsize(xfmt), len(res))
rev = struct.unpack(xfmt, res)[0]
if rev != arg:
self.assertTrue(asy)
def test_calcsize(self):
expected_size = {
'b': 1, 'B': 1,
'h': 2, 'H': 2,
'i': 4, 'I': 4,
'l': 4, 'L': 4,
'q': 8, 'Q': 8,
}
# standard integer sizes
for code, byteorder in iter_integer_formats(('=', '<', '>', '!')):
format = byteorder+code
size = struct.calcsize(format)
self.assertEqual(size, expected_size[code])
# native integer sizes
native_pairs = 'bB', 'hH', 'iI', 'lL', 'nN', 'qQ'
for format_pair in native_pairs:
for byteorder in '', '@':
signed_size = struct.calcsize(byteorder + format_pair[0])
unsigned_size = struct.calcsize(byteorder + format_pair[1])
self.assertEqual(signed_size, unsigned_size)
# bounds for native integer sizes
self.assertEqual(struct.calcsize('b'), 1)
self.assertLessEqual(2, struct.calcsize('h'))
self.assertLessEqual(4, struct.calcsize('l'))
self.assertLessEqual(struct.calcsize('h'), struct.calcsize('i'))
self.assertLessEqual(struct.calcsize('i'), struct.calcsize('l'))
self.assertLessEqual(8, struct.calcsize('q'))
self.assertLessEqual(struct.calcsize('l'), struct.calcsize('q'))
self.assertGreaterEqual(struct.calcsize('n'), struct.calcsize('i'))
self.assertGreaterEqual(struct.calcsize('n'), struct.calcsize('P'))
def test_integers(self):
# Integer tests (bBhHiIlLqQnN).
import binascii
class IntTester(unittest.TestCase):
def __init__(self, format):
super(IntTester, self).__init__(methodName='test_one')
self.format = format
self.code = format[-1]
self.byteorder = format[:-1]
if not self.byteorder in byteorders:
raise ValueError("unrecognized packing byteorder: %s" %
self.byteorder)
self.bytesize = struct.calcsize(format)
self.bitsize = self.bytesize * 8
if self.code in tuple('bhilqn'):
self.signed = True
self.min_value = -(2**(self.bitsize-1))
self.max_value = 2**(self.bitsize-1) - 1
elif self.code in tuple('BHILQN'):
self.signed = False
self.min_value = 0
self.max_value = 2**self.bitsize - 1
else:
raise ValueError("unrecognized format code: %s" %
self.code)
def test_one(self, x, pack=struct.pack,
unpack=struct.unpack,
unhexlify=binascii.unhexlify):
format = self.format
if self.min_value <= x <= self.max_value:
expected = x
if self.signed and x < 0:
expected += 1 << self.bitsize
self.assertGreaterEqual(expected, 0)
expected = '%x' % expected
if len(expected) & 1:
expected = "0" + expected
expected = expected.encode('ascii')
expected = unhexlify(expected)
expected = (b"\x00" * (self.bytesize - len(expected)) +
expected)
if (self.byteorder == '<' or
self.byteorder in ('', '@', '=') and not ISBIGENDIAN):
expected = string_reverse(expected)
self.assertEqual(len(expected), self.bytesize)
# Pack work?
got = pack(format, x)
self.assertEqual(got, expected)
# Unpack work?
retrieved = unpack(format, got)[0]
self.assertEqual(x, retrieved)
# Adding any byte should cause a "too big" error.
self.assertRaises((struct.error, TypeError), unpack, format,
b'\x01' + got)
else:
# x is out of range -- verify pack realizes that.
self.assertRaises((OverflowError, ValueError, struct.error),
pack, format, x)
def run(self):
from random import randrange
# Create all interesting powers of 2.
values = []
for exp in range(self.bitsize + 3):
values.append(1 << exp)
# Add some random values.
for i in range(self.bitsize):
val = 0
for j in range(self.bytesize):
val = (val << 8) | randrange(256)
values.append(val)
# Values absorbed from other tests
values.extend([300, 700000, sys.maxsize*4])
# Try all those, and their negations, and +-1 from
# them. Note that this tests all power-of-2
# boundaries in range, and a few out of range, plus
# +-(2**n +- 1).
for base in values:
for val in -base, base:
for incr in -1, 0, 1:
x = val + incr
self.test_one(x)
# Some error cases.
class NotAnInt:
def __int__(self):
return 42
# Objects with an '__index__' method should be allowed
# to pack as integers. That is assuming the implemented
# '__index__' method returns an 'int'.
class Indexable(object):
def __init__(self, value):
self._value = value
def __index__(self):
return self._value
# If the '__index__' method raises a type error, then
# '__int__' should be used with a deprecation warning.
class BadIndex(object):
def __index__(self):
raise TypeError
def __int__(self):
return 42
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
"a string")
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
randrange)
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
3+42j)
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
NotAnInt())
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
BadIndex())
# Check for legitimate values from '__index__'.
for obj in (Indexable(0), Indexable(10), Indexable(17),
Indexable(42), Indexable(100), Indexable(127)):
try:
struct.pack(format, obj)
except:
self.fail("integer code pack failed on object "
"with '__index__' method")
# Check for bogus values from '__index__'.
for obj in (Indexable(b'a'), Indexable('b'), Indexable(None),
Indexable({'a': 1}), Indexable([1, 2, 3])):
self.assertRaises((TypeError, struct.error),
struct.pack, self.format,
obj)
for code, byteorder in iter_integer_formats():
format = byteorder+code
t = IntTester(format)
t.run()
def test_nN_code(self):
# n and N don't exist in standard sizes
def assertStructError(func, *args, **kwargs):
with self.assertRaises(struct.error) as cm:
func(*args, **kwargs)
self.assertIn("bad char in struct format", str(cm.exception))
for code in 'nN':
for byteorder in ('=', '<', '>', '!'):
format = byteorder+code
assertStructError(struct.calcsize, format)
assertStructError(struct.pack, format, 0)
assertStructError(struct.unpack, format, b"")
def test_p_code(self):
# Test p ("Pascal string") code.
for code, input, expected, expectedback in [
('0p', b'abc', b'', b''),
('p', b'abc', b'\x00', b''),
('1p', b'abc', b'\x00', b''),
('2p', b'abc', b'\x01a', b'a'),
('3p', b'abc', b'\x02ab', b'ab'),
('4p', b'abc', b'\x03abc', b'abc'),
('5p', b'abc', b'\x03abc\x00', b'abc'),
('6p', b'abc', b'\x03abc\x00\x00', b'abc'),
('1000p', b'x'*1000, b'\xff' + b'x'*999, b'x'*255)]:
got = struct.pack(code, input)
self.assertEqual(got, expected)
(got,) = struct.unpack(code, got)
self.assertEqual(got, expectedback)
def test_705836(self):
# SF bug 705836. "<f" and ">f" had a severe rounding bug, where a carry
# from the low-order discarded bits could propagate into the exponent
# field, causing the result to be wrong by a factor of 2.
for base in range(1, 33):
# smaller <- largest representable float less than base.
delta = 0.5
while base - delta / 2.0 != base:
delta /= 2.0
smaller = base - delta
# Packing this rounds away a solid string of trailing 1 bits.
packed = struct.pack("<f", smaller)
unpacked = struct.unpack("<f", packed)[0]
# This failed at base = 2, 4, and 32, with unpacked = 1, 2, and
# 16, respectively.
self.assertEqual(base, unpacked)
bigpacked = struct.pack(">f", smaller)
self.assertEqual(bigpacked, string_reverse(packed))
unpacked = struct.unpack(">f", bigpacked)[0]
self.assertEqual(base, unpacked)
# Largest finite IEEE single.
big = (1 << 24) - 1
big = math.ldexp(big, 127 - 23)
packed = struct.pack(">f", big)
unpacked = struct.unpack(">f", packed)[0]
self.assertEqual(big, unpacked)
# The same, but tack on a 1 bit so it rounds up to infinity.
big = (1 << 25) - 1
big = math.ldexp(big, 127 - 24)
self.assertRaises(OverflowError, struct.pack, ">f", big)
self.assertRaises(OverflowError, struct.pack, "<f", big)
# same for native format, see gh-145633
self.assertRaises(OverflowError, struct.pack, "f", big)
# And for half-floats
big = (1 << 11) - 1
big = math.ldexp(big, 15 - 10)
packed = struct.pack(">e", big)
unpacked = struct.unpack(">e", packed)[0]
self.assertEqual(big, unpacked)
big = (1 << 12) - 1
big = math.ldexp(big, 15 - 11)
self.assertRaises(OverflowError, struct.pack, ">e", big)
self.assertRaises(OverflowError, struct.pack, "<e", big)
self.assertRaises(OverflowError, struct.pack, "e", big)
def test_1530559(self):
for code, byteorder in iter_integer_formats():
format = byteorder + code
self.assertRaises(struct.error, struct.pack, format, 1.0)
self.assertRaises(struct.error, struct.pack, format, 1.5)
self.assertRaises(struct.error, struct.pack, 'P', 1.0)
self.assertRaises(struct.error, struct.pack, 'P', 1.5)
def test_unpack_from(self):
test_string = b'abcd01234'
fmt = '4s'
s = struct.Struct(fmt)
for cls in (bytes, bytearray):
data = cls(test_string)
self.assertEqual(s.unpack_from(data), (b'abcd',))
self.assertEqual(s.unpack_from(data, 2), (b'cd01',))
self.assertEqual(s.unpack_from(data, 4), (b'0123',))
for i in range(6):
self.assertEqual(s.unpack_from(data, i), (data[i:i+4],))
for i in range(6, len(test_string) + 1):
self.assertRaises(struct.error, s.unpack_from, data, i)
for cls in (bytes, bytearray):
data = cls(test_string)
self.assertEqual(struct.unpack_from(fmt, data), (b'abcd',))
self.assertEqual(struct.unpack_from(fmt, data, 2), (b'cd01',))
self.assertEqual(struct.unpack_from(fmt, data, 4), (b'0123',))
for i in range(6):
self.assertEqual(struct.unpack_from(fmt, data, i), (data[i:i+4],))
for i in range(6, len(test_string) + 1):
self.assertRaises(struct.error, struct.unpack_from, fmt, data, i)
# keyword arguments
self.assertEqual(s.unpack_from(buffer=test_string, offset=2),
(b'cd01',))
def _test_pack_into(self, pack_into):
test_string = b'Reykjavik rocks, eow!'
writable_buf = memoryview(array.array('b', b' '*100))
# Test without offset
pack_into(writable_buf, 0, test_string)
from_buf = writable_buf.tobytes()[:len(test_string)]
self.assertEqual(from_buf, test_string)
# Test with offset.
pack_into(writable_buf, 10, test_string)
from_buf = writable_buf.tobytes()[:len(test_string)+10]
self.assertEqual(from_buf, test_string[:10] + test_string)
# Test with negative offset.
pack_into(writable_buf, -30, test_string)
from_buf = writable_buf.tobytes()[-30:-30+len(test_string)]
self.assertEqual(from_buf, test_string)
# Go beyond boundaries.
small_buf = array.array('b', b' '*10)
with self.assertRaises((ValueError, struct.error)):
pack_into(small_buf, 0, test_string)
with self.assertRaises((ValueError, struct.error)):
pack_into(writable_buf, 90, test_string)
with self.assertRaises((ValueError, struct.error)):
pack_into(writable_buf, -10, test_string)
with self.assertRaises((ValueError, struct.error)):
pack_into(writable_buf, 150, test_string)
with self.assertRaises((ValueError, struct.error)):
pack_into(writable_buf, -150, test_string)
# Test invalid buffer.
self.assertRaises(TypeError, pack_into, b' '*100, 0, test_string)
self.assertRaises(TypeError, pack_into, ' '*100, 0, test_string)
self.assertRaises(TypeError, pack_into, [0]*100, 0, test_string)
self.assertRaises(TypeError, pack_into, None, 0, test_string)
self.assertRaises(TypeError, pack_into, writable_buf[::2], 0, test_string)
self.assertRaises(TypeError, pack_into, writable_buf[::-1], 0, test_string)
# Test bogus offset (issue bpo-3694)
with self.assertRaises(TypeError):
pack_into(writable_buf, None, test_string)
with self.assertRaises(TypeError):
pack_into(writable_buf, 0.0, test_string)
with self.assertRaises((IndexError, OverflowError)):
pack_into(writable_buf, 2**1000, test_string)
with self.assertRaises((IndexError, OverflowError)):
pack_into(writable_buf, -2**1000, test_string)
def test_pack_into(self):
s = struct.Struct('21s')
self._test_pack_into(s.pack_into)
def test_pack_into_fn(self):
pack_into = lambda *args: struct.pack_into('21s', *args)
self._test_pack_into(pack_into)
def test_unpack_with_buffer(self):
# SF bug 1563759: struct.unpack doesn't support buffer protocol objects
data1 = array.array('B', b'\x12\x34\x56\x78')
data2 = memoryview(b'\x12\x34\x56\x78') # XXX b'......XXXX......', 6, 4
for data in [data1, data2]:
value, = struct.unpack('>I', data)
self.assertEqual(value, 0x12345678)
def test_bool(self):
class ExplodingBool(object):
def __bool__(self):
raise OSError
for prefix in tuple("<>!=")+('',):
false = (), [], [], '', 0
true = [1], 'test', 5, -1, 0xffffffff+1, 0xffffffff/2
falseFormat = prefix + '?' * len(false)
packedFalse = struct.pack(falseFormat, *false)
unpackedFalse = struct.unpack(falseFormat, packedFalse)
trueFormat = prefix + '?' * len(true)
packedTrue = struct.pack(trueFormat, *true)
unpackedTrue = struct.unpack(trueFormat, packedTrue)
self.assertEqual(len(true), len(unpackedTrue))
self.assertEqual(len(false), len(unpackedFalse))
for t in unpackedFalse:
self.assertFalse(t)
for t in unpackedTrue:
self.assertTrue(t)
packed = struct.pack(prefix+'?', 1)
self.assertEqual(len(packed), struct.calcsize(prefix+'?'))
if len(packed) != 1:
self.assertFalse(prefix, msg='encoded bool is not one byte: %r'
%packed)
try:
struct.pack(prefix + '?', ExplodingBool())
except OSError:
pass
else:
self.fail("Expected OSError: struct.pack(%r, "
"ExplodingBool())" % (prefix + '?'))
for c in [b'\x01', b'\x7f', b'\xff', b'\x0f', b'\xf0']:
self.assertTrue(struct.unpack('>?', c)[0])
self.assertTrue(struct.unpack('<?', c)[0])
self.assertTrue(struct.unpack('=?', c)[0])
self.assertTrue(struct.unpack('@?', c)[0])
def test_count_overflow(self):
hugecount = '{}b'.format(sys.maxsize+1)
self.assertRaises(struct.error, struct.calcsize, hugecount)
hugecount2 = '{}b{}H'.format(sys.maxsize//2, sys.maxsize//2)
self.assertRaises(struct.error, struct.calcsize, hugecount2)
hugecount3 = '{}i{}q'.format(sys.maxsize // 4, sys.maxsize // 8)
self.assertRaises(struct.error, struct.calcsize, hugecount3)
hugecount4 = '{}?s'.format(sys.maxsize)
self.assertRaises(struct.error, struct.calcsize, hugecount4)
hugecount5 = '{}?p'.format(sys.maxsize)
self.assertRaises(struct.error, struct.calcsize, hugecount5)
def test_trailing_counter(self):
store = array.array('b', b' '*100)
# format lists containing only count spec should result in an error
self.assertRaises(struct.error, struct.pack, '12345')
self.assertRaises(struct.error, struct.unpack, '12345', b'')
self.assertRaises(struct.error, struct.pack_into, '12345', store, 0)
self.assertRaises(struct.error, struct.unpack_from, '12345', store, 0)
# Format lists with trailing count spec should result in an error
self.assertRaises(struct.error, struct.pack, 'c12345', 'x')
self.assertRaises(struct.error, struct.unpack, 'c12345', b'x')
self.assertRaises(struct.error, struct.pack_into, 'c12345', store, 0,
'x')
self.assertRaises(struct.error, struct.unpack_from, 'c12345', store,
0)
# Mixed format tests
self.assertRaises(struct.error, struct.pack, '14s42', 'spam and eggs')
self.assertRaises(struct.error, struct.unpack, '14s42',
b'spam and eggs')
self.assertRaises(struct.error, struct.pack_into, '14s42', store, 0,
'spam and eggs')
self.assertRaises(struct.error, struct.unpack_from, '14s42', store, 0)
def test_Struct_reinitialization(self):
# Issue 9422: there was a memory leak when reinitializing a
# Struct instance. This test can be used to detect the leak
# when running with regrtest -L.
s = struct.Struct('>h')
s.__init__('>hh')
self.assertEqual(s.format, '>hh')
packed = b'\x00\x01\x00\x02'
self.assertEqual(s.pack(1, 2), packed)
self.assertEqual(s.unpack(packed), (1, 2))
with self.assertRaises(UnicodeEncodeError):
s.__init__('\udc00')
self.assertEqual(s.format, '>hh')
self.assertEqual(s.pack(1, 2), packed)
self.assertEqual(s.unpack(packed), (1, 2))
with self.assertRaises(struct.error):
s.__init__('$')
self.assertEqual(s.format, '>hh')
self.assertEqual(s.pack(1, 2), packed)
self.assertEqual(s.unpack(packed), (1, 2))
def check_sizeof(self, format_str, number_of_codes):
# The size of 'PyStructObject'
totalsize = support.calcobjsize('2n3P')
# The size taken up by the 'formatcode' dynamic array
totalsize += struct.calcsize('P3n0P') * (number_of_codes + 1)
support.check_sizeof(self, struct.Struct(format_str), totalsize)
@support.cpython_only
def test__sizeof__(self):
for code in integer_codes:
self.check_sizeof(code, 1)
self.check_sizeof('BHILfdspP', 9)
self.check_sizeof('B' * 1234, 1234)
self.check_sizeof('fd', 2)
self.check_sizeof('xxxxxxxxxxxxxx', 0)
self.check_sizeof('100H', 1)
self.check_sizeof('187s', 1)
self.check_sizeof('20p', 1)
self.check_sizeof('0s', 1)
self.check_sizeof('0p', 1)
self.check_sizeof('0c', 0)
def test_boundary_error_message(self):
regex1 = (
r'pack_into requires a buffer of at least 6 '
r'bytes for packing 1 bytes at offset 5 '
r'\(actual buffer size is 1\)'
)
with self.assertRaisesRegex(struct.error, regex1):
struct.pack_into('b', bytearray(1), 5, 1)
regex2 = (
r'unpack_from requires a buffer of at least 6 '
r'bytes for unpacking 1 bytes at offset 5 '
r'\(actual buffer size is 1\)'
)
with self.assertRaisesRegex(struct.error, regex2):
struct.unpack_from('b', bytearray(1), 5)
def test_boundary_error_message_with_negative_offset(self):
byte_list = bytearray(10)
with self.assertRaisesRegex(
struct.error,
r'no space to pack 4 bytes at offset -2'):
struct.pack_into('<I', byte_list, -2, 123)
with self.assertRaisesRegex(
struct.error,
'offset -11 out of range for 10-byte buffer'):
struct.pack_into('<B', byte_list, -11, 123)
with self.assertRaisesRegex(
struct.error,
r'not enough data to unpack 4 bytes at offset -2'):
struct.unpack_from('<I', byte_list, -2)
with self.assertRaisesRegex(
struct.error,
"offset -11 out of range for 10-byte buffer"):
struct.unpack_from('<B', byte_list, -11)
def test_boundary_error_message_with_large_offset(self):
# Test overflows cause by large offset and value size (issue 30245)
regex1 = (
r'pack_into requires a buffer of at least ' + str(sys.maxsize + 4) +
r' bytes for packing 4 bytes at offset ' + str(sys.maxsize) +
r' \(actual buffer size is 10\)'
)
with self.assertRaisesRegex(struct.error, regex1):
struct.pack_into('<I', bytearray(10), sys.maxsize, 1)
regex2 = (
r'unpack_from requires a buffer of at least ' + str(sys.maxsize + 4) +
r' bytes for unpacking 4 bytes at offset ' + str(sys.maxsize) +
r' \(actual buffer size is 10\)'
)
with self.assertRaisesRegex(struct.error, regex2):
struct.unpack_from('<I', bytearray(10), sys.maxsize)
def test_issue29802(self):
# When the second argument of struct.unpack() was of wrong type
# the Struct object was decrefed twice and the reference to
# deallocated object was left in a cache.
with self.assertRaises(TypeError):
struct.unpack('b', 0)
# Shouldn't crash.
self.assertEqual(struct.unpack('b', b'a'), (b'a'[0],))
def test_format_attr(self):
s = struct.Struct('=i2H')
self.assertEqual(s.format, '=i2H')
# use a bytes string
s2 = struct.Struct(s.format.encode())
self.assertEqual(s2.format, s.format)
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_struct_cleans_up_at_runtime_shutdown(self):
code = """if 1:
import struct
class C:
def __init__(self):
self.pack = struct.pack
def __del__(self):
self.pack('I', -42)
struct.x = C()
"""
rc, stdout, stderr = assert_python_ok("-c", code)
self.assertEqual(rc, 0)
self.assertEqual(stdout.rstrip(), b"")
self.assertIn(b"Exception ignored while calling deallocator", stderr)
self.assertIn(b"C.__del__", stderr)
def test__struct_reference_cycle_cleaned_up(self):
# Regression test for python/cpython#94207.
# When we create a new struct module, trigger use of its cache,
# and then delete it ...
_struct_module = import_helper.import_fresh_module("_struct")
module_ref = weakref.ref(_struct_module)
_struct_module.calcsize("b")
del _struct_module
# Then the module should have been garbage collected.
gc.collect()
self.assertIsNone(
module_ref(), "_struct module was not garbage collected")
@support.cpython_only
def test__struct_types_immutable(self):
# See https://github.com/python/cpython/issues/94254
Struct = struct.Struct
unpack_iterator = type(struct.iter_unpack("b", b'x'))
for cls in (Struct, unpack_iterator):
with self.subTest(cls=cls):
with self.assertRaises(TypeError):
cls.x = 1
def test_issue35714(self):
# Embedded null characters should not be allowed in format strings.
for s in '\0', '2\0i', b'\0':
with self.assertRaisesRegex(struct.error,
'embedded null character'):
struct.calcsize(s)
@support.cpython_only
def test_issue98248(self):
def test_error_msg(prefix, int_type, is_unsigned):
fmt_str = prefix + int_type
size = struct.calcsize(fmt_str)
if is_unsigned:
max_ = 2 ** (size * 8) - 1
min_ = 0
else:
max_ = 2 ** (size * 8 - 1) - 1
min_ = -2 ** (size * 8 - 1)
error_msg = f"'{int_type}' format requires {min_} <= number <= {max_}"
for number in [int(-1e50), min_ - 1, max_ + 1, int(1e50)]:
with self.subTest(format_str=fmt_str, number=number):
with self.assertRaisesRegex(struct.error, error_msg):
struct.pack(fmt_str, number)
error_msg = "required argument is not an integer"
not_number = ""
with self.subTest(format_str=fmt_str, number=not_number):
with self.assertRaisesRegex(struct.error, error_msg):
struct.pack(fmt_str, not_number)
for prefix in '@=<>':
for int_type in 'BHILQ':
test_error_msg(prefix, int_type, True)
for int_type in 'bhilq':
test_error_msg(prefix, int_type, False)
int_type = 'N'
test_error_msg('@', int_type, True)
int_type = 'n'
test_error_msg('@', int_type, False)
@support.cpython_only
def test_issue98248_error_propagation(self):
class Div0:
def __index__(self):
1 / 0
def test_error_propagation(fmt_str):
with self.subTest(format_str=fmt_str, exception="ZeroDivisionError"):
with self.assertRaises(ZeroDivisionError):
struct.pack(fmt_str, Div0())
for prefix in '@=<>':
for int_type in 'BHILQbhilq':
test_error_propagation(prefix + int_type)
test_error_propagation('N')
test_error_propagation('n')
def test_struct_subclass_instantiation(self):
# Regression test for https://github.com/python/cpython/issues/112358
class MyStruct(struct.Struct):
def __init__(self):
super().__init__('>h')
my_struct = MyStruct()
self.assertEqual(my_struct.pack(12345), b'\x30\x39')
def test_repr(self):
s = struct.Struct('=i2H')
self.assertEqual(repr(s), f'Struct({s.format!r})')
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_c_complex_round_trip(self):
values = [complex(*_) for _ in combinations([1, -1, 0.0, -0.0, 2,
-3, INF, -INF, NAN], 2)]
for z in values:
for f in ['F', 'D', '>F', '>D', '<F', '<D']:
with self.subTest(z=z, format=f):
round_trip = struct.unpack(f, struct.pack(f, z))[0]
self.assertComplexesAreIdentical(z, round_trip)
@unittest.skipIf(
support.is_android or support.is_apple_mobile,
"Subinterpreters are not supported on Android and iOS"
)
def test_endian_table_init_subinterpreters(self):
# Verify that the _struct extension module can be initialized
# concurrently in subinterpreters (gh-140260).
try:
from concurrent.futures import InterpreterPoolExecutor
except ImportError:
raise unittest.SkipTest("InterpreterPoolExecutor not available")
code = "import struct"
with InterpreterPoolExecutor(max_workers=5) as executor:
results = executor.map(exec, [code] * 5)
self.assertListEqual(list(results), [None] * 5)
@unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: expected at least 1 arguments, got 0
def test_operations_on_half_initialized_Struct(self):
S = struct.Struct.__new__(struct.Struct)
spam = array.array('b', b' ')
self.assertRaises(RuntimeError, S.iter_unpack, spam)
self.assertRaises(RuntimeError, S.pack, 1)
self.assertRaises(RuntimeError, S.pack_into, spam, 1)
self.assertRaises(RuntimeError, S.unpack, spam)
self.assertRaises(RuntimeError, S.unpack_from, spam)
self.assertRaises(RuntimeError, getattr, S, 'format')
self.assertRaises(RuntimeError, S.__sizeof__)
self.assertRaises(RuntimeError, repr, S)
self.assertEqual(S.size, -1)
def test_float_round_trip(self):
for format in (
"f", "<f", ">f",
"d", "<d", ">d",
"e", "<e", ">e",
):
with self.subTest(format=format):
f = struct.unpack(format, struct.pack(format, 1.5))[0]
self.assertEqual(f, 1.5)
f = struct.unpack(format, struct.pack(format, NAN))[0]
self.assertTrue(math.isnan(f), f)
f = struct.unpack(format, struct.pack(format, INF))[0]
self.assertTrue(math.isinf(f), f)
self.assertEqual(math.copysign(1.0, f), 1.0)
f = struct.unpack(format, struct.pack(format, -INF))[0]
self.assertTrue(math.isinf(f), f)
self.assertEqual(math.copysign(1.0, f), -1.0)
class UnpackIteratorTest(unittest.TestCase):
"""
Tests for iterative unpacking (struct.Struct.iter_unpack).
"""
def test_construct(self):
def _check_iterator(it):
self.assertIsInstance(it, abc.Iterator)
self.assertIsInstance(it, abc.Iterable)
s = struct.Struct('>ibcp')
it = s.iter_unpack(b"")
_check_iterator(it)
it = s.iter_unpack(b"1234567")
_check_iterator(it)
# Wrong bytes length
with self.assertRaises(struct.error):
s.iter_unpack(b"123456")
with self.assertRaises(struct.error):
s.iter_unpack(b"12345678")
# Zero-length struct
s = struct.Struct('>')
with self.assertRaises(struct.error):
s.iter_unpack(b"")
with self.assertRaises(struct.error):
s.iter_unpack(b"12")
def test_uninstantiable(self):
iter_unpack_type = type(struct.Struct(">ibcp").iter_unpack(b""))
self.assertRaises(TypeError, iter_unpack_type)
def test_iterate(self):
s = struct.Struct('>IB')
b = bytes(range(1, 16))
it = s.iter_unpack(b)
self.assertEqual(next(it), (0x01020304, 5))
self.assertEqual(next(it), (0x06070809, 10))
self.assertEqual(next(it), (0x0b0c0d0e, 15))
self.assertRaises(StopIteration, next, it)
self.assertRaises(StopIteration, next, it)
def test_arbitrary_buffer(self):
s = struct.Struct('>IB')
b = bytes(range(1, 11))
it = s.iter_unpack(memoryview(b))
self.assertEqual(next(it), (0x01020304, 5))
self.assertEqual(next(it), (0x06070809, 10))
self.assertRaises(StopIteration, next, it)
self.assertRaises(StopIteration, next, it)
def test_length_hint(self):
lh = operator.length_hint
s = struct.Struct('>IB')
b = bytes(range(1, 16))
it = s.iter_unpack(b)
self.assertEqual(lh(it), 3)
next(it)
self.assertEqual(lh(it), 2)
next(it)
self.assertEqual(lh(it), 1)
next(it)
self.assertEqual(lh(it), 0)
self.assertRaises(StopIteration, next, it)
self.assertEqual(lh(it), 0)
def test_module_func(self):
# Sanity check for the global struct.iter_unpack()
it = struct.iter_unpack('>IB', bytes(range(1, 11)))
self.assertEqual(next(it), (0x01020304, 5))
self.assertEqual(next(it), (0x06070809, 10))
self.assertRaises(StopIteration, next, it)
self.assertRaises(StopIteration, next, it)
def test_half_float(self):
_testcapi = import_helper.import_module('_testcapi')
# Little-endian examples from:
# http://en.wikipedia.org/wiki/Half_precision_floating-point_format
format_bits_float__cleanRoundtrip_list = [
(b'\x00\x3c', 1.0),
(b'\x00\xc0', -2.0),
(b'\xff\x7b', 65504.0), # (max half precision)
(b'\x00\x04', 2**-14), # ~= 6.10352 * 10**-5 (min pos normal)
(b'\x01\x00', 2**-24), # ~= 5.96046 * 10**-8 (min pos subnormal)
(b'\x00\x00', 0.0),
(b'\x00\x80', -0.0),
(b'\x00\x7c', float('+inf')),
(b'\x00\xfc', float('-inf')),
(b'\x55\x35', 0.333251953125), # ~= 1/3
]
for le_bits, f in format_bits_float__cleanRoundtrip_list:
be_bits = le_bits[::-1]
self.assertEqual(f, struct.unpack('<e', le_bits)[0])
self.assertEqual(le_bits, struct.pack('<e', f))
self.assertEqual(f, struct.unpack('>e', be_bits)[0])
self.assertEqual(be_bits, struct.pack('>e', f))
if sys.byteorder == 'little':
self.assertEqual(f, struct.unpack('e', le_bits)[0])
self.assertEqual(le_bits, struct.pack('e', f))
else:
self.assertEqual(f, struct.unpack('e', be_bits)[0])
self.assertEqual(be_bits, struct.pack('e', f))
# Check for NaN handling:
format_bits__nan_list = [
('<e', b'\x01\xfc'),
('<e', b'\x00\xfe'),
('<e', b'\xff\xff'),
('<e', b'\x01\x7c'),
('<e', b'\x00\x7e'),
('<e', b'\xff\x7f'),
]
for formatcode, bits in format_bits__nan_list:
self.assertTrue(math.isnan(struct.unpack('<e', bits)[0]))
self.assertTrue(math.isnan(struct.unpack('>e', bits[::-1])[0]))
# Check that packing produces a bit pattern representing a quiet NaN:
# all exponent bits and the msb of the fraction should all be 1.
if _testcapi.nan_msb_is_signaling:
# HP PA RISC and some MIPS CPUs use 0 for quiet, see:
# https://en.wikipedia.org/wiki/NaN#Encoding
expected = 0x7c
else:
expected = 0x7e
packed = struct.pack('<e', math.nan)
self.assertEqual(packed[1] & 0x7e, expected)
packed = struct.pack('<e', -math.nan)
self.assertEqual(packed[1] & 0x7e, expected)
# Checks for round-to-even behavior
format_bits_float__rounding_list = [
('>e', b'\x00\x01', 2.0**-25 + 2.0**-35), # Rounds to minimum subnormal
('>e', b'\x00\x00', 2.0**-25), # Underflows to zero (nearest even mode)
('>e', b'\x00\x00', 2.0**-26), # Underflows to zero
('>e', b'\x03\xff', 2.0**-14 - 2.0**-24), # Largest subnormal.
('>e', b'\x03\xff', 2.0**-14 - 2.0**-25 - 2.0**-65),
('>e', b'\x04\x00', 2.0**-14 - 2.0**-25),
('>e', b'\x04\x00', 2.0**-14), # Smallest normal.
('>e', b'\x3c\x01', 1.0+2.0**-11 + 2.0**-16), # rounds to 1.0+2**(-10)
('>e', b'\x3c\x00', 1.0+2.0**-11), # rounds to 1.0 (nearest even mode)
('>e', b'\x3c\x00', 1.0+2.0**-12), # rounds to 1.0
('>e', b'\x7b\xff', 65504), # largest normal
('>e', b'\x7b\xff', 65519), # rounds to 65504
('>e', b'\x80\x01', -2.0**-25 - 2.0**-35), # Rounds to minimum subnormal
('>e', b'\x80\x00', -2.0**-25), # Underflows to zero (nearest even mode)
('>e', b'\x80\x00', -2.0**-26), # Underflows to zero
('>e', b'\xbc\x01', -1.0-2.0**-11 - 2.0**-16), # rounds to 1.0+2**(-10)
('>e', b'\xbc\x00', -1.0-2.0**-11), # rounds to 1.0 (nearest even mode)
('>e', b'\xbc\x00', -1.0-2.0**-12), # rounds to 1.0
('>e', b'\xfb\xff', -65519), # rounds to 65504
]
for formatcode, bits, f in format_bits_float__rounding_list:
self.assertEqual(bits, struct.pack(formatcode, f))
# This overflows, and so raises an error
format_bits_float__roundingError_list = [
# Values that round to infinity.
('>e', 65520.0),
('>e', 65536.0),
('>e', 1e300),
('>e', -65520.0),
('>e', -65536.0),
('>e', -1e300),
('<e', 65520.0),
('<e', 65536.0),
('<e', 1e300),
('<e', -65520.0),
('<e', -65536.0),
('<e', -1e300),
]
for formatcode, f in format_bits_float__roundingError_list:
self.assertRaises(OverflowError, struct.pack, formatcode, f)
# Double rounding
format_bits_float__doubleRoundingError_list = [
('>e', b'\x67\xff', 0x1ffdffffff * 2**-26), # should be 2047, if double-rounded 64>32>16, becomes 2048
]
for formatcode, bits, f in format_bits_float__doubleRoundingError_list:
self.assertEqual(bits, struct.pack(formatcode, f))
if __name__ == '__main__':
unittest.main()
View Git Blame Copy Permalink