<s> import json <EOL> import os <EOL> import re <EOL> from django . conf import settings <EOL> from django . contrib . auth . models import Permission <EOL> from django . test import TransactionTestCase <EOL> from django_digest . test import Client as DigestClient <EOL> from tempfile import NamedTemporaryFile <EOL> from django . contrib . auth . models import User <EOL> from django_digest . test import DigestAuth <EOL> from django . contrib . auth import authenticate <EOL> from rest_framework . test import APIRequestFactory <EOL> from onadata . apps . api . models import OrganizationProfile , Project <EOL> from onadata . apps . api . viewsets . metadata_viewset import MetaDataViewSet <EOL> from onadata . apps . api . viewsets . organization_profile_viewset import OrganizationProfileViewSet <EOL> from onadata . apps . api . viewsets . project_viewset import ProjectViewSet <EOL> from onadata . apps . main . models import UserProfile , MetaData <EOL> from onadata . apps . main import tests as main_tests <EOL> from onadata . apps . logger . models import Instance , XForm , Attachment <EOL> from onadata . libs . serializers . project_serializer import ProjectSerializer <EOL> from onadata . apps . logger . views import submission <EOL> class TestAbstractViewSet ( TransactionTestCase ) : <EOL> surveys = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> main_directory = os . path . dirname ( main_tests . __file__ ) <EOL> profile_data = { <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:email>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <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>' <EOL> } <EOL> def setUp ( self ) : <EOL> super ( TestAbstractViewSet , self ) . setUp ( ) <EOL> self . factory = APIRequestFactory ( ) <EOL> self . _login_user_and_profile ( ) <EOL> self . maxDiff = None <EOL> def user_profile_data ( self ) : <EOL> return { <EOL> '<STR_LIT:id>' : self . user . pk , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:username>' : u'<STR_LIT>' , <EOL> '<STR_LIT:name>' : u'<STR_LIT>' , <EOL> '<STR_LIT:email>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : self . user . profile . gravatar , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:user>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : { } , <EOL> } <EOL> def _set_api_permissions ( self , user ) : <EOL> add_userprofile = Permission . objects . get ( <EOL> content_type__app_label = '<STR_LIT>' , content_type__model = '<STR_LIT>' , <EOL> codename = '<STR_LIT>' ) <EOL> user . user_permissions . add ( add_userprofile ) <EOL> def _create_user_profile ( self , extra_post_data = { } ) : <EOL> self . profile_data = dict ( <EOL> self . profile_data . items ( ) + extra_post_data . items ( ) ) <EOL> user , created = User . objects . get_or_create ( <EOL> username = self . profile_data [ '<STR_LIT:username>' ] , <EOL> first_name = self . profile_data [ '<STR_LIT:name>' ] , <EOL> email = self . profile_data [ '<STR_LIT:email>' ] ) <EOL> user . set_password ( self . profile_data [ '<STR_LIT>' ] ) <EOL> user . save ( ) <EOL> new_profile , created = UserProfile . objects . get_or_create ( <EOL> user = user , name = self . profile_data [ '<STR_LIT:name>' ] , <EOL> city = self . profile_data [ '<STR_LIT>' ] , <EOL> country = self . profile_data [ '<STR_LIT>' ] , <EOL> organization = self . profile_data [ '<STR_LIT>' ] , <EOL> home_page = self . profile_data [ '<STR_LIT>' ] , <EOL> twitter = self . profile_data [ '<STR_LIT>' ] , <EOL> require_auth = False <EOL> ) <EOL> return new_profile <EOL> def _login_user_and_profile ( self , extra_post_data = { } ) : <EOL> profile = self . _create_user_profile ( extra_post_data ) <EOL> self . user = profile . user <EOL> self . assertTrue ( <EOL> self . client . login ( username = self . user . username , <EOL> password = self . profile_data [ '<STR_LIT>' ] ) ) <EOL> self . extra = { <EOL> '<STR_LIT>' : '<STR_LIT>' % self . user . auth_token } <EOL> def _org_create ( self ) : <EOL> view = OrganizationProfileViewSet . as_view ( { <EOL> '<STR_LIT>' : '<STR_LIT:list>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> request = self . factory . get ( '<STR_LIT:/>' , ** self . extra ) <EOL> response = view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> data = { <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:name>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:description>' : u'<STR_LIT>' , <EOL> '<STR_LIT:address>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> } <EOL> request = self . factory . post ( <EOL> '<STR_LIT:/>' , data = json . dumps ( data ) , <EOL> content_type = "<STR_LIT:application/json>" , ** self . extra ) <EOL> response = view ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> data [ '<STR_LIT:url>' ] = '<STR_LIT>' <EOL> data [ '<STR_LIT:user>' ] = '<STR_LIT>' <EOL> data [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertDictContainsSubset ( data , response . data ) <EOL> self . company_data = response . data <EOL> self . organization = OrganizationProfile . objects . get ( <EOL> user__username = data [ '<STR_LIT>' ] ) <EOL> def _project_create ( self , project_data = { } , merge = True ) : <EOL> view = ProjectViewSet . as_view ( { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> if merge : <EOL> data = { <EOL> '<STR_LIT:name>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' % self . user . username , <EOL> '<STR_LIT>' : { '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : False <EOL> } <EOL> data . update ( project_data ) <EOL> else : <EOL> data = project_data <EOL> request = self . factory . post ( <EOL> '<STR_LIT:/>' , data = json . dumps ( data ) , <EOL> content_type = "<STR_LIT:application/json>" , ** self . extra ) <EOL> response = view ( request , owner = self . user . username ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . project = Project . objects . filter ( <EOL> name = data [ '<STR_LIT:name>' ] , created_by = self . user ) [ <NUM_LIT:0> ] <EOL> data [ '<STR_LIT:url>' ] = '<STR_LIT>' % self . project . pk <EOL> self . assertDictContainsSubset ( data , response . data ) <EOL> request . user = self . user <EOL> self . project_data = ProjectSerializer ( <EOL> self . project , context = { '<STR_LIT>' : request } ) . data <EOL> def _publish_xls_form_to_project ( self , publish_data = { } , merge = True , <EOL> public = False ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _project_create ( ) <EOL> elif self . project . created_by != self . user : <EOL> self . _project_create ( ) <EOL> view = ProjectViewSet . as_view ( { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> project_id = self . project . pk <EOL> if merge : <EOL> data = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> % self . project . organization . username , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:description>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:title>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' <EOL> } <EOL> data . update ( publish_data ) <EOL> else : <EOL> data = publish_data <EOL> path = os . path . join ( <EOL> settings . ONADATA_DIR , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> with open ( path ) as xls_file : <EOL> post_data = { '<STR_LIT>' : xls_file } <EOL> request = self . factory . post ( '<STR_LIT:/>' , data = post_data , ** self . extra ) <EOL> response = view ( request , pk = project_id ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . xform = XForm . objects . all ( ) . order_by ( '<STR_LIT>' ) . reverse ( ) [ <NUM_LIT:0> ] <EOL> data . update ( { <EOL> '<STR_LIT:url>' : <EOL> '<STR_LIT>' % ( self . xform . pk ) <EOL> } ) <EOL> if public : <EOL> data [ '<STR_LIT>' ] = data [ '<STR_LIT>' ] = True <EOL> self . assertDictContainsSubset ( data , response . data ) <EOL> self . form_data = response . data <EOL> def _add_uuid_to_submission_xml ( self , path , xform ) : <EOL> tmp_file = NamedTemporaryFile ( delete = False ) <EOL> split_xml = None <EOL> with open ( path ) as _file : <EOL> split_xml = re . split ( r'<STR_LIT>' , _file . read ( ) ) <EOL> split_xml [ <NUM_LIT:1> : <NUM_LIT:1> ] = [ <EOL> '<STR_LIT>' % xform . uuid <EOL> ] <EOL> tmp_file . write ( '<STR_LIT>' . join ( split_xml ) ) <EOL> path = tmp_file . name <EOL> tmp_file . close ( ) <EOL> return path <EOL> def _make_submission ( self , path , username = None , add_uuid = False , <EOL> forced_submission_time = None , <EOL> client = None , media_file = None , auth = None ) : <EOL> self . factory = APIRequestFactory ( ) <EOL> if auth is None : <EOL> auth = DigestAuth ( self . profile_data [ '<STR_LIT:username>' ] , <EOL> self . profile_data [ '<STR_LIT>' ] ) <EOL> tmp_file = None <EOL> if add_uuid : <EOL> path = self . _add_uuid_to_submission_xml ( path , self . xform ) <EOL> with open ( path ) as f : <EOL> post_data = { '<STR_LIT>' : f } <EOL> if media_file is not None : <EOL> post_data [ '<STR_LIT>' ] = media_file <EOL> if username is None : <EOL> username = self . user . username <EOL> url_prefix = '<STR_LIT>' % username if username else '<STR_LIT>' <EOL> url = '<STR_LIT>' % url_prefix <EOL> request = self . factory . post ( url , post_data ) <EOL> request . user = authenticate ( username = auth . username , <EOL> password = auth . password ) <EOL> self . response = submission ( request , username = username ) <EOL> if auth and self . response . status_code == <NUM_LIT> : <EOL> request . META . update ( auth ( request . META , self . response ) ) <EOL> self . response = submission ( request , username = username ) <EOL> if forced_submission_time : <EOL> instance = Instance . objects . order_by ( '<STR_LIT>' ) . all ( ) [ <NUM_LIT:0> ] <EOL> instance . date_created = forced_submission_time <EOL> instance . save ( ) <EOL> instance . parsed_instance . save ( ) <EOL> if add_uuid : <EOL> os . unlink ( tmp_file . name ) <EOL> def _make_submissions ( self , username = None , add_uuid = False , <EOL> should_store = True ) : <EOL> """<STR_LIT>""" <EOL> paths = [ os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , s , s + '<STR_LIT>' ) for s in self . surveys ] <EOL> pre_count = Instance . objects . count ( ) <EOL> auth = DigestAuth ( self . profile_data [ '<STR_LIT:username>' ] , <EOL> self . profile_data [ '<STR_LIT>' ] ) <EOL> for path in paths : <EOL> self . _make_submission ( path , username , add_uuid , auth = auth ) <EOL> post_count = pre_count + len ( self . surveys ) if should_store else pre_count <EOL> self . assertEqual ( Instance . objects . count ( ) , post_count ) <EOL> self . assertEqual ( self . xform . instances . count ( ) , post_count ) <EOL> xform = XForm . objects . get ( pk = self . xform . pk ) <EOL> self . assertEqual ( xform . num_of_submissions , post_count ) <EOL> self . assertEqual ( xform . user . profile . num_of_submissions , post_count ) <EOL> def _submit_transport_instance_w_attachment ( self , <EOL> survey_at = <NUM_LIT:0> , <EOL> media_file = None ) : <EOL> s = self . surveys [ survey_at ] <EOL> if not media_file : <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> self . _make_submission ( os . path . join ( <EOL> self . main_directory , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , s , s + '<STR_LIT>' ) , media_file = f ) <EOL> attachment = Attachment . objects . all ( ) . reverse ( ) [ <NUM_LIT:0> ] <EOL> self . attachment = attachment <EOL> def _post_form_metadata ( self , data , test = True ) : <EOL> count = MetaData . objects . count ( ) <EOL> view = MetaDataViewSet . as_view ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> request = self . factory . post ( '<STR_LIT:/>' , data , ** self . extra ) <EOL> response = view ( request ) <EOL> if test : <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> another_count = MetaData . objects . count ( ) <EOL> self . assertEqual ( another_count , count + <NUM_LIT:1> ) <EOL> self . metadata = MetaData . objects . get ( pk = response . data [ '<STR_LIT:id>' ] ) <EOL> self . metadata_data = response . data <EOL> return response <EOL> def _add_form_metadata ( self , xform , data_type , data_value , path = None ) : <EOL> data = { <EOL> '<STR_LIT>' : data_type , <EOL> '<STR_LIT>' : data_value , <EOL> '<STR_LIT>' : xform . pk <EOL> } <EOL> if path and data_value : <EOL> with open ( path ) as media_file : <EOL> data . update ( { <EOL> '<STR_LIT>' : media_file , <EOL> } ) <EOL> self . _post_form_metadata ( data ) <EOL> else : <EOL> self . _post_form_metadata ( data ) <EOL> def _get_digest_client ( self ) : <EOL> self . user . profile . require_auth = True <EOL> self . user . profile . save ( ) <EOL> client = DigestClient ( ) <EOL> client . set_authorization ( self . profile_data [ '<STR_LIT:username>' ] , <EOL> self . profile_data [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> return client </s>
