[py-svn] r62004 - py/branch/pytestplugin/py/test/dsession/testing

hpk at codespeak.net hpk at codespeak.net
Wed Feb 18 13:59:12 CET 2009


Author: hpk
Date: Wed Feb 18 13:59:10 2009
New Revision: 62004

Modified:
   py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py
Log:
remove usage of suptest.py file


Modified: py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py
==============================================================================
--- py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py	(original)
+++ py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py	Wed Feb 18 13:59:10 2009
@@ -1,4 +1,3 @@
-from py.__.test.testing.suptest import InlineCollection
 from py.__.test.dsession.dsession import DSession, LoopState
 from py.__.test.dsession.hostmanage import Host, makehostup
 from py.__.test.runner import basic_collect_report 
@@ -24,17 +23,17 @@
     while queue.qsize():
         print queue.get()
 
-class TestDSession(InlineCollection):
-    def test_fixoptions(self):
-        config = self.parseconfig("--exec=xxx")
+class TestDSession:
+    def test_fixoptions(self, fstester):
+        config = fstester.parseconfig("--exec=xxx")
         config.initsession().fixoptions()
         assert config.option.numprocesses == 1
-        config = self.parseconfig("--exec=xxx", '-n3')
+        config = fstester.parseconfig("--exec=xxx", '-n3')
         config.initsession().fixoptions()
         assert config.option.numprocesses == 3
 
-    def test_add_remove_host(self):
-        item = self.getitem("def test_func(): pass")
+    def test_add_remove_host(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         rep = run(item)
         session = DSession(item._config)
         host = Host("localhost")
@@ -48,8 +47,8 @@
         assert item not in session.item2host
         py.test.raises(Exception, "session.removehost(host)")
 
-    def test_senditems_removeitems(self):
-        item = self.getitem("def test_func(): pass")
+    def test_senditems_removeitems(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         rep = run(item)
         session = DSession(item._config)
         host = Host("localhost")
@@ -62,8 +61,8 @@
         assert not session.host2pending[host] 
         assert not session.item2host
 
-    def test_triggertesting_collect(self):
-        modcol = self.getmodulecol("""
+    def test_triggertesting_collect(self, tsession):
+        modcol = tsession.getmodulecol("""
             def test_func():
                 pass
         """)
@@ -71,11 +70,12 @@
         session.triggertesting([modcol])
         kw = session.queue.get(block=False)
         name, rep = kw.items()[0]
+        assert name == 'collectionreport'
         assert isinstance(rep, event.CollectionReport)
         assert len(rep.result) == 1
 
-    def test_triggertesting_item(self):
-        item = self.getitem("def test_func(): pass")
+    def test_triggertesting_item(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
         host1 = Host("localhost")
         host1.node = MockNode()
@@ -95,24 +95,24 @@
         assert isinstance(ev, event.RescheduleItems)
         assert ev.items == [item]
 
-    def test_keyboardinterrupt(self):
-        item = self.getitem("def test_func(): pass")
+    def test_keyboardinterrupt(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
         def raise_(timeout=None): raise KeyboardInterrupt()
         session.queue.get = raise_
         exitstatus = session.loop([])
         assert exitstatus == outcome.EXIT_INTERRUPTED
 
-    def test_internalerror(self):
-        item = self.getitem("def test_func(): pass")
+    def test_internalerror(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
         def raise_(): raise ValueError()
         session.queue.get = raise_
         exitstatus = session.loop([])
         assert exitstatus == outcome.EXIT_INTERNALERROR
 
-    def test_rescheduleevent(self):
-        item = self.getitem("def test_func(): pass")
+    def test_rescheduleevent(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
         host1 = Host("localhost")
         host1.node = MockNode()
@@ -134,8 +134,8 @@
         assert loopstate.shuttingdown 
         assert not loopstate.testsfailed 
 
-    def test_no_hosts_remaining_for_tests(self):
-        item = self.getitem("def test_func(): pass")
+    def test_no_hosts_remaining_for_tests(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         # setup a session with one host
         session = DSession(item._config)
         host1 = Host("localhost")
@@ -152,8 +152,8 @@
         dumpqueue(session.queue)
         assert loopstate.exitstatus == outcome.EXIT_NOHOSTS
 
-    def test_hostdown_causes_reschedule_pending(self, EventRecorder):
-        modcol = self.getmodulecol("""
+    def test_hostdown_causes_reschedule_pending(self, tsession, EventRecorder):
+        modcol = tsession.getmodulecol("""
             def test_crash(): 
                 assert 0
             def test_fail(): 
@@ -186,8 +186,8 @@
         assert str(testrep.longrepr).find("crashed") != -1
         assert str(testrep.longrepr).find(host.hostname) != -1
 
-    def test_hostup_adds_to_available(self):
-        item = self.getitem("def test_func(): pass")
+    def test_hostup_adds_to_available(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         # setup a session with two hosts 
         session = DSession(item._config)
         host1 = Host("localhost")
@@ -199,8 +199,8 @@
         session.loop_once(loopstate)
         assert len(session.host2pending) == 1
 
-    def test_event_propagation(self, EventRecorder):
-        item = self.getitem("def test_func(): pass")
+    def test_event_propagation(self, tsession, EventRecorder):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
       
         ev = event.NOP()
@@ -229,18 +229,18 @@
         dumpqueue(session.queue)
         return session, loopstate.exitstatus 
 
-    def test_exit_completed_tests_ok(self):
-        item = self.getitem("def test_func(): pass")
+    def test_exit_completed_tests_ok(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session, exitstatus = self.runthrough(item)
         assert exitstatus == outcome.EXIT_OK
 
-    def test_exit_completed_tests_fail(self):
-        item = self.getitem("def test_func(): 0/0")
+    def test_exit_completed_tests_fail(self, tsession):
+        item = tsession.getitem("def test_func(): 0/0")
         session, exitstatus = self.runthrough(item)
         assert exitstatus == outcome.EXIT_TESTSFAILED
 
-    def test_exit_on_first_failing(self):
-        modcol = self.getmodulecol("""
+    def test_exit_on_first_failing(self, tsession):
+        modcol = tsession.getmodulecol("""
             def test_fail(): 
                 assert 0
             def test_pass(): 
@@ -267,8 +267,8 @@
         assert loopstate.testsfailed
         assert loopstate.shuttingdown
 
-    def test_shuttingdown_filters_events(self, EventRecorder):
-        item = self.getitem("def test_func(): pass")
+    def test_shuttingdown_filters_events(self, tsession, EventRecorder):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
         host = Host("localhost")
         session.addhost(host)
@@ -283,8 +283,8 @@
         session.loop_once(loopstate)
         assert evrec.getfirstnamed('hostdown') == ev
 
-    def test_filteritems(self, EventRecorder):
-        modcol = self.getmodulecol("""
+    def test_filteritems(self, tsession, EventRecorder):
+        modcol = tsession.getmodulecol("""
             def test_fail(): 
                 assert 0
             def test_pass(): 
@@ -312,8 +312,8 @@
         assert isinstance(ev, event.Deselected)
         assert ev.items == [items[1]]
 
-    def test_hostdown_shutdown_after_completion(self):
-        item = self.getitem("def test_func(): pass")
+    def test_hostdown_shutdown_after_completion(self, tsession):
+        item = tsession.getitem("def test_func(): pass")
         session = DSession(item._config)
 
         host = Host("localhost")
@@ -330,8 +330,8 @@
         session.loop_once(loopstate)
         assert loopstate.exitstatus == 0
 
-    def test_nopending_but_collection_remains(self):
-        modcol = self.getmodulecol("""
+    def test_nopending_but_collection_remains(self, tsession):
+        modcol = tsession.getmodulecol("""
             def test_fail(): 
                 assert 0
             def test_pass(): 



More information about the pytest-commit mailing list