• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

Python middleware.filter_factory函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中swift3.middleware.filter_factory函数的典型用法代码示例。如果您正苦于以下问题:Python filter_factory函数的具体用法?Python filter_factory怎么用?Python filter_factory使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了filter_factory函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_registered_defaults

 def test_registered_defaults(self):
     filter_factory(CONF)
     swift_info = utils.get_swift_info()
     self.assertTrue('swift3' in swift_info)
     self.assertEqual(swift_info['swift3'].get('max_bucket_listing'),
                      CONF.max_bucket_listing)
     self.assertEqual(swift_info['swift3'].get('max_parts_listing'),
                      CONF.max_parts_listing)
     self.assertEqual(swift_info['swift3'].get('max_upload_part_num'),
                      CONF.max_upload_part_num)
     self.assertEqual(swift_info['swift3'].get('max_multi_delete_objects'),
                      CONF.max_multi_delete_objects)
开发者ID:sivakrishnan47,项目名称:swift3,代码行数:12,代码来源:test_middleware.py


示例2: test_bucket_GET_passthroughs

 def test_bucket_GET_passthroughs(self):
     class FakeApp(object):
         def __call__(self, env, start_response):
             self.query_string = env['QUERY_STRING']
             start_response('200 OK', [])
             return '[]'
     fake_app = FakeApp()
     local_app = swift3.filter_factory({})(fake_app)
     bucket_name = 'junk'
     req = Request.blank('/%s' % bucket_name,
             environ={'REQUEST_METHOD': 'GET', 'QUERY_STRING':
                      'delimiter=a&marker=b&prefix=c'},
             headers={'Authorization': 'AWS test:tester:hmac'})
     resp = local_app(req.environ, lambda *args: None)
     dom = xml.dom.minidom.parseString("".join(resp))
     self.assertEquals(dom.getElementsByTagName('Prefix')[0].
             childNodes[0].nodeValue, 'c')
     self.assertEquals(dom.getElementsByTagName('Marker')[0].
             childNodes[0].nodeValue, 'b')
     self.assertEquals(dom.getElementsByTagName('Delimiter')[0].
             childNodes[0].nodeValue, 'a')
     args = dict(cgi.parse_qsl(fake_app.query_string))
     self.assertEquals(args['delimiter'], 'a')
     self.assertEquals(args['marker'], 'b')
     self.assertEquals(args['prefix'], 'c')
开发者ID:blaxter,项目名称:swift3,代码行数:25,代码来源:test_swift3.py


示例3: test_object_DELETE

 def test_object_DELETE(self):
     local_app = swift3.filter_factory({})(FakeAppObject(204))
     req = Request.blank('/bucket/object',
                         environ={'REQUEST_METHOD': 'DELETE'},
                         headers={'Authorization': 'AWS test:tester:hmac'})
     resp = local_app(req.environ, local_app.app.do_start_response)
     self.assertEquals(local_app.app.response_args[0].split()[0], '204')
开发者ID:blaxter,项目名称:swift3,代码行数:7,代码来源:test_swift3.py


示例4: test_bucket_GET_max_keys

    def test_bucket_GET_max_keys(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.query_string = env["QUERY_STRING"]
                start_response("200 OK", [])
                return "[]"

        fake_app = FakeApp()
        local_app = swift3.filter_factory({})(fake_app)
        bucket_name = "junk"

        req = Request.blank(
            "/%s" % bucket_name,
            environ={"REQUEST_METHOD": "GET", "QUERY_STRING": "max-keys=5"},
            headers={"Authorization": "AWS test:tester:hmac"},
        )
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName("MaxKeys")[0].childNodes[0].nodeValue, "5")
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assert_(args["limit"] == "6")

        req = Request.blank(
            "/%s" % bucket_name,
            environ={"REQUEST_METHOD": "GET", "QUERY_STRING": "max-keys=5000"},
            headers={"Authorization": "AWS test:tester:hmac"},
        )
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName("MaxKeys")[0].childNodes[0].nodeValue, "1000")
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assertEquals(args["limit"], "1001")
开发者ID:mygoda,项目名称:openstack,代码行数:32,代码来源:test_swift3.py


