<s> import mock <EOL> from six import text_type <EOL> from uuid import UUID <EOL> from pyramid . httpexceptions import HTTPBadRequest <EOL> from cliquet . errors import ERRORS , http_error <EOL> from cliquet . storage import exceptions as storage_exceptions <EOL> from cliquet . tests . support import FormattedErrorMixin <EOL> from cliquet . utils import hmac_digest <EOL> from kinto . tests . support import ( BaseWebTest , unittest , get_user_headers , <EOL> MINIMALIST_RECORD ) <EOL> class DefaultBucketViewTest ( FormattedErrorMixin , BaseWebTest , <EOL> unittest . TestCase ) : <EOL> bucket_url = '<STR_LIT>' <EOL> collection_url = '<STR_LIT>' <EOL> def test_default_bucket_exists_and_has_user_id ( self ) : <EOL> bucket = self . app . get ( self . bucket_url , headers = self . headers ) <EOL> result = bucket . json <EOL> settings = self . app . app . registry . settings <EOL> hmac_secret = settings [ '<STR_LIT>' ] <EOL> bucket_id = hmac_digest ( hmac_secret , self . principal ) [ : <NUM_LIT:32> ] <EOL> self . assertEqual ( result [ '<STR_LIT:data>' ] [ '<STR_LIT:id>' ] , text_type ( UUID ( bucket_id ) ) ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] , [ self . principal ] ) <EOL> def test_default_bucket_can_still_be_explicitly_created ( self ) : <EOL> bucket = { '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> resp = self . app . put_json ( self . bucket_url , bucket , headers = self . headers ) <EOL> result = resp . json <EOL> self . assertIn ( '<STR_LIT>' , result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_default_bucket_collections_are_automatically_created ( self ) : <EOL> self . app . get ( self . collection_url , headers = self . headers , status = <NUM_LIT:200> ) <EOL> def test_adding_a_task_for_bob_doesnt_add_it_for_alice ( self ) : <EOL> record = MINIMALIST_RECORD . copy ( ) <EOL> resp = self . app . post_json ( self . collection_url + '<STR_LIT>' , <EOL> record , headers = get_user_headers ( '<STR_LIT>' ) ) <EOL> record_id = self . collection_url + '<STR_LIT>' + resp . json [ '<STR_LIT:data>' ] [ '<STR_LIT:id>' ] <EOL> resp = self . app . get ( record_id , headers = get_user_headers ( '<STR_LIT>' ) , <EOL> status = <NUM_LIT> ) <EOL> def test_unauthenticated_bucket_access_raises_json_401 ( self ) : <EOL> resp = self . app . get ( self . bucket_url , status = <NUM_LIT> ) <EOL> self . assertEquals ( resp . json [ '<STR_LIT:message>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_bucket_id_is_an_uuid_with_dashes ( self ) : <EOL> bucket = self . app . get ( self . bucket_url , headers = self . headers ) <EOL> bucket_id = bucket . json [ '<STR_LIT:data>' ] [ '<STR_LIT:id>' ] <EOL> self . assertIn ( '<STR_LIT:->' , bucket_id ) <EOL> try : <EOL> UUID ( bucket_id ) <EOL> except ValueError : <EOL> self . fail ( '<STR_LIT>' % bucket_id ) <EOL> def test_second_call_on_default_bucket_doesnt_raise_a_412 ( self ) : <EOL> self . app . get ( self . bucket_url , headers = self . headers ) <EOL> self . app . get ( self . bucket_url , headers = self . headers ) <EOL> def test_second_call_on_default_bucket_collection_doesnt_raise_a_412 ( self ) : <EOL> self . app . get ( self . collection_url , headers = self . headers ) <EOL> self . app . get ( self . collection_url , headers = self . headers ) <EOL> def test_querystring_parameters_are_taken_into_account ( self ) : <EOL> self . app . get ( self . collection_url + '<STR_LIT>' , <EOL> headers = self . headers , <EOL> status = <NUM_LIT> ) <EOL> def test_option_is_possible_without_authentication_for_default ( self ) : <EOL> headers = '<STR_LIT>' <EOL> self . app . options ( self . collection_url + '<STR_LIT>' , <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT>' : headers } ) <EOL> def test_cors_headers_are_provided_on_errors ( self ) : <EOL> resp = self . app . post_json ( self . collection_url + '<STR_LIT>' , <EOL> MINIMALIST_RECORD , <EOL> headers = self . headers ) <EOL> current = resp . json [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] <EOL> headers = self . headers . copy ( ) <EOL> headers . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( '<STR_LIT>' % current ) . encode ( '<STR_LIT:utf-8>' ) <EOL> } ) <EOL> resp = self . app . get ( self . collection_url + '<STR_LIT>' , <EOL> headers = headers , status = <NUM_LIT> ) <EOL> self . assertIn ( '<STR_LIT>' , resp . headers ) <EOL> self . assertIn ( '<STR_LIT>' , resp . headers [ '<STR_LIT>' ] ) <EOL> def test_etag_is_present_and_exposed_in_304_error ( self ) : <EOL> resp = self . app . post_json ( self . collection_url + '<STR_LIT>' , <EOL> MINIMALIST_RECORD , <EOL> headers = self . headers ) <EOL> current = resp . json [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] <EOL> headers = self . headers . copy ( ) <EOL> headers . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( '<STR_LIT>' % current ) . encode ( '<STR_LIT:utf-8>' ) <EOL> } ) <EOL> resp = self . app . get ( self . collection_url + '<STR_LIT>' , <EOL> headers = headers , status = <NUM_LIT> ) <EOL> self . assertIn ( '<STR_LIT>' , resp . headers ) <EOL> self . assertIn ( '<STR_LIT>' , resp . headers ) <EOL> self . assertIn ( '<STR_LIT>' , resp . headers [ '<STR_LIT>' ] ) <EOL> def test_bucket_id_starting_with_default_can_still_be_created ( self ) : <EOL> resp = self . app . put ( <EOL> self . bucket_url . replace ( '<STR_LIT:default>' , '<STR_LIT>' ) , <EOL> headers = self . headers , status = <NUM_LIT> ) <EOL> bucket_id = resp . json [ '<STR_LIT:data>' ] [ '<STR_LIT:id>' ] <EOL> self . assertEquals ( bucket_id , '<STR_LIT>' ) <EOL> collection_url = '<STR_LIT>' <EOL> self . app . put ( <EOL> collection_url , <EOL> headers = self . headers , <EOL> status = <NUM_LIT> ) <EOL> resp = self . app . get ( '<STR_LIT>' , <EOL> headers = self . headers ) <EOL> self . assertEquals ( resp . json [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] , '<STR_LIT:default>' ) <EOL> def test_default_bucket_objects_are_checked_only_once_in_batch ( self ) : <EOL> batch = { '<STR_LIT>' : [ ] } <EOL> nb_create = <NUM_LIT> <EOL> for i in range ( nb_create ) : <EOL> request = { '<STR_LIT>' : '<STR_LIT:POST>' , <EOL> '<STR_LIT:path>' : self . collection_url + '<STR_LIT>' , <EOL> '<STR_LIT:body>' : MINIMALIST_RECORD } <EOL> batch [ '<STR_LIT>' ] . append ( request ) <EOL> with mock . patch . object ( self . storage , '<STR_LIT>' , <EOL> wraps = self . storage . create ) as patched : <EOL> self . app . post_json ( '<STR_LIT>' , batch , headers = self . headers ) <EOL> self . assertEqual ( patched . call_count , nb_create + <NUM_LIT:2> ) <EOL> def test_parent_collection_is_taken_from_the_one_created_in_batch ( self ) : <EOL> batch = { '<STR_LIT>' : [ ] } <EOL> nb_create = <NUM_LIT> <EOL> for i in range ( nb_create ) : <EOL> request = { '<STR_LIT>' : '<STR_LIT:POST>' , <EOL> '<STR_LIT:path>' : self . collection_url + '<STR_LIT>' , <EOL> '<STR_LIT:body>' : MINIMALIST_RECORD } <EOL> batch [ '<STR_LIT>' ] . append ( request ) <EOL> with mock . patch . object ( self . storage , '<STR_LIT>' , <EOL> wraps = self . storage . get ) as patched : <EOL> self . app . post_json ( '<STR_LIT>' , batch , headers = self . headers ) <EOL> self . assertEqual ( patched . call_count , <NUM_LIT:0> ) <EOL> def test_parent_collection_is_taken_from_the_one_checked_in_batch ( self ) : <EOL> self . app . put ( self . collection_url , headers = self . headers , status = <NUM_LIT> ) <EOL> batch = { '<STR_LIT>' : [ ] } <EOL> nb_create = <NUM_LIT> <EOL> for i in range ( nb_create ) : <EOL> request = { '<STR_LIT>' : '<STR_LIT:POST>' , <EOL> '<STR_LIT:path>' : self . collection_url + '<STR_LIT>' , <EOL> '<STR_LIT:body>' : MINIMALIST_RECORD } <EOL> batch [ '<STR_LIT>' ] . append ( request ) <EOL> with mock . patch . object ( self . storage , '<STR_LIT>' , <EOL> wraps = self . storage . get ) as patched : <EOL> self . app . post_json ( '<STR_LIT>' , batch , headers = self . headers ) <EOL> self . assertEqual ( patched . call_count , <NUM_LIT:0> ) <EOL> def test_collection_id_is_validated ( self ) : <EOL> collection_url = '<STR_LIT>' <EOL> self . app . get ( collection_url , headers = self . headers , status = <NUM_LIT> ) <EOL> def test_collection_id_does_not_support_unicode ( self ) : <EOL> collection_url = '<STR_LIT>' <EOL> self . app . get ( collection_url , headers = self . headers , status = <NUM_LIT> ) <EOL> def test_405_is_a_valid_formatted_error ( self ) : <EOL> response = self . app . post ( self . collection_url , <EOL> headers = self . headers , status = <NUM_LIT> ) <EOL> self . assertFormattedError ( <EOL> response , <NUM_LIT> , ERRORS . METHOD_NOT_ALLOWED , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> def test_formatted_error_are_passed_through ( self ) : <EOL> response = http_error ( HTTPBadRequest ( ) , <EOL> errno = ERRORS . INVALID_PARAMETERS , <EOL> message = '<STR_LIT>' ) <EOL> with mock . patch . object ( self . storage , '<STR_LIT>' ) as mocked : <EOL> mocked . side_effect = [ <EOL> { "<STR_LIT:id>" : "<STR_LIT:abc>" , "<STR_LIT>" : <NUM_LIT> } , <EOL> { "<STR_LIT:id>" : "<STR_LIT:abc>" , "<STR_LIT>" : <NUM_LIT> } , <EOL> response <EOL> ] <EOL> resp = self . app . post ( self . collection_url + '<STR_LIT>' , <EOL> headers = self . headers , <EOL> status = <NUM_LIT> ) <EOL> self . assertEqual ( resp . body , response . body ) <EOL> class HelloViewTest ( BaseWebTest , unittest . TestCase ) : <EOL> def test_returns_bucket_id_and_url_if_authenticated ( self ) : <EOL> response = self . app . get ( '<STR_LIT:/>' , headers = self . headers ) <EOL> self . assertEqual ( response . json [ '<STR_LIT:user>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_flush_capability_if_enabled ( self ) : <EOL> resp = self . app . get ( '<STR_LIT:/>' ) <EOL> capabilities = resp . json [ '<STR_LIT>' ] <EOL> self . assertIn ( '<STR_LIT>' , capabilities ) <EOL> _events = [ ] <EOL> def load_from_config ( config , prefix ) : <EOL> def listener ( event ) : <EOL> _events . append ( event ) <EOL> return listener <EOL> class EventsTest ( BaseWebTest , unittest . TestCase ) : <EOL> def tearDown ( self ) : <EOL> super ( EventsTest , self ) . tearDown ( ) <EOL> del _events [ : ] <EOL> def get_app_settings ( self , extra = None ) : <EOL> settings = super ( EventsTest , self ) . get_app_settings ( extra ) <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT>' , <EOL> settings [ '<STR_LIT>' ] = ( <EOL> '<STR_LIT>' ) <EOL> return settings <EOL> def test_an_event_is_sent_on_implicit_bucket_creation ( self ) : <EOL> bucket_url = '<STR_LIT>' <EOL> self . app . get ( bucket_url , headers = self . headers ) <EOL> assert len ( _events ) == <NUM_LIT:1> <EOL> payload = _events [ - <NUM_LIT:1> ] . payload <EOL> assert payload [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert payload [ '<STR_LIT:action>' ] == '<STR_LIT>' <EOL> def test_an_event_is_sent_on_implicit_collection_creation ( self ) : <EOL> collection_url = '<STR_LIT>' <EOL> self . app . get ( collection_url , headers = self . headers ) <EOL> assert len ( _events ) == <NUM_LIT:2> <EOL> payload = _events [ - <NUM_LIT:1> ] . payload <EOL> assert payload [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert payload [ '<STR_LIT:action>' ] == '<STR_LIT>' <EOL> def test_events_sent_on_bucket_and_collection_creation ( self ) : <EOL> records_uri = '<STR_LIT>' <EOL> self . app . get ( records_uri , headers = self . headers ) <EOL> assert len ( _events ) == <NUM_LIT:2> <EOL> class ReadonlyDefaultBucket ( BaseWebTest , unittest . TestCase ) : <EOL> def get_app_settings ( self , extras = None ) : <EOL> settings = super ( ReadonlyDefaultBucket , self ) . get_app_settings ( extras ) <EOL> settings [ '<STR_LIT>' ] = True <EOL> return settings <EOL> def test_implicit_creation_is_rejected ( self ) : <EOL> self . app . get ( '<STR_LIT>' , headers = self . headers , status = <NUM_LIT> ) <EOL> class BackendErrorTest ( BaseWebTest , unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( BackendErrorTest , self ) . setUp ( ) <EOL> self . patcher = mock . patch . object ( <EOL> self . storage , '<STR_LIT>' , <EOL> side_effect = storage_exceptions . BackendError ( ) ) <EOL> self . addCleanup ( self . patcher . stop ) <EOL> def test_implicit_bucket_creation_raises_503_if_backend_fails ( self ) : <EOL> self . patcher . start ( ) <EOL> self . app . get ( '<STR_LIT>' , headers = self . headers , status = <NUM_LIT> ) <EOL> def test_implicit_collection_creation_raises_503_if_backend_fails ( self ) : <EOL> self . app . get ( '<STR_LIT>' , headers = self . headers ) <EOL> self . patcher . start ( ) <EOL> self . app . get ( '<STR_LIT>' , <EOL> headers = self . headers , status = <NUM_LIT> ) </s>
