def test_get_host():
    """Host lookup"""
    env = {'HTTP_X_FORWARDED_HOST': 'example.org',
           'SERVER_NAME': 'bullshit', 'HOST_NAME': 'ignore me dammit'}
    assert get_host(env) == 'example.org'
    assert get_host(create_environ('/', 'http://example.org')) \
        == 'example.org'

def initdb():
    # A request_context is required to use these helper functions
    with new_app(app).request_context(create_environ()):
        db.drop_all()
        db.create_all()
        readaction = schema.Action("read")
        insertaction = schema.Action("insert")
        deleteaction = schema.Action("delete")
        editaction = schema.Action("edit")
        db.session.add(readaction)
        db.session.add(insertaction)
        db.session.add(deleteaction)
        db.session.add(editaction)
        administrator = schema.User(username="admin",
                                    password="admin",
                                    firstname="admin",
                                    lastname="admin",
                                    email="[email protected]")
        administrator.actions.append(readaction)
        administrator.actions.append(insertaction)
        administrator.actions.append(deleteaction)
        administrator.actions.append(editaction)

        db.session.add(administrator)
        db.session.commit()

    def __init__(self, path="/", query_string=None, method='GET',
                 content_type=None, content_length=0, form_data=None,
                 environ_overrides=None):
        """
        For parameter reference see the documentation of the werkzeug
        package, especially the functions `url_encode` and `create_environ`.
        """
        input_stream = None

        if form_data is not None:
            form_data = url_encode(form_data)
            content_type = 'application/x-www-form-urlencoded'
            content_length = len(form_data)
            input_stream = StringIO(form_data)
        environ = create_environ(path=path, query_string=query_string,
                                 method=method, input_stream=input_stream,
                                 content_type=content_type,
                                 content_length=content_length)

        environ['HTTP_USER_AGENT'] = 'MoinMoin/TestRequest'
        # must have reverse lookup or tests will be extremely slow:
        environ['REMOTE_ADDR'] = '127.0.0.1'

        if environ_overrides:
            environ.update(environ_overrides)

        super(TestRequest, self).__init__(environ)

 def test_request_context(self, *args, **kwargs):
     """Creates a WSGI environment from the given values (see
     :func:`werkzeug.create_environ` for more information, this
     function accepts the same arguments).
     """
     from werkzeug import create_environ
     return self.request_context(create_environ(*args, **kwargs))

def test_path_info_extraction():
    """PATH INFO extraction feature"""
    x = extract_path_info('http://example.com/app', '/app/hello')
    assert x == u'/hello'
    x = extract_path_info('http://example.com/app',
                          'https://example.com/app/hello')
    assert x == u'/hello'
    x = extract_path_info('http://example.com/app/',
                          'https://example.com/app/hello')
    assert x == u'/hello'
    x = extract_path_info('http://example.com/app/',
                          'https://example.com/app')
    assert x == u'/'
    x = extract_path_info(u'http://☃.net/', u'/fööbär')
    assert x == u'/fööbär'
    x = extract_path_info(u'http://☃.net/x', u'http://☃.net/x/fööbär')
    assert x == u'/fööbär'

    env = create_environ(u'/fööbär', u'http://☃.net/x/')
    x = extract_path_info(env, u'http://☃.net/x/fööbär')
    assert x == u'/fööbär'

    x = extract_path_info('http://example.com/app/',
                          'https://example.com/a/hello')
    assert x is None
    x = extract_path_info('http://example.com/app/',
                          'https://example.com/app/hello',
                          collapse_http_schemes=False)
    assert x is None

def test_path_info_from_request_uri_fix():
    """Test the PathInfoFromRequestUriFix fixer"""
    app = fixers.PathInfoFromRequestUriFix(path_check_app)
    for key in 'REQUEST_URI', 'REQUEST_URL', 'UNENCODED_URL':
        env = dict(create_environ(), SCRIPT_NAME='/test', PATH_INFO='/?????')
        env[key] = '/test/foo%25bar?drop=this'
        response = Response.from_app(app, env)
        assert response.data == 'PATH_INFO: /foo%bar\nSCRIPT_NAME: /test'

def test_lighttpd_cgi_root_fix():
    """Test the LighttpdCGIRootFix fixer"""
    app = fixers.LighttpdCGIRootFix(path_check_app)
    response = Response.from_app(app, dict(create_environ(),
        SCRIPT_NAME='/foo',
        PATH_INFO='/bar'
    ))
    assert response.data == 'PATH_INFO: /foo/bar\nSCRIPT_NAME: '

