[svn r58316] (iko, pedronis)

move the FileLogSession into the py.lib proper, activated with the option --resultlog (suggestions
for a better name are welcome)

- added its tests
- plus a functional/integration test in test_config in the style of the one for eventlog

--HG--
branch : trunk

py/test/defaultconftest.py

@@ -113,4 +113,7 @@ def adddefaultoptions(config):
         Option('', '--session',
                action="store", dest="session", default=None,
                help="lookup given sessioname in conftest.py files and use it."),
+        Option('--resultlog', action="store",
+               default=None, dest="resultlog",
+               help="path for machine-readable result log")        
     )

py/test/resultlog.py

@@ -0,0 +1,54 @@
+import py
+from py.__.test import event
+
+def generic_path(item):
+    chain = item.listchain()
+    gpath = [chain[0].name]
+    fspath = chain[0].fspath
+    fspart = False
+    for node in chain[1:]:
+        newfspath = node.fspath
+        if newfspath == fspath:
+            if fspart:
+                gpath.append(':')
+                fspart = False
+            else:
+                gpath.append('.')            
+        else:
+            gpath.append('/')
+            fspart = True
+        name = node.name
+        if name[0] in '([':
+            gpath.pop()
+        gpath.append(name)
+        fspath = newfspath
+    return ''.join(gpath)
+
+class ResultLog(object):
+
+    def __init__(self, bus, logfile):
+        bus.subscribe(self.log_event_to_file)
+        self.logfile = logfile #open(logpath, 'w') # line buffering ?
+
+    def write_log_entry(self, shortrepr, name, longrepr):
+        print >>self.logfile, "%s %s" % (shortrepr, name)
+        for line in longrepr.splitlines():
+            print >>self.logfile, " %s" % line
+
+    def log_outcome(self, ev):
+        outcome = ev.outcome
+        gpath = generic_path(ev.colitem)
+        self.write_log_entry(outcome.shortrepr, gpath, str(outcome.longrepr))
+
+    def log_event_to_file(self, ev):
+        if isinstance(ev, event.ItemTestReport):
+            self.log_outcome(ev)
+        elif isinstance(ev, event.CollectionReport):
+            if not ev.passed:
+                self.log_outcome(ev)
+        elif isinstance(ev, event.InternalException):
+            path = ev.repr.reprcrash.path # fishing :(
+            self.write_log_entry('!', path, str(ev.repr))
+        
+
+

py/test/session.py

@@ -9,6 +9,7 @@ import py
 from py.__.test import event, outcome
 from py.__.test.event import EventBus
 import py.__.test.custompdb
+from py.__.test.resultlog import ResultLog
 
 # used for genitems()
 from py.__.test.outcome import Exit
@@ -34,6 +35,10 @@ class Session(object):
                 print >>f, ev
                 f.flush()
             self.bus.subscribe(eventwrite)
+        resultlog = self.config.option.resultlog
+        if resultlog:
+            f = py.path.local(resultlog).open('w')
+            self.resultlog = ResultLog(self.bus, f)
 
     def fixoptions(self):
         """ check, fix and determine conflicting options. """

py/test/testing/test_config.py

@@ -181,6 +181,27 @@ class TestSessionAndOptions(suptest.FileCreation):
         s = eventlog.read()
         assert s.find("TestrunStart") != -1
 
