Function _test

recipes/Python/576519_Thread_pool_same_API/recipe-576519.py:653–784

Some tests

()

def _test():
    """Some tests"""
    import thread, time

    def return42():
        return 42

    def f(x):
        return x*x

    def work(seconds):
        print "[%d] Start to work for %fs..." % (thread.get_ident(), seconds)
        time.sleep(seconds)
        print "[%d] Work done (%fs)." % (thread.get_ident(), seconds)
        return str(seconds)

    ### Test copy/pasted from multiprocessing
    pool = Pool(9)                # start 4 worker threads

    # edge cases
    assert pool.apply_async(return42, []).get() == 42
    assert pool.apply(return42, []) == 42
    assert pool.map(return42, []) == []
    assert list(pool.imap(return42, iter([]))) == []
    assert list(pool.imap_unordered(return42, iter([]))) == []
    assert pool.map_async(return42, []).get() == []
    assert list(pool.imap_async(return42, iter([])).get()) == []
    assert list(pool.imap_unordered_async(return42, iter([])).get()) == []

    # basic tests
    result = pool.apply_async(f, (10,))   # evaluate "f(10)" asynchronously
    assert result.get(timeout=1) == 100   # ... unless slow computer

    assert pool.map(f, range(10)) == map(f, range(10))

    it = pool.imap(f, range(10))
    assert it.next() == 0
    assert it.next() == 1
    assert it.next(timeout=1) == 4  # ... unless slow computer

    # Test apply_sync exceptions
    result = pool.apply_async(time.sleep, (3,))
    try:
        print result.get(timeout=1)           # raises `TimeoutError`
    except TimeoutError:
        print "Good. Got expected timeout exception."
    else:
        assert False, "Expected exception !"
    assert result.get() == None  # sleep() returns None

    def cb(s):
        print "Result ready: %s" % s

    # Test imap()
    assert list(pool.imap(work, xrange(10, 3, -1), chunksize=4)) == map(
        str, range(10, 3, -1))

    # Test imap_unordered()