def test_broken_multipart():
    """Broken multipart does not break the applicaiton"""
    data = (
        '--foo\r\n'
        'Content-Disposition: form-data; name="test"; filename="test.txt"\r\n'
        'Content-Transfer-Encoding: base64\r\n'
        'Content-Type: text/plain\r\n\r\n'
        'broken base 64'
        '--foo--'
    )
    _, form, files = parse_form_data(create_environ(data=data, method='POST',
                                     content_type='multipart/form-data; boundary=foo'))
    assert not files
    assert not form

    assert_raises(ValueError, parse_form_data, create_environ(data=data, method='POST',
                  content_type='multipart/form-data; boundary=foo'),
                  silent=False)

 def __init__(self, url=None, pagename=''):
     if url is None:
         url = 'http://localhost:0/' # just some somehow valid dummy URL
     environ = create_environ(base_url=url) # XXX not sure about base_url, but makes "make underlay" work
     environ['HTTP_USER_AGENT'] = 'CLI/Script'
     environ['wsgi.input'] = sys.stdin
     request = Request(environ)
     super(ScriptContext, self).__init__(request)
     from MoinMoin import wsgiapp
     wsgiapp.init(self)

def test_parse_form_data_get_without_content():
    """GET requests without data, content type and length returns no data"""
    env = create_environ('/foo', 'http://example.org/', method='GET')
    del env['CONTENT_TYPE']
    del env['CONTENT_LENGTH']

    stream, form, files = parse_form_data(env)
    assert stream.read() == ""
    assert len(form) == 0
    assert len(files) == 0

def test_wrapper_internals():
    """Test internals of the wrappers"""
    from werkzeug import Request
    req = Request.from_values(data={'foo': 'bar'}, method='POST')
    req._load_form_data()
    assert req.form.to_dict() == {'foo': 'bar'}

    # second call does not break
    req._load_form_data()
    assert req.form.to_dict() == {'foo': 'bar'}

    # check reprs
    assert repr(req) == "<Request 'http://localhost/' [POST]>"
    resp = Response()
    assert repr(resp) == '<Response 0 bytes [200 OK]>'
    resp.data = 'Hello World!'
    assert repr(resp) == '<Response 12 bytes [200 OK]>'
    resp.response = iter(['Test'])
    assert repr(resp) == '<Response streamed [200 OK]>'

    # unicode data does not set content length
    response = Response([u'Hällo Wörld'])
    headers = response.get_wsgi_headers(create_environ())
    assert 'Content-Length' not in headers

    response = Response(['Hällo Wörld'])
    headers = response.get_wsgi_headers(create_environ())
    assert 'Content-Length' in headers

    # check for internal warnings
    print 'start'
    filterwarnings('error', category=Warning)
    response = Response()
    environ = create_environ()
    response.response = 'What the...?'
    assert_raises(Warning, lambda: list(response.iter_encoded()))
    assert_raises(Warning, lambda: list(response.get_app_iter(environ)))
    response.direct_passthrough = True
    assert_raises(Warning, lambda: list(response.iter_encoded()))
    assert_raises(Warning, lambda: list(response.get_app_iter(environ)))
    resetwarnings()

def test_header_rewriter_fix():
    """Test the HeaderRewriterFix fixer"""
    @Request.application
    def application(request):
        return Response("", headers=[
            ('X-Foo', 'bar')
        ])
    application = fixers.HeaderRewriterFix(application, ('X-Foo',), (('X-Bar', '42'),))
    response = Response.from_app(application, create_environ())
    assert response.headers['Content-Type'] == 'text/plain; charset=utf-8'
    assert 'X-Foo' not in response.headers
    assert response.headers['X-Bar'] == '42'

 def _create_environ(self, url, method, data, refer, content_type=None):
     """Return an environ to request *url*, including cookies."""
     environ_args = dict(self._wsgi_server, method=method)
     base_url = self._referrer if refer else self._base_url
     environ_args.update(self._canonicalize_url(url, base_url))
     environ_args.update(self._prep_input(method, data, content_type))
     environ = create_environ(**environ_args)
     if refer and self._referrer:
         environ['HTTP_REFERER'] = self._referrer
     environ.setdefault('REMOTE_ADDR', '127.0.0.1')
     self._cookie_jar.export_to_environ(environ)
     return environ

    def request_context(self, *args, **kw):
        """Create a request context for use in tests. The arguments passed will
        be used to create a WSGI environment to create a request instance (see
        :func:`werkzeug.create_environ` for more information). This method must
        be used with the ``with`` statement.

        For example::

            with self.request_context():
                do_something_with(request)
        """
        from werkzeug import create_environ
        return self.test_app.request_context(create_environ(*args, **kw))

