<s> import datetime , json , mock , unittest <EOL> from django . conf . urls import patterns , url <EOL> from django . test . utils import override_settings <EOL> from rest_framework import status <EOL> from rest_framework . test import APITestCase <EOL> @ override_settings ( <EOL> AWS_UPLOAD_SECRET_ACCESS_KEY = '<STR_LIT>' , <EOL> AWS_UPLOAD_BUCKET = '<STR_LIT>' , <EOL> AWS_UPLOAD_PREFIX_FUNC = lambda x : '<STR_LIT>' , <EOL> AWS_STORAGE_BUCKET_NAME = '<STR_LIT>' , <EOL> APPEND_SLASH = False <EOL> ) <EOL> class TestCompletionViewWithoutAccessControl ( APITestCase ) : <EOL> '''<STR_LIT>''' <EOL> from drf_to_s3 . views import fine_uploader_views <EOL> urls = patterns ( '<STR_LIT>' , <EOL> url ( r'<STR_LIT>' , fine_uploader_views . FineUploadCompletionView . as_view ( ) ) , <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_upload_notification_returns_success ( self , copy ) : <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> self . assertEquals ( len ( resp . content ) , <NUM_LIT:0> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_upload_notification_copies_to_new_key ( self , uuid4 , copy ) : <EOL> uuid4 . return_value = new_key = '<STR_LIT>' <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . client . post ( '<STR_LIT>' , notification ) <EOL> copy . assert_called_once_with ( <EOL> src_bucket = notification [ '<STR_LIT>' ] , <EOL> src_key = notification [ '<STR_LIT:key>' ] , <EOL> dst_bucket = '<STR_LIT>' , <EOL> dst_key = new_key <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_upload_notification_preserves_extension_for_new_key ( self , uuid4 , copy ) : <EOL> uuid4 . return_value = new_key = '<STR_LIT>' <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . client . post ( '<STR_LIT>' , notification ) <EOL> copy . assert_called_once_with ( <EOL> src_bucket = notification [ '<STR_LIT>' ] , <EOL> src_key = notification [ '<STR_LIT:key>' ] , <EOL> dst_bucket = '<STR_LIT>' , <EOL> dst_key = new_key + '<STR_LIT>' <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_upload_notification_returns_error_for_nonexistent_key ( self , copy ) : <EOL> from drf_to_s3 import s3 <EOL> copy . side_effect = s3 . ObjectNotFoundException <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> content = json . loads ( resp . content ) <EOL> self . assertEquals ( content [ '<STR_LIT:error>' ] , '<STR_LIT>' ) <EOL> def test_that_upload_notification_returns_error_for_invalid_data ( self ) : <EOL> notification = { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> content = json . loads ( resp . content ) <EOL> expected_error = '<STR_LIT>' <EOL> self . assertEquals ( content [ '<STR_LIT:error>' ] , expected_error ) <EOL> @ override_settings ( <EOL> AWS_UPLOAD_SECRET_ACCESS_KEY = '<STR_LIT>' , <EOL> AWS_UPLOAD_BUCKET = '<STR_LIT>' , <EOL> AWS_STORAGE_BUCKET_NAME = '<STR_LIT>' , <EOL> APPEND_SLASH = False <EOL> ) <EOL> class TestCompletionViewSessionAuth ( APITestCase ) : <EOL> from drf_to_s3 . views import fine_uploader_views <EOL> urls = patterns ( '<STR_LIT>' , <EOL> url ( r'<STR_LIT>' , fine_uploader_views . FineUploadCompletionView . as_view ( ) ) , <EOL> ) <EOL> def setUp ( self ) : <EOL> from . util import get_user_model <EOL> self . username = '<STR_LIT>' <EOL> self . password = '<STR_LIT>' <EOL> user = get_user_model ( ) . objects . create_user ( <EOL> username = self . username , <EOL> password = self . password <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_upload_notification_with_hashed_session_key_returns_success ( self , copy ) : <EOL> self . client . login ( <EOL> username = self . username , <EOL> password = self . password <EOL> ) <EOL> prefix = self . username <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : prefix + '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> self . assertEquals ( len ( resp . content ) , <NUM_LIT:0> ) <EOL> def test_that_upload_notification_without_prefix_fails ( self ) : <EOL> self . client . login ( <EOL> username = self . username , <EOL> password = self . password <EOL> ) <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> content = json . loads ( resp . content ) <EOL> self . assertTrue ( content [ '<STR_LIT:error>' ] . startswith ( '<STR_LIT>' ) ) <EOL> def test_that_upload_notification_without_login_fails ( self ) : <EOL> prefix = self . username <EOL> notification = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:key>' : prefix + '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> content = json . loads ( resp . content ) <EOL> self . assertEquals ( content [ '<STR_LIT:error>' ] , '<STR_LIT>' ) <EOL> @ override_settings ( <EOL> AWS_UPLOAD_SECRET_ACCESS_KEY = '<STR_LIT>' , <EOL> AWS_UPLOAD_BUCKET = '<STR_LIT>' , <EOL> AWS_STORAGE_BUCKET_NAME = '<STR_LIT>' , <EOL> APPEND_SLASH = False <EOL> ) <EOL> class TestCompletionViewSessionAuth ( APITestCase ) : <EOL> from drf_to_s3 . views import api_client_views <EOL> urls = patterns ( '<STR_LIT>' , <EOL> url ( r'<STR_LIT>' , api_client_views . APIUploadCompletionView . as_view ( ) ) <EOL> ) <EOL> def setUp ( self ) : <EOL> from . util import get_user_model <EOL> self . username = '<STR_LIT>' <EOL> self . password = '<STR_LIT>' <EOL> user = get_user_model ( ) . objects . create_user ( <EOL> username = self . username , <EOL> password = self . password <EOL> ) <EOL> self . client . login ( <EOL> username = self . username , <EOL> password = self . password <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_api_upload_notification_returns_success ( self , copy ) : <EOL> prefix = self . username <EOL> notification = { <EOL> '<STR_LIT:key>' : prefix + '<STR_LIT>' , <EOL> '<STR_LIT:filename>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_200_OK ) <EOL> self . assertEquals ( len ( resp . content ) , <NUM_LIT:0> ) <EOL> def test_that_api_upload_notification_returns_error_for_invalid_data ( self ) : <EOL> notification = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> content = json . loads ( resp . content ) <EOL> self . assertEquals ( len ( content . keys ( ) ) , <NUM_LIT:2> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_that_api_upload_notification_returns_error_for_nonexistent_key ( self , copy ) : <EOL> from drf_to_s3 import s3 <EOL> copy . side_effect = s3 . ObjectNotFoundException <EOL> prefix = self . username <EOL> notification = { <EOL> '<STR_LIT:key>' : prefix + '<STR_LIT>' , <EOL> '<STR_LIT:filename>' : '<STR_LIT>' , <EOL> } <EOL> resp = self . client . post ( '<STR_LIT>' , notification ) <EOL> self . assertEquals ( resp . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> content = json . loads ( resp . content ) <EOL> self . assertEquals ( content [ '<STR_LIT>' ] , '<STR_LIT>' ) </s>
