Python ratelimit.filter_factory函数代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

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

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

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

示例1: test_registered_defaults

    def test_registered_defaults(self):

        def check_key_is_absent(key):
            try:
                swift_info[key]
            except KeyError as err:
                if key not in str(err):
                    raise

        test_limits = {'account_ratelimit': 1,
                       'max_sleep_time_seconds': 60,
                       'container_ratelimit_0': 0,
                       'container_ratelimit_10': 10,
                       'container_ratelimit_50': 50,
                       'container_listing_ratelimit_0': 0,
                       'container_listing_ratelimit_10': 10,
                       'container_listing_ratelimit_50': 50}

        ratelimit.filter_factory(test_limits)('have to pass in an app')
        swift_info = utils.get_swift_info()
        self.assertIn('ratelimit', swift_info)
        self.assertEqual(swift_info['ratelimit']
                         ['account_ratelimit'], 1.0)
        self.assertEqual(swift_info['ratelimit']
                         ['max_sleep_time_seconds'], 60.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][0][0], 0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][0][1], 0.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][1][0], 10)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][1][1], 10.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][2][0], 50)
        self.assertEqual(swift_info['ratelimit']
                         ['container_ratelimits'][2][1], 50.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][0][0], 0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][0][1], 0.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][1][0], 10)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][1][1], 10.0)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][2][0], 50)
        self.assertEqual(swift_info['ratelimit']
                         ['container_listing_ratelimits'][2][1], 50.0)

        # these were left out on purpose
        for key in ['log_sleep_time_seconds', 'clock_accuracy',
                    'rate_buffer_seconds', 'ratelimit_whitelis',
                    'ratelimit_blacklist']:
            check_key_is_absent(key)

开发者ID:mahak，项目名称:swift，代码行数:55，代码来源:test_ratelimit.py

示例2: test_ratelimit_blacklist

    def test_ratelimit_blacklist(self):
        global time_ticker
        current_rate = 2
        conf_dict = {'account_ratelimit': current_rate,
                     'max_sleep_time_seconds': 2,
                     'account_whitelist': 'a',
                     'account_blacklist': 'b'}
        self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
        self.test_ratelimit.logger = FakeLogger()
        self.test_ratelimit.BLACK_LIST_SLEEP = 0
        ratelimit.http_connect = mock_http_connect(204)
        req = Request.blank('/v/b/c')
        req.environ['swift.cache'] = FakeMemcache()

        class rate_caller(Thread):

            def __init__(self, parent):
                Thread.__init__(self)
                self.parent = parent

            def run(self):
                self.result = self.parent.test_ratelimit(req.environ,
                                                         start_response)
        nt = 5
        threads = []
        for i in range(nt):
            rc = rate_caller(self)
            rc.start()
            threads.append(rc)
        for thread in threads:
            thread.join()
        the_497s = [
            t for t in threads if ''.join(t.result).startswith('Your account')]
        self.assertEquals(len(the_497s), 5)
        self.assertEquals(time_ticker, 0)

开发者ID:BReduardokramer，项目名称:swift，代码行数:35，代码来源:test_ratelimit.py

示例3: test_ratelimit_max_rate_double_container

    def test_ratelimit_max_rate_double_container(self):
        global time_ticker
        global time_override
        current_rate = 2
        conf_dict = {'container_ratelimit_0': current_rate,
                     'clock_accuracy': 100,
                     'max_sleep_time_seconds': 1}
        self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
        self.test_ratelimit.log_sleep_time_seconds = .00001
        req = Request.blank('/v1/a/c/o')
        req.method = 'PUT'
        req.environ['swift.cache'] = FakeMemcache()
        req.environ['swift.cache'].set(
            get_cache_key('a', 'c'),
            {'object_count': 1})

        time_override = [0, 0, 0, 0, None]
        # simulates 4 requests coming in at same time, then sleeping
        with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                        lambda *args, **kwargs: {}):
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(.1)
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], b'Slow down')
            mock_sleep(.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], b'Slow down')
            mock_sleep(.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], b'Some Content')

开发者ID:mahak，项目名称:swift，代码行数:32，代码来源:test_ratelimit.py