def fetch_body(app, path):
    environ = create_environ(path=path)

    def start_response(status, headers):
        start_response.code = int(status.split()[0])
    start_response.code = None

    content = ''.join(list(app(environ, start_response)))
    if start_response.code == 200:
        return content
    elif start_response.code // 100 == 3:
        abort(404)
    else:
        abort(start_response.code)

def test_shared_data_middleware():
    """Shared data middleware"""
    def null_application(environ, start_response):
        start_response('404 NOT FOUND', [('Content-Type', 'text/plain')])
        yield 'NOT FOUND'
    app = SharedDataMiddleware(null_application, {
        '/':        path.join(path.dirname(__file__), 'res'),
        '/sources': path.join(path.dirname(__file__), 'res'),
        '/pkg':     ('werkzeug.debug', 'shared')
    })

    for p in '/test.txt', '/sources/test.txt':
        app_iter, status, headers = run_wsgi_app(app, create_environ(p))
        assert status == '200 OK'
        assert ''.join(app_iter).strip() == 'FOUND'

    app_iter, status, headers = run_wsgi_app(app, create_environ('/pkg/body.tmpl'))
    contents = ''.join(app_iter)
    assert 'Werkzeug Debugger' in contents

    app_iter, status, headers = run_wsgi_app(app, create_environ('/missing'))
    assert status == '404 NOT FOUND'
    assert ''.join(app_iter).strip() == 'NOT FOUND'

def test_parse_form_data_put_without_content():
    """A PUT without a Content-Type header returns empty data

    Both rfc1945 and rfc2616 (1.0 and 1.1) say "Any HTTP/[1.0/1.1] message
    containing an entity-body SHOULD include a Content-Type header field
    defining the media type of that body."  In the case where either
    headers are omitted, parse_form_data should still work.
    """
    env = create_environ('/foo', 'http://example.org/', method='PUT')
    del env['CONTENT_TYPE']
    del env['CONTENT_LENGTH']

    stream, form, files = parse_form_data(env)
    assert stream.read() == ""
    assert len(form) == 0
    assert len(files) == 0

def test_proxy_fix():
    """Test the ProxyFix fixer"""
    @fixers.ProxyFix
    @Request.application
    def app(request):
        return Response('%s|%s' % (
            request.remote_addr,
            # do not use request.host as this fixes too :)
            request.environ['HTTP_HOST']
        ))
    response = Response.from_app(app, dict(create_environ(),
        HTTP_X_FORWARDED_HOST='example.com',
        HTTP_X_FORWARDED_FOR='1.2.3.4, 5.6.7.8',
        REMOTE_ADDR='127.0.0.1',
        HTTP_HOST='fake'
    ))
    assert response.data == '1.2.3.4|example.com'

    def test_request_context(self, *args, **kwargs):
        """Creates a WSGI environment from the given values (see
        :func:`werkzeug.create_environ` for more information, this
        function accepts the same arguments).
        """
        from werkzeug import create_environ
        environ_overrides = kwargs.setdefault('environ_overrides', {})
        if self.config.get('SERVER_NAME'):
            server_name = self.config.get('SERVER_NAME')
            if ':' not in server_name:
                http_host, http_port = server_name, '80'
            else:
                http_host, http_port = server_name.split(':', 1)

            environ_overrides.setdefault('SERVER_NAME', server_name)
            environ_overrides.setdefault('HTTP_HOST', server_name)
            environ_overrides.setdefault('SERVER_PORT', http_port)
        return self.request_context(create_environ(*args, **kwargs))

    def open(self, path='/', base_url=None, query_string=None, method='GET',
             data=None, input_stream=None, content_type=None,
             content_length=0, errors_stream=None, multithread=False,
             multiprocess=False, run_once=False, environ_overrides=None,
             buffered=True):

        parsed = urlparse(path)
        if parsed.scheme:
            if base_url is None:
                base_url = parsed.scheme + '://' + parsed.netloc
            if query_string is None:
                query_string = parsed.query
            path = parsed.path

        if (input_stream is None and
            data is not None and
            method in ('PUT', 'POST')):
            input_stream, content_length, content_type = \
                self._prep_input(input_stream, data, content_type)

        if base_url is None:
            base_url = self.base_url or self.state.base_url

        environ = create_environ(path, base_url, query_string, method,
                                 input_stream, content_type, content_length,
                                 errors_stream, multithread,
                                 multiprocess, run_once)

        current_state = self.state
        current_state.prepare_environ(environ)
        if environ_overrides:
            environ.update(environ_overrides)

        logger.info("%s %s" % (method, request_uri(environ)))
        rv = run_wsgi_app(self.application, environ, buffered=buffered)

        response = _APIClientResponse(*rv)
        response.state = new_state = current_state.copy()
        new_state.process_response(response, environ)
        return response