1 # Tests the buffer access syntax functionality by constructing

2 # mock buffer objects.

3 #

4 # Note that the buffers are mock objects created for testing

5 # the buffer access behaviour -- for instance there is no flag

6 # checking in the buffer objects (why test our test case?), rather

7 # what we want to test is what is passed into the flags argument.

8 #

9

10 from __future__ import unicode_literals

11

12 cimport stdlib

13 cimport python_buffer

14 cimport stdio

15 cimport cython

16

17 from python_ref cimport PyObject

18

19 __test__ = {}

20

21 import re

22 exclude = []#re.compile('object').search]

23

24 def testcase(func):

25 for e in exclude:

26 if e(func.__name__):

27 return func

28 __test__[func.__name__] = func.__doc__

29 return func

30

31 def testcas(a):

32 pass

33

34

35 #

36 # Buffer acquire and release tests

37 #

38

39 def nousage():

40 """

41 The challenge here is just compilation.

42 """

43 cdef object[int, ndim=2] buf

44

45 def printbuf():

46 """

47 Just compilation.

48 """

49 cdef object[int, ndim=2] buf

50 print buf

51

52 @testcase

53 def acquire_release(o1, o2):

54 """

55 >>> A = IntMockBuffer("A", range(6))

56 >>> B = IntMockBuffer("B", range(6))

57 >>> acquire_release(A, B)

58 acquired A

59 released A

60 acquired B

61 released B

62 >>> acquire_release(None, None)

63 >>> acquire_release(None, B)

64 acquired B

65 released B

66 """

67 cdef object[int] buf

68 buf = o1

69 buf = o2

70

71 @testcase

72 def acquire_raise(o):

73 """

74 Apparently, doctest won't handle mixed exceptions and print

75 stats, so need to circumvent this.

76

77 >>> A = IntMockBuffer("A", range(6))

78 >>> A.resetlog()

79 >>> acquire_raise(A)

80 Traceback (most recent call last):

81 ...

82 Exception: on purpose

83 >>> A.printlog()

84 acquired A

85 released A

86

87 """

88 cdef object[int] buf

89 buf = o

90 raise Exception("on purpose")

91

92 @testcase

93 def acquire_failure1():

94 """

95 >>> acquire_failure1()

96 acquired working

97 0 3

98 0 3

99 released working

100 """

101 cdef object[int] buf

102 buf = IntMockBuffer("working", range(4))

103 print buf[0], buf[3]

104 try:

105 buf = ErrorBuffer()

106 assert False

107 except Exception:

108 print buf[0], buf[3]

109

110 @testcase

111 def acquire_failure2():

112 """

113 >>> acquire_failure2()

114 acquired working

115 0 3

116 0 3

117 released working

118 """

119 cdef object[int] buf = IntMockBuffer("working", range(4))

120 print buf[0], buf[3]

121 try:

122 buf = ErrorBuffer()

123 assert False

124 except Exception:

125 print buf[0], buf[3]

126

127 @testcase

128 def acquire_failure3():

129 """

130 >>> acquire_failure3()

131 acquired working

132 0 3

133 released working

134 acquired working

135 0 3

136 released working

137 """

138 cdef object[int] buf

139 buf = IntMockBuffer("working", range(4))

140 print buf[0], buf[3]

141 try:

142 buf = 3

143 assert False

144 except Exception:

145 print buf[0], buf[3]

146

147 @testcase

148 def acquire_failure4():

149 """

150 >>> acquire_failure4()

151 acquired working

152 0 3

153 released working

154 acquired working

155 0 3

156 released working

157 """

158 cdef object[int] buf = IntMockBuffer("working", range(4))

159 print buf[0], buf[3]

160 try:

161 buf = 2

162 assert False

163 except Exception:

164 print buf[0], buf[3]

165

166 @testcase

167 def acquire_failure5():

168 """

169 >>> acquire_failure5()

170 Traceback (most recent call last):

171 ...

172 ValueError: Buffer acquisition failed on assignment; and then reacquiring the old buffer failed too!

173 """

174 cdef object[int] buf

175 buf = IntMockBuffer("working", range(4))

176 buf.fail = True

177 buf = 3

178

179

180 @testcase

181 def acquire_nonbuffer1(first, second=None):

182 """

183 >>> acquire_nonbuffer1(3)

184 Traceback (most recent call last):

185 ...

186 TypeError: 'int' does not have the buffer interface

187 >>> acquire_nonbuffer1(type)

188 Traceback (most recent call last):

189 ...

190 TypeError: 'type' does not have the buffer interface

191 >>> acquire_nonbuffer1(None, 2)

192 Traceback (most recent call last):

193 ...

194 TypeError: 'int' does not have the buffer interface

195 """

196 cdef object[int] buf

197 buf = first

198 buf = second

199

200 @testcase

201 def acquire_nonbuffer2():

202 """

203 >>> acquire_nonbuffer2()

204 acquired working

205 0 3

206 released working

207 acquired working

208 0 3

209 released working

210 """

211 cdef object[int] buf = IntMockBuffer("working", range(4))

212 print buf[0], buf[3]

213 try:

214 buf = ErrorBuffer

215 assert False

216 except Exception:

217 print buf[0], buf[3]

218

219

220 @testcase

221 def as_argument(object[int] bufarg, int n):

222 """

223 >>> A = IntMockBuffer("A", range(6))

224 >>> as_argument(A, 6)

225 acquired A

226 0 1 2 3 4 5 END

227 released A

228 """

229 cdef int i

230 for i in range(n):

231 print bufarg[i],

232 print 'END'

233

234 @testcase

235 def as_argument_defval(object[int] bufarg=IntMockBuffer('default', range(6)), int n=6):

