[py-svn] r56812 - in py/branch/event/py/test2: . rep rep/testing testing

hpk at codespeak.net hpk at codespeak.net
Mon Jul 28 15:42:00 CEST 2008


Author: hpk
Date: Mon Jul 28 15:41:59 2008
New Revision: 56812

Modified:
   py/branch/event/py/test2/config.py
   py/branch/event/py/test2/rep/base.py
   py/branch/event/py/test2/rep/terminal.py
   py/branch/event/py/test2/rep/testing/test_basereporter.py
   py/branch/event/py/test2/rep/testing/test_terminal.py
   py/branch/event/py/test2/session.py
   py/branch/event/py/test2/testing/suptest.py
Log:
implementing CollectOnly reporting + some (test and otherwise) refactoring 


Modified: py/branch/event/py/test2/config.py
==============================================================================
--- py/branch/event/py/test2/config.py	(original)
+++ py/branch/event/py/test2/config.py	Mon Jul 28 15:41:59 2008
@@ -142,14 +142,8 @@
         cls = self._getsessionclass()
         session = cls(self)
         session.fixoptions()
-        reporter = self._getreporterclass()
-        session.reporter = reporter(self)
         return session
 
-    def _getreporterclass(self):
-        from py.__.test2.rep.terminal import TerminalReporter
-        return TerminalReporter
-
     def _getsessionclass(self): 
         """ return Session class determined from cmdline options
             and looked up in initial config modules. 

Modified: py/branch/event/py/test2/rep/base.py
==============================================================================
--- py/branch/event/py/test2/rep/base.py	(original)
+++ py/branch/event/py/test2/rep/base.py	Mon Jul 28 15:41:59 2008
@@ -4,10 +4,10 @@
     def __init__(self):
         self._outcome2rep = {}
 
-    def activate(self, session):
-        session.bus.subscribe(self.processevent)
-    def deactivate(self, session):
-        session.bus.unsubscribe(self.processevent)
+    def activate(self, bus):
+        bus.subscribe(self.processevent)
+    def deactivate(self, bus):
+        bus.unsubscribe(self.processevent)
 
     def processevent(self, ev):
         evname = ev.__class__.__name__ 

Modified: py/branch/event/py/test2/rep/terminal.py
==============================================================================
--- py/branch/event/py/test2/rep/terminal.py	(original)
+++ py/branch/event/py/test2/rep/terminal.py	Mon Jul 28 15:41:59 2008
@@ -1,18 +1,18 @@
 import py
 import sys
 from py.__.test2 import repevent
+from py.__.test2.rep.base import BaseReporter
+from py.__.test2.outcome import Skipped as Skipped2
+from py.__.test.outcome import Skipped
 
-class TerminalReporter(object):
+class TerminalReporter(BaseReporter):
     def __init__(self, config, out=None):
+        super(TerminalReporter, self).__init__()
         self.config = config
         self.currentfspath = None 
         self.out = py.io.TerminalWriter(sys.stdout)
-        self._failed = []
-        self._numskipped = 0
-        self._numpassed = 0
-        config.bus.subscribe(self.consume)
-
-    def consume(self, ev):
+    
+    def processevent(self, ev):
         if isinstance(ev, repevent.ItemTestReport):
             fspath, modpath = ev.repr_path
             if fspath != self.currentfspath:
@@ -20,24 +20,15 @@
                 self.out.write(fspath + " ")
                 self.currentfspath = fspath
             self.out.write(self._getletter(ev))
-            self.record(ev)
         elif isinstance(ev, repevent.SessionFinish):
             self.out.line()
-            self.print_failures()
-            self.out.sep("=", "%d tests of which %d failed, %d skipped" %(
-                         len(self._failed) + self._numskipped + self._numpassed,
-                         len(self._failed), self._numskipped))
+            #self.print_failures()
+            #self.out.sep("=", "%d tests of which %d failed, %d skipped" %(
+            #             len(self._failed) + self._numskipped + self._numpassed,
+            #             len(self._failed), self._numskipped))
         elif isinstance(ev, repevent.SessionStart):
             self.out.sep("=", "session starts")
 
-    def record(self, ev):
-        if ev.failed:
-            self._failed.append(ev)
-        elif ev.skipped:
-            self._numskipped += 1
-        else: 
-            self._numpassed += 1
-
     def print_failures(self):
         if not self._failed:
             return
@@ -52,5 +43,37 @@
             return "s"
         elif ev.failed_setup:
             return "f"
+        elif ev.failed_crashed:
+            return "C"
         elif ev.failed:
             return "F"
+
+class CollectonlyReporter(BaseReporter):
+    INDENT = "  "
+
+    def __init__(self, config, out):
+        super(CollectonlyReporter, self).__init__()
+        self.config = config
+        self.out = py.io.TerminalWriter(out)
+        self.indent = ""
+
+    def rep_CollectionStart(self, ev):
+        self.out.line(str(ev.collector))
+        self.indent += self.INDENT
+    
+    def rep_ItemStart(self, event):
+        self.out.line(self.indent + str(event.item))
+
+    def rep_CollectionFinish(self, ev):
+        excinfo = ev.excinfo
+        if excinfo is not None:
+            self.out.line(self.indent + "!!! %s !!!" % excinfo.exconly())
+        self.indent = self.indent[:-len(self.INDENT)]
+
+    #def rep_DeselectedTest(self, event):
+    #    self.out.line(" " * self.indent + "- skipped -")
+
+    #def summary(self):
+    #    self.out.sep("=", "Total time: %.1f" % (self.timeend - self.timestart))
+        
+        

Modified: py/branch/event/py/test2/rep/testing/test_basereporter.py
==============================================================================
--- py/branch/event/py/test2/rep/testing/test_basereporter.py	(original)
+++ py/branch/event/py/test2/rep/testing/test_basereporter.py	Mon Jul 28 15:41:59 2008
@@ -3,19 +3,15 @@
 from py.__.test2.eventbus import EventBus
 from py.__.test2 import repevent
 
-class MockSession:
-    def __init__(self):
-        self.bus = EventBus()
-
 class TestBaseReporter:
     def test_activate(self):
-        session = MockSession()
+        bus = EventBus()
         rep = BaseReporter()
-        rep.activate(session)
-        assert session.bus._subscribers
-        assert rep.processevent in session.bus._subscribers
-        rep.deactivate(session)
-        assert not session.bus._subscribers
+        rep.activate(bus)
+        assert bus._subscribers
+        assert rep.processevent in bus._subscribers
+        rep.deactivate(bus)
+        assert not bus._subscribers
 
     def test_dispatch_to_matching_method(self):
         l = []

Modified: py/branch/event/py/test2/rep/testing/test_terminal.py
==============================================================================
--- py/branch/event/py/test2/rep/testing/test_terminal.py	(original)
+++ py/branch/event/py/test2/rep/testing/test_terminal.py	Mon Jul 28 15:41:59 2008
@@ -1,35 +1,96 @@
 import py
-from py.__.test2.rep.terminal import TerminalReporter
+from py.__.test2.rep.terminal import TerminalReporter, CollectonlyReporter
 from py.__.test2 import repevent
+from py.__.test2.testing import suptest
+from py.__.test2.genitem import genitems
 
-class TestInit:
-    def test_is_default(self):
-        config = py.test2.config._reparse(['xxx'])
-        assert config._getreporterclass() is TerminalReporter
+def setup_module(mod):
+    mod.tmpdir = py.test.ensuretemp(mod.__name__)
 
-    def test_session_has_reporter(self):
-        config = py.test2.config._reparse(['xxx'])
-        session = config.initsession()
-        assert isinstance(session.reporter, TerminalReporter)
-        assert session.reporter.config is config
+class InlineCollect:
+    def setup_class(cls):
+        cls.clstmpdir = tmpdir.join(cls.__name__)
+        
+    def setup_method(self, method):
+        self.tmpdir = self.clstmpdir.join(method.__name__)
 
-    def test_reporter_is_subscribed(self):
-        config = py.test2.config._reparse(['xxx'])
-        session = config.initsession()
-        rep = session.reporter
-        for subscriber in config.bus._subscribers:
-            if subscriber == rep.consume:
-                break
-        else:
-            assert 0, "not subscribed" 
+    def getmodulecol(self, configargs, source):
+        self.tmpdir.ensure("__init__.py")
+        path = self.tmpdir.ensure(self.tmpdir.purebasename + ".py")
+        path.write(py.code.Source(source))
+        config = py.test2.config._reparse([path] + configargs)
+        return config._getcollector(path)
 
-class TestEvents:
-    def test_session_starts(self):
+class TestTerminal:
+    def test_session_reporter_subscription(self):
         config = py.test2.config._reparse(['xxx'])
         session = config.initsession()
+        session.sessionstarts()
         rep = session.reporter
-        rep.consume(repevent.SessionStart(session))
-        
-        py.test.skip("xxx")
-        
-        
+        assert isinstance(rep, TerminalReporter)
+        assert rep.processevent in config.bus._subscribers
+        session.sessionfinishes()
+        assert rep.processevent not in config.bus._subscribers
+
+class TestCollectonly(InlineCollect):
+    def test_collectonly(self):
+        modcol = self.getmodulecol(['--collectonly'], """
+            def test_func():
+                pass
+        """)
+        stringio = py.std.cStringIO.StringIO()
+        rep = CollectonlyReporter(modcol._config, out=stringio)
+        indent = rep.indent
+        rep.processevent(repevent.CollectionStart(modcol))
+        s = popvalue(stringio)
+        assert s == "<Module 'test_collectonly.py'>"
+
+        item = modcol.join("test_func")
+        rep.processevent(repevent.ItemStart(item))
+        s = popvalue(stringio)
+        assert s == "  <Function 'test_func'>"
+             
+        rep.processevent(repevent.CollectionFinish(modcol))
+        assert rep.indent == indent 
+
+    def test_collectonly_skipped_module(self):
+        modcol = self.getmodulecol(['--collectonly'], """
+            import py
+            py.test2.skip("nomod")
+        """)
+        stringio = py.std.cStringIO.StringIO()
+        rep = CollectonlyReporter(modcol._config, out=stringio)
+        rep.activate(modcol._config.bus)
+        cols = list(genitems(modcol._config, [modcol]))
+        assert len(cols) == 0
+        stringio.seek(0)
+        lines = stringio.readlines()
+        lines = map(str.rstrip, lines)
+        print lines
+        suptest.assert_lines_contain_lines(lines, """
+            <Module 'test_collectonly_skipped_module.py'>
+              !!! Skipped: 'nomod' !!!
+        """)
+
+    def test_collectonly_failed_module(self):
+        modcol = self.getmodulecol(['--collectonly'], """
+            raise ValueError(0)
+        """)
+        stringio = py.std.cStringIO.StringIO()
+        rep = CollectonlyReporter(modcol._config, out=stringio)
+        rep.activate(modcol._config.bus)
+        cols = list(genitems(modcol._config, [modcol]))
+        assert len(cols) == 0
+        stringio.seek(0)
+        lines = stringio.readlines()
+        lines = map(str.rstrip, lines)
+        print lines
+        suptest.assert_lines_contain_lines(lines, """
+            <Module 'test_collectonly_failed_module.py'>
+              !!! ValueError: 0 !!!
+        """)
+
+def popvalue(stringio):
+    value = stringio.getvalue().rstrip()
+    stringio.truncate(0)
+    return value

Modified: py/branch/event/py/test2/session.py
==============================================================================
--- py/branch/event/py/test2/session.py	(original)
+++ py/branch/event/py/test2/session.py	Mon Jul 28 15:41:59 2008
@@ -44,6 +44,10 @@
 
     def sessionstarts(self):
         """ setup any neccessary resources ahead of the test run. """
+        from py.__.test2.rep.terminal import TerminalReporter
+        rep = TerminalReporter(self)
+        rep.activate(self.config.bus)
+        self.reporter = rep 
         self.config.bus.notify(repevent.SessionStart(self))
         if not self.config.option.nomagic:
             py.magic.invoke(assertion=1)
@@ -63,10 +67,7 @@
             py.magic.revoke(assertion=1)
         self.config.bus.notify(repevent.SessionFinish(self))
         self.config.bus.unsubscribe(self._processfailures)
-        try:
-            self.config.bus.unsubscribe(self.reporter.consume)
-        except AttributeError:
-            pass
+        self.reporter.deactivate(self.config.bus)
         return self._failurelist 
 
     def main(self):

Modified: py/branch/event/py/test2/testing/suptest.py
==============================================================================
--- py/branch/event/py/test2/testing/suptest.py	(original)
+++ py/branch/event/py/test2/testing/suptest.py	Mon Jul 28 15:41:59 2008
@@ -130,10 +130,14 @@
     """ assert that lines2 are contained (linearly) in lines1. 
         return a list of extralines found.
     """
+    if isinstance(lines2, str):
+        lines2 = py.code.Source(lines2)
     if isinstance(lines2, py.code.Source):
-        lines2 = lines2.lines 
+        lines2 = lines2.strip().lines
+
     extralines = []
     lines1 = lines1[:]
+    nextline = None
     for line in lines2:
         while lines1:
             nextline = lines1.pop(0)
@@ -148,14 +152,11 @@
                 raise AssertionError("expected line not found: %r" % line)
     extralines.extend(lines1)
     return extralines 
-    
-
 
 # XXX below some code to help with inlining examples 
 #     as source code.  
  
-class XXXBaseInlineCollectionTests:
-    
+class TestBaseInlineCollection:
     def setup_method(self, method):
         self.tmpdir = tmpdir.join("%s_%s_%s" % 
             (__name__, self.__class__.__name__, method.__name__))



More information about the pytest-commit mailing list