-
-
Notifications
You must be signed in to change notification settings - Fork 34.2k
Expand file tree
/
Copy pathtest_buffer.py
More file actions
4860 lines (4070 loc) · 175 KB
/
test_buffer.py
File metadata and controls
4860 lines (4070 loc) · 175 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#
# The ndarray object from _testbuffer.c is a complete implementation of
# a PEP-3118 buffer provider. It is independent from NumPy's ndarray
# and the tests don't require NumPy.
#
# If NumPy is present, some tests check both ndarray implementations
# against each other.
#
# Most ndarray tests also check that memoryview(ndarray) behaves in
# the same way as the original. Thus, a substantial part of the
# memoryview tests is now in this module.
#
# Written and designed by Stefan Krah for Python 3.3.
#
import contextlib
import unittest
from test import support
from test.support import os_helper
import inspect
from itertools import permutations, product
from random import randrange, sample, choice
import warnings
import sys, array, io, os
from decimal import Decimal
from fractions import Fraction
from test.support import warnings_helper
try:
from _testbuffer import *
except ImportError:
ndarray = None
try:
import struct
except ImportError:
struct = None
try:
import ctypes
except ImportError:
ctypes = None
try:
with os_helper.EnvironmentVarGuard() as os.environ, \
warnings.catch_warnings():
from numpy import ndarray as numpy_array
except ImportError:
numpy_array = None
try:
import _testcapi
except ImportError:
_testcapi = None
SHORT_TEST = True
# ======================================================================
# Random lists by format specifier
# ======================================================================
# Native format chars and their ranges.
NATIVE = {
'?':0, 'c':0, 'b':0, 'B':0,
'h':0, 'H':0, 'i':0, 'I':0,
'l':0, 'L':0, 'n':0, 'N':0,
'e':0, 'f':0, 'd':0, 'P':0
}
# NumPy does not have 'n' or 'N':
if numpy_array:
del NATIVE['n']
del NATIVE['N']
if struct:
try:
# Add "qQ" if present in native mode.
struct.pack('Q', 2**64-1)
NATIVE['q'] = 0
NATIVE['Q'] = 0
except struct.error:
pass
# Standard format chars and their ranges.
STANDARD = {
'?':(0, 2), 'c':(0, 1<<8),
'b':(-(1<<7), 1<<7), 'B':(0, 1<<8),
'h':(-(1<<15), 1<<15), 'H':(0, 1<<16),
'i':(-(1<<31), 1<<31), 'I':(0, 1<<32),
'l':(-(1<<31), 1<<31), 'L':(0, 1<<32),
'q':(-(1<<63), 1<<63), 'Q':(0, 1<<64),
'e':(-65519, 65520), 'f':(-(1<<63), 1<<63),
'd':(-(1<<1023), 1<<1023)
}
def native_type_range(fmt):
"""Return range of a native type."""
if fmt == 'c':
lh = (0, 256)
elif fmt == '?':
lh = (0, 2)
elif fmt == 'e':
lh = (-65519, 65520)
elif fmt == 'f':
lh = (-(1<<63), 1<<63)
elif fmt == 'd':
lh = (-(1<<1023), 1<<1023)
else:
for exp in (128, 127, 64, 63, 32, 31, 16, 15, 8, 7):
try:
struct.pack(fmt, (1<<exp)-1)
break
except struct.error:
pass
lh = (-(1<<exp), 1<<exp) if exp & 1 else (0, 1<<exp)
return lh
fmtdict = {
'':NATIVE,
'@':NATIVE,
'<':STANDARD,
'>':STANDARD,
'=':STANDARD,
'!':STANDARD
}
if struct:
for fmt in fmtdict['@']:
fmtdict['@'][fmt] = native_type_range(fmt)
# Format codes supported by the memoryview object
MEMORYVIEW = NATIVE.copy()
# Format codes supported by array.array
ARRAY = NATIVE.copy()
for k in NATIVE:
if not k in "bBhHiIlLfd":
del ARRAY[k]
BYTEFMT = NATIVE.copy()
for k in NATIVE:
if not k in "Bbc":
del BYTEFMT[k]
fmtdict['m'] = MEMORYVIEW
fmtdict['@m'] = MEMORYVIEW
fmtdict['a'] = ARRAY
fmtdict['b'] = BYTEFMT
fmtdict['@b'] = BYTEFMT
# Capabilities of the test objects:
MODE = 0
MULT = 1
cap = { # format chars # multiplier
'ndarray': (['', '@', '<', '>', '=', '!'], ['', '1', '2', '3']),
'array': (['a'], ['']),
'numpy': ([''], ['']),
'memoryview': (['@m', 'm'], ['']),
'bytefmt': (['@b', 'b'], ['']),
}
def randrange_fmt(mode, char, obj):
"""Return random item for a type specified by a mode and a single
format character."""
x = randrange(*fmtdict[mode][char])
if char == 'c':
x = bytes([x])
if obj == 'numpy' and x == b'\x00':
# https://github.com/numpy/numpy/issues/2518
x = b'\x01'
if char == '?':
x = bool(x)
if char in 'efd':
x = struct.pack(char, x)
x = struct.unpack(char, x)[0]
return x
def gen_item(fmt, obj):
"""Return single random item."""
mode, chars = fmt.split('#')
x = []
for c in chars:
x.append(randrange_fmt(mode, c, obj))
return x[0] if len(x) == 1 else tuple(x)
def gen_items(n, fmt, obj):
"""Return a list of random items (or a scalar)."""
if n == 0:
return gen_item(fmt, obj)
lst = [0] * n
for i in range(n):
lst[i] = gen_item(fmt, obj)
return lst
def struct_items(n, obj):
mode = choice(cap[obj][MODE])
xfmt = mode + '#'
fmt = mode.strip('amb')
nmemb = randrange(2, 10) # number of struct members
for _ in range(nmemb):
char = choice(tuple(fmtdict[mode]))
multiplier = choice(cap[obj][MULT])
xfmt += (char * int(multiplier if multiplier else 1))
fmt += (multiplier + char)
items = gen_items(n, xfmt, obj)
item = gen_item(xfmt, obj)
return fmt, items, item
def randitems(n, obj='ndarray', mode=None, char=None):
"""Return random format, items, item."""
if mode is None:
mode = choice(cap[obj][MODE])
if char is None:
char = choice(tuple(fmtdict[mode]))
multiplier = choice(cap[obj][MULT])
fmt = mode + '#' + char * int(multiplier if multiplier else 1)
items = gen_items(n, fmt, obj)
item = gen_item(fmt, obj)
fmt = mode.strip('amb') + multiplier + char
return fmt, items, item
def iter_mode(n, obj='ndarray'):
"""Iterate through supported mode/char combinations."""
for mode in cap[obj][MODE]:
for char in fmtdict[mode]:
yield randitems(n, obj, mode, char)
def iter_format(nitems, testobj='ndarray'):
"""Yield (format, items, item) for all possible modes and format
characters plus one random compound format string."""
for t in iter_mode(nitems, testobj):
yield t
if testobj != 'ndarray':
return
yield struct_items(nitems, testobj)
def is_byte_format(fmt):
return 'c' in fmt or 'b' in fmt or 'B' in fmt
def is_memoryview_format(fmt):
"""format suitable for memoryview"""
x = len(fmt)
return ((x == 1 or (x == 2 and fmt[0] == '@')) and
fmt[x-1] in MEMORYVIEW)
NON_BYTE_FORMAT = [c for c in fmtdict['@'] if not is_byte_format(c)]
# ======================================================================
# Multi-dimensional tolist(), slicing and slice assignments
# ======================================================================
def atomp(lst):
"""Tuple items (representing structs) are regarded as atoms."""
return not isinstance(lst, list)
def listp(lst):
return isinstance(lst, list)
def prod(lst):
"""Product of list elements."""
if len(lst) == 0:
return 0
x = lst[0]
for v in lst[1:]:
x *= v
return x
def strides_from_shape(ndim, shape, itemsize, layout):
"""Calculate strides of a contiguous array. Layout is 'C' or
'F' (Fortran)."""
if ndim == 0:
return ()
if layout == 'C':
strides = list(shape[1:]) + [itemsize]
for i in range(ndim-2, -1, -1):
strides[i] *= strides[i+1]
else:
strides = [itemsize] + list(shape[:-1])
for i in range(1, ndim):
strides[i] *= strides[i-1]
return strides
def _ca(items, s):
"""Convert flat item list to the nested list representation of a
multidimensional C array with shape 's'."""
if atomp(items):
return items
if len(s) == 0:
return items[0]
lst = [0] * s[0]
stride = len(items) // s[0] if s[0] else 0
for i in range(s[0]):
start = i*stride
lst[i] = _ca(items[start:start+stride], s[1:])
return lst
def _fa(items, s):
"""Convert flat item list to the nested list representation of a
multidimensional Fortran array with shape 's'."""
if atomp(items):
return items
if len(s) == 0:
return items[0]
lst = [0] * s[0]
stride = s[0]
for i in range(s[0]):
lst[i] = _fa(items[i::stride], s[1:])
return lst
def carray(items, shape):
if listp(items) and not 0 in shape and prod(shape) != len(items):
raise ValueError("prod(shape) != len(items)")
return _ca(items, shape)
def farray(items, shape):
if listp(items) and not 0 in shape and prod(shape) != len(items):
raise ValueError("prod(shape) != len(items)")
return _fa(items, shape)
def indices(shape):
"""Generate all possible tuples of indices."""
iterables = [range(v) for v in shape]
return product(*iterables)
def getindex(ndim, ind, strides):
"""Convert multi-dimensional index to the position in the flat list."""
ret = 0
for i in range(ndim):
ret += strides[i] * ind[i]
return ret
def transpose(src, shape):
"""Transpose flat item list that is regarded as a multi-dimensional
matrix defined by shape: dest...[k][j][i] = src[i][j][k]... """
if not shape:
return src
ndim = len(shape)
sstrides = strides_from_shape(ndim, shape, 1, 'C')
dstrides = strides_from_shape(ndim, shape[::-1], 1, 'C')
dest = [0] * len(src)
for ind in indices(shape):
fr = getindex(ndim, ind, sstrides)
to = getindex(ndim, ind[::-1], dstrides)
dest[to] = src[fr]
return dest
def _flatten(lst):
"""flatten list"""
if lst == []:
return lst
if atomp(lst):
return [lst]
return _flatten(lst[0]) + _flatten(lst[1:])
def flatten(lst):
"""flatten list or return scalar"""
if atomp(lst): # scalar
return lst
return _flatten(lst)
def slice_shape(lst, slices):
"""Get the shape of lst after slicing: slices is a list of slice
objects."""
if atomp(lst):
return []
return [len(lst[slices[0]])] + slice_shape(lst[0], slices[1:])
def multislice(lst, slices):
"""Multi-dimensional slicing: slices is a list of slice objects."""
if atomp(lst):
return lst
return [multislice(sublst, slices[1:]) for sublst in lst[slices[0]]]
def m_assign(llst, rlst, lslices, rslices):
"""Multi-dimensional slice assignment: llst and rlst are the operands,
lslices and rslices are lists of slice objects. llst and rlst must
have the same structure.
For a two-dimensional example, this is not implemented in Python:
llst[0:3:2, 0:3:2] = rlst[1:3:1, 1:3:1]
Instead we write:
lslices = [slice(0,3,2), slice(0,3,2)]
rslices = [slice(1,3,1), slice(1,3,1)]
multislice_assign(llst, rlst, lslices, rslices)
"""
if atomp(rlst):
return rlst
rlst = [m_assign(l, r, lslices[1:], rslices[1:])
for l, r in zip(llst[lslices[0]], rlst[rslices[0]])]
llst[lslices[0]] = rlst
return llst
def cmp_structure(llst, rlst, lslices, rslices):
"""Compare the structure of llst[lslices] and rlst[rslices]."""
lshape = slice_shape(llst, lslices)
rshape = slice_shape(rlst, rslices)
if (len(lshape) != len(rshape)):
return -1
for i in range(len(lshape)):
if lshape[i] != rshape[i]:
return -1
if lshape[i] == 0:
return 0
return 0
def multislice_assign(llst, rlst, lslices, rslices):
"""Return llst after assigning: llst[lslices] = rlst[rslices]"""
if cmp_structure(llst, rlst, lslices, rslices) < 0:
raise ValueError("lvalue and rvalue have different structures")
return m_assign(llst, rlst, lslices, rslices)
# ======================================================================
# Random structures
# ======================================================================
#
# PEP-3118 is very permissive with respect to the contents of a
# Py_buffer. In particular:
#
# - shape can be zero
# - strides can be any integer, including zero
# - offset can point to any location in the underlying
# memory block, provided that it is a multiple of
# itemsize.
#
# The functions in this section test and verify random structures
# in full generality. A structure is valid iff it fits in the
# underlying memory block.
#
# The structure 't' (short for 'tuple') is fully defined by:
#
# t = (memlen, itemsize, ndim, shape, strides, offset)
#
def verify_structure(memlen, itemsize, ndim, shape, strides, offset):
"""Verify that the parameters represent a valid array within
the bounds of the allocated memory:
char *mem: start of the physical memory block
memlen: length of the physical memory block
offset: (char *)buf - mem
"""
if offset % itemsize:
return False
if offset < 0 or offset+itemsize > memlen:
return False
if any(v % itemsize for v in strides):
return False
if ndim <= 0:
return ndim == 0 and not shape and not strides
if 0 in shape:
return True
imin = sum(strides[j]*(shape[j]-1) for j in range(ndim)
if strides[j] <= 0)
imax = sum(strides[j]*(shape[j]-1) for j in range(ndim)
if strides[j] > 0)
return 0 <= offset+imin and offset+imax+itemsize <= memlen
def get_item(lst, indices):
for i in indices:
lst = lst[i]
return lst
def memory_index(indices, t):
"""Location of an item in the underlying memory."""
memlen, itemsize, ndim, shape, strides, offset = t
p = offset
for i in range(ndim):
p += strides[i]*indices[i]
return p
def is_overlapping(t):
"""The structure 't' is overlapping if at least one memory location
is visited twice while iterating through all possible tuples of
indices."""
memlen, itemsize, ndim, shape, strides, offset = t
visited = 1<<memlen
for ind in indices(shape):
i = memory_index(ind, t)
bit = 1<<i
if visited & bit:
return True
visited |= bit
return False
def rand_structure(itemsize, valid, maxdim=5, maxshape=16, shape=()):
"""Return random structure:
(memlen, itemsize, ndim, shape, strides, offset)
If 'valid' is true, the returned structure is valid, otherwise invalid.
If 'shape' is given, use that instead of creating a random shape.
"""
if not shape:
ndim = randrange(maxdim+1)
if (ndim == 0):
if valid:
return itemsize, itemsize, ndim, (), (), 0
else:
nitems = randrange(1, 16+1)
memlen = nitems * itemsize
offset = -itemsize if randrange(2) == 0 else memlen
return memlen, itemsize, ndim, (), (), offset
minshape = 2
n = randrange(100)
if n >= 95 and valid:
minshape = 0
elif n >= 90:
minshape = 1
shape = [0] * ndim
for i in range(ndim):
shape[i] = randrange(minshape, maxshape+1)
else:
ndim = len(shape)
maxstride = 5
n = randrange(100)
zero_stride = True if n >= 95 and n & 1 else False
strides = [0] * ndim
strides[ndim-1] = itemsize * randrange(-maxstride, maxstride+1)
if not zero_stride and strides[ndim-1] == 0:
strides[ndim-1] = itemsize
for i in range(ndim-2, -1, -1):
maxstride *= shape[i+1] if shape[i+1] else 1
if zero_stride:
strides[i] = itemsize * randrange(-maxstride, maxstride+1)
else:
strides[i] = ((1,-1)[randrange(2)] *
itemsize * randrange(1, maxstride+1))
imin = imax = 0
if not 0 in shape:
imin = sum(strides[j]*(shape[j]-1) for j in range(ndim)
if strides[j] <= 0)
imax = sum(strides[j]*(shape[j]-1) for j in range(ndim)
if strides[j] > 0)
nitems = imax - imin
if valid:
offset = -imin * itemsize
memlen = offset + (imax+1) * itemsize
else:
memlen = (-imin + imax) * itemsize
offset = -imin-itemsize if randrange(2) == 0 else memlen
return memlen, itemsize, ndim, shape, strides, offset
def randslice_from_slicelen(slicelen, listlen):
"""Create a random slice of len slicelen that fits into listlen."""
maxstart = listlen - slicelen
start = randrange(maxstart+1)
maxstep = (listlen - start) // slicelen if slicelen else 1
step = randrange(1, maxstep+1)
stop = start + slicelen * step
s = slice(start, stop, step)
_, _, _, control = slice_indices(s, listlen)
if control != slicelen:
raise RuntimeError
return s
def randslice_from_shape(ndim, shape):
"""Create two sets of slices for an array x with shape 'shape'
such that shapeof(x[lslices]) == shapeof(x[rslices])."""
lslices = [0] * ndim
rslices = [0] * ndim
for n in range(ndim):
l = shape[n]
slicelen = randrange(1, l+1) if l > 0 else 0
lslices[n] = randslice_from_slicelen(slicelen, l)
rslices[n] = randslice_from_slicelen(slicelen, l)
return tuple(lslices), tuple(rslices)
def rand_aligned_slices(maxdim=5, maxshape=16):
"""Create (lshape, rshape, tuple(lslices), tuple(rslices)) such that
shapeof(x[lslices]) == shapeof(y[rslices]), where x is an array
with shape 'lshape' and y is an array with shape 'rshape'."""
ndim = randrange(1, maxdim+1)
minshape = 2
n = randrange(100)
if n >= 95:
minshape = 0
elif n >= 90:
minshape = 1
all_random = True if randrange(100) >= 80 else False
lshape = [0]*ndim; rshape = [0]*ndim
lslices = [0]*ndim; rslices = [0]*ndim
for n in range(ndim):
small = randrange(minshape, maxshape+1)
big = randrange(minshape, maxshape+1)
if big < small:
big, small = small, big
# Create a slice that fits the smaller value.
if all_random:
start = randrange(-small, small+1)
stop = randrange(-small, small+1)
step = (1,-1)[randrange(2)] * randrange(1, small+2)
s_small = slice(start, stop, step)
_, _, _, slicelen = slice_indices(s_small, small)
else:
slicelen = randrange(1, small+1) if small > 0 else 0
s_small = randslice_from_slicelen(slicelen, small)
# Create a slice of the same length for the bigger value.
s_big = randslice_from_slicelen(slicelen, big)
if randrange(2) == 0:
rshape[n], lshape[n] = big, small
rslices[n], lslices[n] = s_big, s_small
else:
rshape[n], lshape[n] = small, big
rslices[n], lslices[n] = s_small, s_big
return lshape, rshape, tuple(lslices), tuple(rslices)
def randitems_from_structure(fmt, t):
"""Return a list of random items for structure 't' with format
'fmtchar'."""
memlen, itemsize, _, _, _, _ = t
return gen_items(memlen//itemsize, '#'+fmt, 'numpy')
def ndarray_from_structure(items, fmt, t, flags=0):
"""Return ndarray from the tuple returned by rand_structure()"""
memlen, itemsize, ndim, shape, strides, offset = t
return ndarray(items, shape=shape, strides=strides, format=fmt,
offset=offset, flags=ND_WRITABLE|flags)
def numpy_array_from_structure(items, fmt, t):
"""Return numpy_array from the tuple returned by rand_structure()"""
memlen, itemsize, ndim, shape, strides, offset = t
buf = bytearray(memlen)
for j, v in enumerate(items):
struct.pack_into(fmt, buf, j*itemsize, v)
return numpy_array(buffer=buf, shape=shape, strides=strides,
dtype=fmt, offset=offset)
# ======================================================================
# memoryview casts
# ======================================================================
def cast_items(exporter, fmt, itemsize, shape=None):
"""Interpret the raw memory of 'exporter' as a list of items with
size 'itemsize'. If shape=None, the new structure is assumed to
be 1-D with n * itemsize = bytelen. If shape is given, the usual
constraint for contiguous arrays prod(shape) * itemsize = bytelen
applies. On success, return (items, shape). If the constraints
cannot be met, return (None, None). If a chunk of bytes is interpreted
as NaN as a result of float conversion, return ('nan', None)."""
bytelen = exporter.nbytes
if shape:
if prod(shape) * itemsize != bytelen:
return None, shape
elif shape == []:
if exporter.ndim == 0 or itemsize != bytelen:
return None, shape
else:
n, r = divmod(bytelen, itemsize)
shape = [n]
if r != 0:
return None, shape
mem = exporter.tobytes()
byteitems = [mem[i:i+itemsize] for i in range(0, len(mem), itemsize)]
items = []
for v in byteitems:
item = struct.unpack(fmt, v)[0]
if item != item:
return 'nan', shape
items.append(item)
return (items, shape) if shape != [] else (items[0], shape)
def gencastshapes():
"""Generate shapes to test casting."""
for n in range(32):
yield [n]
ndim = randrange(4, 6)
minshape = 1 if randrange(100) > 80 else 2
yield [randrange(minshape, 5) for _ in range(ndim)]
ndim = randrange(2, 4)
minshape = 1 if randrange(100) > 80 else 2
yield [randrange(minshape, 5) for _ in range(ndim)]
# ======================================================================
# Actual tests
# ======================================================================
def genslices(n):
"""Generate all possible slices for a single dimension."""
return product(range(-n, n+1), range(-n, n+1), range(-n, n+1))
def genslices_ndim(ndim, shape):
"""Generate all possible slice tuples for 'shape'."""
iterables = [genslices(shape[n]) for n in range(ndim)]
return product(*iterables)
def rslice(n, allow_empty=False):
"""Generate random slice for a single dimension of length n.
If zero=True, the slices may be empty, otherwise they will
be non-empty."""
minlen = 0 if allow_empty or n == 0 else 1
slicelen = randrange(minlen, n+1)
return randslice_from_slicelen(slicelen, n)
def rslices(n, allow_empty=False):
"""Generate random slices for a single dimension."""
for _ in range(5):
yield rslice(n, allow_empty)
def rslices_ndim(ndim, shape, iterations=5):
"""Generate random slice tuples for 'shape'."""
# non-empty slices
for _ in range(iterations):
yield tuple(rslice(shape[n]) for n in range(ndim))
# possibly empty slices
for _ in range(iterations):
yield tuple(rslice(shape[n], allow_empty=True) for n in range(ndim))
# invalid slices
yield tuple(slice(0,1,0) for _ in range(ndim))
def rpermutation(iterable, r=None):
pool = tuple(iterable)
r = len(pool) if r is None else r
yield tuple(sample(pool, r))
def ndarray_print(nd):
"""Print ndarray for debugging."""
try:
x = nd.tolist()
except (TypeError, NotImplementedError):
x = nd.tobytes()
if isinstance(nd, ndarray):
offset = nd.offset
flags = nd.flags
else:
offset = 'unknown'
flags = 'unknown'
print("ndarray(%s, shape=%s, strides=%s, suboffsets=%s, offset=%s, "
"format='%s', itemsize=%s, flags=%s)" %
(x, nd.shape, nd.strides, nd.suboffsets, offset,
nd.format, nd.itemsize, flags))
sys.stdout.flush()
ITERATIONS = 100
MAXDIM = 5
MAXSHAPE = 10
if SHORT_TEST:
ITERATIONS = 10
MAXDIM = 3
MAXSHAPE = 4
genslices = rslices
genslices_ndim = rslices_ndim
permutations = rpermutation
@unittest.skipUnless(struct, 'struct module required for this test.')
@unittest.skipUnless(ndarray, 'ndarray object required for this test')
class TestBufferProtocol(unittest.TestCase):
def setUp(self):
# The suboffsets tests need sizeof(void *).
self.sizeof_void_p = get_sizeof_void_p()
def verify(self, result, *, obj,
itemsize, fmt, readonly,
ndim, shape, strides,
lst, sliced=False, cast=False):
# Verify buffer contents against expected values.
if shape:
expected_len = prod(shape)*itemsize
else:
if not fmt: # array has been implicitly cast to unsigned bytes
expected_len = len(lst)
else: # ndim = 0
expected_len = itemsize
# Reconstruct suboffsets from strides. Support for slicing
# could be added, but is currently only needed for test_getbuf().
suboffsets = ()
if result.suboffsets:
self.assertGreater(ndim, 0)
suboffset0 = 0
for n in range(1, ndim):
if shape[n] == 0:
break
if strides[n] <= 0:
suboffset0 += -strides[n] * (shape[n]-1)
suboffsets = [suboffset0] + [-1 for v in range(ndim-1)]
# Not correct if slicing has occurred in the first dimension.
stride0 = self.sizeof_void_p
if strides[0] < 0:
stride0 = -stride0
strides = [stride0] + list(strides[1:])
self.assertIs(result.obj, obj)
self.assertEqual(result.nbytes, expected_len)
self.assertEqual(result.itemsize, itemsize)
self.assertEqual(result.format, fmt)
self.assertIs(result.readonly, readonly)
self.assertEqual(result.ndim, ndim)
self.assertEqual(result.shape, tuple(shape))
if not (sliced and suboffsets):
self.assertEqual(result.strides, tuple(strides))
self.assertEqual(result.suboffsets, tuple(suboffsets))
if isinstance(result, ndarray) or is_memoryview_format(fmt):
rep = result.tolist() if fmt else result.tobytes()
self.assertEqual(rep, lst)
if not fmt: # array has been cast to unsigned bytes,
return # the remaining tests won't work.
# PyBuffer_GetPointer() is the definition how to access an item.
# If PyBuffer_GetPointer(indices) is correct for all possible
# combinations of indices, the buffer is correct.
#
# Also test tobytes() against the flattened 'lst', with all items
# packed to bytes.
if not cast: # casts chop up 'lst' in different ways
b = bytearray()
buf_err = None
for ind in indices(shape):
try:
item1 = get_pointer(result, ind)
item2 = get_item(lst, ind)
if isinstance(item2, tuple):
x = struct.pack(fmt, *item2)
else:
x = struct.pack(fmt, item2)
b.extend(x)
except BufferError:
buf_err = True # re-exporter does not provide full buffer
break
self.assertEqual(item1, item2)
if not buf_err:
# test tobytes()
self.assertEqual(result.tobytes(), b)
# test hex()
m = memoryview(result)
h = "".join("%02x" % c for c in b)
self.assertEqual(m.hex(), h)
# lst := expected multi-dimensional logical representation
# flatten(lst) := elements in C-order
ff = fmt if fmt else 'B'
flattened = flatten(lst)
# Rules for 'A': if the array is already contiguous, return
# the array unaltered. Otherwise, return a contiguous 'C'
# representation.
for order in ['C', 'F', 'A']:
expected = result
if order == 'F':
if not is_contiguous(result, 'A') or \
is_contiguous(result, 'C'):
# For constructing the ndarray, convert the
# flattened logical representation to Fortran order.
trans = transpose(flattened, shape)
expected = ndarray(trans, shape=shape, format=ff,
flags=ND_FORTRAN)
else: # 'C', 'A'
if not is_contiguous(result, 'A') or \
is_contiguous(result, 'F') and order == 'C':
# The flattened list is already in C-order.
expected = ndarray(flattened, shape=shape, format=ff)
contig = get_contiguous(result, PyBUF_READ, order)
self.assertEqual(contig.tobytes(), b)
self.assertTrue(cmp_contig(contig, expected))
if ndim == 0:
continue
nmemb = len(flattened)
ro = 0 if readonly else ND_WRITABLE
### See comment in test_py_buffer_to_contiguous for an
### explanation why these tests are valid.
# To 'C'
contig = py_buffer_to_contiguous(result, 'C', PyBUF_FULL_RO)
self.assertEqual(len(contig), nmemb * itemsize)
initlst = [struct.unpack_from(fmt, contig, n*itemsize)
for n in range(nmemb)]
if len(initlst[0]) == 1:
initlst = [v[0] for v in initlst]
y = ndarray(initlst, shape=shape, flags=ro, format=fmt)
self.assertEqual(memoryview(y), memoryview(result))
contig_bytes = memoryview(result).tobytes()
self.assertEqual(contig_bytes, contig)
contig_bytes = memoryview(result).tobytes(order=None)
self.assertEqual(contig_bytes, contig)
contig_bytes = memoryview(result).tobytes(order='C')
self.assertEqual(contig_bytes, contig)
# To 'F'
contig = py_buffer_to_contiguous(result, 'F', PyBUF_FULL_RO)
self.assertEqual(len(contig), nmemb * itemsize)
initlst = [struct.unpack_from(fmt, contig, n*itemsize)
for n in range(nmemb)]
if len(initlst[0]) == 1:
initlst = [v[0] for v in initlst]
y = ndarray(initlst, shape=shape, flags=ro|ND_FORTRAN,
format=fmt)
self.assertEqual(memoryview(y), memoryview(result))
contig_bytes = memoryview(result).tobytes(order='F')
self.assertEqual(contig_bytes, contig)
# To 'A'
contig = py_buffer_to_contiguous(result, 'A', PyBUF_FULL_RO)
self.assertEqual(len(contig), nmemb * itemsize)
initlst = [struct.unpack_from(fmt, contig, n*itemsize)
for n in range(nmemb)]
if len(initlst[0]) == 1:
initlst = [v[0] for v in initlst]
f = ND_FORTRAN if is_contiguous(result, 'F') else 0
y = ndarray(initlst, shape=shape, flags=f|ro, format=fmt)
self.assertEqual(memoryview(y), memoryview(result))
contig_bytes = memoryview(result).tobytes(order='A')
self.assertEqual(contig_bytes, contig)
if is_memoryview_format(fmt):
try:
m = memoryview(result)
except BufferError: # re-exporter does not provide full information
return
ex = result.obj if isinstance(result, memoryview) else result
def check_memoryview(m, expected_readonly=readonly):
self.assertIs(m.obj, ex)
self.assertEqual(m.nbytes, expected_len)
self.assertEqual(m.itemsize, itemsize)
self.assertEqual(m.format, fmt)
self.assertEqual(m.readonly, expected_readonly)
self.assertEqual(m.ndim, ndim)
self.assertEqual(m.shape, tuple(shape))
if not (sliced and suboffsets):
self.assertEqual(m.strides, tuple(strides))
self.assertEqual(m.suboffsets, tuple(suboffsets))
if ndim == 0:
self.assertRaises(TypeError, len, m)
else:
self.assertEqual(len(m), len(lst))
rep = result.tolist() if fmt else result.tobytes()
self.assertEqual(rep, lst)
self.assertEqual(m, result)
check_memoryview(m)
with m.toreadonly() as mm:
check_memoryview(mm, expected_readonly=True)
m.tobytes() # Releasing mm didn't release m
def verify_getbuf(self, orig_ex, ex, req, sliced=False):
def match(req, flag):
return ((req&flag) == flag)
if (# writable request to read-only exporter
(ex.readonly and match(req, PyBUF_WRITABLE)) or
# cannot match explicit contiguity request
(match(req, PyBUF_C_CONTIGUOUS) and not ex.c_contiguous) or
(match(req, PyBUF_F_CONTIGUOUS) and not ex.f_contiguous) or
(match(req, PyBUF_ANY_CONTIGUOUS) and not ex.contiguous) or
# buffer needs suboffsets
(not match(req, PyBUF_INDIRECT) and ex.suboffsets) or
# buffer without strides must be C-contiguous
(not match(req, PyBUF_STRIDES) and not ex.c_contiguous) or
# PyBUF_SIMPLE|PyBUF_FORMAT and PyBUF_WRITABLE|PyBUF_FORMAT
(not match(req, PyBUF_ND) and match(req, PyBUF_FORMAT))):