示例4: test_ratelimit_max_rate_multiple_acc

    def test_ratelimit_max_rate_multiple_acc(self):
        num_calls = 4
        current_rate = 2
        conf_dict = {"account_ratelimit": current_rate, "max_sleep_time_seconds": 2}
        fake_memcache = FakeMemcache()

        the_app = ratelimit.filter_factory(conf_dict)(FakeApp())
        the_app.memcache_client = fake_memcache
        req = lambda: None
        req.method = "PUT"
        req.environ = {}

        class rate_caller(threading.Thread):
            def __init__(self, name):
                self.myname = name
                threading.Thread.__init__(self)

            def run(self):
                for j in range(num_calls):
                    self.result = the_app.handle_ratelimit(req, self.myname, "c", None)

        with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
            nt = 15
            begin = time.time()
            threads = []
            for i in range(nt):
                rc = rate_caller("a%s" % i)
                rc.start()
                threads.append(rc)
            for thread in threads:
                thread.join()

            time_took = time.time() - begin
            self.assertEqual(1.5, round(time_took, 1))

开发者ID:notmyname，项目名称:swift，代码行数:34，代码来源:test_ratelimit.py

示例5: test_ratelimit_max_rate_double_container_listing

    def test_ratelimit_max_rate_double_container_listing(self):
        global time_ticker
        global time_override
        current_rate = 2
        conf_dict = {"container_listing_ratelimit_0": current_rate, "clock_accuracy": 100, "max_sleep_time_seconds": 1}
        self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
        ratelimit.http_connect = mock_http_connect(204)
        self.test_ratelimit.log_sleep_time_seconds = 0.00001
        req = Request.blank("/v/a/c")
        req.method = "GET"
        req.environ["swift.cache"] = FakeMemcache()
        req.environ["swift.cache"].set(get_cache_key("a", "c"), {"object_count": 1})

        with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
            time_override = [0, 0, 0, 0, None]
            # simulates 4 requests coming in at same time, then sleeping
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "Slow down")
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "Slow down")
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "204 No Content")
            mc = self.test_ratelimit.memcache_client
            try:
                self.test_ratelimit.memcache_client = None
                self.assertEqual(self.test_ratelimit.handle_ratelimit(req, "n", "c", None), None)
            finally:
                self.test_ratelimit.memcache_client = mc

开发者ID:notmyname，项目名称:swift，代码行数:34，代码来源:test_ratelimit.py

示例6: test_ratelimit_max_rate_double

    def test_ratelimit_max_rate_double(self):
        global time_ticker
        global time_override
        current_rate = 2
        conf_dict = {"account_ratelimit": current_rate, "clock_accuracy": 100, "max_sleep_time_seconds": 1}
        self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
        ratelimit.http_connect = mock_http_connect(204)
        self.test_ratelimit.log_sleep_time_seconds = 0.00001
        req = Request.blank("/v/a/c")
        req.method = "PUT"
        req.environ["swift.cache"] = FakeMemcache()

        time_override = [0, 0, 0, 0, None]
        # simulates 4 requests coming in at same time, then sleeping
        with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "Slow down")
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "Slow down")
            mock_sleep(0.1)
            r = self.test_ratelimit(req.environ, start_response)
            self.assertEqual(r[0], "204 No Content")

开发者ID:notmyname，项目名称:swift，代码行数:27，代码来源:test_ratelimit.py

示例7: test_get_ratelimitable_key_tuples

    def test_get_ratelimitable_key_tuples(self):
        current_rate = 13
        conf_dict = {"account_ratelimit": current_rate, "container_ratelimit_3": 200}
        fake_memcache = FakeMemcache()
        fake_memcache.store[get_cache_key("a", "c")] = {"object_count": "5"}
        the_app = ratelimit.filter_factory(conf_dict)(FakeApp())
        the_app.memcache_client = fake_memcache
        req = lambda: None
        req.environ = {"swift.cache": fake_memcache, "PATH_INFO": "/v1/a/c/o"}
        with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
            req.method = "DELETE"
            self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", None, None)), 0)
            req.method = "PUT"
            self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", "c", None)), 1)
            req.method = "DELETE"
            self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", "c", None)), 1)
            req.method = "GET"
            self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", "c", "o")), 0)
            req.method = "PUT"
            self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", "c", "o")), 1)

        req.method = "PUT"
        self.assertEqual(len(the_app.get_ratelimitable_key_tuples(req, "a", "c", None, global_ratelimit=10)), 2)
        self.assertEqual(
            the_app.get_ratelimitable_key_tuples(req, "a", "c", None, global_ratelimit=10)[1],
            ("ratelimit/global-write/a", 10),
        )

        req.method = "PUT"
        self.assertEqual(
            len(the_app.get_ratelimitable_key_tuples(req, "a", "c", None, global_ratelimit="notafloat")), 1
        )

开发者ID:notmyname，项目名称:swift，代码行数:32，代码来源:test_ratelimit.py

