<s> """<STR_LIT>""" <EOL> import httplib <EOL> import json <EOL> import StringIO <EOL> import stubout <EOL> import time <EOL> import unittest <EOL> import webob <EOL> from xml . dom import minidom <EOL> import nova . context <EOL> from nova . api . openstack import limits <EOL> from nova . api . openstack import views <EOL> from nova import test <EOL> TEST_LIMITS = [ <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:1> , limits . PER_MINUTE ) , <EOL> limits . Limit ( "<STR_LIT:POST>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:7> , limits . PER_MINUTE ) , <EOL> limits . Limit ( "<STR_LIT:POST>" , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:3> , limits . PER_MINUTE ) , <EOL> limits . Limit ( "<STR_LIT>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:10> , limits . PER_MINUTE ) , <EOL> limits . Limit ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:5> , limits . PER_MINUTE ) , <EOL> ] <EOL> class BaseLimitTestSuite ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . time = <NUM_LIT:0.0> <EOL> self . stubs = stubout . StubOutForTesting ( ) <EOL> self . stubs . Set ( limits . Limit , "<STR_LIT>" , self . _get_time ) <EOL> self . absolute_limits = { } <EOL> def stub_get_project_quotas ( context , project_id ) : <EOL> return self . absolute_limits <EOL> self . stubs . Set ( nova . quota , "<STR_LIT>" , <EOL> stub_get_project_quotas ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stubs . UnsetAll ( ) <EOL> def _get_time ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . time <EOL> class LimitsControllerV10Test ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> self . controller = limits . create_resource ( '<STR_LIT:1.0>' ) <EOL> def _get_index_request ( self , accept_header = "<STR_LIT:application/json>" ) : <EOL> """<STR_LIT>""" <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> request . accept = accept_header <EOL> request . environ [ "<STR_LIT>" ] = ( None , { <EOL> "<STR_LIT:action>" : "<STR_LIT:index>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ) <EOL> context = nova . context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . environ [ "<STR_LIT>" ] = context <EOL> return request <EOL> def _populate_limits ( self , request ) : <EOL> """<STR_LIT>""" <EOL> _limits = [ <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:10> , <NUM_LIT> ) . display ( ) , <EOL> limits . Limit ( "<STR_LIT:POST>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:5> , <NUM_LIT> * <NUM_LIT> ) . display ( ) , <EOL> ] <EOL> request . environ [ "<STR_LIT>" ] = _limits <EOL> return request <EOL> def _setup_absolute_limits ( self ) : <EOL> self . absolute_limits = { <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : <NUM_LIT:8> , <EOL> '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> def test_empty_index_json ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body ) <EOL> def test_index_json ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( ) <EOL> request = self . _populate_limits ( request ) <EOL> self . _setup_absolute_limits ( ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ] , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT:8> , <EOL> "<STR_LIT>" : <NUM_LIT:2> ** <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> } , <EOL> } , <EOL> } <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body ) <EOL> def test_empty_index_xml ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( "<STR_LIT>" ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = minidom . parseString ( """<STR_LIT>""" . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> body = minidom . parseString ( response . body . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected . toxml ( ) , body . toxml ( ) ) <EOL> def test_index_xml ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( "<STR_LIT>" ) <EOL> request = self . _populate_limits ( request ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = minidom . parseString ( """<STR_LIT>""" . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> body = minidom . parseString ( response . body . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected . toxml ( ) , body . toxml ( ) ) <EOL> class LimitsControllerV11Test ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> self . controller = limits . create_resource ( '<STR_LIT>' ) <EOL> self . maxDiff = None <EOL> def _get_index_request ( self , accept_header = "<STR_LIT:application/json>" ) : <EOL> """<STR_LIT>""" <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> request . accept = accept_header <EOL> request . environ [ "<STR_LIT>" ] = ( None , { <EOL> "<STR_LIT:action>" : "<STR_LIT:index>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ) <EOL> context = nova . context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . environ [ "<STR_LIT>" ] = context <EOL> return request <EOL> def _populate_limits ( self , request ) : <EOL> """<STR_LIT>""" <EOL> _limits = [ <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:10> , <NUM_LIT> ) . display ( ) , <EOL> limits . Limit ( "<STR_LIT:POST>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:5> , <NUM_LIT> * <NUM_LIT> ) . display ( ) , <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> <NUM_LIT:5> , <NUM_LIT> ) . display ( ) , <EOL> ] <EOL> request . environ [ "<STR_LIT>" ] = _limits <EOL> return request <EOL> def test_empty_index_json ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body ) <EOL> def test_index_json ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( ) <EOL> request = self . _populate_limits ( request ) <EOL> self . absolute_limits = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> response = request . get_response ( self . controller ) <EOL> expected = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> } , <EOL> ] , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> } , <EOL> ] , <EOL> } , <EOL> ] , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> } , <EOL> } , <EOL> } <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body ) <EOL> def _populate_limits_diff_regex ( self , request ) : <EOL> """<STR_LIT>""" <EOL> _limits = [ <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:10> , <NUM_LIT> ) . display ( ) , <EOL> limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:10> , <NUM_LIT> ) . display ( ) , <EOL> ] <EOL> request . environ [ "<STR_LIT>" ] = _limits <EOL> return request <EOL> def test_index_diff_regex ( self ) : <EOL> """<STR_LIT>""" <EOL> request = self . _get_index_request ( ) <EOL> request = self . _populate_limits_diff_regex ( request ) <EOL> response = request . get_response ( self . controller ) <EOL> expected = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> } , <EOL> ] , <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:GET>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> } , <EOL> ] , <EOL> } , <EOL> ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body ) <EOL> def _test_index_absolute_limits_json ( self , expected ) : <EOL> request = self . _get_index_request ( ) <EOL> response = request . get_response ( self . controller ) <EOL> body = json . loads ( response . body ) <EOL> self . assertEqual ( expected , body [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_index_ignores_extra_absolute_limits_json ( self ) : <EOL> self . absolute_limits = { '<STR_LIT>' : <NUM_LIT> } <EOL> self . _test_index_absolute_limits_json ( { } ) <EOL> def test_index_absolute_ram_json ( self ) : <EOL> self . absolute_limits = { '<STR_LIT>' : <NUM_LIT> } <EOL> self . _test_index_absolute_limits_json ( { '<STR_LIT>' : <NUM_LIT> } ) <EOL> def test_index_absolute_cores_json ( self ) : <EOL> self . absolute_limits = { '<STR_LIT>' : <NUM_LIT> } <EOL> self . _test_index_absolute_limits_json ( { '<STR_LIT>' : <NUM_LIT> } ) <EOL> def test_index_absolute_instances_json ( self ) : <EOL> self . absolute_limits = { '<STR_LIT>' : <NUM_LIT> } <EOL> self . _test_index_absolute_limits_json ( { '<STR_LIT>' : <NUM_LIT> } ) <EOL> def test_index_absolute_metadata_json ( self ) : <EOL> self . absolute_limits = { '<STR_LIT>' : <NUM_LIT> } <EOL> expected = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> self . _test_index_absolute_limits_json ( expected ) <EOL> def test_index_absolute_injected_files ( self ) : <EOL> self . absolute_limits = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> expected = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> self . _test_index_absolute_limits_json ( expected ) <EOL> class TestLimiter ( limits . Limiter ) : <EOL> pass <EOL> class LimitMiddlewareTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> @ webob . dec . wsgify <EOL> def _empty_app ( self , request ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> _limits = '<STR_LIT>' <EOL> self . app = limits . RateLimitingMiddleware ( self . _empty_app , _limits , <EOL> "<STR_LIT>" % <EOL> self . __class__ . __module__ ) <EOL> def test_limit_class ( self ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( self . app . _limiter , TestLimiter ) <EOL> def test_good_request ( self ) : <EOL> """<STR_LIT>""" <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> def test_limited_request_json ( self ) : <EOL> """<STR_LIT>""" <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( response . status_int , <NUM_LIT> ) <EOL> body = json . loads ( response . body ) <EOL> expected = "<STR_LIT>" <EOL> value = body [ "<STR_LIT>" ] [ "<STR_LIT>" ] . strip ( ) <EOL> self . assertEqual ( value , expected ) <EOL> def test_limited_request_xml ( self ) : <EOL> """<STR_LIT>""" <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> request . accept = "<STR_LIT>" <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( response . status_int , <NUM_LIT> ) <EOL> root = minidom . parseString ( response . body ) . childNodes [ <NUM_LIT:0> ] <EOL> expected = "<STR_LIT>" <EOL> details = root . getElementsByTagName ( "<STR_LIT>" ) <EOL> self . assertEqual ( details . length , <NUM_LIT:1> ) <EOL> value = details . item ( <NUM_LIT:0> ) . firstChild . data . strip ( ) <EOL> self . assertEqual ( value , expected ) <EOL> class LimitTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def test_GET_no_delay ( self ) : <EOL> """<STR_LIT>""" <EOL> limit = limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> delay = limit ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( None , delay ) <EOL> self . assertEqual ( <NUM_LIT:0> , limit . next_request ) <EOL> self . assertEqual ( <NUM_LIT:0> , limit . last_request ) <EOL> def test_GET_delay ( self ) : <EOL> """<STR_LIT>""" <EOL> limit = limits . Limit ( "<STR_LIT:GET>" , "<STR_LIT:*>" , "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> delay = limit ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( None , delay ) <EOL> delay = limit ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , delay ) <EOL> self . assertEqual ( <NUM_LIT:1> , limit . next_request ) <EOL> self . assertEqual ( <NUM_LIT:0> , limit . last_request ) <EOL> self . time += <NUM_LIT:4> <EOL> delay = limit ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( None , delay ) <EOL> self . assertEqual ( <NUM_LIT:4> , limit . next_request ) <EOL> self . assertEqual ( <NUM_LIT:4> , limit . last_request ) <EOL> class ParseLimitsTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def test_invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , limits . Limiter . parse_limits , <EOL> '<STR_LIT>' ) <EOL> def test_bad_rule ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , limits . Limiter . parse_limits , <EOL> '<STR_LIT>' ) <EOL> def test_missing_arg ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , limits . Limiter . parse_limits , <EOL> '<STR_LIT>' ) <EOL> def test_bad_value ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , limits . Limiter . parse_limits , <EOL> '<STR_LIT>' ) <EOL> def test_bad_unit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , limits . Limiter . parse_limits , <EOL> '<STR_LIT>' ) <EOL> def test_multiple_rules ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> l = limits . Limiter . parse_limits ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except ValueError , e : <EOL> assert False , str ( e ) <EOL> self . assertEqual ( len ( l ) , <NUM_LIT:4> ) <EOL> expected = [ '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT:POST>' , '<STR_LIT>' ] <EOL> self . assertEqual ( [ t . verb for t in l ] , expected ) <EOL> expected = [ '<STR_LIT:*>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertEqual ( [ t . uri for t in l ] , expected ) <EOL> expected = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertEqual ( [ t . regex for t in l ] , expected ) <EOL> expected = [ <NUM_LIT:20> , <NUM_LIT:10> , <NUM_LIT:5> , <NUM_LIT:1> ] <EOL> self . assertEqual ( [ t . value for t in l ] , expected ) <EOL> expected = [ limits . PER_MINUTE , limits . PER_HOUR , <EOL> limits . PER_SECOND , limits . PER_DAY ] <EOL> self . assertEqual ( [ t . unit for t in l ] , expected ) <EOL> class LimiterTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> userlimits = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . limiter = limits . Limiter ( TEST_LIMITS , ** userlimits ) <EOL> def _check ( self , num , verb , url , username = None ) : <EOL> """<STR_LIT>""" <EOL> for x in xrange ( num ) : <EOL> yield self . limiter . check_for_delay ( verb , url , username ) [ <NUM_LIT:0> ] <EOL> def _check_sum ( self , num , verb , url , username = None ) : <EOL> """<STR_LIT>""" <EOL> results = self . _check ( num , verb , url , username ) <EOL> return sum ( item for item in results if item ) <EOL> def test_no_delay_GET ( self ) : <EOL> """<STR_LIT>""" <EOL> delay = self . limiter . check_for_delay ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , ( None , None ) ) <EOL> def test_no_delay_PUT ( self ) : <EOL> """<STR_LIT>""" <EOL> delay = self . limiter . check_for_delay ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , ( None , None ) ) <EOL> def test_delay_PUT ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:10> + [ <NUM_LIT> ] <EOL> results = list ( self . _check ( <NUM_LIT:11> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> def test_delay_POST ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:7> <EOL> results = list ( self . _check ( <NUM_LIT:7> , "<STR_LIT:POST>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> expected = <NUM_LIT> / <NUM_LIT> <EOL> results = self . _check_sum ( <NUM_LIT:1> , "<STR_LIT:POST>" , "<STR_LIT>" ) <EOL> self . failUnlessAlmostEqual ( expected , results , <NUM_LIT:8> ) <EOL> def test_delay_GET ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:11> <EOL> results = list ( self . _check ( <NUM_LIT:11> , "<STR_LIT:GET>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> def test_delay_PUT_servers ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:5> + [ <NUM_LIT> ] <EOL> results = list ( self . _check ( <NUM_LIT:6> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> expected = [ None ] * <NUM_LIT:4> + [ <NUM_LIT> ] <EOL> results = list ( self . _check ( <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> def test_delay_PUT_wait ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:10> + [ <NUM_LIT> ] <EOL> results = list ( self . _check ( <NUM_LIT:11> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> self . time += <NUM_LIT> <EOL> expected = [ None , <NUM_LIT> ] <EOL> results = list ( self . _check ( <NUM_LIT:2> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> def test_multiple_delays ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:10> + [ <NUM_LIT> ] * <NUM_LIT:10> <EOL> results = list ( self . _check ( <NUM_LIT:20> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> self . time += <NUM_LIT:1.0> <EOL> expected = [ <NUM_LIT> ] * <NUM_LIT:10> <EOL> results = list ( self . _check ( <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> def test_user_limit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . limiter . levels [ '<STR_LIT>' ] , [ ] ) <EOL> def test_multiple_users ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = [ None ] * <NUM_LIT:10> + [ <NUM_LIT> ] * <NUM_LIT:10> <EOL> results = list ( self . _check ( <NUM_LIT:20> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> expected = [ None ] * <NUM_LIT:10> + [ <NUM_LIT> ] * <NUM_LIT:5> <EOL> results = list ( self . _check ( <NUM_LIT:15> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> expected = [ None ] * <NUM_LIT:20> <EOL> results = list ( self . _check ( <NUM_LIT:20> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> self . time += <NUM_LIT:1.0> <EOL> expected = [ <NUM_LIT> ] * <NUM_LIT:10> <EOL> results = list ( self . _check ( <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> self . time += <NUM_LIT:1.0> <EOL> expected = [ <NUM_LIT> ] * <NUM_LIT:5> <EOL> results = list ( self . _check ( <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected , results ) <EOL> class WsgiLimiterTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> self . app = limits . WsgiLimiter ( TEST_LIMITS ) <EOL> def _request_data ( self , verb , path ) : <EOL> """<STR_LIT>""" <EOL> return json . dumps ( { "<STR_LIT>" : verb , "<STR_LIT:path>" : path } ) <EOL> def _request ( self , verb , url , username = None ) : <EOL> """<STR_LIT>""" <EOL> if username : <EOL> request = webob . Request . blank ( "<STR_LIT>" % username ) <EOL> else : <EOL> request = webob . Request . blank ( "<STR_LIT:/>" ) <EOL> request . method = "<STR_LIT:POST>" <EOL> request . body = self . _request_data ( verb , url ) <EOL> response = request . get_response ( self . app ) <EOL> if "<STR_LIT>" in response . headers : <EOL> self . assertEqual ( response . status_int , <NUM_LIT> ) <EOL> return response . headers [ "<STR_LIT>" ] <EOL> self . assertEqual ( response . status_int , <NUM_LIT> ) <EOL> def test_invalid_methods ( self ) : <EOL> """<STR_LIT>""" <EOL> requests = [ ] <EOL> for method in [ "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> request = webob . Request . blank ( "<STR_LIT:/>" , method = method ) <EOL> response = request . get_response ( self . app ) <EOL> self . assertEqual ( response . status_int , <NUM_LIT> ) <EOL> def test_good_url ( self ) : <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , None ) <EOL> def test_escaping ( self ) : <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , None ) <EOL> def test_response_to_delays ( self ) : <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , None ) <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , '<STR_LIT>' ) <EOL> def test_response_to_delays_usernames ( self ) : <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , None ) <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , None ) <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , '<STR_LIT>' ) <EOL> delay = self . _request ( "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , '<STR_LIT>' ) <EOL> class FakeHttplibSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , response_string ) : <EOL> """<STR_LIT>""" <EOL> self . _buffer = StringIO . StringIO ( response_string ) <EOL> def makefile ( self , _mode , _other ) : <EOL> """<STR_LIT>""" <EOL> return self . _buffer <EOL> class FakeHttplibConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app , host ) : <EOL> """<STR_LIT>""" <EOL> self . app = app <EOL> self . host = host <EOL> def request ( self , method , path , body = "<STR_LIT>" , headers = None ) : <EOL> """<STR_LIT>""" <EOL> if not headers : <EOL> headers = { } <EOL> req = webob . Request . blank ( path ) <EOL> req . method = method <EOL> req . headers = headers <EOL> req . host = self . host <EOL> req . body = body <EOL> resp = str ( req . get_response ( self . app ) ) <EOL> resp = "<STR_LIT>" % resp <EOL> sock = FakeHttplibSocket ( resp ) <EOL> self . http_response = httplib . HTTPResponse ( sock ) <EOL> self . http_response . begin ( ) <EOL> def getresponse ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . http_response <EOL> def wire_HTTPConnection_to_WSGI ( host , app ) : <EOL> """<STR_LIT>""" <EOL> class HTTPConnectionDecorator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , wrapped ) : <EOL> self . wrapped = wrapped <EOL> def __call__ ( self , connection_host , * args , ** kwargs ) : <EOL> if connection_host == host : <EOL> return FakeHttplibConnection ( app , host ) <EOL> else : <EOL> return self . wrapped ( connection_host , * args , ** kwargs ) <EOL> httplib . HTTPConnection = HTTPConnectionDecorator ( httplib . HTTPConnection ) <EOL> class WsgiLimiterProxyTest ( BaseLimitTestSuite ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> BaseLimitTestSuite . setUp ( self ) <EOL> self . app = limits . WsgiLimiter ( TEST_LIMITS ) <EOL> wire_HTTPConnection_to_WSGI ( "<STR_LIT>" , self . app ) <EOL> self . proxy = limits . WsgiLimiterProxy ( "<STR_LIT>" ) <EOL> def test_200 ( self ) : <EOL> """<STR_LIT>""" <EOL> delay = self . proxy . check_for_delay ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , ( None , None ) ) <EOL> def test_403 ( self ) : <EOL> """<STR_LIT>""" <EOL> delay = self . proxy . check_for_delay ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( delay , ( None , None ) ) <EOL> delay , error = self . proxy . check_for_delay ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> error = error . strip ( ) <EOL> expected = ( "<STR_LIT>" , "<STR_LIT>" "<STR_LIT>" ) <EOL> self . assertEqual ( ( delay , error ) , expected ) <EOL> class LimitsViewBuilderV11Test ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . view_builder = views . limits . ViewBuilderV11 ( ) <EOL> self . rate_limits = [ { "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : <NUM_LIT:50> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> } ] <EOL> self . absolute_limits = { "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT:5> } <EOL> def tearDown ( self ) : <EOL> pass <EOL> def test_build_limits ( self ) : <EOL> expected_limits = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { "<STR_LIT:value>" : <NUM_LIT:50> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } ] , <EOL> "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT:5> } } } <EOL> output = self . view_builder . build ( self . rate_limits , <EOL> self . absolute_limits ) <EOL> self . assertDictMatch ( output , expected_limits ) <EOL> def test_build_limits_empty_limits ( self ) : <EOL> expected_limits = { "<STR_LIT>" : { "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } } } <EOL> abs_limits = { } <EOL> rate_limits = [ ] <EOL> output = self . view_builder . build ( rate_limits , abs_limits ) <EOL> self . assertDictMatch ( output , expected_limits ) <EOL> class LimitsXMLSerializationTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . maxDiff = None <EOL> def tearDown ( self ) : <EOL> pass <EOL> def test_index ( self ) : <EOL> serializer = limits . LimitsXMLSerializer ( ) <EOL> fixture = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT:*>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT:value>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT:value>" : <NUM_LIT:50> , <EOL> "<STR_LIT>" : "<STR_LIT:POST>" , <EOL> "<STR_LIT>" : <NUM_LIT:10> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } ] , <EOL> "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT>" : <NUM_LIT> } } } <EOL> output = serializer . serialize ( fixture , '<STR_LIT:index>' ) <EOL> actual = minidom . parseString ( output . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> expected = minidom . parseString ( """<STR_LIT>""" . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected . toxml ( ) , actual . toxml ( ) ) <EOL> def test_index_no_limits ( self ) : <EOL> serializer = limits . LimitsXMLSerializer ( ) <EOL> fixture = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } } } <EOL> output = serializer . serialize ( fixture , '<STR_LIT:index>' ) <EOL> actual = minidom . parseString ( output . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> expected = minidom . parseString ( """<STR_LIT>""" . replace ( "<STR_LIT:U+0020>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( expected . toxml ( ) , actual . toxml ( ) ) </s>