236 """

237 >>> as_argument_defval()

238 acquired default

239 0 1 2 3 4 5 END

240 released default

241 >>> A = IntMockBuffer("A", range(6))

242 >>> as_argument_defval(A, 6)

243 acquired A

244 0 1 2 3 4 5 END

245 released A

246 """

247 cdef int i

248 for i in range(n):

249 print bufarg[i],

250 print 'END'

251

252 @testcase

253 def cdef_assignment(obj, n):

254 """

255 >>> A = IntMockBuffer("A", range(6))

256 >>> cdef_assignment(A, 6)

257 acquired A

258 0 1 2 3 4 5 END

259 released A

260

261 """

262 cdef object[int] buf = obj

263 cdef int i

264 for i in range(n):

265 print buf[i],

266 print 'END'

267

268 @testcase

269 def forin_assignment(objs, int pick):

270 """

271 >>> A = IntMockBuffer("A", range(6))

272 >>> B = IntMockBuffer("B", range(6))

273 >>> forin_assignment([A, B, A, A], 2)

274 acquired A

275 2

276 released A

277 acquired B

278 2

279 released B

280 acquired A

281 2

282 released A

283 acquired A

284 2

285 released A

286 """

287 cdef object[int] buf

288 for buf in objs:

289 print buf[pick]

290

291 @testcase

292 def cascaded_buffer_assignment(obj):

293 """

294 >>> A = IntMockBuffer("A", range(6))

295 >>> cascaded_buffer_assignment(A)

296 acquired A

297 acquired A

298 released A

299 released A

300 """

301 cdef object[int] a, b

302 a = b = obj

303

304 @testcase

305 def tuple_buffer_assignment1(a, b):

306 """

307 >>> A = IntMockBuffer("A", range(6))

308 >>> B = IntMockBuffer("B", range(6))

309 >>> tuple_buffer_assignment1(A, B)

310 acquired A

311 acquired B

312 released A

313 released B

314 """

315 cdef object[int] x, y

316 x, y = a, b

317

318 @testcase

319 def tuple_buffer_assignment2(tup):

320 """

321 >>> A = IntMockBuffer("A", range(6))

322 >>> B = IntMockBuffer("B", range(6))

323 >>> tuple_buffer_assignment2((A, B))

324 acquired A

325 acquired B

326 released A

327 released B

328 """

329 cdef object[int] x, y

330 x, y = tup

331

332 @testcase

333 def explicitly_release_buffer():

334 """

335 >>> explicitly_release_buffer()

336 acquired A

337 released A

338 After release

339 """

340 cdef object[int] x = IntMockBuffer("A", range(10))

341 x = None

342 print "After release"

343

344 #

345 # Format strings

346 #

347 @testcase

348 def alignment_string(object[int] buf):

349 """

350 >>> alignment_string(IntMockBuffer(None, [1,2], format="@i"))

351 2

352 >>> alignment_string(IntMockBuffer(None, [1,2], format="@i@@"))

353 2

354 >>> alignment_string(IntMockBuffer(None, [1,2], format=">i"))

355 Traceback (most recent call last):

356 ...

357 ValueError: Buffer acquisition error: Only native byte order, size and alignment supported.

358 >>> alignment_string(IntMockBuffer(None, [1,2], format="<i"))

359 Traceback (most recent call last):

360 ...

361 ValueError: Buffer acquisition error: Only native byte order, size and alignment supported.

362 >>> alignment_string(IntMockBuffer(None, [1,2], format="=i"))

363 Traceback (most recent call last):

364 ...

365 ValueError: Buffer acquisition error: Only native byte order, size and alignment supported.

366 >>> alignment_string(IntMockBuffer(None, [1,2], format="!i"))

367 Traceback (most recent call last):

368 ...

369 ValueError: Buffer acquisition error: Only native byte order, size and alignment supported.

370 """

371 print buf[1]

372

373 @testcase

374 def wrong_string(object[int] buf):

375 """

376 >>> wrong_string(IntMockBuffer(None, [1,2], format="iasdf"))

377 Traceback (most recent call last):

378 ...

379 ValueError: Buffer format string specifies more data than 'int' can hold (expected end, got 'asdf')

380 >>> wrong_string(IntMockBuffer(None, [1,2], format="$$"))

381 Traceback (most recent call last):

382 ...

383 ValueError: Buffer datatype mismatch (expected 'i', got '$$')

384 """

385 print buf[1]

386

387 #

388 # Getting items and index bounds checking

389 #

390 @testcase

391 def get_int_2d(object[int, ndim=2] buf, int i, int j):

392 """

393 >>> C = IntMockBuffer("C", range(6), (2,3))

394 >>> get_int_2d(C, 1, 1)

395 acquired C

396 released C

397 4

398

399 Check negative indexing:

400 >>> get_int_2d(C, -1, 0)

401 acquired C

402 released C

403 3

404 >>> get_int_2d(C, -1, -2)

405 acquired C

406 released C

407 4

408 >>> get_int_2d(C, -2, -3)

409 acquired C

410 released C

411 0

412

413 Out-of-bounds errors:

414 >>> get_int_2d(C, 2, 0)

415 Traceback (most recent call last):

416 ...

417 IndexError: Out of bounds on buffer access (axis 0)

418 >>> get_int_2d(C, 0, -4)

419 Traceback (most recent call last):

420 ...

421 IndexError: Out of bounds on buffer access (axis 1)

422 """

423 return buf[i, j]

424

425 @testcase

426 def get_int_2d_uintindex(object[int, ndim=2] buf, unsigned int i, unsigned int j):