示例8: test_registered_defaults

    def test_registered_defaults(self):
        def check_key_is_absnet(key):
            try:
                swift_info[key]
            except KeyError as err:
                if key not in err:
                    raise

        test_limits = {
            "account_ratelimit": 1,
            "max_sleep_time_seconds": 60,
            "container_ratelimit_0": 0,
            "container_ratelimit_10": 10,
            "container_ratelimit_50": 50,
            "container_listing_ratelimit_0": 0,
            "container_listing_ratelimit_10": 10,
            "container_listing_ratelimit_50": 50,
        }

        ratelimit.filter_factory(test_limits)("have to pass in an app")
        swift_info = utils.get_swift_info()
        self.assertTrue("ratelimit" in swift_info)
        self.assertEqual(swift_info["ratelimit"]["account_ratelimit"], 1.0)
        self.assertEqual(swift_info["ratelimit"]["max_sleep_time_seconds"], 60.0)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][0][0], 0)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][0][1], 0.0)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][1][0], 10)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][1][1], 10.0)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][2][0], 50)
        self.assertEqual(swift_info["ratelimit"]["container_ratelimits"][2][1], 50.0)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][0][0], 0)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][0][1], 0.0)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][1][0], 10)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][1][1], 10.0)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][2][0], 50)
        self.assertEqual(swift_info["ratelimit"]["container_listing_ratelimits"][2][1], 50.0)

        # these were left out on purpose
        for key in [
            "log_sleep_time_seconds",
            "clock_accuracy",
            "rate_buffer_seconds",
            "ratelimit_whitelis",
            "ratelimit_blacklist",
        ]:
            check_key_is_absnet(key)

开发者ID:notmyname，项目名称:swift，代码行数:46，代码来源:test_ratelimit.py

示例9: test_get_maxrate

 def test_get_maxrate(self):
     conf_dict = {"container_ratelimit_10": 200, "container_ratelimit_50": 100, "container_ratelimit_75": 30}
     test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     test_ratelimit.logger = FakeLogger()
     self.assertEqual(ratelimit.get_maxrate(test_ratelimit.container_ratelimits, 0), None)
     self.assertEqual(ratelimit.get_maxrate(test_ratelimit.container_ratelimits, 5), None)
     self.assertEqual(ratelimit.get_maxrate(test_ratelimit.container_ratelimits, 10), 200)
     self.assertEqual(ratelimit.get_maxrate(test_ratelimit.container_ratelimits, 60), 72)
     self.assertEqual(ratelimit.get_maxrate(test_ratelimit.container_ratelimits, 160), 30)

开发者ID:notmyname，项目名称:swift，代码行数:9，代码来源:test_ratelimit.py

