Making the test to run.

Author: greut

async/test/mod/test_zlib.py

@@ -22,17 +22,17 @@ def test_constants(self):
         assert hasattr(zlib, "Z_MEM_ERROR")
         assert hasattr(zlib, "Z_BUF_ERROR")
         assert hasattr(zlib, "Z_VERSION_ERROR")
-        
-        
+
+
     def test_status(self):
         # test the newly introduced status code
         data = struct.pack(">L", (1<<31) + (1<<15) + (1<<2))
         assert len(data) == 4
-        
+
         # compress
         cobj = zlib.compressobj(zlib.Z_BEST_SPEED)
         assert cobj.status == zlib.Z_STATUS_UNSET
-        
+
         cchunk = ''
         for c in data:
             cchunk += cobj.compress(c)
@@ -41,34 +41,34 @@ def test_status(self):
         # its not yet done, but soon it will
         cchunk += cobj.flush()
         assert cobj.status == zlib.Z_STREAM_END
-        
+
         # zip should have added a few bytes of info
         assert len(cchunk) > len(data)
-        
-        
+
+
         # decompress - need status to determine decompession finished
         dcobj = zlib.decompressobj()
         idata = ''                      # inflated data
         for i, c in enumerate(cchunk):
             idata += dcobj.decompress(c)
             assert dcobj.status == zlib.Z_OK
-            
+
             # break if we have it
             if len(idata) == len(data):
                 break
         # END for each character
         assert idata == data
-        
+
         # we should still have some bytes left
         assert i < len(cchunk) - 1
-        
+
         # feed the remaining data, we don't expect to decompress anything, but
         # want to see the status change
         while dcobj.status == zlib.Z_OK:
             i += 1
             assert len(dcobj.decompress(cchunk[i])) == 0
         # END deplete compressed stream
-        
+
         # now we are done
         assert dcobj.status == zlib.Z_STREAM_END
         assert i == len(cchunk) - 1

async/test/task.py

@@ -55,18 +55,18 @@ def _assert(self, pc, fc, check_scheduled=False):
         return self
 
 
-class TestThreadTask(_TestTaskBase, IteratorThreadTask):
+class FixtureThreadTask(_TestTaskBase, IteratorThreadTask):
     pass
 
 
-class TestFailureThreadTask(TestThreadTask):
+class FixtureFailureThreadTask(FixtureThreadTask):
     """Fails after X items"""
     def __init__(self, *args, **kwargs):
         self.fail_after = kwargs.pop('fail_after')
-        super(TestFailureThreadTask, self).__init__(*args, **kwargs)
+        super(FixtureFailureThreadTask, self).__init__(*args, **kwargs)
 
     def do_fun(self, item):
-        item = TestThreadTask.do_fun(self, item)
+        item = FixtureThreadTask.do_fun(self, item)
 
         self.lock.acquire()
         try:
@@ -78,15 +78,15 @@ def do_fun(self, item):
         return item
 
 
-class TestChannelThreadTask(_TestTaskBase, ChannelThreadTask):
+class FixtureChannelThreadTask(_TestTaskBase, ChannelThreadTask):
     """Apply a transformation on items read from an input channel"""
     def __init__(self, *args, **kwargs):
         self.fail_after = kwargs.pop('fail_after', 0)
-        super(TestChannelThreadTask, self).__init__(*args, **kwargs)
+        super(FixtureChannelThreadTask, self).__init__(*args, **kwargs)
 
     def do_fun(self, item):
         """return tuple(i, i*2)"""
-        item = super(TestChannelThreadTask, self).do_fun(item)
+        item = super(FixtureChannelThreadTask, self).do_fun(item)
 
         # fail after support
         if self.fail_after:
@@ -106,22 +106,22 @@ def do_fun(self, item):
         # END handle tuple
 
 
-class TestPerformanceThreadTask(ChannelThreadTask):
+class FixturePerformanceThreadTask(ChannelThreadTask):
     """Applies no operation to the item, and does not lock, measuring
     the actual throughput of the system"""
 
     def do_fun(self, item):
         return item
 
 
-class TestVerifyChannelThreadTask(_TestTaskBase, ChannelThreadTask):
+class FixtureVerifyChannelThreadTask(_TestTaskBase, ChannelThreadTask):
     """An input channel task, which verifies the result of its input channels,
     should be last in the chain.
     Id must be int"""
 
     def do_fun(self, item):
         """return tuple(i, i*2)"""
-        item = super(TestVerifyChannelThreadTask, self).do_fun(item)
+        item = super(FixtureVerifyChannelThreadTask, self).do_fun(item)
 
         # make sure the computation order matches
         assert isinstance(item, tuple), "input was no tuple: %s" % item
@@ -141,7 +141,7 @@ def make_proxy_method(t):
     return lambda item: wt.do_fun(item)
 
 def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_offset=0,