427 """

428 Unsigned indexing:

429 >>> C = IntMockBuffer("C", range(6), (2,3))

430 >>> get_int_2d_uintindex(C, 0, 0)

431 acquired C

432 released C

433 0

434 >>> get_int_2d_uintindex(C, 1, 2)

435 acquired C

436 released C

437 5

438 """

439 # This is most interesting with regards to the C code

440 # generated.

441 return buf[i, j]

442

443 @testcase

444 def set_int_2d(object[int, ndim=2] buf, int i, int j, int value):

445 """

446 Uses get_int_2d to read back the value afterwards. For pure

447 unit test, one should support reading in MockBuffer instead.

448

449 >>> C = IntMockBuffer("C", range(6), (2,3))

450 >>> set_int_2d(C, 1, 1, 10)

451 acquired C

452 released C

453 >>> get_int_2d(C, 1, 1)

454 acquired C

455 released C

456 10

457

458 Check negative indexing:

459 >>> set_int_2d(C, -1, 0, 3)

460 acquired C

461 released C

462 >>> get_int_2d(C, -1, 0)

463 acquired C

464 released C

465 3

466

467 >>> set_int_2d(C, -1, -2, 8)

468 acquired C

469 released C

470 >>> get_int_2d(C, -1, -2)

471 acquired C

472 released C

473 8

474

475 >>> set_int_2d(C, -2, -3, 9)

476 acquired C

477 released C

478 >>> get_int_2d(C, -2, -3)

479 acquired C

480 released C

481 9

482

483 Out-of-bounds errors:

484 >>> set_int_2d(C, 2, 0, 19)

485 Traceback (most recent call last):

486 ...

487 IndexError: Out of bounds on buffer access (axis 0)

488 >>> set_int_2d(C, 0, -4, 19)

489 Traceback (most recent call last):

490 ...

491 IndexError: Out of bounds on buffer access (axis 1)

492

493 """

494 buf[i, j] = value

495

496 @testcase

497 def list_comprehension(object[int] buf, len):

498 """

499 >>> list_comprehension(IntMockBuffer(None, [1,2,3]), 3)

500 1|2|3

501 """

502 cdef int i

503 print u"|".join([unicode(buf[i]) for i in range(len)])

504

505 #

506 # The negative_indices buffer option

507 #

508 @testcase

509 def no_negative_indices(object[int, negative_indices=False] buf, int idx):

510 """

511 The most interesting thing here is to inspect the C source and

512 make sure optimal code is produced.

513

514 >>> A = IntMockBuffer(None, range(6))

515 >>> no_negative_indices(A, 3)

516 3

517 >>> no_negative_indices(A, -1)

518 Traceback (most recent call last):

519 ...

520 IndexError: Out of bounds on buffer access (axis 0)

521 """

522 return buf[idx]

523

524 #

525 # Buffer type mismatch examples. Varying the type and access

526 # method simultaneously, the odds of an interaction is virtually

527 # zero.

528 #

529 @testcase

530 def fmtst1(buf):

531 """

532 >>> fmtst1(IntMockBuffer("A", range(3)))

533 Traceback (most recent call last):

534 ...

535 ValueError: Buffer datatype mismatch (expected 'f', got 'i')

536 """

537 cdef object[float] a = buf

538

539 @testcase

540 def fmtst2(object[int] buf):

541 """

542 >>> fmtst2(FloatMockBuffer("A", range(3)))

543 Traceback (most recent call last):

544 ...

545 ValueError: Buffer datatype mismatch (expected 'i', got 'f')

546 """

547

548 @testcase

549 def ndim1(object[int, ndim=2] buf):

550 """

551 >>> ndim1(IntMockBuffer("A", range(3)))

552 Traceback (most recent call last):

553 ...

554 ValueError: Buffer has wrong number of dimensions (expected 2, got 1)

555 """

556

557 #

558 # Test which flags are passed.

559 #

560 @testcase

561 def readonly(obj):

562 """

563 >>> R = UnsignedShortMockBuffer("R", range(27), shape=(3, 3, 3))

564 >>> readonly(R)

565 acquired R

566 25

567 released R

568 >>> [str(x) for x in R.recieved_flags] # Works in both py2 and py3

569 ['FORMAT', 'INDIRECT', 'ND', 'STRIDES']

570 """

571 cdef object[unsigned short int, ndim=3] buf = obj

572 print buf[2, 2, 1]

573

574 @testcase

575 def writable(obj):

576 """

577 >>> R = UnsignedShortMockBuffer("R", range(27), shape=(3, 3, 3))

578 >>> writable(R)

579 acquired R

580 released R

581 >>> [str(x) for x in R.recieved_flags] # Py2/3

582 ['FORMAT', 'INDIRECT', 'ND', 'STRIDES', 'WRITABLE']

583 """

584 cdef object[unsigned short int, ndim=3] buf = obj

585 buf[2, 2, 1] = 23

586

587 @testcase

588 def strided(object[int, ndim=1, mode='strided'] buf):

589 """

590 >>> A = IntMockBuffer("A", range(4))

591 >>> strided(A)

592 acquired A

593 released A

594 2

595 >>> [str(x) for x in A.recieved_flags] # Py2/3

596 ['FORMAT', 'ND', 'STRIDES']

597

598 Check that the suboffsets were patched back prior to release.

599 >>> A.release_ok

600 True

601 """

602 return buf[2]

603

604 @testcase

605 def c_contig(object[int, ndim=1, mode='c'] buf):

606 """

607 >>> A = IntMockBuffer(None, range(4))

608 >>> c_contig(A)

609 2

610 >>> [str(x) for x in A.recieved_flags]

611 ['FORMAT', 'ND', 'STRIDES', 'C_CONTIGUOUS']

612 """

