Class test_numexpr

numexpr3/tests/test_numexpr.py:37–293

Testing with 1 thread for serial operation

LARGE_SIZE = 80000.0

class test_numexpr(unittest.TestCase):
    '''Testing with 1 thread for serial operation'''

    def setUp(self, N_threads=1, ssize=SMALL_SIZE):
        logger.info( "**Run NumExpr() tests with {} thread(s) over {} element arrays**".format(N_threads, ssize) )
        ne3.set_nthreads(N_threads)
        self.ssize = ssize

    def test_scalars(self):
        logger.info( 'Test scalars' )
        a = np.array([1., 2, 3])
        b = np.array([4., 5, 6])
        c = np.array([7., 8, 9])
        out = ne3.NumExpr( '2.0 * a + 3.0 * b * c' )()
        npt.assert_array_almost_equal( out, np.array([86., 124., 168.]))

    def test_changing_array_size(self):
        logger.info('Test input with keywords, changing input array size')
        a = np.array([1., 2., 3.])
        b = np.array([4., 5., 6.])
        a2 = np.arange(self.ssize).astype(a.dtype)
        b2 = np.arange(3.0, self.ssize+3.0).astype(b.dtype)
        out = ne3.NumExpr( 'a*b' )(a=a2, b=b2)
        npt.assert_array_almost_equal( out, a2*b2 )

    def test_changing_array_shape(self):
        logger.info('Test changing input array shape')
        a = np.array([1., 2., 3.])
        b = np.array([4., 5., 6.])
        a2 = np.arange(self.ssize).astype(a.dtype).reshape( int(self.ssize/20), 4, 5)
        b2 = np.arange(3.0, self.ssize+3.0).astype(b.dtype).reshape( int(self.ssize/20), 4, 5)
        out = ne3.NumExpr( 'a*b' )(a=a2, b=b2)
        npt.assert_array_almost_equal( out, a2*b2 )

    def test_verify_input(self):
        logger.info('Test input with verify=True')
        a = np.array([1., 2., 3.])
        b = np.array([4., 5., 6.])
        func = ne3.NumExpr( 'a*b' )
        a = np.arange(self.ssize).astype(a.dtype)
        b = np.arange(3.0, self.ssize+3.0).astype(b.dtype)
        out = func(verify=True)
        npt.assert_array_almost_equal( out, a*b )

    def test_weakref_expiry(self):
        # I have no idea how to turn on gc in unittest. I can't find anything
        # in unittest that does that:
        # https://github.com/python/cpython/tree/master/Lib/unittest
        # This script works fine when run independantly.
        '''
        import gc
        gc.enable()
        logger.warning('Test expiry of weak reference')
        x = np.arange(self.ssize)
        func = ne3.NumExpr( 'x+x' )
        logging.warning( 'x is tracked: ' + str(gc.is_tracked(x) ) )
        del x  # kill original array, should expire weak ref in func.registers
        gc.collect(generation=2)