"Fossies" - the Fresh Open Source Software Archive

Member "buildbot-2.5.1/buildbot/test/unit/test_process_logobserver.py" (24 Nov 2019, 7525 Bytes) of package /linux/misc/buildbot-2.5.1.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 # This file is part of Buildbot.  Buildbot is free software: you can
    2 # redistribute it and/or modify it under the terms of the GNU General Public
    3 # License as published by the Free Software Foundation, version 2.
    4 #
    5 # This program is distributed in the hope that it will be useful, but WITHOUT
    6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
    7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
    8 # details.
    9 #
   10 # You should have received a copy of the GNU General Public License along with
   11 # this program; if not, write to the Free Software Foundation, Inc., 51
   12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
   13 #
   14 # Copyright Buildbot Team Members
   15 
   16 
   17 import mock
   18 
   19 from twisted.internet import defer
   20 from twisted.trial import unittest
   21 
   22 from buildbot.process import log
   23 from buildbot.process import logobserver
   24 from buildbot.test.fake import fakemaster
   25 from buildbot.test.util.misc import TestReactorMixin
   26 
   27 
   28 class MyLogObserver(logobserver.LogObserver):
   29 
   30     def __init__(self):
   31         self.obs = []
   32 
   33     def outReceived(self, data):
   34         self.obs.append(('out', data))
   35 
   36     def errReceived(self, data):
   37         self.obs.append(('err', data))
   38 
   39     def headerReceived(self, data):
   40         self.obs.append(('hdr', data))
   41 
   42     def finishReceived(self):
   43         self.obs.append(('fin',))
   44 
   45 
   46 class TestLogObserver(TestReactorMixin, unittest.TestCase):
   47 
   48     def setUp(self):
   49         self.setUpTestReactor()
   50         self.master = fakemaster.make_master(self, wantData=True)
   51 
   52     @defer.inlineCallbacks
   53     def test_sequence(self):
   54         logid = yield self.master.data.updates.addLog(1, 'mine', 's')
   55         _log = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
   56         lo = MyLogObserver()
   57         lo.setLog(_log)
   58 
   59         yield _log.addStdout('hello\n')
   60         yield _log.addStderr('cruel\n')
   61         yield _log.addStdout('world\n')
   62         yield _log.addStdout('multi\nline\nchunk\n')
   63         yield _log.addHeader('HDR\n')
   64         yield _log.finish()
   65 
   66         self.assertEqual(lo.obs, [
   67             ('out', 'hello\n'),
   68             ('err', 'cruel\n'),
   69             ('out', 'world\n'),
   70             ('out', 'multi\nline\nchunk\n'),
   71             ('hdr', 'HDR\n'),
   72             ('fin',),
   73         ])
   74 
   75 
   76 class MyLogLineObserver(logobserver.LogLineObserver):
   77 
   78     def __init__(self):
   79         super().__init__()
   80         self.obs = []
   81 
   82     def outLineReceived(self, data):
   83         self.obs.append(('out', data))
   84 
   85     def errLineReceived(self, data):
   86         self.obs.append(('err', data))
   87 
   88     def headerLineReceived(self, data):
   89         self.obs.append(('hdr', data))
   90 
   91     def finishReceived(self):
   92         self.obs.append(('fin',))
   93 
   94 
   95 class TestLineConsumerLogObesrver(TestReactorMixin, unittest.TestCase):
   96 
   97     def setUp(self):
   98         self.setUpTestReactor()
   99         self.master = fakemaster.make_master(self, wantData=True)
  100 
  101     @defer.inlineCallbacks
  102     def do_test_sequence(self, consumer):
  103         logid = yield self.master.data.updates.addLog(1, 'mine', 's')
  104         _log = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
  105         lo = logobserver.LineConsumerLogObserver(consumer)
  106         lo.setLog(_log)
  107 
  108         yield _log.addStdout('hello\n')
  109         yield _log.addStderr('cruel\n')
  110         yield _log.addStdout('multi\nline\nchunk\n')
  111         yield _log.addHeader('H1\nH2\n')
  112         yield _log.finish()
  113 
  114     @defer.inlineCallbacks
  115     def test_sequence_finish(self):
  116         results = []
  117 
  118         def consumer():
  119             while True:
  120                 try:
  121                     stream, line = yield
  122                     results.append((stream, line))
  123                 except GeneratorExit:
  124                     results.append('finish')
  125                     raise
  126         yield self.do_test_sequence(consumer)
  127 
  128         self.assertEqual(results, [
  129             ('o', 'hello'),
  130             ('e', 'cruel'),
  131             ('o', 'multi'),
  132             ('o', 'line'),
  133             ('o', 'chunk'),
  134             ('h', 'H1'),
  135             ('h', 'H2'),
  136             'finish',
  137         ])
  138 
  139     @defer.inlineCallbacks
  140     def test_sequence_no_finish(self):
  141         results = []
  142 
  143         def consumer():
  144             while True:
  145                 stream, line = yield
  146                 results.append((stream, line))
  147         yield self.do_test_sequence(consumer)
  148 
  149         self.assertEqual(results, [
  150             ('o', 'hello'),
  151             ('e', 'cruel'),
  152             ('o', 'multi'),
  153             ('o', 'line'),
  154             ('o', 'chunk'),
  155             ('h', 'H1'),
  156             ('h', 'H2'),
  157         ])
  158 
  159 
  160 class TestLogLineObserver(TestReactorMixin, unittest.TestCase):
  161 
  162     def setUp(self):
  163         self.setUpTestReactor()
  164         self.master = fakemaster.make_master(self, wantData=True)
  165 
  166     @defer.inlineCallbacks
  167     def test_sequence(self):
  168         logid = yield self.master.data.updates.addLog(1, 'mine', 's')
  169         _log = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
  170         lo = MyLogLineObserver()
  171         lo.setLog(_log)
  172 
  173         yield _log.addStdout('hello\n')
  174         yield _log.addStderr('cruel\n')
  175         yield _log.addStdout('multi\nline\nchunk\n')
  176         yield _log.addHeader('H1\nH2\n')
  177         yield _log.finish()
  178 
  179         self.assertEqual(lo.obs, [
  180             ('out', 'hello'),
  181             ('err', 'cruel'),
  182             ('out', 'multi'),
  183             ('out', 'line'),
  184             ('out', 'chunk'),
  185             ('hdr', 'H1'),
  186             ('hdr', 'H2'),
  187             ('fin',),
  188         ])
  189 
  190     def test_old_setMaxLineLength(self):
  191         # this method is gone, but used to be documented, so it's still
  192         # callable.  Just don't fail.
  193         lo = MyLogLineObserver()
  194         lo.setMaxLineLength(120939403)
  195 
  196 
  197 class TestOutputProgressObserver(TestReactorMixin, unittest.TestCase):
  198 
  199     def setUp(self):
  200         self.setUpTestReactor()
  201         self.master = fakemaster.make_master(self, wantData=True)
  202 
  203     @defer.inlineCallbacks
  204     def test_sequence(self):
  205         logid = yield self.master.data.updates.addLog(1, 'mine', 's')
  206         _log = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
  207         lo = logobserver.OutputProgressObserver('stdio')
  208         step = mock.Mock()
  209         lo.setStep(step)
  210         lo.setLog(_log)
  211 
  212         yield _log.addStdout('hello\n')
  213         step.setProgress.assert_called_with('stdio', 6)
  214         yield _log.finish()
  215 
  216 
  217 class TestBufferObserver(TestReactorMixin, unittest.TestCase):
  218 
  219     def setUp(self):
  220         self.setUpTestReactor()
  221         self.master = fakemaster.make_master(self, wantData=True)
  222 
  223     @defer.inlineCallbacks
  224     def do_test_sequence(self, lo):
  225         logid = yield self.master.data.updates.addLog(1, 'mine', 's')
  226         _log = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
  227         lo.setLog(_log)
  228 
  229         yield _log.addStdout('hello\n')
  230         yield _log.addStderr('cruel\n')
  231         yield _log.addStdout('multi\nline\nchunk\n')
  232         yield _log.addHeader('H1\nH2\n')
  233         yield _log.finish()
  234 
  235     @defer.inlineCallbacks
  236     def test_stdout_only(self):
  237         lo = logobserver.BufferLogObserver(wantStdout=True, wantStderr=False)
  238         yield self.do_test_sequence(lo)
  239         self.assertEqual(lo.getStdout(), 'hello\nmulti\nline\nchunk\n')
  240         self.assertEqual(lo.getStderr(), '')
  241 
  242     @defer.inlineCallbacks
  243     def test_both(self):
  244         lo = logobserver.BufferLogObserver(wantStdout=True, wantStderr=True)
  245         yield self.do_test_sequence(lo)
  246         self.assertEqual(lo.getStdout(), 'hello\nmulti\nline\nchunk\n')
  247         self.assertEqual(lo.getStderr(), 'cruel\n')