613 return buf[2]

614

615 @testcase

616 def c_contig_2d(object[int, ndim=2, mode='c'] buf):

617 """

618 Multi-dim has seperate implementation

619

620 >>> A = IntMockBuffer(None, range(12), shape=(3,4))

621 >>> c_contig_2d(A)

622 7

623 >>> [str(x) for x in A.recieved_flags]

624 ['FORMAT', 'ND', 'STRIDES', 'C_CONTIGUOUS']

625 """

626 return buf[1, 3]

627

628 @testcase

629 def f_contig(object[int, ndim=1, mode='fortran'] buf):

630 """

631 >>> A = IntMockBuffer(None, range(4))

632 >>> f_contig(A)

633 2

634 >>> [str(x) for x in A.recieved_flags]

635 ['FORMAT', 'ND', 'STRIDES', 'F_CONTIGUOUS']

636 """

637 return buf[2]

638

639 @testcase

640 def f_contig_2d(object[int, ndim=2, mode='fortran'] buf):

641 """

642 Must set up strides manually to ensure Fortran ordering.

643

644 >>> A = IntMockBuffer(None, range(12), shape=(4,3), strides=(1, 4))

645 >>> f_contig_2d(A)

646 7

647 >>> [str(x) for x in A.recieved_flags]

648 ['FORMAT', 'ND', 'STRIDES', 'F_CONTIGUOUS']

649 """

650 return buf[3, 1]

651

652 #

653 # Test compiler options for bounds checking. We create an array with a

654 # safe "boundary" (memory

655 # allocated outside of what it published) and then check whether we get back

656 # what we stored in the memory or an error.

657

658 @testcase

659 def safe_get(object[int] buf, int idx):

660 """

661 >>> A = IntMockBuffer(None, range(10), shape=(3,), offset=5)

662

663 Validate our testing buffer...

664 >>> safe_get(A, 0)

665 5

666 >>> safe_get(A, 2)

667 7

668 >>> safe_get(A, -3)

669 5

670

671 Access outside it. This is already done above for bounds check

672 testing but we include it to tell the story right.

673

674 >>> safe_get(A, -4)

675 Traceback (most recent call last):

676 ...

677 IndexError: Out of bounds on buffer access (axis 0)

678 >>> safe_get(A, 3)

679 Traceback (most recent call last):

680 ...

681 IndexError: Out of bounds on buffer access (axis 0)

682 """

683 return buf[idx]

684

685 @testcase

686 @cython.boundscheck(False) # outer decorators should take precedence

687 @cython.boundscheck(True)

688 def unsafe_get(object[int] buf, int idx):

689 """

690 Access outside of the area the buffer publishes.

691 >>> A = IntMockBuffer(None, range(10), shape=(3,), offset=5)

692 >>> unsafe_get(A, -4)

693 4

694 >>> unsafe_get(A, -5)

695 3

696 >>> unsafe_get(A, 3)

697 8

698 """

699 return buf[idx]

700

701 @testcase

702 @cython.boundscheck(False)

703 def unsafe_get_nonegative(object[int, negative_indices=False] buf, int idx):

704 """

705 Also inspect the C source to see that it is optimal...

706

707 >>> A = IntMockBuffer(None, range(10), shape=(3,), offset=5)

708 >>> unsafe_get_nonegative(A, -2)

709 3

710 """

711 return buf[idx]

712

713 @testcase

714 def mixed_get(object[int] buf, int unsafe_idx, int safe_idx):

715 """

716 >>> A = IntMockBuffer(None, range(10), shape=(3,), offset=5)

717 >>> mixed_get(A, -4, 0)

718 (4, 5)

719 >>> mixed_get(A, 0, -4)

720 Traceback (most recent call last):

721 ...

722 IndexError: Out of bounds on buffer access (axis 0)

723 """

724 with cython.boundscheck(False):

725 one = buf[unsafe_idx]

726 with cython.boundscheck(True):

727 two = buf[safe_idx]

728 return (one, two)

729

730 #

731 # Coercions

732 #

733 @testcase

734 def coercions(object[unsigned char] uc):

735 """

736 TODO

737 """

738 print type(uc[0])

739 uc[0] = -1

740 print uc[0]

741 uc[0] = <int>3.14

742 print uc[0]

743

744 cdef char* ch = "asfd"

745 cdef object[object] objbuf

746 objbuf[3] = ch

747

748

749 #

750 # Testing that accessing data using various types of buffer access

751 # all works.

752 #

753

754 def printbuf_int(object[int] buf, shape):

755 # Utility func

756 cdef int i

757 for i in range(shape[0]):

758 print buf[i],

759 print 'END'

760

761

762 @testcase

763 def printbuf_int_2d(o, shape):

764 """

765 Strided:

766

767 >>> printbuf_int_2d(IntMockBuffer("A", range(6), (2,3)), (2,3))

768 acquired A

769 0 1 2 END

770 3 4 5 END

771 released A

772 >>> printbuf_int_2d(IntMockBuffer("A", range(100), (3,3), strides=(20,5)), (3,3))

773 acquired A

774 0 5 10 END

775 20 25 30 END

776 40 45 50 END

777 released A

778

779 Indirect:

780 >>> printbuf_int_2d(IntMockBuffer("A", [[1,2],[3,4]]), (2,2))

781 acquired A

782 1 2 END

783 3 4 END

784 released A

785 """

786 # should make shape builtin

787 cdef object[int, ndim=2] buf

788 buf = o

789 cdef int i, j

790 for i in range(shape[0]):

791 for j in range(shape[1]):

792 print buf[i, j],

793 print 'END'

794

795 @testcase

