"Fossies" - the Fresh Open Source Software Archive

Member "buildbot-2.5.1/buildbot/test/unit/test_www_service.py" (24 Nov 2019, 10249 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 calendar
   17 import datetime
   18 
   19 import jwt
   20 
   21 import mock
   22 
   23 from twisted.cred import strcred
   24 from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
   25 from twisted.internet import defer
   26 from twisted.trial import unittest
   27 from twisted.web._auth.wrapper import HTTPAuthSessionWrapper
   28 from twisted.web.server import Request
   29 
   30 from buildbot.test.unit import test_www_hooks_base
   31 from buildbot.test.util import www
   32 from buildbot.test.util.misc import TestReactorMixin
   33 from buildbot.www import auth
   34 from buildbot.www import change_hook
   35 from buildbot.www import resource
   36 from buildbot.www import rest
   37 from buildbot.www import service
   38 
   39 
   40 class NeedsReconfigResource(resource.Resource):
   41 
   42     needsReconfig = True
   43     reconfigs = 0
   44 
   45     def reconfigResource(self, config):
   46         NeedsReconfigResource.reconfigs += 1
   47 
   48 
   49 class Test(TestReactorMixin, www.WwwTestMixin, unittest.TestCase):
   50 
   51     def setUp(self):
   52         self.setUpTestReactor()
   53         self.master = self.make_master(url='h:/a/b/')
   54         self.svc = self.master.www = service.WWWService()
   55         self.svc.setServiceParent(self.master)
   56 
   57     def makeConfig(self, **kwargs):
   58         w = dict(port=None, auth=auth.NoAuth(), logfileName='l')
   59         w.update(kwargs)
   60         new_config = mock.Mock()
   61         new_config.www = w
   62         new_config.buildbotURL = 'h:/'
   63         self.master.config = new_config
   64         return new_config
   65 
   66     @defer.inlineCallbacks
   67     def test_reconfigService_no_port(self):
   68         new_config = self.makeConfig()
   69         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
   70 
   71         self.assertEqual(self.svc.site, None)
   72 
   73     @defer.inlineCallbacks
   74     def test_reconfigService_reconfigResources(self):
   75         new_config = self.makeConfig(port=8080)
   76         self.patch(rest, 'RestRootResource', NeedsReconfigResource)
   77         NeedsReconfigResource.reconfigs = 0
   78 
   79         # first time, reconfigResource gets called along with setupSite
   80         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
   81         self.assertEqual(NeedsReconfigResource.reconfigs, 1)
   82 
   83         # and the next time, setupSite isn't called, but reconfigResource is
   84         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
   85         self.assertEqual(NeedsReconfigResource.reconfigs, 2)
   86 
   87     @defer.inlineCallbacks
   88     def test_reconfigService_port(self):
   89         new_config = self.makeConfig(port=20)
   90         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
   91 
   92         self.assertNotEqual(self.svc.site, None)
   93         self.assertNotEqual(self.svc.port_service, None)
   94         self.assertEqual(self.svc.port, 20)
   95 
   96     @defer.inlineCallbacks
   97     def test_reconfigService_expiration_time(self):
   98         new_config = self.makeConfig(port=80, cookie_expiration_time=datetime.timedelta(minutes=1))
   99         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
  100 
  101         self.assertNotEqual(self.svc.site, None)
  102         self.assertNotEqual(self.svc.port_service, None)
  103         self.assertEqual(service.BuildbotSession.expDelay, datetime.timedelta(minutes=1))
  104 
  105     @defer.inlineCallbacks
  106     def test_reconfigService_port_changes(self):
  107         new_config = self.makeConfig(port=20)
  108         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
  109 
  110         newer_config = self.makeConfig(port=999)
  111         yield self.svc.reconfigServiceWithBuildbotConfig(newer_config)
  112 
  113         self.assertNotEqual(self.svc.site, None)
  114         self.assertNotEqual(self.svc.port_service, None)
  115         self.assertEqual(self.svc.port, 999)
  116 
  117     @defer.inlineCallbacks
  118     def test_reconfigService_port_changes_to_none(self):
  119         new_config = self.makeConfig(port=20)
  120         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
  121 
  122         newer_config = self.makeConfig()
  123         yield self.svc.reconfigServiceWithBuildbotConfig(newer_config)
  124 
  125         # (note the site sticks around)
  126         self.assertEqual(self.svc.port_service, None)
  127         self.assertEqual(self.svc.port, None)
  128 
  129     def test_setupSite(self):
  130         self.svc.setupSite(self.makeConfig())
  131         site = self.svc.site
  132 
  133         # check that it has the right kind of resources attached to its
  134         # root
  135         root = site.resource
  136         req = mock.Mock()
  137         self.assertIsInstance(root.getChildWithDefault(b'api', req),
  138                               rest.RestRootResource)
  139 
  140     def test_setupSiteWithProtectedHook(self):
  141         checker = InMemoryUsernamePasswordDatabaseDontUse()
  142         checker.addUser("guest", "password")
  143 
  144         self.svc.setupSite(self.makeConfig(
  145             change_hook_dialects={'base': True},
  146             change_hook_auth=[checker]))
  147         site = self.svc.site
  148 
  149         # check that it has the right kind of resources attached to its
  150         # root
  151         root = site.resource
  152         req = mock.Mock()
  153         self.assertIsInstance(root.getChildWithDefault(b'change_hook', req),
  154                               HTTPAuthSessionWrapper)
  155 
  156     @defer.inlineCallbacks
  157     def test_setupSiteWithHook(self):
  158         new_config = self.makeConfig(
  159             change_hook_dialects={'base': True})
  160         self.svc.setupSite(new_config)
  161         site = self.svc.site
  162 
  163         # check that it has the right kind of resources attached to its
  164         # root
  165         root = site.resource
  166         req = mock.Mock()
  167         ep = root.getChildWithDefault(b'change_hook', req)
  168         self.assertIsInstance(ep,
  169                               change_hook.ChangeHookResource)
  170 
  171         # not yet configured
  172         self.assertEqual(ep.dialects, {})
  173 
  174         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
  175 
  176         # now configured
  177         self.assertEqual(ep.dialects, {'base': True})
  178 
  179         rsrc = self.svc.site.resource.getChildWithDefault(b'change_hook', mock.Mock())
  180         path = b'/change_hook/base'
  181         request = test_www_hooks_base._prepare_request({})
  182         self.master.data.updates.addChange = mock.Mock()
  183         yield self.render_resource(rsrc, path, request=request)
  184         self.master.data.updates.addChange.assert_called()
  185 
  186     @defer.inlineCallbacks
  187     def test_setupSiteWithHookAndAuth(self):
  188         fn = self.mktemp()
  189         with open(fn, 'w') as f:
  190             f.write("user:pass")
  191         new_config = self.makeConfig(
  192             port=8080,
  193             plugins={},
  194             change_hook_dialects={'base': True},
  195             change_hook_auth=[strcred.makeChecker("file:" + fn)])
  196         self.svc.setupSite(new_config)
  197 
  198         yield self.svc.reconfigServiceWithBuildbotConfig(new_config)
  199         rsrc = self.svc.site.resource.getChildWithDefault(b'', mock.Mock())
  200 
  201         res = yield self.render_resource(rsrc, b'')
  202         self.assertIn(b'{"type": "file"}', res)
  203 
  204         rsrc = self.svc.site.resource.getChildWithDefault(
  205             b'change_hook', mock.Mock())
  206         res = yield self.render_resource(rsrc, b'/change_hook/base')
  207         # as UnauthorizedResource is in private namespace, we cannot use
  208         # assertIsInstance :-(
  209         self.assertIn('UnauthorizedResource', repr(res))
  210 
  211 
  212 class TestBuildbotSite(unittest.SynchronousTestCase):
  213     SECRET = 'secret'
  214 
  215     def setUp(self):
  216         self.site = service.BuildbotSite(None, "logs", 0, 0)
  217         self.site.setSessionSecret(self.SECRET)
  218 
  219     def test_getSession_from_bad_jwt(self):
  220         """ if the cookie is bad (maybe from previous version of buildbot),
  221             then we should raise KeyError for consumption by caller,
  222             and log the JWT error
  223         """
  224         with self.assertRaises(KeyError):
  225             self.site.getSession("xxx")
  226         self.flushLoggedErrors(jwt.exceptions.DecodeError)
  227 
  228     def test_getSession_from_correct_jwt(self):
  229         payload = {'user_info': {'some': 'payload'}}
  230         uid = jwt.encode(payload, self.SECRET, algorithm=service.SESSION_SECRET_ALGORITHM)
  231         session = self.site.getSession(uid)
  232         self.assertEqual(session.user_info, {'some': 'payload'})
  233 
  234     def test_getSession_from_expired_jwt(self):
  235         # expired one week ago
  236         exp = datetime.datetime.utcnow() - datetime.timedelta(weeks=1)
  237         exp = calendar.timegm(datetime.datetime.timetuple(exp))
  238         payload = {'user_info': {'some': 'payload'}, 'exp': exp}
  239         uid = jwt.encode(payload, self.SECRET, algorithm=service.SESSION_SECRET_ALGORITHM)
  240         with self.assertRaises(KeyError):
  241             self.site.getSession(uid)
  242 
  243     def test_getSession_with_no_user_info(self):
  244         payload = {'foo': 'bar'}
  245         uid = jwt.encode(payload, self.SECRET, algorithm=service.SESSION_SECRET_ALGORITHM)
  246         with self.assertRaises(KeyError):
  247             self.site.getSession(uid)
  248 
  249     def test_makeSession(self):
  250         session = self.site.makeSession()
  251         self.assertEqual(session.user_info, {'anonymous': True})
  252 
  253     def test_updateSession(self):
  254         session = self.site.makeSession()
  255 
  256         class FakeChannel:
  257             transport = None
  258 
  259             def isSecure(self):
  260                 return False
  261 
  262             def getPeer(self):
  263                 return None
  264 
  265             def getHost(self):
  266                 return None
  267 
  268         request = Request(FakeChannel(), False)
  269         request.sitepath = [b"bb"]
  270         session.updateSession(request)
  271         self.assertEqual(len(request.cookies), 1)
  272         name, value = request.cookies[0].split(b";")[0].split(b"=")
  273         decoded = jwt.decode(value, self.SECRET,
  274                              algorithms=[service.SESSION_SECRET_ALGORITHM])
  275         self.assertEqual(decoded['user_info'], {'anonymous': True})
  276         self.assertIn('exp', decoded)