<s> import mock <EOL> from st2common . models . db . liveaction import LiveActionDB <EOL> from st2common . models . db . notification import NotificationSchema , NotificationSubSchema <EOL> from st2common . persistence . liveaction import LiveAction <EOL> from st2common . transport . publishers import PoolPublisher <EOL> from st2common . util import date as date_utils <EOL> from st2tests import DbTestCase <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> class LiveActionModelTest ( DbTestCase ) : <EOL> def test_liveaction_crud_no_notify ( self ) : <EOL> created = LiveActionDB ( ) <EOL> created . action = '<STR_LIT>' <EOL> created . description = '<STR_LIT>' <EOL> created . status = '<STR_LIT>' <EOL> created . parameters = { } <EOL> saved = LiveActionModelTest . _save_liveaction ( created ) <EOL> retrieved = LiveAction . get_by_id ( saved . id ) <EOL> self . assertEqual ( saved . action , retrieved . action , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( retrieved . notify , None ) <EOL> self . assertTrue ( retrieved . end_timestamp is None ) <EOL> retrieved . end_timestamp = date_utils . get_datetime_utc_now ( ) <EOL> updated = LiveAction . add_or_update ( retrieved ) <EOL> self . assertTrue ( updated . end_timestamp == retrieved . end_timestamp ) <EOL> LiveActionModelTest . _delete ( [ retrieved ] ) <EOL> try : <EOL> retrieved = LiveAction . get_by_id ( saved . id ) <EOL> except ValueError : <EOL> retrieved = None <EOL> self . assertIsNone ( retrieved , '<STR_LIT>' ) <EOL> def test_liveaction_create_with_notify_on_complete_only ( self ) : <EOL> created = LiveActionDB ( ) <EOL> created . action = '<STR_LIT>' <EOL> created . description = '<STR_LIT>' <EOL> created . status = '<STR_LIT>' <EOL> created . parameters = { } <EOL> notify_db = NotificationSchema ( ) <EOL> notify_sub_schema = NotificationSubSchema ( ) <EOL> notify_sub_schema . message = '<STR_LIT>' <EOL> notify_sub_schema . data = { <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> '<STR_LIT:bar>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } <EOL> notify_db . on_complete = notify_sub_schema <EOL> created . notify = notify_db <EOL> saved = LiveActionModelTest . _save_liveaction ( created ) <EOL> retrieved = LiveAction . get_by_id ( saved . id ) <EOL> self . assertEqual ( saved . action , retrieved . action , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( notify_sub_schema . message , retrieved . notify . on_complete . message ) <EOL> self . assertDictEqual ( notify_sub_schema . data , retrieved . notify . on_complete . data ) <EOL> self . assertListEqual ( notify_sub_schema . routes , retrieved . notify . on_complete . routes ) <EOL> self . assertEqual ( retrieved . notify . on_success , None ) <EOL> self . assertEqual ( retrieved . notify . on_failure , None ) <EOL> def test_liveaction_create_with_notify_on_success_only ( self ) : <EOL> created = LiveActionDB ( ) <EOL> created . action = '<STR_LIT>' <EOL> created . description = '<STR_LIT>' <EOL> created . status = '<STR_LIT>' <EOL> created . parameters = { } <EOL> notify_db = NotificationSchema ( ) <EOL> notify_sub_schema = NotificationSubSchema ( ) <EOL> notify_sub_schema . message = '<STR_LIT>' <EOL> notify_sub_schema . data = { <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> '<STR_LIT:bar>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } <EOL> notify_db . on_success = notify_sub_schema <EOL> created . notify = notify_db <EOL> saved = LiveActionModelTest . _save_liveaction ( created ) <EOL> retrieved = LiveAction . get_by_id ( saved . id ) <EOL> self . assertEqual ( saved . action , retrieved . action , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( notify_sub_schema . message , <EOL> retrieved . notify . on_success . message ) <EOL> self . assertDictEqual ( notify_sub_schema . data , retrieved . notify . on_success . data ) <EOL> self . assertListEqual ( notify_sub_schema . routes , retrieved . notify . on_success . routes ) <EOL> self . assertEqual ( retrieved . notify . on_failure , None ) <EOL> self . assertEqual ( retrieved . notify . on_complete , None ) <EOL> def test_liveaction_create_with_notify_both_on_success_and_on_error ( self ) : <EOL> created = LiveActionDB ( ) <EOL> created . action = '<STR_LIT>' <EOL> created . description = '<STR_LIT>' <EOL> created . status = '<STR_LIT>' <EOL> created . parameters = { } <EOL> on_success = NotificationSubSchema ( message = '<STR_LIT>' ) <EOL> on_failure = NotificationSubSchema ( message = '<STR_LIT>' ) <EOL> created . notify = NotificationSchema ( on_success = on_success , <EOL> on_failure = on_failure ) <EOL> saved = LiveActionModelTest . _save_liveaction ( created ) <EOL> retrieved = LiveAction . get_by_id ( saved . id ) <EOL> self . assertEqual ( saved . action , retrieved . action , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( on_success . message , retrieved . notify . on_success . message ) <EOL> self . assertEqual ( on_failure . message , retrieved . notify . on_failure . message ) <EOL> self . assertEqual ( retrieved . notify . on_complete , None ) <EOL> @ staticmethod <EOL> def _save_liveaction ( liveaction ) : <EOL> return LiveAction . add_or_update ( liveaction ) <EOL> @ staticmethod <EOL> def _delete ( model_objects ) : <EOL> for model_object in model_objects : <EOL> model_object . delete ( ) </s>