796 def printbuf_float(o, shape):

797 """

798 >>> printbuf_float(FloatMockBuffer("F", [1.0, 1.25, 0.75, 1.0]), (4,))

799 acquired F

800 1.0 1.25 0.75 1.0 END

801 released F

802 """

803

804 # should make shape builtin

805 cdef object[float] buf

806 buf = o

807 cdef int i, j

808 for i in range(shape[0]):

809 print buf[i],

810 print "END"

811

812

813 #

814 # Test assignments

815 #

816 @testcase

817 def inplace_operators(object[int] buf):

818 """

819 >>> buf = IntMockBuffer(None, [2, 2])

820 >>> inplace_operators(buf)

821 >>> printbuf_int(buf, (2,))

822 0 3 END

823 """

824 cdef int j = 0

825 buf[1] += 1

826 buf[j] *= 2

827 buf[0] -= 4

828

829

830

831 #

832 # Typedefs

833 #

834 # Test three layers of typedefs going through a h file for plain int, and

835 # simply a header file typedef for floats and unsigned.

836

837 ctypedef int td_cy_int

838 cdef extern from "bufaccess.h":

839 ctypedef td_cy_int td_h_short # Defined as short, but Cython doesn't know this!

840 ctypedef float td_h_double # Defined as double

841 ctypedef unsigned int td_h_ushort # Defined as unsigned short

842 ctypedef td_h_short td_h_cy_short

843

844 @testcase

845 def printbuf_td_cy_int(object[td_cy_int] buf, shape):

846 """

847 >>> printbuf_td_cy_int(IntMockBuffer(None, range(3)), (3,))

848 0 1 2 END

849 >>> printbuf_td_cy_int(ShortMockBuffer(None, range(3)), (3,))

850 Traceback (most recent call last):

851 ...

852 ValueError: Buffer datatype mismatch (rejecting on 'h')

853

854 """

855 cdef int i

856 for i in range(shape[0]):

857 print buf[i],

858 print 'END'

859

860 @testcase

861 def printbuf_td_h_short(object[td_h_short] buf, shape):

862 """

863 >>> printbuf_td_h_short(ShortMockBuffer(None, range(3)), (3,))

864 0 1 2 END

865 >>> printbuf_td_h_short(IntMockBuffer(None, range(3)), (3,))

866 Traceback (most recent call last):

867 ...

868 ValueError: Buffer datatype mismatch (rejecting on 'i')

869 """

870 cdef int i

871 for i in range(shape[0]):

872 print buf[i],

873 print 'END'

874

875 @testcase

876 def printbuf_td_h_cy_short(object[td_h_cy_short] buf, shape):

877 """

878 >>> printbuf_td_h_cy_short(ShortMockBuffer(None, range(3)), (3,))

879 0 1 2 END

880 >>> printbuf_td_h_cy_short(IntMockBuffer(None, range(3)), (3,))

881 Traceback (most recent call last):

882 ...

883 ValueError: Buffer datatype mismatch (rejecting on 'i')

884 """

885 cdef int i

886 for i in range(shape[0]):

887 print buf[i],

888 print 'END'

889

890 @testcase

891 def printbuf_td_h_ushort(object[td_h_ushort] buf, shape):

892 """

893 >>> printbuf_td_h_ushort(UnsignedShortMockBuffer(None, range(3)), (3,))

894 0 1 2 END

895 >>> printbuf_td_h_ushort(ShortMockBuffer(None, range(3)), (3,))

896 Traceback (most recent call last):

897 ...

898 ValueError: Buffer datatype mismatch (rejecting on 'h')

899 """

900 cdef int i

901 for i in range(shape[0]):

902 print buf[i],

903 print 'END'

904

905 @testcase

906 def printbuf_td_h_double(object[td_h_double] buf, shape):

907 """

908 >>> printbuf_td_h_double(DoubleMockBuffer(None, [0.25, 1, 3.125]), (3,))

909 0.25 1.0 3.125 END

910 >>> printbuf_td_h_double(FloatMockBuffer(None, [0.25, 1, 3.125]), (3,))

911 Traceback (most recent call last):

912 ...

913 ValueError: Buffer datatype mismatch (rejecting on 'f')

914 """

915 cdef int i

916 for i in range(shape[0]):

917 print buf[i],

918 print 'END'

919

920

921 #

922 # Object access

923 #

924 from python_ref cimport Py_INCREF, Py_DECREF

925 def addref(*args):

926 for item in args: Py_INCREF(item)

927 def decref(*args):

928 for item in args: Py_DECREF(item)

929

930 def get_refcount(x):

931 return (<PyObject*>x).ob_refcnt

932

933 @testcase

934 def printbuf_object(object[object] buf, shape):

935 """

936 Only play with unique objects, interned numbers etc. will have

937 unpredictable refcounts.

938

939 ObjectMockBuffer doesn't do anything about increfing/decrefing,

940 we to the "buffer implementor" refcounting directly in the

941 testcase.

942

943 >>> a, b, c = "globally_unique_string_23234123", {4:23}, [34,3]

944 >>> get_refcount(a), get_refcount(b), get_refcount(c)

945 (2, 2, 2)

946 >>> A = ObjectMockBuffer(None, [a, b, c])

947 >>> printbuf_object(A, (3,))

948 'globally_unique_string_23234123' 2

949 {4: 23} 2

950 [34, 3] 2

951 """

952 cdef int i

953 for i in range(shape[0]):

954 print repr(buf[i]), (<PyObject*>buf[i]).ob_refcnt

955

956 @testcase

957 def assign_to_object(object[object] buf, int idx, obj):