示例5: test_bucket_GET

    def test_bucket_GET(self):
        local_app = swift3.filter_factory({})(FakeAppBucket())
        bucket_name = "junk"
        req = Request.blank(
            "/%s" % bucket_name, environ={"REQUEST_METHOD": "GET"}, headers={"Authorization": "AWS test:tester:hmac"}
        )
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], "200")

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, "ListBucketResult")
        name = dom.getElementsByTagName("Name")[0].childNodes[0].nodeValue
        self.assertEquals(name, bucket_name)

        objects = [n for n in dom.getElementsByTagName("Contents")]

        names = []
        for o in objects:
            if o.childNodes[0].nodeName == "Key":
                names.append(o.childNodes[0].childNodes[0].nodeValue)
            if o.childNodes[1].nodeName == "LastModified":
                self.assertTrue(o.childNodes[1].childNodes[0].nodeValue.endswith("Z"))

        self.assertEquals(len(names), len(FakeAppBucket().objects))
        for i in FakeAppBucket().objects:
            self.assertTrue(i[0] in names)
开发者ID:mygoda,项目名称:openstack,代码行数:26,代码来源:test_swift3.py


示例6: test_bucket_GET_max_keys

    def test_bucket_GET_max_keys(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.query_string = env['QUERY_STRING']
                start_response('200 OK', [])
                return '[]'
        fake_app = FakeApp()
        local_app = swift3.filter_factory({})(fake_app)
        bucket_name = 'junk'

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=5'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('MaxKeys')[0].
                childNodes[0].nodeValue, '5')
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assert_(args['limit'] == '6')

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=5000'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('MaxKeys')[0].
                childNodes[0].nodeValue, '1000')
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assertEquals(args['limit'], '1001')
开发者ID:blaxter,项目名称:swift3,代码行数:31,代码来源:test_swift3.py


