<s> import functools <EOL> from uuid import uuid4 <EOL> from couchdbkit import ResourceNotFound <EOL> from datetime import datetime <EOL> from nose . tools import nottest <EOL> from casexml . apps . case . models import CommCareCase <EOL> from casexml . apps . phone . models import SyncLog <EOL> from corehq . form_processor . backends . sql . dbaccessors import CaseAccessorSQL , FormAccessorSQL <EOL> from corehq . form_processor . backends . sql . processor import FormProcessorSQL <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface , ProcessedForms <EOL> from corehq . form_processor . parsers . form import process_xform_xml <EOL> from corehq . form_processor . utils . general import should_use_sql_backend <EOL> from couchforms . models import XFormInstance <EOL> from dimagi . utils . couch . database import safe_delete <EOL> from corehq . util . test_utils import unit_testing_only , run_with_multiple_configs , RunConfig <EOL> from corehq . form_processor . models import XFormInstanceSQL , CommCareCaseSQL , CaseTransaction , Attachment <EOL> from django . conf import settings <EOL> class FormProcessorTestUtils ( object ) : <EOL> @ classmethod <EOL> @ unit_testing_only <EOL> def delete_all_cases ( cls , domain = None ) : <EOL> assert CommCareCase . get_db ( ) . dbname . startswith ( '<STR_LIT>' ) <EOL> view_kwargs = { } <EOL> if domain : <EOL> view_kwargs = { <EOL> '<STR_LIT>' : [ domain ] , <EOL> '<STR_LIT>' : [ domain , { } ] , <EOL> } <EOL> cls . _delete_all ( <EOL> CommCareCase . get_db ( ) , <EOL> '<STR_LIT>' , <EOL> ** view_kwargs <EOL> ) <EOL> FormProcessorTestUtils . delete_all_sql_cases ( domain ) <EOL> @ staticmethod <EOL> def delete_all_sql_cases ( domain = None ) : <EOL> CaseAccessorSQL . delete_all_cases ( domain ) <EOL> @ classmethod <EOL> @ unit_testing_only <EOL> def delete_all_xforms ( cls , domain = None , user_id = None ) : <EOL> view = '<STR_LIT>' <EOL> view_kwargs = { } <EOL> if domain and user_id : <EOL> view = '<STR_LIT>' <EOL> view_kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , domain , user_id ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , domain , user_id , { } ] , <EOL> } <EOL> elif domain : <EOL> view_kwargs = { <EOL> '<STR_LIT>' : [ domain ] , <EOL> '<STR_LIT>' : [ domain , { } ] <EOL> } <EOL> cls . _delete_all ( <EOL> XFormInstance . get_db ( ) , <EOL> view , <EOL> ** view_kwargs <EOL> ) <EOL> FormProcessorTestUtils . delete_all_sql_forms ( domain , user_id ) <EOL> @ staticmethod <EOL> def delete_all_sql_forms ( domain = None , user_id = None ) : <EOL> FormAccessorSQL . delete_all_forms ( domain , user_id ) <EOL> @ classmethod <EOL> @ unit_testing_only <EOL> def delete_all_sync_logs ( cls ) : <EOL> cls . _delete_all ( SyncLog . get_db ( ) , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> @ unit_testing_only <EOL> def _delete_all ( db , viewname , ** view_kwargs ) : <EOL> deleted = set ( ) <EOL> for row in db . view ( viewname , reduce = False , ** view_kwargs ) : <EOL> doc_id = row [ '<STR_LIT:id>' ] <EOL> if id not in deleted : <EOL> try : <EOL> safe_delete ( db , doc_id ) <EOL> deleted . add ( doc_id ) <EOL> except ResourceNotFound : <EOL> pass <EOL> run_with_all_backends = functools . partial ( <EOL> run_with_multiple_configs , <EOL> run_configs = [ <EOL> RunConfig ( <EOL> settings = { <EOL> '<STR_LIT>' : getattr ( settings , '<STR_LIT>' , False ) , <EOL> } , <EOL> post_run = lambda * args , ** kwargs : args [ <NUM_LIT:0> ] . tearDown ( ) <EOL> ) , <EOL> RunConfig ( <EOL> settings = { <EOL> '<STR_LIT>' : not getattr ( settings , '<STR_LIT>' , False ) , <EOL> } , <EOL> pre_run = lambda * args , ** kwargs : args [ <NUM_LIT:0> ] . setUp ( ) , <EOL> ) , <EOL> ] <EOL> ) <EOL> @ unit_testing_only <EOL> def post_xform ( instance_xml , attachments = None , domain = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> result = process_xform_xml ( domain , instance_xml , attachments = attachments ) <EOL> with result . get_locked_forms ( ) as xforms : <EOL> FormProcessorInterface ( domain ) . save_processed_models ( xforms ) <EOL> return xforms [ <NUM_LIT:0> ] <EOL> @ nottest <EOL> def create_form_for_test ( domain , case_id = None , attachments = None , save = True ) : <EOL> """<STR_LIT>""" <EOL> from corehq . form_processor . utils import get_simple_form_xml <EOL> form_id = uuid4 ( ) . hex <EOL> user_id = '<STR_LIT>' <EOL> utcnow = datetime . utcnow ( ) <EOL> form_xml = get_simple_form_xml ( form_id , case_id ) <EOL> form = XFormInstanceSQL ( <EOL> form_id = form_id , <EOL> xmlns = '<STR_LIT>' , <EOL> received_on = utcnow , <EOL> user_id = user_id , <EOL> domain = domain <EOL> ) <EOL> attachments = attachments or { } <EOL> attachment_tuples = map ( <EOL> lambda a : Attachment ( name = a [ <NUM_LIT:0> ] , raw_content = a [ <NUM_LIT:1> ] , content_type = a [ <NUM_LIT:1> ] . content_type ) , <EOL> attachments . items ( ) <EOL> ) <EOL> attachment_tuples . append ( Attachment ( '<STR_LIT>' , form_xml , '<STR_LIT>' ) ) <EOL> FormProcessorSQL . store_attachments ( form , attachment_tuples ) <EOL> cases = [ ] <EOL> if case_id : <EOL> case = CommCareCaseSQL ( <EOL> case_id = case_id , <EOL> domain = domain , <EOL> type = '<STR_LIT>' , <EOL> owner_id = user_id , <EOL> opened_on = utcnow , <EOL> modified_on = utcnow , <EOL> modified_by = user_id , <EOL> server_modified_on = utcnow , <EOL> ) <EOL> case . track_create ( CaseTransaction . form_transaction ( case , form ) ) <EOL> cases = [ case ] <EOL> if save : <EOL> FormProcessorSQL . save_processed_models ( ProcessedForms ( form , None ) , cases ) <EOL> return form <EOL> @ unit_testing_only <EOL> def set_case_property_directly ( case , property_name , value ) : <EOL> if should_use_sql_backend ( case . domain ) : <EOL> case . case_json [ property_name ] = value <EOL> else : <EOL> setattr ( case , property_name , value ) </s>
