/usr/share/pyshared/async/test/test_example.py is in python-async 0.6.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | """Module containing examples from the documentaiton"""
from lib import *
from async.pool import *
from async.task import *
from async.thread import terminate_threads
class TestExamples(TestBase):
@terminate_threads
def test_usage(self):
p = ThreadPool()
# default size is 0, synchronous mode
assert p.size() == 0
# now tasks would be processed asynchronously
p.set_size(1)
assert p.size() == 1
# A task performing processing on items from an iterator
t = IteratorThreadTask(iter(range(10)), "power", lambda i: i*i)
reader = p.add_task(t)
# read all items - they where procesed by worker 1
items = reader.read()
assert len(items) == 10 and items[0] == 0 and items[-1] == 81
# chaining
t = IteratorThreadTask(iter(range(10)), "power", lambda i: i*i)
reader = p.add_task(t)
# chain both by linking their readers
tmult = ChannelThreadTask(reader, "mult", lambda i: i*2)
result_reader = p.add_task(tmult)
# read all
items = result_reader.read()
assert len(items) == 10 and items[0] == 0 and items[-1] == 162
|