958 """

959 See comments on printbuf_object above.

960

961 >>> a, b = [1, 2, 3], [4, 5, 6]

962 >>> get_refcount(a), get_refcount(b)

963 (2, 2)

964 >>> addref(a)

965 >>> A = ObjectMockBuffer(None, [1, a]) # 1, ...,otherwise it thinks nested lists...

966 >>> get_refcount(a), get_refcount(b)

967 (3, 2)

968 >>> assign_to_object(A, 1, b)

969 >>> get_refcount(a), get_refcount(b)

970 (2, 3)

971 >>> decref(b)

972 """

973 buf[idx] = obj

974

975 #

976 # cast option

977 #

978 @testcase

979 def buffer_cast(object[unsigned int, cast=True] buf, int idx):

980 """

981 Round-trip a signed int through unsigned int buffer access.

982

983 >>> A = IntMockBuffer(None, [-100])

984 >>> buffer_cast(A, 0)

985 -100

986 """

987 cdef unsigned int data = buf[idx]

988 return <int>data

989

990 @testcase

991 def buffer_cast_fails(object[char, cast=True] buf):

992 """

993 Cannot cast between datatype of different sizes.

994

995 >>> buffer_cast_fails(IntMockBuffer(None, [0]))

996 Traceback (most recent call last):

997 ...

998 ValueError: Attempted cast of buffer to datatype of different size.

999 """

1000 return buf[0]

1001

1002

1003 #

1004 # Testcase support code (more tests below!, because of scope rules)

1005 #

1006

1007

1008 available_flags = (

1009 ('FORMAT', python_buffer.PyBUF_FORMAT),

1010 ('INDIRECT', python_buffer.PyBUF_INDIRECT),

1011 ('ND', python_buffer.PyBUF_ND),

1012 ('STRIDES', python_buffer.PyBUF_STRIDES),

1013 ('C_CONTIGUOUS', python_buffer.PyBUF_C_CONTIGUOUS),

1014 ('F_CONTIGUOUS', python_buffer.PyBUF_F_CONTIGUOUS),

1015 ('WRITABLE', python_buffer.PyBUF_WRITABLE)

1016 )

1017

1018 cdef class MockBuffer:

1019 cdef object format, offset

1020 cdef void* buffer

1021 cdef int len, itemsize, ndim

1022 cdef Py_ssize_t* strides

1023 cdef Py_ssize_t* shape

1024 cdef Py_ssize_t* suboffsets

1025 cdef object label, log

1026

1027 cdef readonly object recieved_flags, release_ok

1028 cdef public object fail

1029

1030 def __init__(self, label, data, shape=None, strides=None, format=None, offset=0):

1031 # It is important not to store references to data after the constructor

1032 # as refcounting is checked on object buffers.

1033 self.label = label

1034 self.release_ok = True

1035 self.log = ""

1036 self.offset = offset

1037 self.itemsize = self.get_itemsize()

1038 if format is None: format = self.get_default_format()

1039 if shape is None: shape = (len(data),)

1040 if strides is None:

1041 strides = []

1042 cumprod = 1

1043 rshape = list(shape)

1044 rshape.reverse()

1045 for s in rshape:

1046 strides.append(cumprod)

1047 cumprod *= s

1048 strides.reverse()

1049 strides = [x * self.itemsize for x in strides]

1050 suboffsets = [-1] * len(shape)

1051 datashape = [len(data)]

1052 p = data

1053 while True:

1054 p = p[0]

1055 if isinstance(p, list): datashape.append(len(p))

1056 else: break

1057 if len(datashape) > 1:

1058 # indirect access

1059 self.ndim = len(datashape)

1060 shape = datashape

1061 self.buffer = self.create_indirect_buffer(data, shape)

1062 suboffsets = [0] * (self.ndim-1) + [-1]

1063 strides = [sizeof(void*)] * (self.ndim-1) + [self.itemsize]

1064 self.suboffsets = self.list_to_sizebuf(suboffsets)

1065 else:

1066 # strided and/or simple access

1067 self.buffer = self.create_buffer(data)

1068 self.ndim = len(shape)

1069 self.suboffsets = NULL

1070

1071 self.format = format

1072 self.len = len(data) * self.itemsize

1073

1074 self.strides = self.list_to_sizebuf(strides)

1075 self.shape = self.list_to_sizebuf(shape)

1076

1077 def __dealloc__(self):

1078 stdlib.free(self.strides)

1079 stdlib.free(self.shape)

1080 if self.suboffsets != NULL:

1081 stdlib.free(self.suboffsets)

1082 # must recursively free indirect...

1083 else:

1084 stdlib.free(self.buffer)

1085

1086 cdef void* create_buffer(self, data):

1087 cdef char* buf = <char*>stdlib.malloc(len(data) * self.itemsize)

1088 cdef char* it = buf

1089 for value in data:

1090 self.write(it, value)

1091 it += self.itemsize

1092 return buf

1093

1094 cdef void* create_indirect_buffer(self, data, shape):

1095 cdef void** buf

1096 assert shape[0] == len(data)

1097 if len(shape) == 1:

1098 return self.create_buffer(data)

1099 else:

1100 shape = shape[1:]

1101 buf = <void**>stdlib.malloc(len(data) * sizeof(void*))

1102 for idx, subdata in enumerate(data):

1103 buf[idx] = self.create_indirect_buffer(subdata, shape)

1104 return buf

1105

1106 cdef Py_ssize_t* list_to_sizebuf(self, l):

1107 cdef Py_ssize_t* buf = <Py_ssize_t*>stdlib.malloc(len(l) * sizeof(Py_ssize_t))

1108 for i, x in enumerate(l):

1109 buf[i] = x

1110 return buf

1111

