[py-svn] r13663 - py/branch/execnet-refactoring/testing

[arigo at codespeak.net]

Author: arigo
Date: Tue Jun 21 18:31:49 2005
New Revision: 13663

Modified:
   py/branch/execnet-refactoring/testing/test_gateway.py
Log:
A test for the "sendonly" state.


Modified: py/branch/execnet-refactoring/testing/test_gateway.py
==============================================================================
--- py/branch/execnet-refactoring/testing/test_gateway.py	(original)
+++ py/branch/execnet-refactoring/testing/test_gateway.py	Tue Jun 21 18:31:49 2005
@@ -1,4 +1,4 @@
-import os, sys
+import os, sys, time
 import py
 from py.__.execnet import gateway
 from py.__.conftest import option 
@@ -165,6 +165,37 @@
         assert l[:2] == [42,13]
         assert isinstance(l[2], channel.__class__) 
 
+    def test_channel_callback_stays_active(self, earlyfree=True):
+        # with 'earlyfree==True', this tests the "sendonly" channel state.
+        l = []
+        channel = self.gw.newchannel(receiver=l.append)
+        self.gw.remote_exec(channel=channel, source='''
+            import thread, time
+            def producer(channel):
+                for i in range(5):
+                    time.sleep(0.15)
+                    channel.send(i*100)
+            thread.start_new_thread(producer, (channel,))
+            ''')
+        if earlyfree:
+            del channel
+        for i in range(5):
+            for _ in range(50):     # busy-wait
+                if l:
+                    break
+                if not earlyfree:
+                     assert not channel.isclosed()
+                time.sleep(0.04)
+            else:
+                py.test.fail("timed out waiting for the answer[%d]" % i)
+            res = l.pop(0)
+            assert res == i*100
+        if not earlyfree:
+            channel.waitclose(1.0) #freed automatically at the end of producer()
+
+    def DEBUGGING_test_channel_callback_remote_freed(self):
+        self.test_channel_callback_stays_active(False)
+
     def test_remote_redirect_stdout(self): 
         out = py.std.StringIO.StringIO() 
         handle = self.gw.remote_redirect(stdout=out) 
@@ -215,7 +246,7 @@
         """)
         first = channel.receive() + channel.receive()
         assert first.strip() == 'hello world' 
-        py.test.raises(EOFError, channel.receive) 
+        py.test.raises(EOFError, channel.receive)
 
 #class TestBlockingIssues: 
 #    def test_join_blocked_execution_gateway(self):