<s> import uuid <EOL> from couchdbkit . exceptions import ResourceNotFound <EOL> from django . test import TestCase , SimpleTestCase <EOL> from casexml . apps . case import const <EOL> from casexml . apps . case . cleanup import rebuild_case_from_forms <EOL> from casexml . apps . case . exceptions import MissingServerDate <EOL> from casexml . apps . case . mock import CaseBlock <EOL> from casexml . apps . case . models import CommCareCase , CommCareCaseAction <EOL> from datetime import datetime , timedelta <EOL> from copy import deepcopy <EOL> from casexml . apps . case . tests . util import delete_all_cases <EOL> from casexml . apps . case . util import primary_actions , post_case_blocks <EOL> from corehq . form_processor . backends . couch . update_strategy import CouchCaseUpdateStrategy , _action_sort_key_function <EOL> from corehq . form_processor . exceptions import CaseNotFound <EOL> from corehq . form_processor . interfaces . dbaccessors import CaseAccessors , FormAccessors <EOL> from corehq . form_processor . models import RebuildWithReason <EOL> from corehq . form_processor . tests . utils import run_with_all_backends <EOL> from corehq . form_processor . utils . general import should_use_sql_backend <EOL> from couchforms . models import XFormInstance <EOL> REBUILD_TEST_DOMAIN = '<STR_LIT>' <EOL> def _post_util ( create = False , case_id = None , user_id = None , owner_id = None , <EOL> case_type = None , form_extras = None , close = False , date_modified = None , <EOL> ** kwargs ) : <EOL> form_extras = form_extras or { } <EOL> form_extras [ '<STR_LIT>' ] = REBUILD_TEST_DOMAIN <EOL> uid = lambda : uuid . uuid4 ( ) . hex <EOL> case_id = case_id or uid ( ) <EOL> block = CaseBlock ( create = create , <EOL> case_id = case_id , <EOL> user_id = user_id or uid ( ) , <EOL> owner_id = owner_id or uid ( ) , <EOL> case_type = case_type or '<STR_LIT:test>' , <EOL> date_modified = date_modified , <EOL> update = kwargs , <EOL> close = close ) . as_xml ( ) <EOL> post_case_blocks ( [ block ] , form_extras ) <EOL> return case_id <EOL> class CaseRebuildTest ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> delete_all_cases ( ) <EOL> def _assertListEqual ( self , l1 , l2 , include_ordering = True ) : <EOL> if include_ordering : <EOL> self . assertEqual ( len ( l1 ) , len ( l2 ) ) <EOL> for i in range ( len ( l1 ) ) : <EOL> self . assertEqual ( l1 [ i ] , l2 [ i ] ) <EOL> else : <EOL> self . assertListEqual ( l1 , l2 ) <EOL> def _assertListNotEqual ( self , l1 , l2 , msg = None , include_ordering = True ) : <EOL> try : <EOL> self . _assertListEqual ( l1 , l2 , include_ordering = include_ordering ) <EOL> except self . failureException : <EOL> pass <EOL> else : <EOL> self . fail ( msg ) <EOL> def test_couch_action_equality ( self ) : <EOL> case_id = _post_util ( create = True ) <EOL> _post_util ( case_id = case_id , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' ) <EOL> case = CommCareCase . get ( case_id ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( case . actions ) ) <EOL> self . assertTrue ( case . actions [ <NUM_LIT:0> ] != case . actions [ <NUM_LIT:1> ] ) <EOL> self . assertTrue ( case . actions [ <NUM_LIT:1> ] == case . actions [ <NUM_LIT:1> ] ) <EOL> orig = case . actions [ <NUM_LIT:1> ] <EOL> copy = CommCareCaseAction . wrap ( orig . _doc . copy ( ) ) <EOL> self . assertTrue ( copy != case . actions [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( copy == orig ) <EOL> copy . server_date = copy . server_date + timedelta ( seconds = <NUM_LIT:1> ) <EOL> self . assertTrue ( copy != orig ) <EOL> copy . server_date = orig . server_date <EOL> self . assertTrue ( copy == orig ) <EOL> copy . updated_unknown_properties [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertTrue ( copy != orig ) <EOL> copy . updated_unknown_properties [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertTrue ( copy == orig ) <EOL> copy . updated_unknown_properties [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertTrue ( copy != orig ) <EOL> def test_couch_soft_rebuild ( self ) : <EOL> user_id = '<STR_LIT>' <EOL> now = datetime . utcnow ( ) <EOL> case_id = _post_util ( create = True , user_id = user_id , date_modified = now ) <EOL> _post_util ( case_id = case_id , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' , user_id = user_id , date_modified = now ) <EOL> _post_util ( case_id = case_id , p2 = '<STR_LIT>' , p3 = '<STR_LIT>' , user_id = user_id , date_modified = now ) <EOL> case = CommCareCase . get ( case_id ) <EOL> self . assertEqual ( case . p1 , '<STR_LIT>' ) <EOL> self . assertEqual ( case . p2 , '<STR_LIT>' ) <EOL> self . assertEqual ( case . p3 , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( case . actions ) ) <EOL> a1 = case . actions [ <NUM_LIT:1> ] <EOL> self . assertEqual ( a1 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( a1 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> a2 = case . actions [ <NUM_LIT:2> ] <EOL> self . assertEqual ( a2 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( a2 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> case . actions = [ case . actions [ <NUM_LIT:0> ] , a2 , a1 ] <EOL> case . xform_ids = [ case . xform_ids [ <NUM_LIT:0> ] , case . xform_ids [ <NUM_LIT:2> ] , case . xform_ids [ <NUM_LIT:1> ] ] <EOL> CouchCaseUpdateStrategy ( case ) . soft_rebuild_case ( ) <EOL> self . assertEqual ( case . p1 , '<STR_LIT>' ) <EOL> self . assertEqual ( case . p2 , '<STR_LIT>' ) <EOL> self . assertEqual ( case . p3 , '<STR_LIT>' ) <EOL> def test_couch_action_comparison ( self ) : <EOL> user_id = '<STR_LIT>' <EOL> case_id = _post_util ( create = True , property = '<STR_LIT>' , user_id = user_id ) <EOL> _post_util ( case_id = case_id , property = '<STR_LIT>' , user_id = user_id ) <EOL> _post_util ( case_id = case_id , property = '<STR_LIT>' , user_id = user_id ) <EOL> case = CommCareCase . get ( case_id ) <EOL> create , a1 , a2 , a3 = deepcopy ( list ( case . actions ) ) <EOL> self . assertEqual ( '<STR_LIT>' , case . property ) <EOL> self . assertEqual ( a1 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( a2 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( a3 . updated_unknown_properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def _confirm_action_order ( case , expected_actions ) : <EOL> actual_actions = case . actions [ <NUM_LIT:1> : ] <EOL> for expected , actual in zip ( expected_actions , actual_actions ) : <EOL> self . assertEqual ( expected . updated_unknown_properties [ '<STR_LIT>' ] , <EOL> actual . updated_unknown_properties [ '<STR_LIT>' ] ) <EOL> _confirm_action_order ( case , [ a1 , a2 , a3 ] ) <EOL> update_strategy = CouchCaseUpdateStrategy ( case ) <EOL> update_strategy . soft_rebuild_case ( ) <EOL> _confirm_action_order ( case , [ a1 , a2 , a3 ] ) <EOL> case . actions [ <NUM_LIT:2> ] . server_date = case . actions [ <NUM_LIT:2> ] . server_date + timedelta ( days = <NUM_LIT:1> ) <EOL> update_strategy . soft_rebuild_case ( ) <EOL> _confirm_action_order ( case , [ a1 , a3 , a2 ] ) <EOL> case = CommCareCase . get ( case_id ) <EOL> _confirm_action_order ( case , [ a1 , a2 , a3 ] ) <EOL> case . actions [ <NUM_LIT:2> ] . date = case . actions [ <NUM_LIT:3> ] . date + timedelta ( minutes = <NUM_LIT:1> ) <EOL> CouchCaseUpdateStrategy ( case ) . soft_rebuild_case ( ) <EOL> _confirm_action_order ( case , [ a1 , a3 , a2 ] ) <EOL> case = CommCareCase . get ( case_id ) <EOL> case . actions [ <NUM_LIT:3> ] . server_date = case . actions [ <NUM_LIT:2> ] . server_date <EOL> case . actions [ <NUM_LIT:3> ] . date = case . actions [ <NUM_LIT:2> ] . date <EOL> case . xform_ids = [ a1 . xform_id , a3 . xform_id , a2 . xform_id ] <EOL> CouchCaseUpdateStrategy ( case ) . soft_rebuild_case ( ) <EOL> _confirm_action_order ( case , [ a1 , a3 , a2 ] ) <EOL> case = CommCareCase . get ( case_id ) <EOL> case . actions = [ a1 , create , a2 , a3 ] <EOL> CouchCaseUpdateStrategy ( case ) . soft_rebuild_case ( ) <EOL> _confirm_action_order ( case , [ a1 , a2 , a3 ] ) <EOL> @ run_with_all_backends <EOL> def test_rebuild_empty ( self ) : <EOL> self . assertEqual ( None , rebuild_case_from_forms ( '<STR_LIT>' , '<STR_LIT>' , RebuildWithReason ( reason = '<STR_LIT:test>' ) ) ) <EOL> def test_couch_rebuild_deleted_case ( self ) : <EOL> """<STR_LIT>""" <EOL> case_id = _post_util ( create = True ) <EOL> _post_util ( case_id = case_id , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' ) <EOL> delete_all_cases ( ) <EOL> with self . assertRaises ( CaseNotFound ) : <EOL> CaseAccessors ( REBUILD_TEST_DOMAIN ) . get_case ( case_id ) <EOL> case = rebuild_case_from_forms ( REBUILD_TEST_DOMAIN , case_id , RebuildWithReason ( reason = '<STR_LIT:test>' ) ) <EOL> self . assertEqual ( case . p1 , '<STR_LIT>' ) <EOL> self . assertEqual ( case . p2 , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( primary_actions ( case ) ) ) <EOL> def test_couch_reconcile_actions ( self ) : <EOL> now = datetime . utcnow ( ) <EOL> case_id = _post_util ( create = True , form_extras = { '<STR_LIT>' : now } ) <EOL> _post_util ( case_id = case_id , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' , form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:1> ) } ) <EOL> _post_util ( case_id = case_id , p2 = '<STR_LIT>' , p3 = '<STR_LIT>' , form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:2> ) } ) <EOL> case = CommCareCase . get ( case_id ) <EOL> update_strategy = CouchCaseUpdateStrategy ( case ) <EOL> original_actions = [ deepcopy ( a ) for a in case . actions ] <EOL> original_form_ids = [ id for id in case . xform_ids ] <EOL> self . assertEqual ( <NUM_LIT:3> , len ( original_actions ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( original_form_ids ) ) <EOL> self . _assertListEqual ( original_actions , case . actions ) <EOL> case . actions = [ case . actions [ <NUM_LIT:2> ] , case . actions [ <NUM_LIT:1> ] , case . actions [ <NUM_LIT:0> ] ] <EOL> self . _assertListNotEqual ( original_actions , case . actions ) <EOL> update_strategy . reconcile_actions ( ) <EOL> self . _assertListEqual ( original_actions , case . actions ) <EOL> case . actions = case . actions * <NUM_LIT:3> <EOL> self . assertEqual ( <NUM_LIT:9> , len ( case . actions ) ) <EOL> self . _assertListNotEqual ( original_actions , case . actions ) <EOL> update_strategy . reconcile_actions ( ) <EOL> self . _assertListEqual ( original_actions , case . actions ) <EOL> case . actions = original_actions + [ deepcopy ( case . actions [ <NUM_LIT:2> ] ) ] <EOL> case . actions [ - <NUM_LIT:1> ] . server_date = case . actions [ - <NUM_LIT:1> ] . server_date + timedelta ( seconds = <NUM_LIT:1> ) <EOL> self . _assertListNotEqual ( original_actions , case . actions ) <EOL> update_strategy . reconcile_actions ( ) <EOL> self . _assertListEqual ( original_actions , case . actions ) <EOL> case . actions = original_actions + [ deepcopy ( case . actions [ <NUM_LIT:2> ] ) ] <EOL> case . actions [ - <NUM_LIT:1> ] . updated_unknown_properties [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertEqual ( <NUM_LIT:4> , len ( case . actions ) ) <EOL> self . _assertListNotEqual ( original_actions , case . actions ) <EOL> update_strategy . reconcile_actions ( ) <EOL> self . _assertListNotEqual ( original_actions , case . actions ) <EOL> case = rebuild_case_from_forms ( REBUILD_TEST_DOMAIN , case_id , RebuildWithReason ( reason = '<STR_LIT:test>' ) ) <EOL> self . _assertListEqual ( original_actions , primary_actions ( case ) ) <EOL> self . _assertListEqual ( original_form_ids , case . xform_ids ) <EOL> @ run_with_all_backends <EOL> def test_archiving_only_form ( self ) : <EOL> """<STR_LIT>""" <EOL> case_id = _post_util ( create = True , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' ) <EOL> case_accessors = CaseAccessors ( REBUILD_TEST_DOMAIN ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertFalse ( case . is_deleted ) <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( case . actions ) ) <EOL> else : <EOL> self . assertEqual ( <NUM_LIT:2> , len ( case . actions ) ) <EOL> [ form_id ] = case . xform_ids <EOL> form = FormAccessors ( REBUILD_TEST_DOMAIN ) . get_form ( form_id ) <EOL> form . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertTrue ( case . is_deleted ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( case . actions ) ) <EOL> self . assertTrue ( case . actions [ <NUM_LIT:0> ] . is_case_rebuild ) <EOL> form . unarchive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertFalse ( case . is_deleted ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( case . actions ) ) <EOL> self . assertTrue ( case . actions [ - <NUM_LIT:1> ] . is_case_rebuild ) <EOL> @ run_with_all_backends <EOL> def test_form_archiving ( self ) : <EOL> now = datetime . utcnow ( ) <EOL> case_id = _post_util ( create = True , p1 = '<STR_LIT>' , p2 = '<STR_LIT>' , <EOL> form_extras = { '<STR_LIT>' : now } ) <EOL> _post_util ( case_id = case_id , p2 = '<STR_LIT>' , p3 = '<STR_LIT>' , p4 = '<STR_LIT>' , <EOL> form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:1> ) } ) <EOL> _post_util ( case_id = case_id , p4 = '<STR_LIT>' , p5 = '<STR_LIT>' , close = True , <EOL> form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:2> ) } ) <EOL> case_accessors = CaseAccessors ( REBUILD_TEST_DOMAIN ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> closed_by = case . closed_by <EOL> closed_on = case . closed_on <EOL> self . assertNotEqual ( '<STR_LIT>' , closed_by ) <EOL> self . assertNotEqual ( None , closed_on ) <EOL> def _check_initial_state ( case ) : <EOL> self . assertTrue ( case . closed ) <EOL> self . assertEqual ( closed_by , case . closed_by ) <EOL> self . assertEqual ( closed_on , case . closed_on ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> self . assertEqual ( <NUM_LIT:3> , len ( primary_actions ( case ) ) ) <EOL> else : <EOL> self . assertEqual ( <NUM_LIT:5> , len ( primary_actions ( case ) ) ) <EOL> _check_initial_state ( case ) <EOL> [ f1 , f2 , f3 ] = case . xform_ids <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> [ create , update , close ] = case . actions <EOL> self . assertEqual ( f1 , create . form_id ) <EOL> self . assertEqual ( f2 , update . form_id ) <EOL> self . assertEqual ( f3 , close . form_id ) <EOL> else : <EOL> [ create , u1 , u2 , u3 , close ] = case . actions <EOL> self . assertEqual ( f1 , create . form_id ) <EOL> self . assertEqual ( f1 , u1 . form_id ) <EOL> self . assertEqual ( f2 , u2 . form_id ) <EOL> self . assertEqual ( f3 , u3 . form_id ) <EOL> form_acessors = FormAccessors ( REBUILD_TEST_DOMAIN ) <EOL> f1_doc = form_acessors . get_form ( f1 ) <EOL> f1_doc . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> self . assertEqual ( <NUM_LIT:2> , len ( primary_actions ( case ) ) ) <EOL> else : <EOL> self . assertEqual ( <NUM_LIT:3> , len ( primary_actions ( case ) ) ) <EOL> [ u2 , u3 ] = case . xform_ids <EOL> self . assertEqual ( f2 , u2 ) <EOL> self . assertEqual ( f3 , u3 ) <EOL> self . assertTrue ( case . closed ) <EOL> self . assertFalse ( '<STR_LIT>' in case . dynamic_case_properties ( ) ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def _reset ( form_id ) : <EOL> form_doc = form_acessors . get_form ( form_id ) <EOL> form_doc . unarchive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> _check_initial_state ( case ) <EOL> _reset ( f1 ) <EOL> f2_doc = form_acessors . get_form ( f2 ) <EOL> f2_doc . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> self . assertEqual ( <NUM_LIT:2> , len ( primary_actions ( case ) ) ) <EOL> else : <EOL> self . assertEqual ( <NUM_LIT:4> , len ( primary_actions ( case ) ) ) <EOL> [ u1 , u3 ] = case . xform_ids <EOL> self . assertEqual ( f1 , u1 ) <EOL> self . assertEqual ( f3 , u3 ) <EOL> self . assertTrue ( case . closed ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertFalse ( '<STR_LIT>' in case . dynamic_case_properties ( ) ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> _reset ( f2 ) <EOL> f3_doc = form_acessors . get_form ( f3 ) <EOL> f3_doc . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> if should_use_sql_backend ( REBUILD_TEST_DOMAIN ) : <EOL> self . assertEqual ( <NUM_LIT:2> , len ( primary_actions ( case ) ) ) <EOL> else : <EOL> self . assertEqual ( <NUM_LIT:3> , len ( primary_actions ( case ) ) ) <EOL> [ u1 , u2 ] = case . xform_ids <EOL> self . assertEqual ( f1 , u1 ) <EOL> self . assertEqual ( f2 , u2 ) <EOL> self . assertFalse ( case . closed ) <EOL> self . assertEqual ( '<STR_LIT>' , case . closed_by ) <EOL> self . assertEqual ( None , case . closed_on ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( case . get_case_property ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertFalse ( '<STR_LIT>' in case . dynamic_case_properties ( ) ) <EOL> _reset ( f3 ) <EOL> @ run_with_all_backends <EOL> def test_archie_modified_on ( self ) : <EOL> case_id = uuid . uuid4 ( ) . hex <EOL> now = datetime . utcnow ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> earlier = now - timedelta ( hours = <NUM_LIT:1> ) <EOL> way_earlier = now - timedelta ( days = <NUM_LIT:1> ) <EOL> create_block = CaseBlock ( case_id , create = True , date_modified = way_earlier ) <EOL> post_case_blocks ( <EOL> [ create_block . as_xml ( ) ] , form_extras = { '<STR_LIT>' : way_earlier } <EOL> ) <EOL> update_block = CaseBlock ( case_id , update = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , date_modified = earlier ) <EOL> post_case_blocks ( <EOL> [ update_block . as_xml ( ) ] , form_extras = { '<STR_LIT>' : earlier } <EOL> ) <EOL> case_accessors = CaseAccessors ( REBUILD_TEST_DOMAIN ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertEqual ( earlier , case . modified_on ) <EOL> second_form = FormAccessors ( REBUILD_TEST_DOMAIN ) . get_form ( case . xform_ids [ - <NUM_LIT:1> ] ) <EOL> second_form . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertEqual ( way_earlier , case . modified_on ) <EOL> @ run_with_all_backends <EOL> def test_archive_against_deleted_case ( self ) : <EOL> now = datetime . utcnow ( ) <EOL> case_id = _post_util ( create = True , p1 = '<STR_LIT>' , form_extras = { '<STR_LIT>' : now } ) <EOL> _post_util ( case_id = case_id , p2 = '<STR_LIT>' , <EOL> form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:1> ) } ) <EOL> _post_util ( case_id = case_id , p3 = '<STR_LIT>' , <EOL> form_extras = { '<STR_LIT>' : now + timedelta ( seconds = <NUM_LIT:2> ) } ) <EOL> case_accessors = CaseAccessors ( REBUILD_TEST_DOMAIN ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> case_accessors . soft_delete_cases ( [ case_id ] ) <EOL> [ f1 , f2 , f3 ] = case . xform_ids <EOL> f2_doc = FormAccessors ( REBUILD_TEST_DOMAIN ) . get_form ( f2 ) <EOL> f2_doc . archive ( ) <EOL> case = case_accessors . get_case ( case_id ) <EOL> self . assertTrue ( case . is_deleted ) <EOL> class TestCheckActionOrder ( SimpleTestCase ) : <EOL> def test_already_sorted ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ] ) <EOL> self . assertTrue ( CouchCaseUpdateStrategy ( case ) . check_action_order ( ) ) <EOL> def test_out_of_order ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ] ) <EOL> self . assertFalse ( CouchCaseUpdateStrategy ( case ) . check_action_order ( ) ) <EOL> def test_sorted_with_none ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = None ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ] ) <EOL> self . assertTrue ( CouchCaseUpdateStrategy ( case ) . check_action_order ( ) ) <EOL> def test_out_of_order_with_none ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> CommCareCaseAction ( server_date = None ) , <EOL> CommCareCaseAction ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ] ) <EOL> self . assertFalse ( CouchCaseUpdateStrategy ( case ) . check_action_order ( ) ) <EOL> class TestActionSortKey ( SimpleTestCase ) : <EOL> def test_missing_server_date ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) , <EOL> _make_action ( server_date = None , phone_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) , <EOL> ] ) <EOL> with self . assertRaises ( MissingServerDate ) : <EOL> sorted ( case . actions , key = _action_sort_key_function ( case ) ) <EOL> def test_missing_phone_date ( self ) : <EOL> case = CommCareCase ( actions = [ <EOL> _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) , <EOL> _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , phone_date = None ) , <EOL> ] ) <EOL> with self . assertRaises ( MissingServerDate ) : <EOL> sorted ( case . actions , key = _action_sort_key_function ( case ) ) <EOL> def test_sort_by_server_date ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) ) , <EOL> ( <NUM_LIT:2> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> ) ) ) , <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> ) ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def test_sort_by_server_date_precise ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> ) ) ) , <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def test_default_to_phone_date_on_same_day ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , phone_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> ) ) ) , <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> ) , phone_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def test_create_before_update ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , action_type = const . CASE_ACTION_UPDATE ) ) , <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , action_type = const . CASE_ACTION_CREATE ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def test_update_before_close ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , action_type = const . CASE_ACTION_CLOSE ) ) , <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , action_type = const . CASE_ACTION_UPDATE ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def test_different_usernames_default_to_server_date ( self ) : <EOL> expected_actions = ( <EOL> ( <NUM_LIT:0> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> phone_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> ) , <EOL> user_id = '<STR_LIT>' ) ) , <EOL> ( <NUM_LIT:1> , _make_action ( server_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> ) , <EOL> phone_date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> user_id = '<STR_LIT>' ) ) , <EOL> ) <EOL> self . _test ( expected_actions ) <EOL> def _test ( self , action_tuples ) : <EOL> """<STR_LIT>""" <EOL> case = CommCareCase ( actions = [ tup [ <NUM_LIT:1> ] for tup in action_tuples ] ) <EOL> sorted_actions = sorted ( case . actions , key = _action_sort_key_function ( case ) ) <EOL> for index , action in action_tuples : <EOL> self . assertEqual ( action , sorted_actions [ index ] ) <EOL> EMPTY_DATE = object ( ) <EOL> def _make_action ( server_date , phone_date = EMPTY_DATE , action_type = const . CASE_ACTION_UPDATE , user_id = '<STR_LIT>' ) : <EOL> if phone_date == EMPTY_DATE : <EOL> phone_date = server_date <EOL> return CommCareCaseAction ( action_type = action_type , server_date = server_date , date = phone_date , user_id = user_id ) </s>