1112 def __getbuffer__(MockBuffer self, Py_buffer* buffer, int flags):

1113 if self.fail:

1114 raise ValueError("Failing on purpose")

1115

1116 self.recieved_flags = []

1117 cdef int value

1118 for name, value in available_flags:

1119 if (value & flags) == value:

1120 self.recieved_flags.append(name)

1121

1122 buffer.buf = <void*>(<char*>self.buffer + (<int>self.offset * self.itemsize))

1123 buffer.obj = self

1124 buffer.len = self.len

1125 buffer.readonly = 0

1126 buffer.format = <char*>self.format

1127 buffer.ndim = self.ndim

1128 buffer.shape = self.shape

1129 buffer.strides = self.strides

1130 buffer.suboffsets = self.suboffsets

1131 buffer.itemsize = self.itemsize

1132 buffer.internal = NULL

1133 if self.label:

1134 msg = "acquired %s" % self.label

1135 print msg

1136 self.log += msg + "\n"

1137

1138 def __releasebuffer__(MockBuffer self, Py_buffer* buffer):

1139 if buffer.suboffsets != self.suboffsets:

1140 self.release_ok = False

1141 if self.label:

1142 msg = "released %s" % self.label

1143 print msg

1144 self.log += msg + "\n"

1145

1146 def printlog(self):

1147 print self.log[:-1]

1148

1149 def resetlog(self):

1150 self.log = ""

1151

1152 cdef int write(self, char* buf, object value) except -1: raise Exception()

1153 cdef get_itemsize(self):

1154 print "ERROR, not subclassed", self.__class__

1155 cdef get_default_format(self):

1156 print "ERROR, not subclassed", self.__class__

1157

1158 cdef class CharMockBuffer(MockBuffer):

1159 cdef int write(self, char* buf, object value) except -1:

1160 (<char*>buf)[0] = <int>value

1161 return 0

1162 cdef get_itemsize(self): return sizeof(char)

1163 cdef get_default_format(self): return b"@b"

1164

1165 cdef class IntMockBuffer(MockBuffer):

1166 cdef int write(self, char* buf, object value) except -1:

1167 (<int*>buf)[0] = <int>value

1168 return 0

1169 cdef get_itemsize(self): return sizeof(int)

1170 cdef get_default_format(self): return b"@i"

1171

1172 cdef class UnsignedIntMockBuffer(MockBuffer):

1173 cdef int write(self, char* buf, object value) except -1:

1174 (<unsigned int*>buf)[0] = <unsigned int>value

1175 return 0

1176 cdef get_itemsize(self): return sizeof(unsigned int)

1177 cdef get_default_format(self): return b"@I"

1178

1179 cdef class ShortMockBuffer(MockBuffer):

1180 cdef int write(self, char* buf, object value) except -1:

1181 (<short*>buf)[0] = <short>value

1182 return 0

1183 cdef get_itemsize(self): return sizeof(short)

1184 cdef get_default_format(self): return b"h" # Try without endian specifier

1185

1186 cdef class UnsignedShortMockBuffer(MockBuffer):

1187 cdef int write(self, char* buf, object value) except -1:

1188 (<unsigned short*>buf)[0] = <unsigned short>value

1189 return 0

1190 cdef get_itemsize(self): return sizeof(unsigned short)

1191 cdef get_default_format(self): return b"@1H" # Try with repeat count

1192

1193 cdef class FloatMockBuffer(MockBuffer):

1194 cdef int write(self, char* buf, object value) except -1:

1195 (<float*>buf)[0] = <float>value

1196 return 0

1197 cdef get_itemsize(self): return sizeof(float)

1198 cdef get_default_format(self): return b"f"

1199

1200 cdef class DoubleMockBuffer(MockBuffer):

1201 cdef int write(self, char* buf, object value) except -1:

1202 (<double*>buf)[0] = <double>value

1203 return 0

1204 cdef get_itemsize(self): return sizeof(double)

1205 cdef get_default_format(self): return b"d"

1206

1207 cdef extern from *:

1208 void* addr_of_pyobject "(void*)"(object)

1209

1210 cdef class ObjectMockBuffer(MockBuffer):

1211 cdef int write(self, char* buf, object value) except -1:

1212 (<void**>buf)[0] = addr_of_pyobject(value)

1213 return 0

1214

1215 cdef get_itemsize(self): return sizeof(void*)

1216 cdef get_default_format(self): return b"@O"

1217

1218

1219 cdef class IntStridedMockBuffer(IntMockBuffer):

1220 cdef __cythonbufferdefaults__ = {"mode" : "strided"}

1221

1222 cdef class ErrorBuffer:

1223 cdef object label

1224

1225 def __init__(self, label):

1226 self.label = label

1227

1228 def __getbuffer__(ErrorBuffer self, Py_buffer* buffer, int flags):

1229 raise Exception("acquiring %s" % self.label)

1230

1231 def __releasebuffer__(ErrorBuffer self, Py_buffer* buffer):

1232 raise Exception("releasing %s" % self.label)

1233

1234 #

1235 # Typed buffers

1236 #

1237 @testcase

1238 def typedbuffer1(obj):

1239 """

1240 >>> typedbuffer1(IntMockBuffer("A", range(10)))

1241 acquired A

1242 released A

1243 >>> typedbuffer1(None)

1244 >>> typedbuffer1(4)

1245 Traceback (most recent call last):

1246 ...

1247 TypeError: Cannot convert int to bufaccess.IntMockBuffer

1248 """

1249 cdef IntMockBuffer[int, ndim=1] buf = obj

1250

1251 @testcase

1252 def typedbuffer2(IntMockBuffer[int, ndim=1] obj):

