[py-svn] r36845 - py/branch/rsession-cleanup/test/rsession/testing

[fijal at codespeak.net]

Author: fijal
Date: Tue Jan 16 19:23:55 2007
New Revision: 36845

Modified:
   py/branch/rsession-cleanup/test/rsession/testing/test_rsession.py
Log:
Add a test for tests rescheduling. The hairy part of it is that it may hang if
test rescheduling does not work. Still better than lack of tests.


Modified: py/branch/rsession-cleanup/test/rsession/testing/test_rsession.py
==============================================================================
--- py/branch/rsession-cleanup/test/rsession/testing/test_rsession.py	(original)
+++ py/branch/rsession-cleanup/test/rsession/testing/test_rsession.py	Tue Jan 16 19:23:55 2007
@@ -5,7 +5,7 @@
 import py
 from py.__.test.rsession import report
 from py.__.test.rsession.rsession import RSession, parse_directories,\
-    session_options, remote_options, parse_directories
+    session_options, remote_options, parse_directories, AbstractSession
 from py.__.test.rsession.hostmanage import init_hosts, teardown_hosts,\
      HostInfo
 from py.__.test.rsession.testing.test_slave import funcfail_spec,\
@@ -308,4 +308,59 @@
         assert events[2].host.relpath == 'pytestcache-h2'
         assert events[3].host.hostname == 'h2'
         assert events[3].host.relpath == 'home'
-        
+
+class AbstractNode(object):
+    def __init__(self, done_dict, reporter):
+        self.done_dict = done_dict
+        self.pending = []
+        self.reporter = reporter
+    
+class DummyWorkingNode(AbstractNode):
+    def send(self, item):
+        self.done_dict[item] = True
+        self.reporter(item)
+
+class DummyNotWorkingNode(AbstractNode):
+    def send(self, item):
+        if len(self.pending) > 2:
+            py.test.fail("Too much items in pending")
+        self.pending.insert(0, item)
+
+class TestSession(RSession):
+    def __init__(self):
+        pass
+    
+    def test_dispatch_tests(self):
+        # we need to perform some initialisation here
+        tmpdir = py.test.ensuretemp("dispatch_tests")
+        tmpdir.ensure("__init__.py")
+        source = py.code.Source("""
+        def test_1():
+            pass
+        def test_2():
+            pass
+        def test_3():
+            pass
+        def test_4():
+            pass
+        def test_5():
+            pass
+        def test_6():
+            pass
+        """)
+        tmpdir.ensure("test_one.py").write(source)
+        tmpdir.ensure("test_two.py").write(source)
+        config, args = py.test.Config.parse([str(tmpdir)])
+        self.config = config
+        reports = []
+        self.config.option.keyword = None
+        session_options.bind_config(config)
+        session_options.max_tasks_per_node = 1
+        done_dict = {}
+        wn = DummyWorkingNode(done_dict, reports.append)
+        nwn = DummyNotWorkingNode(done_dict, reports.append)
+        nodes = [wn, nwn]
+        self.dispatch_tests(nodes, args, tmpdir, reports.append, lambda:False,
+            done_dict)
+        assert len([i for i in reports if isinstance(i, py.test.Item)]) == 12
+        assert len(nwn.pending) == 1