'test non-copying numpy array messages'

@skip_pypy
def test_buffer_numpy(self):

    try:
        import numpy
    except ImportError:
        raise SkipTest('requires    numpy')
    if (sys.version_info < (2, 7)):
        raise SkipTest('requires    new-style    buffer    interface    (py    >=    2.7)')
    rand = numpy.random.randint
    shapes = [rand(2, 5) for i in range(5)]
    (a, b) = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
    dtypes = [int, float, '>i4', 'B']
    for i in range(1, (len(shapes) + 1)):
        shape = shapes[:i]
        for dt in dtypes:
            A = numpy.empty(shape, dtype=dt)
            while numpy.isnan(A).any():
                A = numpy.ndarray(shape, dtype=dt)
            a.send(A, copy=False)
            msg = b.recv(copy=False)
            B = numpy.frombuffer(msg, A.dtype).reshape(A.shape)
            self.assertEqual(A.shape, B.shape)
            self.assertTrue((A == B).all())
        A = numpy.empty(shape, dtype=[('a', int), ('b', float), ('c', 'a32')])
        A['a'] = 1024
        A['b'] = 1000000000.0
        A['c'] = 'hello    there'
        a.send(A, copy=False)
        msg = b.recv(copy=False)
        B = numpy.frombuffer(msg, A.dtype).reshape(A.shape)
        self.assertEqual(A.shape, B.shape)
        self.assertTrue((A == B).all())
