"Fossies" - the Fresh Open Source Software Archive

Member "buildbot-2.5.1/buildbot/test/unit/test_data_logchunks.py" (24 Nov 2019, 7952 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 import textwrap
   17 
   18 from twisted.internet import defer
   19 from twisted.trial import unittest
   20 
   21 from buildbot.data import logchunks
   22 from buildbot.data import resultspec
   23 from buildbot.test.fake import fakedb
   24 from buildbot.test.util import endpoint
   25 
   26 
   27 class LogChunkEndpointBase(endpoint.EndpointMixin, unittest.TestCase):
   28 
   29     endpointClass = logchunks.LogChunkEndpoint
   30     resourceTypeClass = logchunks.LogChunk
   31     endpointname = "contents"
   32     log60Lines = ['line zero', 'line 1', 'line TWO', 'line 3', 'line 2**2',
   33                   'another line', 'yet another line']
   34     log61Lines = ['%08d' % i for i in range(100)]
   35 
   36     def setUp(self):
   37         self.setUpEndpoint()
   38         self.db.insertTestData([
   39             fakedb.Builder(id=77),
   40             fakedb.Worker(id=13, name='wrk'),
   41             fakedb.Master(id=88),
   42             fakedb.Buildset(id=8822),
   43             fakedb.BuildRequest(id=82, buildsetid=8822),
   44             fakedb.Build(id=13, builderid=77, masterid=88, workerid=13,
   45                          buildrequestid=82, number=3),
   46             fakedb.Step(id=50, buildid=13, number=9, name='make'),
   47             fakedb.Log(id=60, stepid=50, name='stdio', slug='stdio', type='s',
   48                        num_lines=7),
   49             fakedb.LogChunk(logid=60, first_line=0, last_line=1, compressed=0,
   50                             content=textwrap.dedent("""\
   51                         line zero
   52                         line 1""")),
   53             fakedb.LogChunk(logid=60, first_line=2, last_line=4, compressed=0,
   54                             content=textwrap.dedent("""\
   55                         line TWO
   56                         line 3
   57                         line 2**2""")),
   58             fakedb.LogChunk(logid=60, first_line=5, last_line=5, compressed=0,
   59                             content="another line"),
   60             fakedb.LogChunk(logid=60, first_line=6, last_line=6, compressed=0,
   61                             content="yet another line"),
   62             fakedb.Log(id=61, stepid=50, name='errors', slug='errors',
   63                        type='t', num_lines=100),
   64         ] + [
   65             fakedb.LogChunk(logid=61, first_line=i, last_line=i, compressed=0,
   66                             content="%08d" % i)
   67             for i in range(100)
   68         ] + [
   69             fakedb.Log(id=62, stepid=50, name='notes', slug='notes', type='t',
   70                        num_lines=0),
   71             # logid 62 is empty
   72         ])
   73 
   74     def tearDown(self):
   75         self.tearDownEndpoint()
   76 
   77     @defer.inlineCallbacks
   78     def do_test_chunks(self, path, logid, expLines):
   79         # get the whole thing in one go
   80         logchunk = yield self.callGet(path)
   81         self.validateData(logchunk)
   82         expContent = '\n'.join(expLines) + '\n'
   83         self.assertEqual(logchunk,
   84                          {'logid': logid, 'firstline': 0, 'content': expContent})
   85 
   86         # line-by-line
   87         for i, expLine in enumerate(expLines):
   88             logchunk = yield self.callGet(path,
   89                                           resultSpec=resultspec.ResultSpec(offset=i, limit=1))
   90             self.validateData(logchunk)
   91             self.assertEqual(logchunk,
   92                              {'logid': logid, 'firstline': i, 'content': expLines[i] + '\n'})
   93 
   94         # half and half
   95         mid = int(len(expLines) / 2)
   96         for f, length in (0, mid), (mid, len(expLines) - 1):
   97             logchunk = yield self.callGet(path,
   98                                           resultSpec=resultspec.ResultSpec(offset=f, limit=length - f + 1))
   99             self.validateData(logchunk)
  100             expContent = '\n'.join(expLines[f:length + 1]) + '\n'
  101             self.assertEqual(logchunk,
  102                              {'logid': logid, 'firstline': f, 'content': expContent})
  103 
  104         # truncated at EOF
  105         f, length = len(expLines) - 2, len(expLines) + 10
  106         logchunk = yield self.callGet(path,
  107                                       resultSpec=resultspec.ResultSpec(offset=f, limit=length - f + 1))
  108         self.validateData(logchunk)
  109         expContent = '\n'.join(expLines[-2:]) + '\n'
  110         self.assertEqual(logchunk,
  111                          {'logid': logid, 'firstline': f, 'content': expContent})
  112 
  113         # some illegal stuff
  114         self.assertEqual(
  115             (yield self.callGet(path,
  116                                 resultSpec=resultspec.ResultSpec(offset=-1))), None)
  117         self.assertEqual(
  118             (yield self.callGet(path,
  119                                 resultSpec=resultspec.ResultSpec(offset=10, limit=-1))),
  120             None)
  121 
  122     def test_get_logid_60(self):
  123         return self.do_test_chunks(('logs', 60, self.endpointname), 60,
  124                                    self.log60Lines)
  125 
  126     def test_get_logid_61(self):
  127         return self.do_test_chunks(('logs', 61, self.endpointname), 61,
  128                                    self.log61Lines)
  129 
  130 
  131 class LogChunkEndpoint(LogChunkEndpointBase):
  132 
  133     @defer.inlineCallbacks
  134     def test_get_missing(self):
  135         logchunk = yield self.callGet(('logs', 99, self.endpointname))
  136         self.assertEqual(logchunk, None)
  137 
  138     @defer.inlineCallbacks
  139     def test_get_empty(self):
  140         logchunk = yield self.callGet(('logs', 62, self.endpointname))
  141         self.validateData(logchunk)
  142         self.assertEqual(logchunk['content'], '')
  143 
  144     @defer.inlineCallbacks
  145     def test_get_by_stepid(self):
  146         logchunk = yield self.callGet(
  147             ('steps', 50, 'logs', 'errors', self.endpointname))
  148         self.validateData(logchunk)
  149         self.assertEqual(logchunk['logid'], 61)
  150 
  151     @defer.inlineCallbacks
  152     def test_get_by_buildid(self):
  153         logchunk = yield self.callGet(
  154             ('builds', 13, 'steps', 9, 'logs', 'stdio', self.endpointname))
  155         self.validateData(logchunk)
  156         self.assertEqual(logchunk['logid'], 60)
  157 
  158     @defer.inlineCallbacks
  159     def test_get_by_builder(self):
  160         logchunk = yield self.callGet(
  161             ('builders', 77, 'builds', 3, 'steps', 9,
  162              'logs', 'errors', self.endpointname))
  163         self.validateData(logchunk)
  164         self.assertEqual(logchunk['logid'], 61)
  165 
  166     @defer.inlineCallbacks
  167     def test_get_by_builder_step_name(self):
  168         logchunk = yield self.callGet(
  169             ('builders', 77, 'builds', 3, 'steps', 'make',
  170              'logs', 'errors', self.endpointname))
  171         self.validateData(logchunk)
  172         self.assertEqual(logchunk['logid'], 61)
  173 
  174 
  175 class RawLogChunkEndpoint(LogChunkEndpointBase):
  176 
  177     endpointClass = logchunks.RawLogChunkEndpoint
  178     endpointname = "raw"
  179 
  180     def validateData(self, data):
  181         self.assertIsInstance(data['raw'], str)
  182         self.assertIsInstance(data['mime-type'], str)
  183         self.assertIsInstance(data['filename'], str)
  184 
  185     @defer.inlineCallbacks
  186     def do_test_chunks(self, path, logid, expLines):
  187         # get the whole thing in one go
  188         logchunk = yield self.callGet(path)
  189         self.validateData(logchunk)
  190         if logid == 60:
  191             expContent = '\n'.join([line[1:] for line in expLines])
  192             expFilename = "stdio"
  193         else:
  194             expContent = '\n'.join(expLines) + '\n'
  195             expFilename = "errors"
  196 
  197         self.assertEqual(logchunk,
  198                          {'filename': expFilename, 'mime-type': "text/plain", 'raw': expContent})