python类Request()的实例源码

def test_form_data_ordering(self):
        class MyRequest(wrappers.Request):
            parameter_storage_class = ImmutableOrderedMultiDict

        req = MyRequest.from_values('/?foo=1&bar=0&foo=3')
        self.assert_equal(list(req.args), ['foo', 'bar'])
        self.assert_equal(list(req.args.items(multi=True)), [
            ('foo', '1'),
            ('bar', '0'),
            ('foo', '3')
        ])
        self.assert_is_instance(req.args, ImmutableOrderedMultiDict)
        self.assert_is_instance(req.values, CombinedMultiDict)
        self.assert_equal(req.values['foo'], '1')
        self.assert_equal(req.values.getlist('foo'), ['1', '3'])

def test_modified_url_encoding(self):
        class ModifiedRequest(wrappers.Request):
            url_charset = 'euc-kr'

        req = ModifiedRequest.from_values(u'/?foo=????'.encode('euc-kr'))
        self.assert_strict_equal(req.args['foo'], u'????')

def test_reverse_slash_behavior(self):
        class MyRequest(wrappers.ReverseSlashBehaviorRequestMixin, Request):
            pass
        req = MyRequest.from_values('/foo/bar', 'http://example.com/test')
        assert req.url == 'http://example.com/test/foo/bar'
        assert req.path == 'foo/bar'
        assert req.script_root == '/test/'

        # make sure the routing system works with the slashes in
        # reverse order as well.
        map = routing.Map([routing.Rule('/foo/bar', endpoint='foo')])
        adapter = map.bind_to_environ(req.environ)
        assert adapter.match() == ('foo', {})
        adapter = map.bind(req.host, req.script_root)
        assert adapter.match(req.path) == ('foo', {})

def test_dynamic_charset_request_mixin(self):
        class MyRequest(wrappers.DynamicCharsetRequestMixin, Request):
            pass
        env = {'CONTENT_TYPE': 'text/html'}
        req = MyRequest(env)
        assert req.charset == 'latin1'

        env = {'CONTENT_TYPE': 'text/html; charset=utf-8'}
        req = MyRequest(env)
        assert req.charset == 'utf-8'

        env = {'CONTENT_TYPE': 'application/octet-stream'}
        req = MyRequest(env)
        assert req.charset == 'latin1'
        assert req.url_charset == 'latin1'

        MyRequest.url_charset = 'utf-8'
        env = {'CONTENT_TYPE': 'application/octet-stream'}
        req = MyRequest(env)
        assert req.charset == 'latin1'
        assert req.url_charset == 'utf-8'

        def return_ascii(x):
            return "ascii"
        env = {'CONTENT_TYPE': 'text/plain; charset=x-weird-charset'}
        req = MyRequest(env)
        req.unknown_charset = return_ascii
        assert req.charset == 'ascii'
        assert req.url_charset == 'utf-8'

def application(environ, start_responseonse):
    request = Request(environ)
    if request.method == 'POST':
        response = stats(request)
    else:
        response = upload_file(request)
    return response(environ, start_responseonse)

def wsgi_app(app, environ, start_response):
    # ?????
    request = Request(environ)

    # ??????????????????????
    response = app.dispatch_request(request)

    # ??????
    return response(environ, start_response)

def __call__(self, environ, start_response):
        """
        The WSGI application interface.
        """
        request = Request(environ)
        response = self.handle(request)
        return response(environ, start_response)

def request(**kwargs):
    return Request(EnvironBuilder(**kwargs).get_environ())

def test_parse_body(self):
        request = Request({})
        request.form = {"foo": "bar"}

        body = self.handler.parse_body(request)
        self.assertEqual("bar", body["foo"])

def test_call(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)
        request.form = { 'param': 42 }

        res = handler._call(handler.dummy_endpoint, {}, request)
        self.assertEqual(43, res["incremented"])

        Logger.c.execute("SELECT * FROM logs WHERE category = 'HTTP'")
        row = Logger.c.fetchone()
        self.assertIn("1.2.3.4", row[3])
        self.assertIn("dummy_endpoint", row[3])

def test_call_default(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)

        res = handler._call(handler.dummy_endpoint, {}, request)
        self.assertEqual(124, res["incremented"])

def test_call_cast_parameter(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)
        request.form = { 'param': '42' }

        res = handler._call(handler.dummy_endpoint, {}, request)
        self.assertEqual(43, res["incremented"])

def test_call_fail_cast_parameter(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)
        request.form = { 'param': 'nope' }

        with self.assertRaises(BadRequest):
            handler._call(handler.dummy_endpoint, {}, request)

def test_call_required_args(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)

        with self.assertRaises(BadRequest) as ex:
            handler._call(handler.required, {}, request)

        response = ex.exception.response
        self.assertIn("MISSING_PARAMETER", response.data.decode())
        self.assertIn("param", response.data.decode())

def test_call_with_error(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)

        with self.assertRaises(Forbidden) as ex:
            handler._call(handler.required, {'param': 42}, request)

        response = ex.exception.response
        self.assertIn("NOBUONO", response.data.decode())
        self.assertIn("nononono", response.data.decode())

def test_call_general_attrs(self, name_mock, content_mock, ip_mock):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)

        res = handler._call(handler.myip, {}, request)
        self.assertEqual("1.2.3.4", res)

def test_call_file(self):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)
        request.files = {"file": FileStorage(filename="foo")}

        res = handler._call(handler.file, {}, request)
        self.assertEqual("foo", res)

def test_call_with_decorators(self):
        handler = TestBaseHandler.DummyHandler()
        env = Environ({"wsgi.input": None})
        request = Request(env)
        Database.add_user("token", "", "")
        Database.add_task("poldo", "", "", 1, 1)
        Database.add_input("inputid", "token", "poldo", 1, "", 42)
        Database.add_output("outputid", "inputid", "", 42, "")

        handler._call(handler.with_decorators, {"input_id": "inputid", "output_id":"outputid"}, request)

def test_get_file_name_no_file(self):
        request = Request(Environ())
        request.files = {}

        self.assertIsNone(BaseHandler._get_file_name(request))

def test_get_file_content(self):
        request = Request(Environ())
        stream = _io.BytesIO("hello world".encode())
        request.files = {"file": FileStorage(stream=stream, filename="foo")}

        self.assertEqual("hello world", BaseHandler._get_file_content(request).decode())