<s> from django . test . client import Client <EOL> from couchforms . util import spoof_submission <EOL> import uuid <EOL> from corehq . apps . accounting . tests . utils import DomainSubscriptionMixin <EOL> from corehq . apps . receiverwrapper . util import get_submit_url <EOL> from corehq . apps . domain . shortcuts import create_domain <EOL> from django . core . urlresolvers import reverse <EOL> from corehq . apps . users . models import WebUser <EOL> from couchforms . models import XFormInstance <EOL> from couchexport . export import ExportConfiguration <EOL> import time <EOL> from couchexport . models import ExportSchema <EOL> from corehq . apps . accounting . tests import BaseAccountingTest <EOL> from corehq . apps . accounting . models import SoftwarePlanEdition <EOL> from corehq . apps . domain . models import Domain <EOL> FORM_TEMPLATE = """<STR_LIT>""" <EOL> DOMAIN = "<STR_LIT:test>" <EOL> def get_form ( ) : <EOL> return FORM_TEMPLATE % { "<STR_LIT>" : uuid . uuid4 ( ) . hex } <EOL> def _submit_form ( f = None , domain = DOMAIN ) : <EOL> if f is None : <EOL> f = get_form ( ) <EOL> url = get_submit_url ( domain ) <EOL> return spoof_submission ( url , f ) <EOL> def get_export_response ( client , previous = "<STR_LIT>" , include_errors = False , domain = DOMAIN ) : <EOL> return client . get ( <EOL> reverse ( "<STR_LIT>" , args = [ domain ] ) , <EOL> { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : previous , <EOL> "<STR_LIT>" : include_errors , <EOL> "<STR_LIT>" : "<STR_LIT:html>" , <EOL> "<STR_LIT>" : False <EOL> } <EOL> ) <EOL> def _content ( streaming_response ) : <EOL> return '<STR_LIT>' . join ( streaming_response . streaming_content ) <EOL> class ExportTest ( BaseAccountingTest , DomainSubscriptionMixin ) : <EOL> def _clear_docs ( self ) : <EOL> config = ExportConfiguration ( XFormInstance . get_db ( ) , <EOL> [ DOMAIN , "<STR_LIT>" ] ) <EOL> for form in config . get_docs ( ) : <EOL> XFormInstance . wrap ( form ) . delete ( ) <EOL> def setUp ( self ) : <EOL> self . _clear_docs ( ) <EOL> self . domain = create_domain ( DOMAIN ) <EOL> self . setup_subscription ( self . domain . name , SoftwarePlanEdition . ADVANCED ) <EOL> self . couch_user = WebUser . create ( None , "<STR_LIT:test>" , "<STR_LIT>" ) <EOL> self . couch_user . add_domain_membership ( DOMAIN , is_admin = True ) <EOL> self . couch_user . save ( ) <EOL> def tearDown ( self ) : <EOL> self . couch_user . delete ( ) <EOL> self . _clear_docs ( ) <EOL> self . teardown_subscription ( ) <EOL> super ( ExportTest , self ) . tearDown ( ) <EOL> def testExportTokenMigration ( self ) : <EOL> c = Client ( ) <EOL> c . login ( ** { '<STR_LIT:username>' : '<STR_LIT:test>' , '<STR_LIT:password>' : '<STR_LIT>' } ) <EOL> _submit_form ( ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> resp = get_export_response ( c ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( _content ( resp ) is not None ) <EOL> self . assertTrue ( "<STR_LIT>" in resp ) <EOL> prev_token = resp [ "<STR_LIT>" ] <EOL> prev_checkpoint = ExportSchema . get ( prev_token ) <EOL> assert prev_checkpoint . timestamp <EOL> def testExportTokens ( self ) : <EOL> c = Client ( ) <EOL> c . login ( ** { '<STR_LIT:username>' : '<STR_LIT:test>' , '<STR_LIT:password>' : '<STR_LIT>' } ) <EOL> resp = get_export_response ( c ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> _submit_form ( ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> resp = get_export_response ( c ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( _content ( resp ) is not None ) <EOL> self . assertTrue ( "<STR_LIT>" in resp ) <EOL> prev_token = resp [ "<STR_LIT>" ] <EOL> resp = get_export_response ( c , prev_token ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> _submit_form ( ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> resp = get_export_response ( c , prev_token ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( _content ( resp ) is not None ) <EOL> self . assertTrue ( "<STR_LIT>" in resp ) <EOL> def testExportFilter ( self ) : <EOL> c = Client ( ) <EOL> c . login ( ** { '<STR_LIT:username>' : '<STR_LIT:test>' , '<STR_LIT:password>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( <NUM_LIT> , get_export_response ( c ) . status_code ) <EOL> f = get_form ( ) <EOL> _submit_form ( f ) <EOL> resp = get_export_response ( c ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> initial_content = _content ( resp ) <EOL> _submit_form ( f ) <EOL> resp = get_export_response ( c ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> content = _content ( resp ) <EOL> self . assertEqual ( initial_content . count ( "<STR_LIT>" ) , <EOL> content . count ( "<STR_LIT>" ) ) <EOL> resp = get_export_response ( c , include_errors = True ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( len ( _content ( resp ) ) > len ( initial_content ) ) <EOL> def test_no_subscription ( self ) : <EOL> """<STR_LIT>""" <EOL> community_domain = Domain . get_or_create_with_name ( '<STR_LIT>' , is_active = True ) <EOL> new_user = WebUser . create ( community_domain . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> new_user . save ( ) <EOL> c = Client ( ) <EOL> c . login ( ** { '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT>' } ) <EOL> f = get_form ( ) <EOL> _submit_form ( f , community_domain . name ) <EOL> resp = get_export_response ( c , domain = community_domain . name ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> community_domain . delete ( ) <EOL> new_user . delete ( ) </s>
