def testRequestEnvironment(self):
        req = self._createRequest()
        publish(req, handle_errors=0) # Force expansion of URL variables

        self.assertEqual(str(req.URL), 'http://foobar.com/folder/item')
        self.assertEqual(req.URL['-1'], 'http://foobar.com/folder')
        self.assertEqual(req.URL['-2'], 'http://foobar.com')
        self.assertRaises(KeyError, req.URL.__getitem__, '-3')

        self.assertEqual(req.URL['0'], 'http://foobar.com')
        self.assertEqual(req.URL['1'], 'http://foobar.com/folder')
        self.assertEqual(req.URL['2'], 'http://foobar.com/folder/item')
        self.assertRaises(KeyError, req.URL.__getitem__, '3')

        self.assertEqual(req.URL.get('0'), 'http://foobar.com')
        self.assertEqual(req.URL.get('1'), 'http://foobar.com/folder')
        self.assertEqual(req.URL.get('2'), 'http://foobar.com/folder/item')
        self.assertEqual(req.URL.get('3', 'none'), 'none')

        self.assertEqual(req['SERVER_URL'], 'http://foobar.com')
        self.assertEqual(req['HTTP_HOST'], 'foobar.com')
        self.assertEqual(req['PATH_INFO'], '/folder/item')
        self.assertEqual(req['CONTENT_LENGTH'], '0')
        self.assertRaises(KeyError, req.__getitem__, 'HTTP_AUTHORIZATION')
        self.assertEqual(req['GATEWAY_INTERFACE'], 'TestFooInterface/1.0')
        self.assertEqual(req['HTTP_OFF_THE_WALL'], "Spam 'n eggs")

        self.assertRaises(KeyError, req.__getitem__,
                          'HTTP_WE_DID_NOT_PROVIDE_THIS')

    def _testBaseTags(self, url, expected):
        # Make sure I1 and O1 are visible in the module namespace
        # so that the classes can be pickled.
        import transaction

        pub = BrowserPublication(self.db)

        ztapi.browserView(I1, 'view', DummyView)
        ztapi.setDefaultViewName(I1, 'view')
        ztapi.browserViewProviding(None, TestTraverser, IBrowserPublisher)

        ob = O1()

        ## the following is for running the tests standalone
        principalRegistry.defineDefaultPrincipal(
            'tim', 'timbot', 'ai at its best')

        # now place our object inside the application

        connection = self.db.open()
        app = connection.root()['Application']
        app.somepath = ob
        transaction.commit()
        connection.close()

        defineChecker(app.__class__, NamesChecker(somepath='xxx'))

        req = self._createRequest(url, pub)
        response = req.response

        publish(req, handle_errors=0)

        self.assertEqual(response.getBase(), expected)

 def _publisherResults(self, extra_env={}, body=b""):
     request = self._createRequest(extra_env, body)
     response = request.response
     publish(request, handle_errors=False)
     headers = response.getHeaders()
     return (
         "Status: %s\r\n" % response.getStatusString()
         +
         "\r\n".join([("%s: %s" % h) for h in headers]) + "\r\n\r\n"
         +
         response.consumeBody().decode('utf8')
         )

    def executeRequest(self, task):
        """Overrides HTTPServer.executeRequest()."""
        env = task.getCGIEnvironment()
        env['HTTPS'] = 'ON'
        try:
            del env['HTTP']
        except KeyError:
            pass
        instream = task.request_data.getBodyStream()

        request = self.request_factory(instream, task, env)
        response = request.response
        response.setHeaderOutput(task)
        response.setHTTPTransaction(task)
        publish(request)

 def application(environ, start_response):
     request = BrowserRequest(environ['wsgi.input'], environ)
     request.setPublication(pub)
     request = publish(request)
     response = request.response
     start_response(response.getStatusString(), response.getHeaders())
     return response.consumeBodyIter()

    def __call__(self, environ, start_response):
        """See zope.app.wsgi.interfaces.IWSGIApplication"""
        request = self.requestFactory(environ['wsgi.input'], environ)

        # Let's support post-mortem debugging
        handle_errors = environ.get('wsgi.handleErrors', self.handleErrors)

        request = publish(request, handle_errors=handle_errors)
        response = request.response
        # Get logging info from principal for log use
        logging_info = ILoggingInfo(request.principal, None)
        if logging_info is None:
            message = b'-'
        else:
            message = logging_info.getLogMessage()

        if not PYTHON2:
            # In python 3, convert message bytes to native string
            message = message.decode('latin1')

        environ['wsgi.logging_info'] = message
        if 'REMOTE_USER' not in environ:
            environ['REMOTE_USER'] = message

        # Start the WSGI server response
        start_response(response.getStatusString(), response.getHeaders())

        # Return the result body iterable.
        return response.consumeBodyIter()

    def test_shiftNameToApplication(self):
        r = self._createRequest()
        publish(r, handle_errors=0)
        appurl = r.getApplicationURL()

        # Verify that we can shift. It would be a little more realistic
        # if we could test this during traversal, but the api doesn't
        # let us do that.
        r = self._createRequest(extra_env={"PATH_INFO": "/xxx"})
        publish(r, handle_errors=0)
        r.shiftNameToApplication()
        self.assertEqual(r.getApplicationURL(), appurl+"/xxx")

        # Verify that we can only shift if we've traversed only a single name
        r = self._createRequest(extra_env={"PATH_INFO": "/folder/item"})
        publish(r, handle_errors=0)
        self.assertRaises(ValueError, r.shiftNameToApplication)

    def _execute(self, path, command, split=True, **kw):
        env = {}
        env.update(kw)
        env['command'] = command

        path = self._translate(path)

        if split:
            env['path'], env['name'] = posixpath.split(path)
        else:
            env['path'] = path

        env['credentials'] = self.credentials
        request = self.request_factory(BytesIO(b''), env)

        # Note that publish() calls close() on request, which deletes the
        # response from the request, so that we need to keep track of it.
        # agroszer: 2008.feb.1.: currently the above seems not to be true
        # request will KEEP the response on close()
        # even more if a retry occurs in the publisher,
        # the response will be LOST, so we must accept the returned request
        request = publish(request)
        return request.response.getResult()

    def __call__(self, request_string, handle_errors=True, form=None):
        # Commit work done by previous python code.
        commit()

        # Discard leading white space to make call layout simpler
        request_string = request_string.lstrip()

        # split off and parse the command line
        l = request_string.find('\n')
        command_line = request_string[:l].rstrip()
        request_string = request_string[l + 1:]
        method, path, protocol = command_line.split()

        # If we don't feed bytes to Python 3, it gets stuck in a loop
        # and ultimately raises HTTPException: got more than 100 headers.
        instream = io.BytesIO(request_string.encode("latin-1")
                              if not isinstance(request_string, bytes)
                              else request_string)
        environment = {
            "HTTP_COOKIE": self.httpCookie(path),
            "HTTP_HOST": 'localhost',
            "REQUEST_METHOD": method,
            "SERVER_PROTOCOL": protocol,
            "REMOTE_ADDR": '127.0.0.1',
        }

        headers = [split_header(header)
                   for header in headers_factory(instream).headers]
        for name, value in headers:
            name = ('_'.join(name.upper().split('-')))
            if name not in ('CONTENT_TYPE', 'CONTENT_LENGTH'):
                name = 'HTTP_' + name
            environment[name] = value.rstrip()

        auth_key = 'HTTP_AUTHORIZATION'
        if auth_key in environment:
            environment[auth_key] = auth_header(environment[auth_key])

        old_site = getSite()
        setSite(None)

        request_cls, publication_cls = self.chooseRequestClass(method, path,
                                                               environment)
        app = FunctionalTestSetup().getApplication()

        request = app._request(
            path, instream,
            environment=environment,
            request=request_cls, publication=publication_cls)
        if ISkinnable.providedBy(request):
            # only ISkinnable requests have skins
            setDefaultSkin(request)

        if form is not None:
            if request.form:
                raise ValueError("only one set of form values can be provided")
            request.form = form

        request = publish(request, handle_errors=handle_errors)

        response = ResponseWrapper(
            request.response, path, request,
            omit=('x-content-type-warning', 'x-powered-by'),
            )

        self.saveCookies(response)
        setSite(old_site)

        # sync Python connection:
        getRootFolder()._p_jar.sync()

        return response

 def _publisherResults(self, path, **kw):
     request = self._createRequest(path, **kw)
     response = request.response
     publish(request, handle_errors=False)
     return response._result

 def publish(self, path):
     return publish(self.makeRequest(path)).response