<s> import os <EOL> from django . core . files . uploadedfile import InMemoryUploadedFile <EOL> from django . test import TransactionTestCase <EOL> from django_digest . test import DigestAuth <EOL> from django . contrib . auth . models import AnonymousUser <EOL> import simplejson as json <EOL> from onadata . apps . api . tests . viewsets . test_abstract_viewset import TestAbstractViewSet <EOL> from onadata . apps . api . viewsets . xform_submission_api import XFormSubmissionApi <EOL> from onadata . apps . logger . models import Attachment <EOL> from onadata . libs . permissions import DataEntryRole <EOL> class TestXFormSubmissionApi ( TestAbstractViewSet , TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( self . __class__ , self ) . setUp ( ) <EOL> self . view = XFormSubmissionApi . as_view ( { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } ) <EOL> self . _publish_xls_form_to_project ( ) <EOL> def test_post_submission_anonymous ( self ) : <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( <EOL> '<STR_LIT>' % self . user . username , data ) <EOL> request . user = AnonymousUser ( ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' <EOL> % self . user . username ) <EOL> def test_post_submission_authenticated ( self ) : <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> sf . seek ( <NUM_LIT:0> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_post_submission_uuid_other_user_username_not_provided ( self ) : <EOL> alice_data = { '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT:email>' : '<STR_LIT>' } <EOL> self . _create_user_profile ( alice_data ) <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> path = self . _add_uuid_to_submission_xml ( path , self . xform ) <EOL> with open ( path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> sf . seek ( <NUM_LIT:0> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_post_submission_authenticated_json ( self ) : <EOL> path = os . path . join ( <EOL> os . path . dirname ( os . path . abspath ( __file__ ) ) , <EOL> '<STR_LIT:..>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> with open ( path ) as f : <EOL> data = json . loads ( f . read ( ) ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT:application/json>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_post_submission_authenticated_json_with_geo ( self ) : <EOL> path = os . path . join ( <EOL> os . path . dirname ( os . path . abspath ( __file__ ) ) , <EOL> '<STR_LIT:..>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> with open ( path ) as f : <EOL> data = json . loads ( f . read ( ) ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT:application/json>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_post_submission_authenticated_bad_json ( self ) : <EOL> path = os . path . join ( <EOL> os . path . dirname ( os . path . abspath ( __file__ ) ) , <EOL> '<STR_LIT:..>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> with open ( path ) as f : <EOL> data = json . loads ( f . read ( ) ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT:application/json>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_post_submission_require_auth ( self ) : <EOL> self . user . profile . require_auth = True <EOL> self . user . profile . save ( ) <EOL> count = Attachment . objects . count ( ) <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> sf . seek ( <NUM_LIT:0> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> self . assertEqual ( count + <NUM_LIT:1> , Attachment . objects . count ( ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( <EOL> response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( response . has_header ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response [ '<STR_LIT:Content-Type>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_post_submission_require_auth_anonymous_user ( self ) : <EOL> self . user . profile . require_auth = True <EOL> self . user . profile . save ( ) <EOL> count = Attachment . objects . count ( ) <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( count , Attachment . objects . count ( ) ) <EOL> def test_post_submission_require_auth_other_user ( self ) : <EOL> self . user . profile . require_auth = True <EOL> self . user . profile . save ( ) <EOL> alice_data = { '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT:email>' : '<STR_LIT>' } <EOL> self . _create_user_profile ( alice_data ) <EOL> count = Attachment . objects . count ( ) <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( count , Attachment . objects . count ( ) ) <EOL> sf . seek ( <NUM_LIT:0> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertContains ( <EOL> response , <EOL> '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> def test_post_submission_require_auth_data_entry_role ( self ) : <EOL> self . user . profile . require_auth = True <EOL> self . user . profile . save ( ) <EOL> alice_data = { '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT:email>' : '<STR_LIT>' } <EOL> alice_profile = self . _create_user_profile ( alice_data ) <EOL> DataEntryRole . add ( alice_profile . user , self . xform ) <EOL> count = Attachment . objects . count ( ) <EOL> s = self . surveys [ <NUM_LIT:0> ] <EOL> media_file = "<STR_LIT>" <EOL> path = os . path . join ( self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , media_file ) <EOL> with open ( path ) as f : <EOL> f = InMemoryUploadedFile ( f , '<STR_LIT>' , media_file , '<STR_LIT>' , <EOL> os . path . getsize ( path ) , None ) <EOL> submission_path = os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) <EOL> with open ( submission_path ) as sf : <EOL> data = { '<STR_LIT>' : sf , '<STR_LIT>' : f } <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( count , Attachment . objects . count ( ) ) <EOL> sf . seek ( <NUM_LIT:0> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request , username = self . user . username ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) <EOL> def test_post_submission_json_without_submission_key ( self ) : <EOL> data = { "<STR_LIT:id>" : "<STR_LIT>" } <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> response = self . view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> request = self . factory . post ( '<STR_LIT>' , data , format = '<STR_LIT>' ) <EOL> auth = DigestAuth ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . META . update ( auth ( request . META , response ) ) <EOL> response = self . view ( request ) <EOL> self . assertContains ( response , '<STR_LIT>' , <EOL> status_code = <NUM_LIT> ) </s>