示例7: test_object_PUT_headers

    def test_object_PUT_headers(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.req = Request(env)
                start_response("200 OK", [])
                return []

        app = FakeApp()
        local_app = swift3.filter_factory({})(app)
        req = Request.blank(
            "/bucket/object",
            environ={"REQUEST_METHOD": "PUT"},
            headers={
                "Authorization": "AWS test:tester:hmac",
                "X-Amz-Storage-Class": "REDUCED_REDUNDANCY",
                "X-Amz-Meta-Something": "oh hai",
                "X-Amz-Copy-Source": "/some/source",
                "Content-MD5": "ffoHqOWd280dyE1MT4KuoQ==",
            },
        )
        req.date = datetime.now()
        req.content_type = "text/plain"
        resp = local_app(req.environ, lambda *args: None)
        self.assertEquals(app.req.headers["ETag"], "7dfa07a8e59ddbcd1dc84d4c4f82aea1")
        self.assertEquals(app.req.headers["X-Object-Meta-Something"], "oh hai")
        self.assertEquals(app.req.headers["X-Copy-From"], "/some/source")
开发者ID:mygoda,项目名称:openstack,代码行数:26,代码来源:test_swift3.py


示例8: test_bucket_PUT

 def test_bucket_PUT(self):
     local_app = swift3.filter_factory({})(FakeAppBucket(201))
     req = Request.blank('/bucket',
                         environ={'REQUEST_METHOD': 'PUT'},
                         headers={'Authorization': 'AWS test:tester:hmac'})
     local_app(req.environ, local_app.app.do_start_response)
     self.assertEquals(local_app.app.response_args[0].split()[0], '200')
开发者ID:cwedgwood,项目名称:swift3,代码行数:7,代码来源:test_swift3.py


示例9: test_bucket_PUT

 def test_bucket_PUT(self):
     local_app = swift3.filter_factory({})(FakeAppBucket(201))
     req = Request.blank(
         "/bucket", environ={"REQUEST_METHOD": "PUT"}, headers={"Authorization": "AWS test:tester:hmac"}
     )
     local_app(req.environ, local_app.app.do_start_response)
     self.assertEquals(local_app.app.response_args[0].split()[0], "200")
开发者ID:bloodeagle40234,项目名称:swift3,代码行数:7,代码来源:test_swift3.py


示例10: test_object_acl_GET

 def test_object_acl_GET(self):
     local_app = swift3.filter_factory({})(FakeAppObject())
     req = Request.blank('/bucket/object?acl',
                         environ={'REQUEST_METHOD': 'GET'},
                         headers={'Authorization': 'AWS test:tester:hmac'})
     resp = local_app(req.environ, local_app.app.do_start_response)
     self._check_acl('test:tester', resp)
开发者ID:blaxter,项目名称:swift3,代码行数:7,代码来源:test_swift3.py


示例11: test_bucket_GET

    def test_bucket_GET(self):
        local_app = swift3.filter_factory({})(FakeAppBucket())
        bucket_name = 'junk'
        req = Request.blank('/%s' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'ListBucketResult')
        name = dom.getElementsByTagName('Name')[0].childNodes[0].nodeValue
        self.assertEquals(name, bucket_name)

        objects = [n for n in dom.getElementsByTagName('Contents')]

        names = []
        for o in objects:
            if o.childNodes[0].nodeName == 'Key':
                names.append(o.childNodes[0].childNodes[0].nodeValue)
            if o.childNodes[1].nodeName == 'LastModified':
                self.assertTrue(
                    o.childNodes[1].childNodes[0].nodeValue.endswith('Z'))

        self.assertEquals(len(names), len(FakeAppBucket().objects))
        for i in FakeAppBucket().objects:
            self.assertTrue(i[0] in names)
开发者ID:blaxter,项目名称:swift3,代码行数:27,代码来源:test_swift3.py


示例12: test_bucket_GET_versions

    def test_bucket_GET_versions(self):
        app = Mock(wraps=FakeAppBucket())
        local_app = swift3.filter_factory({})(app)
        bucket_name = 'junk'
        req = Request.blank('/%s?versions' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'ListVersionsResult')
        name = dom.getElementsByTagName('Name')[0].childNodes[0].nodeValue
        self.assertEquals(name, bucket_name)

        objects = [n for n in dom.getElementsByTagName('Version')]
        dobjects = [n for n in dom.getElementsByTagName('DeleteMarker')]
        self.assertEquals(len(objects) + len(dobjects),
                          len(FakeAppBucket().vobjects))
        self.assertEquals(len(dobjects), 1)

        for o in objects:
            for node in o.childNodes:
                if node.nodeName =='LastModified':
                    self.assertTrue(node.childNodes[0].nodeValue.endswith('Z'))
                    break
开发者ID:blaxter,项目名称:zentyal-swift-middleware,代码行数:25,代码来源:test_swift3.py


示例13: setUp

    def setUp(self):
        self.app = FakeApp()
        self.swift = self.app.swift
        self.swift3 = swift3.filter_factory({})(self.app)

        self.swift.register('GET', '/something', swob.HTTPOk, {}, 'FAKE APP')

        self.setup_buckets()
        self.setup_objects()

        self.object_body = 'hello'
        self.response_headers = {'Content-Type': 'text/html',
                                 'Content-Length': len(self.object_body),
                                 'x-object-meta-test': 'swift',
                                 'etag': '1b2cf535f27731c974343645a3985328',
                                 'last-modified': '2011-01-05T02:19:14.275290'}

        self.swift.register('PUT', '/v1/AUTH_test/bucket',
                            swob.HTTPCreated, {}, None)
        self.swift.register('POST', '/v1/AUTH_test/bucket',
                            swob.HTTPNoContent, {}, None)
        self.swift.register('DELETE', '/v1/AUTH_test/bucket',
                            swob.HTTPNoContent, {}, None)

        self.swift.register('GET', '/v1/AUTH_test/bucket/object',
                            swob.HTTPOk, self.response_headers,
                            self.object_body)
        self.swift.register('PUT', '/v1/AUTH_test/bucket/object',
                            swob.HTTPCreated, {}, None)
        self.swift.register('DELETE', '/v1/AUTH_test/bucket/object',
                            swob.HTTPNoContent, {}, None)
开发者ID:kvdveer,项目名称:swift3,代码行数:31,代码来源:test_swift3.py


示例14: test_object_acl_GET

 def test_object_acl_GET(self):
     local_app = swift3.filter_factory({})(FakeAppObject())
     req = Request.blank(
         "/bucket/object?acl", environ={"REQUEST_METHOD": "GET"}, headers={"Authorization": "AWS test:tester:hmac"}
     )
     resp = local_app(req.environ, local_app.app.do_start_response)
     self._check_acl("test:tester", resp)
开发者ID:mygoda,项目名称:openstack,代码行数:7,代码来源:test_swift3.py


示例15: test_object_DELETE

 def test_object_DELETE(self):
     local_app = swift3.filter_factory({})(FakeAppObject(204))
     req = Request.blank(
         "/bucket/object", environ={"REQUEST_METHOD": "DELETE"}, headers={"Authorization": "AWS test:tester:hmac"}
     )
     resp = local_app(req.environ, local_app.app.do_start_response)
     self.assertEquals(local_app.app.response_args[0].split()[0], "204")
开发者ID:mygoda,项目名称:openstack,代码行数:7,代码来源:test_swift3.py


示例16: _test_method_error

 def _test_method_error(self, cl, method, path, status, headers={}):
     local_app = swift3.filter_factory({})(cl(status))
     headers.update({"Authorization": "AWS test:tester:hmac"})
     req = Request.blank(path, environ={"REQUEST_METHOD": method}, headers=headers)
     resp = local_app(req.environ, start_response)
     dom = xml.dom.minidom.parseString("".join(resp))
     self.assertEquals(dom.firstChild.nodeName, "Error")
     return dom.getElementsByTagName("Code")[0].childNodes[0].nodeValue
开发者ID:mygoda,项目名称:openstack,代码行数:8,代码来源:test_swift3.py


示例17: test_object_GET_with_versionId

    def test_object_GET_with_versionId(self):
        local_app = swift3.filter_factory({})(FakeAppObject())
        req = Request.blank('/bucket/object?versionId=123456789.123456',
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:hmac'})
        local_app(req.environ, local_app.app.do_start_response)

        self.assertEqual(local_app.app.env[0]["PATH_INFO"],
                         '/v1/test/bucket_versioned/object$123456789.123456$1')
开发者ID:blaxter,项目名称:zentyal-swift-middleware,代码行数:9,代码来源:test_swift3.py


示例18: test_bucket_PUT_acl_to_existing_container

 def test_bucket_PUT_acl_to_existing_container(self):
     local_app = swift3.filter_factory({})(FakeAppBucket(204))
     req = Request.blank('/bucket?acl',
                         environ={'REQUEST_METHOD': 'PUT'},
                         headers={
                             'Authorization': 'AWS test:tester:hmac',
                             'X-Amz-Acl': 'public-read'})
     resp = local_app(req.environ, local_app.app.do_start_response)
     self.assertEquals(local_app.app.response_args[0].split()[0], '200')
开发者ID:swiftstack,项目名称:swift3,代码行数:9,代码来源:test_swift3.py


示例19: _test_method_error

 def _test_method_error(self, cl, method, path, status):
     local_app = swift3.filter_factory({})(cl(status))
     req = Request.blank(path,
                         environ={'REQUEST_METHOD': method},
                         headers={'Authorization': 'AWS test:tester:hmac'})
     resp = local_app(req.environ, start_response)
     dom = xml.dom.minidom.parseString("".join(resp))
     self.assertEquals(dom.firstChild.nodeName, 'Error')
     return dom.getElementsByTagName('Code')[0].childNodes[0].nodeValue
开发者ID:TMG-nl,项目名称:swift3,代码行数:9,代码来源:test_swift3.py


示例20: test_bucket_versioning_GET

 def test_bucket_versioning_GET(self):
     local_app = swift3.filter_factory({})(FakeAppBucket())
     bucket_name = 'junk'
     req = Request.blank('/%s?versioning' % bucket_name,
                         environ={'REQUEST_METHOD': 'GET'},
                         headers={'Authorization': 'AWS test:tester:hmac'})
     resp = local_app(req.environ, local_app.app.do_start_response)
     dom = xml.dom.minidom.parseString("".join(resp))
     self.assertEquals(dom.firstChild.nodeName, 'VersioningConfiguration')
开发者ID:blaxter,项目名称:swift3,代码行数:9,代码来源:test_swift3.py



注:本文中的swift3.middleware.filter_factory函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python subresource.encode_acl函数代码示例发布时间:2022-05-27
下一篇:
Python etree.tostring函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap