<s> import contextlib <EOL> from oslo_utils import uuidutils <EOL> from taskflow import exceptions as exc <EOL> from taskflow . persistence import models <EOL> from taskflow import states <EOL> from taskflow . types import failure <EOL> class PersistenceTestMixin ( object ) : <EOL> def _get_connection ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def test_task_detail_update_not_existing ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> fd . add ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> td2 = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> fd . add ( td2 ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . update_flow_details ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb . uuid ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> self . assertIsNotNone ( fd2 . find ( td . uuid ) ) <EOL> self . assertIsNotNone ( fd2 . find ( td2 . uuid ) ) <EOL> def test_flow_detail_update_not_existing ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> fd2 = models . FlowDetail ( '<STR_LIT>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd2 ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb . uuid ) <EOL> self . assertIsNotNone ( lb2 . find ( fd . uuid ) ) <EOL> self . assertIsNotNone ( lb2 . find ( fd2 . uuid ) ) <EOL> def test_logbook_save_retrieve_many ( self ) : <EOL> lb_ids = { } <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( i , lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> lb_ids [ lb_id ] = True <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> self . assertRaises ( exc . NotFound , conn . get_logbook , lb_id ) <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lbs = conn . get_logbooks ( ) <EOL> for lb in lbs : <EOL> self . assertIn ( lb . uuid , lb_ids ) <EOL> lb_ids . pop ( lb . uuid ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( lb_ids ) ) <EOL> def test_logbook_save_retrieve ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_meta = { '<STR_LIT:1>' : <NUM_LIT:2> } <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> lb . meta = lb_meta <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> self . assertRaises ( exc . NotFound , conn . get_logbook , lb_id ) <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb = conn . get_logbook ( lb_id ) <EOL> self . assertEqual ( lb_name , lb . name ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( lb ) ) <EOL> self . assertEqual ( lb_meta , lb . meta ) <EOL> self . assertIsNone ( lb . updated_at ) <EOL> self . assertIsNotNone ( lb . created_at ) <EOL> def test_flow_detail_save ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> self . assertRaises ( exc . NotFound , conn . get_logbook , lb_id ) <EOL> self . assertRaises ( exc . NotFound , conn . update_flow_details , fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> def test_flow_detail_meta_update ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> fd . meta = { '<STR_LIT:test>' : <NUM_LIT> } <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> fd . meta [ '<STR_LIT:test>' ] = <NUM_LIT> <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . update_flow_details ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> self . assertEqual ( <NUM_LIT> , fd2 . meta . get ( '<STR_LIT:test>' ) ) <EOL> def test_flow_detail_lazy_fetch ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> td . version = '<STR_LIT>' <EOL> fd . add ( td ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> fd2 = conn . get_flow_details ( fd . uuid , lazy = True ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( fd2 ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( fd ) ) <EOL> def test_task_detail_save ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> fd . add ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> self . assertRaises ( exc . NotFound , conn . update_flow_details , fd ) <EOL> self . assertRaises ( exc . NotFound , conn . update_atom_details , td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( td ) <EOL> def test_task_detail_meta_update ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> td . meta = { '<STR_LIT:test>' : <NUM_LIT> } <EOL> fd . add ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( td ) <EOL> td . meta [ '<STR_LIT:test>' ] = <NUM_LIT> <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . update_atom_details ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> td2 = fd2 . find ( td . uuid ) <EOL> self . assertEqual ( <NUM_LIT> , td2 . meta . get ( '<STR_LIT:test>' ) ) <EOL> self . assertIsInstance ( td2 , models . TaskDetail ) <EOL> def test_task_detail_with_failure ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> try : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> except Exception : <EOL> td . failure = failure . Failure ( ) <EOL> fd . add ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( td ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> td2 = fd2 . find ( td . uuid ) <EOL> self . assertEqual ( '<STR_LIT>' , td2 . failure . exception_str ) <EOL> self . assertIs ( td2 . failure . check ( RuntimeError ) , RuntimeError ) <EOL> self . assertEqual ( td . failure . traceback_str , td2 . failure . traceback_str ) <EOL> self . assertIsInstance ( td2 , models . TaskDetail ) <EOL> def test_logbook_merge_flow_detail ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> lb2 = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd2 = models . FlowDetail ( '<STR_LIT>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb2 . add ( fd2 ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb2 ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb3 = conn . get_logbook ( lb_id ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( lb3 ) ) <EOL> def test_logbook_add_flow_detail ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( lb2 ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( lb ) ) <EOL> self . assertEqual ( fd . name , lb2 . find ( fd . uuid ) . name ) <EOL> def test_logbook_lazy_fetch ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id , lazy = True ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( lb2 ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( lb ) ) <EOL> def test_logbook_add_task_detail ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> td = models . TaskDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> td . version = '<STR_LIT>' <EOL> fd . add ( td ) <EOL> lb . add ( fd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( lb2 ) ) <EOL> tasks = <NUM_LIT:0> <EOL> for fd in lb : <EOL> tasks += len ( fd ) <EOL> self . assertEqual ( <NUM_LIT:1> , tasks ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> td2 = fd2 . find ( td . uuid ) <EOL> self . assertIsNot ( td2 , None ) <EOL> self . assertEqual ( '<STR_LIT>' , td2 . name ) <EOL> self . assertEqual ( '<STR_LIT>' , td2 . version ) <EOL> self . assertEqual ( states . EXECUTE , td2 . intention ) <EOL> def test_logbook_delete ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> self . assertRaises ( exc . NotFound , conn . destroy_logbook , lb_id ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> self . assertIsNotNone ( lb2 ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . destroy_logbook ( lb_id ) <EOL> self . assertRaises ( exc . NotFound , conn . destroy_logbook , lb_id ) <EOL> def test_task_detail_retry_type_ ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> rd = models . RetryDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> rd . intention = states . REVERT <EOL> fd . add ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> rd2 = fd2 . find ( rd . uuid ) <EOL> self . assertEqual ( states . REVERT , rd2 . intention ) <EOL> self . assertIsInstance ( rd2 , models . RetryDetail ) <EOL> def test_retry_detail_save_with_task_failure ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> rd = models . RetryDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> fail = failure . Failure . from_exception ( RuntimeError ( '<STR_LIT>' ) ) <EOL> rd . results . append ( ( <NUM_LIT> , { '<STR_LIT>' : fail } ) ) <EOL> fd . add ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> rd2 = fd2 . find ( rd . uuid ) <EOL> self . assertIsInstance ( rd2 , models . RetryDetail ) <EOL> fail2 = rd2 . results [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . get ( '<STR_LIT>' ) <EOL> self . assertIsInstance ( fail2 , failure . Failure ) <EOL> self . assertTrue ( fail . matches ( fail2 ) ) <EOL> def test_retry_detail_save_intention ( self ) : <EOL> lb_id = uuidutils . generate_uuid ( ) <EOL> lb_name = '<STR_LIT>' % ( lb_id ) <EOL> lb = models . LogBook ( name = lb_name , uuid = lb_id ) <EOL> fd = models . FlowDetail ( '<STR_LIT:test>' , uuid = uuidutils . generate_uuid ( ) ) <EOL> lb . add ( fd ) <EOL> rd = models . RetryDetail ( "<STR_LIT>" , uuid = uuidutils . generate_uuid ( ) ) <EOL> fd . add ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . save_logbook ( lb ) <EOL> conn . update_flow_details ( fd ) <EOL> conn . update_atom_details ( rd ) <EOL> rd . intention = states . REVERT <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> conn . update_atom_details ( rd ) <EOL> with contextlib . closing ( self . _get_connection ( ) ) as conn : <EOL> lb2 = conn . get_logbook ( lb_id ) <EOL> fd2 = lb2 . find ( fd . uuid ) <EOL> rd2 = fd2 . find ( rd . uuid ) <EOL> self . assertEqual ( states . REVERT , rd2 . intention ) <EOL> self . assertIsInstance ( rd2 , models . RetryDetail ) </s>
