#NEWLINE# def test_integrity_error_on_okc_id_handled_by_safe_upsert( ) : #NEWLINE# #INDENT# session = db . Session ( ) #NEWLINE# other_session = db . Session ( ) #NEWLINE# m1 = model . MessageThread ( okc_id = 1 , initiator_id = 1 , respondent_id = 1 ) #NEWLINE# m2 = model . MessageThread ( okc_id = 1 , initiator_id = 1 , respondent_id = 2 ) #NEWLINE# other_session . add ( commit_other_session ) #NEWLINE# find_all_no_txn = model . MessageThread . find_all_no_txn #NEWLINE# def commit_other_session( * args , **kwargs ) : #NEWLINE# #INDENT# other_session . commit ( ) #NEWLINE# model . MessageThread . find_all_no_txn = find_all_no_txn #NEWLINE# return [ ] #NEWLINE# with mock . patch . object ( model . MessageThread , 'find_all_no_txn' , commit_other_session ) : #NEWLINE# #INDENT# model . MessageThread . safe_upsert ( [ m2 ] , id_key = 'okc_id' ) #NEWLINE# #UNINDENT# #UNINDENT# other_session . commit ( ) #NEWLINE# session . commit ( )