-                    feedercls=TestThreadTask, transformercls=TestChannelThreadTask,
+                    feedercls=FixtureThreadTask, transformercls=FixtureChannelThreadTask,
                     include_verifier=True):
     """Create a task chain of feeder, count transformers and order verifcator
     to the pool p, like t1 -> t2 -> t3
@@ -183,7 +183,7 @@ def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_of
     # END setup failure
 
     if include_verifier:
-        verifier = TestVerifyChannelThreadTask(inrc, 'verifier', None)
+        verifier = FixtureVerifyChannelThreadTask(inrc, 'verifier', None)
         #verifier.fun = verifier.do_fun
         verifier.fun = make_proxy_method(verifier)
         vrc = p.add_task(verifier)
@@ -194,7 +194,7 @@ def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_of
     # END handle include verifier
     return tasks, rcs
 
-def make_iterator_task(ni, taskcls=TestThreadTask, **kwargs):
+def make_iterator_task(ni, taskcls=FixtureThreadTask, **kwargs):
     """:return: task which yields ni items
     :param taskcls: the actual iterator type to use
     :param kwargs: additional kwargs to be passed to the task"""

async/test/test_performance.py

@@ -31,7 +31,7 @@ def test_base(self):
                 for read_mode in range(2):
                     ts, rcs = add_task_chain(pool, ni, count=num_transformers,
                                                 feedercls=IteratorThreadTask,
-                                                transformercls=TestPerformanceThreadTask,
+                                                transformercls=FixturePerformanceThreadTask,
                                                 include_verifier=False)
 
                     mode_info = "read(0)"

async/test/test_pool.py

@@ -202,7 +202,7 @@ def _assert_single_task(self, p, async=False):
         # test failure after ni / 2 items
         # This makes sure it correctly closes the channel on failure to prevent blocking
         nri = ni/2
-        task = make_task(TestFailureThreadTask, fail_after=ni/2)
+        task = make_task(FixtureFailureThreadTask, fail_after=ni/2)
         rc = p.add_task(task)
         assert len(rc.read()) == nri
         assert task.is_done()
@@ -412,7 +412,7 @@ def test_base(self):
         # SINGLE TASK SERIAL SYNC MODE
         ##############################
         # put a few unrelated tasks that we forget about - check ref counts and cleanup
-        t1, t2 = TestThreadTask(iter(list()), "nothing1", None), TestThreadTask(iter(list()), "nothing2", None)
+        t1, t2 = FixtureThreadTask(iter(list()), "nothing1", None), FixtureThreadTask(iter(list()), "nothing2", None)
         urc1 = p.add_task(t1)
         urc2 = p.add_task(t2)
         assert p.num_tasks() == 2
@@ -433,7 +433,7 @@ def test_base(self):
         assert p.num_tasks() == 0
         assert sys.getrefcount(t2) == 2
 
-        t3 = TestChannelThreadTask(urc2, "channel", None)
+        t3 = FixtureChannelThreadTask(urc2, "channel", None)
         urc3 = p.add_task(t3)
         assert p.num_tasks() == 1
         del(urc3)

async/test/test_thread.py

@@ -11,11 +11,12 @@
 except ImportError:
     from Queue import Queue
 
+import sys
 import time
 
-class TestWorker(WorkerThread):
+class FixtureWorker(WorkerThread):
     def __init__(self, *args, **kwargs):
-        super(TestWorker, self).__init__(*args, **kwargs)
+        super(FixtureWorker, self).__init__(*args, **kwargs)
         self.reset()
 
     def fun(self, arg):
@@ -37,12 +38,18 @@ class TestThreads(TestBase):
 
     @terminate_threads
     def test_worker_thread(self):
-        worker = TestWorker()
+        worker = FixtureWorker()
         assert isinstance(worker.start(), WorkerThread)
 
         # test different method types
         standalone_func = lambda *args, **kwargs: worker.fun(*args, **kwargs)
-        for function in (TestWorker.fun, worker.fun, standalone_func):
+        functions = (worker.fun, standalone_func,)
+        if sys.version_info < (3, 0):
+            # unbound methods are gone from Python 3, it's not possible to
+            # reconstruct the `self` from it.
+            functions = functions + (FixtureWorker.fun,)
+
+        for function in functions:
             worker.inq.put((function, 1))
             time.sleep(0.01)
             worker.make_assertion()

async/thread.py

@@ -4,6 +4,9 @@
 # the New BSD License: http://www.opensource.org/licenses/bsd-license.php
 # -*- coding: utf-8 -*-
 """Module with threading utilities"""
+
+from __future__ import print_function
+
 __docformat__ = "restructuredtext"
 import threading
 import inspect
@@ -182,7 +185,8 @@ def run(self):
                         rval = routine(arg)
                     else:
                         # ignore unknown items
-                        sys.stderr.write("%s: task %s was not understood - terminating\n" % (self.getName(), str(tasktuple)))
+                        print("%s: task %s was not understood - terminating" % (self.getName(), str(tasktuple)),
+                              file=sys.stderr)
                         break
                     # END make routine call
                 finally:
@@ -194,7 +198,9 @@ def run(self):
             except StopProcessing:
                 break
             except Exception as e:
-                sys.stderr.write("%s: Task %s raised unhandled exception: %s - this really shouldn't happen !\n" % (self.getName(), str(tasktuple), str(e)))
+                print("%s: Task %s raised unhandled exception: %s - this really shouldn't happen !"
+                      % (self.getName(), str(tasktuple), str(e)),
+                      file=sys.stderr)
                 continue    # just continue
             # END routine exception handling