• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

Python executionstate.ActionExecutionState类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中st2common.persistence.executionstate.ActionExecutionState类的典型用法代码示例。如果您正苦于以下问题：Python ActionExecutionState类的具体用法？Python ActionExecutionState怎么用？Python ActionExecutionState使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了ActionExecutionState类的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_keep_state_object_on_error_at_update_result

    def test_keep_state_object_on_error_at_update_result(self):
        tracker = self._get_tracker()
        querier = tracker.get_querier('test_querymodule')
        querier._delete_state_object = mock.Mock(return_value=None)
        querier.delete_state_object_on_error = False

        # Ensure state objects are present.
        state1 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state1.yaml'].id)
        state2 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state2.yaml'].id)
        self.assertIsNotNone(state1)
        self.assertIsNotNone(state2)

        with mock.patch.object(
                querier.__class__, '_update_action_results',
                mock.MagicMock(side_effect=Exception('Mock update exception.'))):
            tracker._bootstrap()
            eventlet.sleep(1)

            exec_id = str(ResultsTrackerTests.states['state1.yaml'].execution_id)
            exec_db = LiveAction.get_by_id(exec_id)
            self.assertDictEqual(exec_db.result, {})

            exec_id = str(ResultsTrackerTests.states['state2.yaml'].execution_id)
            exec_db = LiveAction.get_by_id(exec_id)
            self.assertDictEqual(exec_db.result, {})

            tracker.shutdown()

        # Ensure deletes are not called.
        querier._delete_state_object.assert_not_called()
开发者ID:nzlosh，项目名称:st2，代码行数:30，代码来源:test_resultstracker.py


示例2: test_execution_cancellation

    def test_execution_cancellation(self):
        tracker = self._get_tracker()
        querier = tracker.get_querier('test_querymodule')
        querier._delete_state_object = mock.Mock(return_value=None)
        runners_utils.invoke_post_run = mock.Mock(return_value=None)

        # Ensure state objects are present.
        state1 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state1.yaml'].id)
        state2 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state2.yaml'].id)
        self.assertIsNotNone(state1)
        self.assertIsNotNone(state2)

        with mock.patch.object(
                querier.__class__, 'query',
                mock.MagicMock(return_value=(action_constants.LIVEACTION_STATUS_CANCELED, {}))):
            tracker._bootstrap()
            eventlet.sleep(2)

            exec_id = str(ResultsTrackerTests.states['state1.yaml'].execution_id)
            exec_db = LiveAction.get_by_id(exec_id)
            self.assertDictEqual(exec_db.result, {})

            exec_id = str(ResultsTrackerTests.states['state2.yaml'].execution_id)
            exec_db = LiveAction.get_by_id(exec_id)
            self.assertDictEqual(exec_db.result, {})

            tracker.shutdown()

        # Ensure deletes are called.
        self.assertEqual(2, querier._delete_state_object.call_count)

        # Ensure invoke_post_run is called.
        self.assertEqual(2, runners_utils.invoke_post_run.call_count)
开发者ID:nzlosh，项目名称:st2，代码行数:33，代码来源:test_resultstracker.py


示例3: _delete_state_object

 def _delete_state_object(self, query_context):
     state_db = ActionExecutionState.get_by_id(query_context.id)
     if state_db is not None:
         try:
             ActionExecutionState.delete(state_db)
         except:
             LOG.exception('Failed clearing state object: %s', state_db)
开发者ID:ruslantum，项目名称:st2，代码行数:7，代码来源:base.py


示例4: _update_state_models

 def _update_state_models(cls):
     states = ResultsTrackerTests.states
     state1 = ActionExecutionState.get_by_id(states['state1.yaml'].id)
     state1.execution_id = ResultsTrackerTests.liveactions['liveaction1.yaml'].id
     state2 = ActionExecutionState.get_by_id(states['state2.yaml'].id)
     state2.execution_id = ResultsTrackerTests.liveactions['liveaction2.yaml'].id
     ResultsTrackerTests.states['state1.yaml'] = ActionExecutionState.add_or_update(state1)
     ResultsTrackerTests.states['state2.yaml'] = ActionExecutionState.add_or_update(state2)
开发者ID:nzlosh，项目名称:st2，代码行数:8，代码来源:test_resultstracker.py


示例5: remove_query

def remove_query(liveaction_id):
    state_db = ActionExecutionState.query(execution_id=liveaction_id)

    if not state_db:
        return False

    ActionExecutionState.delete(state_db, publish=False, dispatch_trigger=False)

    return True