1253 """

1254 >>> typedbuffer2(IntMockBuffer("A", range(10)))

1255 acquired A

1256 released A

1257 >>> typedbuffer2(None)

1258 >>> typedbuffer2(4)

1259 Traceback (most recent call last):

1260 ...

1261 TypeError: Argument 'obj' has incorrect type (expected bufaccess.IntMockBuffer, got int)

1262 """

1263 pass

1264

1265 #

1266 # Test __cythonbufferdefaults__

1267 #

1268 @testcase

1269 def bufdefaults1(IntStridedMockBuffer[int, ndim=1] buf):

1270 """

1271 For IntStridedMockBuffer, mode should be

1272 "strided" by defaults which should show

1273 up in the flags.

1274

1275 >>> A = IntStridedMockBuffer("A", range(10))

1276 >>> bufdefaults1(A)

1277 acquired A

1278 released A

1279 >>> [str(x) for x in A.recieved_flags]

1280 ['FORMAT', 'ND', 'STRIDES']

1281 """

1282 pass

1283

1284

1285 #

1286 # Structs

1287 #

1288 cdef struct MyStruct:

1289 char a

1290 char b

1291 long long int c

1292 int d

1293 int e

1294

1295 cdef struct SmallStruct:

1296 int a

1297 int b

1298

1299 cdef struct NestedStruct:

1300 SmallStruct x

1301 SmallStruct y

1302 int z

1303

1304 cdef class MyStructMockBuffer(MockBuffer):

1305 cdef int write(self, char* buf, object value) except -1:

1306 cdef MyStruct* s

1307 s = <MyStruct*>buf;

1308 s.a, s.b, s.c, s.d, s.e = value

1309 return 0

1310

1311 cdef get_itemsize(self): return sizeof(MyStruct)

1312 cdef get_default_format(self): return b"2bq2i"

1313

1314 cdef class NestedStructMockBuffer(MockBuffer):

1315 cdef int write(self, char* buf, object value) except -1:

1316 cdef NestedStruct* s

1317 s = <NestedStruct*>buf;

1318 s.x.a, s.x.b, s.y.a, s.y.b, s.z = value

1319 return 0

1320

1321 cdef get_itemsize(self): return sizeof(NestedStruct)

1322 cdef get_default_format(self): return b"2T{ii}i"

1323

1324 @testcase

1325 def basic_struct(object[MyStruct] buf):

1326 """

1327 >>> basic_struct(MyStructMockBuffer(None, [(1, 2, 3, 4, 5)]))

1328 1 2 3 4 5

1329 >>> basic_struct(MyStructMockBuffer(None, [(1, 2, 3, 4, 5)], format="bbqii"))

1330 1 2 3 4 5

1331 >>> basic_struct(MyStructMockBuffer(None, [(1, 2, 3, 4, 5)], format="i"))

1332 Traceback (most recent call last):

1333 ...

1334 ValueError: Buffer datatype mismatch (expected 'b', got 'i')

1335 """

1336 print buf[0].a, buf[0].b, buf[0].c, buf[0].d, buf[0].e

1337

1338 @testcase

1339 def nested_struct(object[NestedStruct] buf):

1340 """

1341 >>> nested_struct(NestedStructMockBuffer(None, [(1, 2, 3, 4, 5)]))

1342 1 2 3 4 5

1343 >>> nested_struct(NestedStructMockBuffer(None, [(1, 2, 3, 4, 5)], format="T{ii}T{2i}i"))

1344 1 2 3 4 5

1345 >>> nested_struct(NestedStructMockBuffer(None, [(1, 2, 3, 4, 5)], format="iiiii"))

1346 Traceback (most recent call last):

1347 ...

1348 ValueError: Expected start of SmallStruct, got 'iiiii'

1349 """

1350 print buf[0].x.a, buf[0].x.b, buf[0].y.a, buf[0].y.b, buf[0].z

1351

1352

1353 cdef struct LongComplex:

1354 long double real

1355 long double imag

1356

1357 cdef struct MixedComplex:

1358 long double real

1359 float imag

1360

1361 cdef class LongComplexMockBuffer(MockBuffer):

1362 cdef int write(self, char* buf, object value) except -1:

1363 cdef LongComplex* s

1364 s = <LongComplex*>buf;

1365 s.real, s.imag = value

1366 return 0

1367

1368 cdef get_itemsize(self): return sizeof(LongComplex)

1369 cdef get_default_format(self): return b"Zg"

1370

1371 @testcase

1372 def complex_struct_dtype(object[LongComplex] buf):

1373 """

1374 Note that the format string is "Zg" rather than "2g"...

1375 >>> complex_struct_dtype(LongComplexMockBuffer(None, [(0, -1)]))

1376 0.0 -1.0

1377 """

1378 print buf[0].real, buf[0].imag

1379

1380 @testcase

1381 def mixed_complex_struct_dtype(object[MixedComplex] buf):

1382 """

1383 Triggering a specific execution path for this case.

1384

1385 >>> mixed_complex_struct_dtype(LongComplexMockBuffer(None, [(0, -1)]))

1386 Traceback (most recent call last):

1387 ...

1388 ValueError: Cannot store complex number in 'MixedComplex' as 'long double' differs from 'float' in size.

1389 """

1390 print buf[0].real, buf[0].imag

1391

1392 @testcase

1393 def complex_struct_inplace(object[LongComplex] buf):

1394 """

1395 >>> complex_struct_inplace(LongComplexMockBuffer(None, [(0, -1)]))

1396 1.0 1.0

1397 """

1398 buf[0].real += 1

1399 buf[0].imag += 2

1400 print buf[0].real, buf[0].imag

1401