示例10: test_no_memcache

 def test_no_memcache(self):
     current_rate = 13
     num_calls = 5
     conf_dict = {'account_ratelimit': current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     ratelimit.http_connect = mock_http_connect(204)
     req = Request.blank('/v/a')
     req.environ['swift.cache'] = None
     make_app_call = lambda: self.test_ratelimit(req.environ,
                                                 start_response)
     self._run(make_app_call, num_calls, current_rate)

开发者ID:pombredanne，项目名称:swift，代码行数:11，代码来源:test_ratelimit.py

示例11: test_no_memcache

 def test_no_memcache(self):
     current_rate = 13
     num_calls = 5
     conf_dict = {'account_ratelimit': current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     req = Request.blank('/v1/a')
     req.environ['swift.cache'] = None
     make_app_call = lambda: self.test_ratelimit(req.environ,
                                                 start_response)
     begin = time.time()
     self._run(make_app_call, num_calls, current_rate, check_time=False)
     time_took = time.time() - begin
     self.assertEqual(round(time_took, 1), 0)  # no memcache, no limiting

开发者ID:mahak，项目名称:swift，代码行数:13，代码来源:test_ratelimit.py

示例12: test_ratelimit_old_memcache_format

 def test_ratelimit_old_memcache_format(self):
     current_rate = 13
     conf_dict = {"account_ratelimit": current_rate, "container_ratelimit_3": 200}
     fake_memcache = FakeMemcache()
     fake_memcache.store[get_cache_key("a", "c")] = {"container_size": 5}
     the_app = ratelimit.filter_factory(conf_dict)(FakeApp())
     the_app.memcache_client = fake_memcache
     req = lambda: None
     req.method = "PUT"
     req.environ = {"PATH_INFO": "/v1/a/c/o", "swift.cache": fake_memcache}
     with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
         tuples = the_app.get_ratelimitable_key_tuples(req, "a", "c", "o")
         self.assertEqual(tuples, [("ratelimit/a/c", 200.0)])

开发者ID:notmyname，项目名称:swift，代码行数:13，代码来源:test_ratelimit.py

示例13: test_ratelimit_old_white_black_list

 def test_ratelimit_old_white_black_list(self):
     global time_ticker
     current_rate = 2
     conf_dict = {
         "account_ratelimit": current_rate,
         "max_sleep_time_seconds": 2,
         "account_whitelist": "a",
         "account_blacklist": "b",
     }
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     req = Request.blank("/")
     with mock.patch.object(self.test_ratelimit, "memcache_client", FakeMemcache()):
         self.assertEqual(self.test_ratelimit.handle_ratelimit(req, "a", "c", "o"), None)
         self.assertEqual(self.test_ratelimit.handle_ratelimit(req, "b", "c", "o").status_int, 497)

开发者ID:notmyname，项目名称:swift，代码行数:14，代码来源:test_ratelimit.py

示例14: test_ratelimit_set_incr

 def test_ratelimit_set_incr(self):
     current_rate = 5
     num_calls = 50
     conf_dict = {"account_ratelimit": current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     ratelimit.http_connect = mock_http_connect(204)
     req = Request.blank("/v/a/c")
     req.method = "PUT"
     req.environ["swift.cache"] = FakeMemcache()
     req.environ["swift.cache"].init_incr_return_neg = True
     make_app_call = lambda: self.test_ratelimit(req.environ, start_response)
     begin = time.time()
     with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
         self._run(make_app_call, num_calls, current_rate, check_time=False)
         self.assertEqual(round(time.time() - begin, 1), 9.8)

开发者ID:notmyname，项目名称:swift，代码行数:15，代码来源:test_ratelimit.py

示例15: test_ratelimit_set_incr

 def test_ratelimit_set_incr(self):
     current_rate = 5
     num_calls = 50
     conf_dict = {'account_ratelimit': current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     ratelimit.http_connect = mock_http_connect(204)
     req = Request.blank('/v/a/c')
     req.method = 'PUT'
     req.environ['swift.cache'] = FakeMemcache()
     req.environ['swift.cache'].init_incr_return_neg = True
     make_app_call = lambda: self.test_ratelimit(req.environ,
                                                 start_response)
     begin = time.time()
     self._run(make_app_call, num_calls, current_rate, check_time=False)
     self.assertEquals(round(time.time() - begin, 1), 9.8)

开发者ID:Guoweiwei1130，项目名称:openstack，代码行数:15，代码来源:test_ratelimit.py

示例16: test_restarting_memcache

 def test_restarting_memcache(self):
     current_rate = 2
     num_calls = 5
     conf_dict = {'account_ratelimit': current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     ratelimit.http_connect = mock_http_connect(204)
     req = Request.blank('/v/a')
     req.environ['swift.cache'] = FakeMemcache()
     req.environ['swift.cache'].error_on_incr = True
     make_app_call = lambda: self.test_ratelimit(req.environ,
                                                 start_response)
     begin = time.time()
     self._run(make_app_call, num_calls, current_rate, check_time=False)
     time_took = time.time() - begin
     self.assert_(round(time_took, 1) == 0) # no memcache, no limiting

开发者ID:pombredanne，项目名称:swift，代码行数:15，代码来源:test_ratelimit.py

示例17: test_get_ratelimitable_key_tuples

    def test_get_ratelimitable_key_tuples(self):
        current_rate = 13
        conf_dict = {'account_ratelimit': current_rate,
                     'container_ratelimit_3': 200}
        fake_memcache = FakeMemcache()
        fake_memcache.store[get_container_memcache_key('a', 'c')] = \
            {'object_count': '5'}
        the_app = ratelimit.filter_factory(conf_dict)(FakeApp())
        the_app.memcache_client = fake_memcache
        req = lambda: None
        req.environ = {'swift.cache': fake_memcache, 'PATH_INFO': '/v1/a/c/o'}
        with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                        lambda *args, **kwargs: {}):
            req.method = 'DELETE'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', None, None)), 0)
            req.method = 'PUT'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', None)), 1)
            req.method = 'DELETE'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', None)), 1)
            req.method = 'GET'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', 'o')), 0)
            req.method = 'PUT'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', 'o')), 1)

        def get_fake_ratelimit(*args, **kwargs):
            return {'sysmeta': {'global-write-ratelimit': 10}}

        with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                        get_fake_ratelimit):
            req.method = 'PUT'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', None)), 2)
            self.assertEquals(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', None)[1], ('ratelimit/global-write/a', 10))

        def get_fake_ratelimit(*args, **kwargs):
            return {'sysmeta': {'global-write-ratelimit': 'notafloat'}}

        with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                        get_fake_ratelimit):
            req.method = 'PUT'
            self.assertEquals(len(the_app.get_ratelimitable_key_tuples(
                req, 'a', 'c', None)), 1)

开发者ID:BReduardokramer，项目名称:swift，代码行数:48，代码来源:test_ratelimit.py

示例18: test_ratelimit_old_memcache_format

 def test_ratelimit_old_memcache_format(self):
     current_rate = 13
     conf_dict = {'account_ratelimit': current_rate,
                  'container_ratelimit_3': 200}
     fake_memcache = FakeMemcache()
     fake_memcache.store[get_cache_key('a', 'c')] = \
         {'container_size': 5}
     the_app = ratelimit.filter_factory(conf_dict)(FakeApp())
     the_app.memcache_client = fake_memcache
     req = lambda: None
     req.method = 'PUT'
     req.environ = {'PATH_INFO': '/v1/a/c/o', 'swift.cache': fake_memcache}
     with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                     lambda *args, **kwargs: {}):
         tuples = the_app.get_ratelimitable_key_tuples(req, 'a', 'c', 'o')
         self.assertEqual(tuples, [('ratelimit/a/c', 200.0)])

开发者ID:mahak，项目名称:swift，代码行数:16，代码来源:test_ratelimit.py

示例19: test_account_ratelimit

 def test_account_ratelimit(self):
     current_rate = 5
     num_calls = 50
     conf_dict = {"account_ratelimit": current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     ratelimit.http_connect = mock_http_connect(204)
     with mock.patch("swift.common.middleware.ratelimit.get_container_info", lambda *args, **kwargs: {}):
         with mock.patch("swift.common.middleware.ratelimit.get_account_info", lambda *args, **kwargs: {}):
             for meth, exp_time in [("DELETE", 9.8), ("GET", 0), ("POST", 0), ("PUT", 9.8)]:
                 req = Request.blank("/v/a%s/c" % meth)
                 req.method = meth
                 req.environ["swift.cache"] = FakeMemcache()
                 make_app_call = lambda: self.test_ratelimit(req.environ, start_response)
                 begin = time.time()
                 self._run(make_app_call, num_calls, current_rate, check_time=bool(exp_time))
                 self.assertEqual(round(time.time() - begin, 1), exp_time)
                 self._reset_time()

开发者ID:notmyname，项目名称:swift，代码行数:17，代码来源:test_ratelimit.py

示例20: test_restarting_memcache

 def test_restarting_memcache(self):
     current_rate = 2
     num_calls = 5
     conf_dict = {'account_ratelimit': current_rate}
     self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp())
     req = Request.blank('/v1/a/c')
     req.method = 'PUT'
     req.environ['swift.cache'] = FakeMemcache()
     req.environ['swift.cache'].error_on_incr = True
     make_app_call = lambda: self.test_ratelimit(req.environ,
                                                 start_response)
     begin = time.time()
     with mock.patch('swift.common.middleware.ratelimit.get_account_info',
                     lambda *args, **kwargs: {}):
         self._run(make_app_call, num_calls, current_rate, check_time=False)
         time_took = time.time() - begin
         self.assertEqual(round(time_took, 1), 0)  # no memcache, no limit

开发者ID:mahak，项目名称:swift，代码行数:17，代码来源:test_ratelimit.py

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python etree.fromstring函数代码示例发布时间：2022-05-27

Python keystoneauth.filter_factory函数代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13784|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10045|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4071|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4039|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3827|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3505|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3028|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2641|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2633|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2293|2022-01-22

客服电话

电子邮件

Python ratelimit.filter_factory函数代码示例

示例1: test_registered_defaults

示例2: test_ratelimit_blacklist

示例3: test_ratelimit_max_rate_double_container

示例4: test_ratelimit_max_rate_multiple_acc

示例5: test_ratelimit_max_rate_double_container_listing

示例6: test_ratelimit_max_rate_double

示例7: test_get_ratelimitable_key_tuples

示例8: test_registered_defaults

示例9: test_get_maxrate

示例10: test_no_memcache

示例11: test_no_memcache

示例12: test_ratelimit_old_memcache_format

示例13: test_ratelimit_old_white_black_list

示例14: test_ratelimit_set_incr

示例15: test_ratelimit_set_incr

示例16: test_restarting_memcache

示例17: test_get_ratelimitable_key_tuples

示例18: test_ratelimit_old_memcache_format

示例19: test_account_ratelimit

示例20: test_restarting_memcache

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053