<s> import datetime <EOL> import uuid <EOL> from contextlib import contextmanager <EOL> from couchdbkit import ResourceNotFound <EOL> from django . test import SimpleTestCase , TestCase <EOL> from mock import patch <EOL> from casexml . apps . case . mock import CaseBlock <EOL> from casexml . apps . case . models import CommCareCase <EOL> from casexml . apps . case . sharedmodels import CommCareCaseAttachment <EOL> from casexml . apps . case . tests . util import delete_all_cases , delete_all_xforms <EOL> from casexml . apps . case . xml import V2 <EOL> from corehq . apps . app_manager . tests . util import TestXmlMixin <EOL> from corehq . apps . hqcase . tasks import explode_cases <EOL> from corehq . apps . hqcase . utils import make_creating_casexml , submit_case_blocks <EOL> from corehq . apps . users . models import CommCareUser <EOL> from corehq . apps . domain . models import Domain <EOL> from corehq . form_processor . interfaces . dbaccessors import CaseAccessors <EOL> from corehq . form_processor . tests . utils import run_with_all_backends <EOL> TESTS = ( <EOL> ( <EOL> CommCareCase ( <EOL> _id = '<STR_LIT>' , <EOL> domain = '<STR_LIT:foo>' , <EOL> type = '<STR_LIT>' , <EOL> closed = False , <EOL> user_id = '<STR_LIT>' , <EOL> modified_on = datetime . datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:20> , <NUM_LIT:0> , <NUM_LIT:11> , <NUM_LIT:2> ) , <EOL> owner_id = '<STR_LIT>' , <EOL> name = '<STR_LIT>' , <EOL> version = '<STR_LIT>' , <EOL> indices = [ ] , <EOL> case_attachments = { <EOL> '<STR_LIT>' : CommCareCaseAttachment ( <EOL> attachment_from = u'<STR_LIT>' , <EOL> attachment_name = None , <EOL> attachment_properties = { '<STR_LIT:width>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> attachment_size = <NUM_LIT> , <EOL> attachment_src = u'<STR_LIT>' , <EOL> doc_type = u'<STR_LIT>' , <EOL> identifier = u'<STR_LIT>' , <EOL> server_md5 = None , <EOL> server_mime = u'<STR_LIT>' , <EOL> ) <EOL> } , <EOL> age = '<STR_LIT>' , <EOL> ) , <EOL> { u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> """<STR_LIT>""" <EOL> ) , <EOL> ) <EOL> @ contextmanager <EOL> def mock_fetch_case_attachment ( case_id , attachments ) : <EOL> class MockCachedObject ( object ) : <EOL> def __init__ ( self , attachment_file ) : <EOL> self . attachment_file = attachment_file <EOL> def get ( self , ** kwargs ) : <EOL> return None , open ( self . attachment_file ) <EOL> def get_cached_case_attachment ( domain , _case_id , attachment_id ) : <EOL> if case_id == _case_id and attachment_id in attachments : <EOL> return MockCachedObject ( attachments [ attachment_id ] ) <EOL> else : <EOL> raise ResourceNotFound ( ) <EOL> with patch ( '<STR_LIT>' , get_cached_case_attachment ) : <EOL> yield <EOL> class ExplodeCasesTest ( SimpleTestCase , TestXmlMixin ) : <EOL> maxDiff = <NUM_LIT> <EOL> def test_make_creating_casexml ( self ) : <EOL> for input , files , output in TESTS : <EOL> with mock_fetch_case_attachment ( input . case_id , files ) : <EOL> case_block , attachments = make_creating_casexml ( <EOL> '<STR_LIT>' , input , '<STR_LIT>' ) <EOL> self . assertXmlEqual ( case_block , output ) <EOL> self . assertDictEqual ( <EOL> { key : value . read ( ) for key , value in attachments . items ( ) } , <EOL> { value : open ( value ) . read ( ) for key , value in files . items ( ) } <EOL> ) <EOL> class ExplodeCasesDbTest ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> delete_all_cases ( ) <EOL> cls . domain = Domain ( name = '<STR_LIT:foo>' ) <EOL> cls . domain . save ( ) <EOL> cls . user = CommCareUser . create ( cls . domain . name , '<STR_LIT>' , '<STR_LIT:password>' ) <EOL> cls . user_id = cls . user . _id <EOL> def setUp ( self ) : <EOL> self . accessor = CaseAccessors ( self . domain . name ) <EOL> delete_all_cases ( ) <EOL> delete_all_xforms ( ) <EOL> def tearDown ( self ) : <EOL> delete_all_cases ( ) <EOL> delete_all_xforms ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . user . delete ( ) <EOL> cls . domain . delete ( ) <EOL> @ run_with_all_backends <EOL> def test_simple ( self ) : <EOL> caseblock = CaseBlock ( <EOL> create = True , <EOL> case_id = uuid . uuid4 ( ) . hex , <EOL> user_id = self . user_id , <EOL> owner_id = self . user_id , <EOL> case_type = '<STR_LIT>' , <EOL> ) . as_string ( ) <EOL> submit_case_blocks ( [ caseblock ] , self . domain . name ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . accessor . get_case_ids_in_domain ( ) ) ) <EOL> explode_cases ( self . user_id , self . domain . name , <NUM_LIT:10> ) <EOL> case_ids = self . accessor . get_case_ids_in_domain ( ) <EOL> cases_back = list ( self . accessor . iter_cases ( case_ids ) ) <EOL> self . assertEqual ( <NUM_LIT:10> , len ( cases_back ) ) <EOL> for case in cases_back : <EOL> self . assertEqual ( self . user_id , case . owner_id ) <EOL> @ run_with_all_backends <EOL> def test_skip_user_case ( self ) : <EOL> caseblock = CaseBlock ( <EOL> create = True , <EOL> case_id = uuid . uuid4 ( ) . hex , <EOL> user_id = self . user_id , <EOL> owner_id = self . user_id , <EOL> case_type = '<STR_LIT>' , <EOL> ) . as_string ( ) <EOL> submit_case_blocks ( [ caseblock ] , self . domain . name ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . accessor . get_case_ids_in_domain ( ) ) ) <EOL> explode_cases ( self . user_id , self . domain . name , <NUM_LIT:10> ) <EOL> case_ids = self . accessor . get_case_ids_in_domain ( ) <EOL> cases_back = list ( self . accessor . iter_cases ( case_ids ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( cases_back ) ) <EOL> for case in cases_back : <EOL> self . assertEqual ( self . user_id , case . owner_id ) <EOL> @ run_with_all_backends <EOL> def test_parent_child ( self ) : <EOL> parent_id = uuid . uuid4 ( ) . hex <EOL> parent_type = '<STR_LIT>' <EOL> parent_block = CaseBlock ( <EOL> create = True , <EOL> case_id = parent_id , <EOL> user_id = self . user_id , <EOL> owner_id = self . user_id , <EOL> case_type = parent_type , <EOL> ) . as_string ( ) <EOL> child_id = uuid . uuid4 ( ) . hex <EOL> child_block = CaseBlock ( <EOL> create = True , <EOL> case_id = child_id , <EOL> user_id = self . user_id , <EOL> owner_id = self . user_id , <EOL> case_type = '<STR_LIT>' , <EOL> index = { '<STR_LIT>' : ( parent_type , parent_id ) } , <EOL> ) . as_string ( ) <EOL> submit_case_blocks ( [ parent_block , child_block ] , self . domain . name ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( self . accessor . get_case_ids_in_domain ( ) ) ) <EOL> explode_cases ( self . user_id , self . domain . name , <NUM_LIT:5> ) <EOL> case_ids = self . accessor . get_case_ids_in_domain ( ) <EOL> cases_back = list ( self . accessor . iter_cases ( case_ids ) ) <EOL> self . assertEqual ( <NUM_LIT:10> , len ( cases_back ) ) <EOL> parent_cases = { p . case_id : p for p in filter ( lambda case : case . type == parent_type , cases_back ) } <EOL> self . assertEqual ( <NUM_LIT:5> , len ( parent_cases ) ) <EOL> child_cases = filter ( lambda case : case . type == '<STR_LIT>' , cases_back ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( child_cases ) ) <EOL> child_indices = [ child . indices [ <NUM_LIT:0> ] . referenced_id for child in child_cases ] <EOL> self . assertEqual ( len ( child_cases ) , len ( set ( child_indices ) ) ) <EOL> for child in child_cases : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( child . indices ) ) <EOL> self . assertTrue ( child . indices [ <NUM_LIT:0> ] . referenced_id in parent_cases ) </s>