开发者ID:StackStorm，项目名称:st2，代码行数:9，代码来源:queries.py


示例6: test_state_crud

 def test_state_crud(self):
     saved = ActionExecutionStateTests._create_save_actionstate()
     retrieved = ActionExecutionState.get_by_id(saved.id)
     self.assertDictEqual(saved.query_context, retrieved.query_context)
     self.assertEqual(saved.query_module, retrieved.query_module)
     ActionExecutionStateTests._delete(model_objects=[retrieved])
     try:
         retrieved = ActionExecutionState.get_by_id(saved.id)
     except StackStormDBObjectNotFoundError:
         retrieved = None
     self.assertIsNone(retrieved, 'managed to retrieve after failure.')
开发者ID:StackStorm，项目名称:st2，代码行数:11，代码来源:test_db_action_state.py


示例7: setup_query

def setup_query(liveaction_id, runnertype_db, query_context):
    if not getattr(runnertype_db, 'query_module', None):
        raise Exception('The runner "%s" does not have a query module.' % runnertype_db.name)

    state_db = ActionExecutionStateDB(
        execution_id=liveaction_id,
        query_module=runnertype_db.query_module,
        query_context=query_context
    )

    ActionExecutionState.add_or_update(state_db)
开发者ID:StackStorm，项目名称:st2，代码行数:11，代码来源:queries.py


示例8: test_state_db_created_for_polling_async_actions

    def test_state_db_created_for_polling_async_actions(self):
        runner_container = get_runner_container()

        params = {
            'actionstr': 'foo',
            'actionint': 20,
            'async_test': True
        }

        liveaction_db = self._get_liveaction_model(
            RunnerContainerTest.polling_async_action_db,
            params
        )

        liveaction_db = LiveAction.add_or_update(liveaction_db)
        executions.create_execution_object(liveaction_db)

        # Assert that execution ran without exceptions.
        runner_container.dispatch(liveaction_db)
        states = ActionExecutionState.get_all()
        found = [state for state in states if state.execution_id == liveaction_db.id]

        self.assertTrue(len(found) > 0, 'There should be a state db object.')
        self.assertTrue(len(found) == 1, 'There should only be one state db object.')
        self.assertTrue(found[0].query_context is not None)
        self.assertTrue(found[0].query_module is not None)
开发者ID:lyandut，项目名称:st2，代码行数:26，代码来源:test_runner_container.py


示例9: _create_execution_state

 def _create_execution_state(self, liveaction_id, runnertype_db, query_context):
     state_db = ActionExecutionStateDB(
         execution_id=liveaction_id, query_module=runnertype_db.query_module, query_context=query_context
     )
     try:
         return ActionExecutionState.add_or_update(state_db)
     except:
         LOG.exception("Unable to create execution state db for liveaction_id %s." % liveaction_id)
         return None
开发者ID:pixelrebel，项目名称:st2，代码行数:9，代码来源:base.py


示例10: test_action_deleted

    def test_action_deleted(self):
        tracker = self._get_tracker()
        action_db_utils.get_runnertype_by_name = mock.Mock(return_value=None)

        # Ensure state objects are present.
        state1 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state1.yaml'].id)
        state2 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state2.yaml'].id)
        self.assertIsNotNone(state1)
        self.assertIsNotNone(state2)

        # Process the state objects.
        tracker._bootstrap()
        eventlet.sleep(1)

        exec_id = str(ResultsTrackerTests.states['state1.yaml'].execution_id)
        exec_db = LiveAction.get_by_id(exec_id)
        self.assertTrue(exec_db.result['called_with'][exec_id] is not None,
                        exec_db.result)

        exec_id = str(ResultsTrackerTests.states['state2.yaml'].execution_id)
        exec_db = LiveAction.get_by_id(exec_id)
        self.assertTrue(exec_db.result['called_with'][exec_id] is not None,
                        exec_db.result)

        tracker.shutdown()

        # Ensure state objects are deleted.
        self.assertRaises(
            StackStormDBObjectNotFoundError,
            ActionExecutionState.get_by_id,
            ResultsTrackerTests.states['state1.yaml'].id
        )

        self.assertRaises(
            StackStormDBObjectNotFoundError,
            ActionExecutionState.get_by_id,
            ResultsTrackerTests.states['state2.yaml'].id
        )

        # Ensure get_runnertype_by_name in invoke_post_run is not called.
        action_db_utils.get_runnertype_by_name.assert_not_called()
