www.py

# This file is part of Buildbot.  Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members

import json
import os
import pkg_resources
from io import BytesIO
from io import StringIO
from urllib.parse import parse_qs
from urllib.parse import unquote as urlunquote
from uuid import uuid1

import mock

from twisted.internet import defer
from twisted.web import server

from buildbot.test.fake import fakemaster
from buildbot.util import bytes2unicode
from buildbot.util import unicode2bytes
from buildbot.www import auth
from buildbot.www import authz


class FakeSession:

    def __init__(self):
        self.user_info = {"anonymous": True}

    def updateSession(self, request):
        pass


class FakeRequest:
    written = b''
    finished = False
    redirected_to = None
    rendered_resource = None
    failure = None
    method = b'GET'
    path = b'/req.path'
    responseCode = 200

    def __init__(self, path=None):
        self.headers = {}
        self.input_headers = {}
        self.prepath = []
        x = path.split(b'?', 1)
        if len(x) == 1:
            self.path = path
            self.args = {}
        else:
            path, argstring = x
            self.path = path
            self.args = parse_qs(argstring, 1)
        self.uri = self.path
        self.postpath = []
        for p in path[1:].split(b'/'):
            path = urlunquote(bytes2unicode(p))
            self.postpath.append(unicode2bytes(path))

        self.deferred = defer.Deferred()

    def write(self, data):
        self.written = self.written + data

    def redirect(self, url):
        self.redirected_to = url

    def render(self, rsrc):
        rendered_resource = rsrc
        self.deferred.callback(rendered_resource)

    def finish(self):
        self.finished = True
        if self.redirected_to is not None:
            self.deferred.callback(dict(redirected=self.redirected_to))
        else:
            self.deferred.callback(self.written)

    def setResponseCode(self, code, text=None):
        # twisted > 16 started to assert this
        assert isinstance(code, int)
        self.responseCode = code
        self.responseText = text

    def setHeader(self, hdr, value):
        assert isinstance(hdr, bytes)
        assert isinstance(value, bytes)
        self.headers.setdefault(hdr, []).append(value)

    def getHeader(self, key):
        assert isinstance(key, bytes)
        return self.input_headers.get(key)

    def processingFailed(self, f):
        self.deferred.errback(f)

    def notifyFinish(self):
        d = defer.Deferred()

        @self.deferred.addBoth
        def finished(res):
            d.callback(res)
            return res
        return d

    def getSession(self):
        return self.session


class RequiresWwwMixin:
    # mix this into a TestCase to skip if buildbot-www is not installed

    if not list(pkg_resources.iter_entry_points('buildbot.www', 'base')):
        if 'BUILDBOT_TEST_REQUIRE_WWW' in os.environ:
            raise RuntimeError('$BUILDBOT_TEST_REQUIRE_WWW is set but '
                               'buildbot-www is not installed')
        skip = 'buildbot-www not installed'


class WwwTestMixin(RequiresWwwMixin):
    UUID = str(uuid1())

    def make_master(self, url=None, **kwargs):
        master = fakemaster.make_master(self, wantData=True)
        self.master = master
        master.www = mock.Mock()  # to handle the resourceNeedsReconfigs call
        master.www.getUserInfos = lambda _: getattr(
            self.master.session, "user_info", {"anonymous": True})
        cfg = dict(port=None, auth=auth.NoAuth(), authz=authz.Authz())
        cfg.update(kwargs)
        master.config.www = cfg
        if url is not None:
            master.config.buildbotURL = url
        self.master.session = FakeSession()
        self.master.authz = cfg["authz"]
        self.master.authz.setMaster(self.master)
        return master

    def make_request(self, path=None, method=b'GET'):
        self.request = FakeRequest(path)
        self.request.session = self.master.session
        self.request.method = method
        return self.request

    def render_resource(self, rsrc, path=b'/', accept=None, method=b'GET',
                        origin=None, access_control_request_method=None,
                        extraHeaders=None, request=None,
                        content=None, content_type=None):
        if not request:
            request = self.make_request(path, method=method)
            if accept:
                request.input_headers[b'accept'] = accept
            if origin:
                request.input_headers[b'origin'] = origin
            if access_control_request_method:
                request.input_headers[b'access-control-request-method'] = \
                    access_control_request_method
            if extraHeaders is not None:
                request.input_headers.update(extraHeaders)
            if content_type is not None:
                request.input_headers.update({b'content-type': content_type})
                request.content = BytesIO(content)

        rv = rsrc.render(request)
        if rv != server.NOT_DONE_YET:
            if rv is not None:
                request.write(rv)
            request.finish()
        return request.deferred

    @defer.inlineCallbacks
    def render_control_resource(self, rsrc, path=b'/', params=None,
                                requestJson=None, action="notfound", id=None,
                                content_type=b'application/json'):
        # pass *either* a request or postpath
        if params is None:
            params = {}
        id = id or self.UUID
        request = self.make_request(path)
        request.method = b"POST"
        request.content = StringIO(requestJson or json.dumps(
            {"jsonrpc": "2.0", "method": action, "params": params, "id": id}))
        request.input_headers = {b'content-type': content_type}
        rv = rsrc.render(request)
        if rv == server.NOT_DONE_YET:
            rv = yield request.deferred

        res = json.loads(bytes2unicode(rv))
        self.assertIn("jsonrpc", res)
        self.assertEqual(res["jsonrpc"], "2.0")
        if not requestJson:
            # requestJson is used for invalid requests, so don't expect ID
            self.assertIn("id", res)
            self.assertEqual(res["id"], id)

    def assertRequest(self, content=None, contentJson=None, contentType=None,
                      responseCode=None, contentDisposition=None, headers=None):
        if headers is None:
            headers = {}
        got, exp = {}, {}
        if content is not None:
            got['content'] = self.request.written
            exp['content'] = content
        if contentJson is not None:
            got['contentJson'] = json.loads(
                bytes2unicode(self.request.written))
            exp['contentJson'] = contentJson
        if contentType is not None:
            got['contentType'] = self.request.headers[b'content-type']
            exp['contentType'] = [contentType]
        if responseCode is not None:
            got['responseCode'] = str(self.request.responseCode)
            exp['responseCode'] = str(responseCode)
        for header, value in headers.items():
            got[header] = self.request.headers.get(header)
            exp[header] = value
        self.assertEqual(got, exp)