+    def test_session_resultlog(self):
+        from py.__.test.collect import Item
+        from py.__.test.runner import OutcomeRepr
+
+        resultlog = self.tmpdir.join("test_session_resultlog")
+        config = py.test.config._reparse([self.tmpdir, 
+                                          '--resultlog=%s' % resultlog])
+
+        session = config.initsession()
+
+        item = Item("a", config=config)
+        outcome = OutcomeRepr('execute', '.', '')
+        rep_ev = event.ItemTestReport(item, passed=outcome)
+      
+        session.bus.notify(rep_ev)
+
+        session.resultlog.logfile.flush()
+        
+        s = resultlog.read()
+        assert s.find(". a") != -1        
+
     def test_tracedir_tracer(self):
         tracedir = self.tmpdir.join("tracedir")
         config = py.test.config._reparse([self.tmpdir, 

py/test/testing/test_resultlog.py

@@ -0,0 +1,184 @@
+import os, StringIO
+
+import py
+
+from py.__.test import resultlog
+from py.__.test.collect import Node, Item, FSCollector
+from py.__.test.event import EventBus
+from py.__.test.event import ItemTestReport, CollectionReport
+from py.__.test.event import InternalException
+from py.__.test.runner import OutcomeRepr
+
+
+class Fake(object):
+    def __init__(self, **kwds):
+        self.__dict__.update(kwds)
+
+
+def test_generic_path():
+    p1 = Node('a', config='dummy')
+    assert p1.fspath is None
+    p2 = Node('B', parent=p1)
+    p3 = Node('()', parent = p2)
+    item = Item('c', parent = p3)
+
+    res = resultlog.generic_path(item)
+    assert res == 'a.B().c'
+
+    p0 = FSCollector('proj/test', config='dummy')
+    p1 = FSCollector('proj/test/a', parent=p0)
+    p2 = Node('B', parent=p1)
+    p3 = Node('()', parent = p2)
+    p4 = Node('c', parent=p3)
+    item = Item('[1]', parent = p4)
+
+    res = resultlog.generic_path(item)
+    assert res == 'test/a:B().c[1]'
+
+
+def make_item(*names):
+    node = None
+    config = "dummy"
+    for name in names[:-1]:
+        if '/' in name:
+            node = FSCollector(name, parent=node, config=config)
+        else:
+            node = Node(name, parent=node, config=config)
+    if names[-1] is None:
+        return node
+    return Item(names[-1], parent=node)
+
+class TestResultLog(object):
+
+    def test_create(self):
+        bus = EventBus()
+        logfile = object()
+        
+        reslog = resultlog.ResultLog(bus, logfile)
+        assert len(bus._subscribers) == 1
+        assert reslog.logfile is logfile
+
+    def test_write_log_entry(self):
+        reslog = resultlog.ResultLog(EventBus(), None)
+
+        reslog.logfile = StringIO.StringIO()
+        reslog.write_log_entry('.', 'name', '')  
+        entry = reslog.logfile.getvalue()
+        assert entry[-1] == '\n'        
+        entry_lines = entry.splitlines()
+        assert len(entry_lines) == 1
+        assert entry_lines[0] == '. name'
+
+        reslog.logfile = StringIO.StringIO()
+        reslog.write_log_entry('s', 'name', 'Skipped')  
+        entry = reslog.logfile.getvalue()
+        assert entry[-1] == '\n'        
+        entry_lines = entry.splitlines()
+        assert len(entry_lines) == 2
+        assert entry_lines[0] == 's name'
+        assert entry_lines[1] == ' Skipped'
+
+        reslog.logfile = StringIO.StringIO()
+        reslog.write_log_entry('s', 'name', 'Skipped\n')  
+        entry = reslog.logfile.getvalue()
+        assert entry[-1] == '\n'        
+        entry_lines = entry.splitlines()
+        assert len(entry_lines) == 2
+        assert entry_lines[0] == 's name'
+        assert entry_lines[1] == ' Skipped'
+
+        reslog.logfile = StringIO.StringIO()
+        longrepr = ' tb1\n tb 2\nE tb3\nSome Error'
+        reslog.write_log_entry('F', 'name', longrepr)
+        entry = reslog.logfile.getvalue()
+        assert entry[-1] == '\n'        
+        entry_lines = entry.splitlines()
+        assert len(entry_lines) == 5
+        assert entry_lines[0] == 'F name'
+        assert entry_lines[1:] == [' '+line for line in longrepr.splitlines()]
+    
+    def test_log_outcome(self):
+        reslog = resultlog.ResultLog(EventBus(), StringIO.StringIO())
+
+        colitem = make_item('some', 'path', 'a', 'b')
+
+        try:
+            raise ValueError
+        except ValueError:
+            the_repr = py.code.ExceptionInfo().getrepr()
+
+        outcome=OutcomeRepr('execute', 'F', the_repr)
+        ev = Fake(colitem=colitem, outcome=outcome)
+
+        reslog.log_outcome(ev)
+
+        entry = reslog.logfile.getvalue()
+        entry_lines = entry.splitlines()
+
+        assert entry_lines[0] == 'F some.path.a.b'
+        assert entry_lines[-1][0] == ' '
+        assert 'ValueError' in entry
+
+    def test_item_test_passed(self):
+        bus = EventBus()
+        reslog = resultlog.ResultLog(bus, StringIO.StringIO())
+
+        colitem = make_item('proj/test', 'proj/test/mod', 'a', 'b')
+
+        outcome=OutcomeRepr('execute', '.', '')
+        rep_ev = ItemTestReport(colitem, passed=outcome)
+
+        bus.notify(rep_ev)
+
+        lines = reslog.logfile.getvalue().splitlines()
+        assert len(lines) == 1
+        line = lines[0]
+        assert line.startswith(". ")
+        assert line[2:] == 'test/mod:a.b'
+
+    def test_collection_report(self):
+        bus = EventBus()        
+        reslog = resultlog.ResultLog(bus, None)
+
+        reslog.logfile = StringIO.StringIO()
+        colitem = make_item('proj/test', 'proj/test/mod', 'A', None)
+        outcome=OutcomeRepr('execute', '', '')
+        rep_ev = CollectionReport(colitem, object(), passed=outcome)
+
+        bus.notify(rep_ev)
+
+        entry = reslog.logfile.getvalue()
+        assert not entry
+
+        reslog.logfile = StringIO.StringIO()
+        outcome=OutcomeRepr('execute', 'F', 'Some Error')
+        rep_ev = CollectionReport(colitem, object(), failed=outcome)        
+
+        bus.notify(rep_ev)
+
+        lines = reslog.logfile.getvalue().splitlines()        
+        assert len(lines) == 2
+        assert lines[0] == 'F test/mod:A'
+
+    def test_internal_exception(self):
+        # they are produced for example by a teardown failing
+        # at the end of the run
+        bus = EventBus()
+        reslog = resultlog.ResultLog(bus, StringIO.StringIO())        
+
+        try:
+            raise ValueError
+        except ValueError:
+            excinfo = py.code.ExceptionInfo()
+
+        internal = InternalException(excinfo)
+
+        bus.notify(internal)
+
+        entry = reslog.logfile.getvalue()
+        entry_lines = entry.splitlines()
+
+        assert entry_lines[0].startswith('! ')
+        assert os.path.basename(__file__)[:-1] in entry_lines[0] #.py/.pyc
+        assert entry_lines[-1][0] == ' '
+        assert 'ValueError' in entry