开发者ID:nzlosh，项目名称:st2，代码行数:41，代码来源:test_resultstracker.py


示例11: test_query_process

    def test_query_process(self):
        tracker = results_tracker.get_tracker()
        runners_utils.invoke_post_run = mock.Mock(return_value=None)

        # Ensure state objects are present.
        state1 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state1.yaml'].id)
        state2 = ActionExecutionState.get_by_id(ResultsTrackerTests.states['state2.yaml'].id)
        self.assertIsNotNone(state1)
        self.assertIsNotNone(state2)

        # Process the state objects.
        tracker._bootstrap()
        eventlet.sleep(1)

        exec_id = str(ResultsTrackerTests.states['state1.yaml'].execution_id)
        exec_db = LiveAction.get_by_id(exec_id)
        self.assertTrue(exec_db.result['called_with'][exec_id] is not None,
                        exec_db.result)

        exec_id = str(ResultsTrackerTests.states['state2.yaml'].execution_id)
        exec_db = LiveAction.get_by_id(exec_id)
        self.assertTrue(exec_db.result['called_with'][exec_id] is not None,
                        exec_db.result)

        tracker.shutdown()

        # Ensure state objects are deleted.
        self.assertRaises(
            StackStormDBObjectNotFoundError,
            ActionExecutionState.get_by_id,
            ResultsTrackerTests.states['state1.yaml'].id
        )

        self.assertRaises(
            StackStormDBObjectNotFoundError,
            ActionExecutionState.get_by_id,
            ResultsTrackerTests.states['state2.yaml'].id
        )

        # Ensure invoke_post_run is called.
        self.assertEqual(2, runners_utils.invoke_post_run.call_count)
开发者ID:lyandut，项目名称:st2，代码行数:41，代码来源:test_resultstracker.py


示例12: test_state_db_creation_async_actions

    def test_state_db_creation_async_actions(self):
        runner_container = get_runner_container()
        params = {"actionstr": "foo", "actionint": 20, "async_test": True}
        liveaction_db = self._get_action_exec_db_model(RunnerContainerTest.async_action_db, params)
        liveaction_db = LiveAction.add_or_update(liveaction_db)
        executions.create_execution_object(liveaction_db)
        # Assert that execution ran without exceptions.
        runner_container.dispatch(liveaction_db)
        states = ActionExecutionState.get_all()

        found = None
        for state in states:
            if state.execution_id == liveaction_db.id:
                found = state
        self.assertTrue(found is not None, "There should be a state db object.")
        self.assertTrue(found.query_context is not None)
        self.assertTrue(found.query_module is not None)
开发者ID:emptywee，项目名称:st2，代码行数:17，代码来源:test_runner_container.py


示例13: _bootstrap

    def _bootstrap(self):
        all_states = ActionExecutionState.get_all()
        LOG.info('Found %d pending states in db.' % len(all_states))

        query_contexts_dict = defaultdict(list)
        for state_db in all_states:
            try:
                context = QueryContext.from_model(state_db)
            except:
                LOG.exception('Invalid state object: %s', state_db)
                continue
            query_module_name = state_db.query_module
            querier = self.get_querier(query_module_name)

            if querier is not None:
                query_contexts_dict[querier].append(context)

        for querier, contexts in six.iteritems(query_contexts_dict):
            LOG.info('Found %d pending actions for query module %s', len(contexts), querier)
            querier.add_queries(query_contexts=contexts)
开发者ID:AlexeyDeyneko，项目名称:st2，代码行数:20，代码来源:resultstracker.py


示例14: _create_save_actionstate

 def _create_save_actionstate():
     created = ActionExecutionStateDB()
     created.query_context = {'id': 'some_external_service_id'}
     created.query_module = 'dummy.modules.query1'
     created.execution_id = bson.ObjectId()
     return ActionExecutionState.add_or_update(created)
开发者ID:StackStorm，项目名称:st2，代码行数:6，代码来源:test_db_action_state.py


示例15: get_state

 def get_state(cls, exec_db):
     state = ActionExecutionStateDB(execution_id=str(exec_db.id), query_context={'id': 'foo'},
                                    query_module='test_querymodule')
     return ActionExecutionState.add_or_update(state)
开发者ID:StackStorm，项目名称:st2，代码行数:4，代码来源:test_action_state_consumer.py



注：本文中的st2common.persistence.executionstate.ActionExecutionState类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python keyvalue.KeyValuePair类代码示例发布时间：2022-05-27
下一篇：
Python execution.ActionExecution类代码示例发布时间：2022-05-27
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap