[py-svn] r31713 - in py/branch/distributed/py/test: . rsession rsession/testing testing

fijal at codespeak.net fijal at codespeak.net
Sun Aug 27 12:51:46 CEST 2006


Author: fijal
Date: Sun Aug 27 12:51:43 2006
New Revision: 31713

Modified:
   py/branch/distributed/py/test/collect.py
   py/branch/distributed/py/test/rsession/report.py
   py/branch/distributed/py/test/rsession/rsession.py
   py/branch/distributed/py/test/rsession/testing/test_rsession.py
   py/branch/distributed/py/test/testing/test_collect.py
Log:
Fix some bugs, added select-by-keyword command line argument.


Modified: py/branch/distributed/py/test/collect.py
==============================================================================
--- py/branch/distributed/py/test/collect.py	(original)
+++ py/branch/distributed/py/test/collect.py	Sun Aug 27 12:51:43 2006
@@ -176,28 +176,52 @@
                 newl.append(x.name) 
         return ".".join(newl) 
 
-    def tryiter(self, yieldtype=None, reporterror=None):
+    # XXX: Copied from session
+    def skipbykeyword(self, keyword): 
+        if not keyword:
+            return
+        chain = self.listchain()
+        for key in filter(None, keyword.split()): 
+            eor = key[:1] == '-'
+            if eor:
+                key = key[1:]
+            if not (eor ^ self._matchonekeyword(key, chain)):
+                py.test.skip("test not selected by keyword %r" %(keyword,))
+
+    def _matchonekeyword(self, key, chain): 
+        for subitem in chain: 
+            if subitem.haskeyword(key): 
+                return True 
+        return False
+
+    def tryiter(self, yieldtype=None, reporterror=None, keyword=None):
         """ yield stop item instances from flattening the collector. 
             XXX deprecated: this way of iteration is not safe in all
-            cases. 
+            cases. Mostly fixed, need to introduce skipped-by-keyword
         """ 
         
         if yieldtype is None: 
             yieldtype = py.test.Item 
-        if isinstance(self, yieldtype): 
-            yield self
+        if isinstance(self, yieldtype):
+            try:
+                self.skipbykeyword(keyword)
+                yield self
+            except py.test.Item.Skipped:
+                if reporterror is not None:
+                    excinfo = py.code.ExceptionInfo()
+                    reporterror((excinfo, self))
         else:
             if not isinstance(self, py.test.Item):
                 try:
                     for x in self.run(): 
-                        for y in self.join(x).tryiter(yieldtype): 
+                        for y in self.join(x).tryiter(yieldtype, reporterror, keyword): 
                             yield y
                 except KeyboardInterrupt:
                     raise
                 except: 
                     if reporterror is not None: 
                         excinfo = py.code.ExceptionInfo()
-                        reporterror(excinfo) 
+                        reporterror((excinfo, self)) 
 
     def _prepare(self): 
         if not hasattr(self, '_name2items'): 

Modified: py/branch/distributed/py/test/rsession/report.py
==============================================================================
--- py/branch/distributed/py/test/rsession/report.py	(original)
+++ py/branch/distributed/py/test/rsession/report.py	Sun Aug 27 12:51:43 2006
@@ -86,5 +86,15 @@
     def __init__(self, nodes):
         self.nodes = nodes
 
+class SkippedTryiter(ReportEvent):
+    def __init__(self, excinfo, item):
+        self.excinfo = excinfo
+        self.item = item
+
+class FailedTryiter(ReportEvent):
+    def __init__(self, excinfo, item):
+        self.excinfo = excinfo
+        self.item = item
+
 # tryiter, main dispatch loop, something else (setup-teardown stuff)
 # 

Modified: py/branch/distributed/py/test/rsession/rsession.py
==============================================================================
--- py/branch/distributed/py/test/rsession/rsession.py	(original)
+++ py/branch/distributed/py/test/rsession/rsession.py	Sun Aug 27 12:51:43 2006
@@ -119,15 +119,19 @@
         texts = {}
         for event in self.skipped_tests_outcome:
             colitem = event.item
-            outcome = event.outcome
-            text = outcome.skipped
-            itemname = event.channel.gateway.sshaddress + ":" + \
-                "/".join(colitem.listnames())
+            if isinstance(event, report.ReceivedItemOutcome):
+                outcome = event.outcome
+                text = outcome.skipped
+                itemname = event.channel.gateway.sshaddress + ":" + \
+                    "/".join(colitem.listnames())
+            elif isinstance(event, report.SkippedTryiter):
+                text = str(event.excinfo.value)
+                itemname = "/".join(colitem.listnames())
             if text not in texts:
                 texts[text] = [itemname]
             else:
                 texts[text].append(itemname)
-        
+            
         if texts:
             self.out.line()
             self.out.sep('_', 'reasons for skipped tests')
@@ -157,6 +161,14 @@
         self.out.sep("=", " %d test run%s%s in %.2fs " % 
             (total, skipped_str, failed_str, self.timeend - self.timestart))
     
+    def report_SkippedTryiter(self, event):
+        #event.outcome.excinfo.source = 
+        self.skipped_tests_outcome.append(event)
+    
+    def report_FailedTryiter(self, event):
+        pass
+        # XXX: right now we do not do anything with it
+    
     def report_ReceivedItemOutcome(self, event):
         host = event.channel.gateway.sshaddress
         if event.outcome.passed:
@@ -199,10 +211,19 @@
 
         destrelpath = "pytestcache"
         nodes = init_hosts(reporter, sshhosts, destrelpath, pkgdir)
+        
+        def reporterror(data):
+            excinfo, item = data
+            if excinfo.type is py.test.Item.Skipped:
+                reporter(report.SkippedTryiter(excinfo, item))
+            else:
+                reporter(report.FailedTryiter(excinfo, item))
+        
+        keyword = self.config.option.keyword
 
         def itemgen():
             for x in colitems: 
-                for y in x.tryiter():
+                for y in x.tryiter(reporterror = reporterror, keyword = keyword):
                     yield y 
         
         itemgenerator = itemgen() 

Modified: py/branch/distributed/py/test/rsession/testing/test_rsession.py
==============================================================================
--- py/branch/distributed/py/test/rsession/testing/test_rsession.py	(original)
+++ py/branch/distributed/py/test/rsession/testing/test_rsession.py	Sun Aug 27 12:51:43 2006
@@ -50,6 +50,23 @@
     assert col_two.listnames() == [pkgdir.dirpath().basename, 
                                    "py", "path", "__init__.py"]
 
+def test_example_tryiter():
+    events = []
+    tmpdir = py.test.ensuretemp("tryitertest")
+    tmpdir.ensure("a", "__init__.py")
+    tmpdir.ensure("conftest.py").write(py.code.Source("""
+        import py
+        py.test.skip("Reason")
+    """))
+    tmpdir.ensure("a", "test_empty.py").write(py.code.Source("""
+        def test_empty():
+            pass
+    """))
+    rootcol = py.test.collect.Directory(tmpdir)
+    data = list(rootcol.tryiter(reporterror=events.append))
+    assert len(events) == 1
+    assert str(events[0][0].value) == "Reason"
+
 class TestWithRealSshHosts: 
     def setup_class(cls):
         from py.__.test.rsession.conftest import option 
@@ -89,7 +106,7 @@
         channel.send(None)
         res = channel.receive()
         assert res == "ok"
-
+    
     def test_example_distribution(self): 
         # XXX find a better way for the below 
         tmpdir = py.path.local(py.__file__).dirpath().dirpath()

Modified: py/branch/distributed/py/test/testing/test_collect.py
==============================================================================
--- py/branch/distributed/py/test/testing/test_collect.py	(original)
+++ py/branch/distributed/py/test/testing/test_collect.py	Sun Aug 27 12:51:43 2006
@@ -366,7 +366,8 @@
     l = []
     list(col.tryiter(reporterror=l.append))
     assert len(l) == 1
-    assert isinstance(l[0], py.code.ExceptionInfo)
+    excinfo, item = l[0]
+    assert isinstance(excinfo, py.code.ExceptionInfo)
 
 def test_tryiter_handles_keyboardinterrupt(): 
     tmp = py.test.ensuretemp("tryiterkeyboard")



More information about the pytest-commit mailing list