KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use action_info method in autotest

Best Python code snippet using autotest_python

workflow_actions.py

Source:workflow_actions.py

1import json2import logging3import time4from threading import Thread, Event5from dtable_events.app.event_redis import RedisClient6from dtable_events.automations.general_actions import ActionInvalid, AddRecordToOtherTableAction, BaseContext, NotifyAction, SendEmailAction, \7    SendWechatAction, SendDingtalkAction, UpdateAction, AddRowAction, LockRecordAction, LinkRecordsAction, \8    RunPythonScriptAction9from dtable_events.db import init_db_session_class10logger = logging.getLogger(__name__)11def do_workflow_actions(task_id, node_id, db_session):12    sql = '''13    SELECT dw.dtable_uuid, dw.token, dw.workflow_config, dwt.row_id FROM dtable_workflows dw14    JOIN dtable_workflow_tasks dwt ON dw.id = dwt.dtable_workflow_id15    WHERE dwt.id=:task_id16    '''17    task_item = db_session.execute(sql, {'task_id': task_id}).fetchone()18    if not task_item:19        return20    dtable_uuid = task_item.dtable_uuid21    workflow_config = json.loads(task_item.workflow_config)22    workflow_token = task_item.token23    table_id = workflow_config.get('table_id')24    workflow_name = workflow_config.get('workflow_name')25    row_id = task_item.row_id26    try:27        context = BaseContext(dtable_uuid, table_id, db_session, caller='workflow')28    except Exception as e:29        logger.error('task: %s node: %s dtable_uuid: %s context error: %s', task_id, node_id, dtable_uuid, e)30        return31    nodes = workflow_config.get('nodes', [])32    node = None33    for tmp_node in nodes:34        if tmp_node['_id'] == node_id:35            node = tmp_node36            break37    if not node:38        return39    actions = node.get('actions', [])40    converted_row = context.get_converted_row(table_id, row_id)41    if not converted_row:42        return43    for action_info in actions:44        logger.debug('start action: %s', action_info.get('type'))45        try:46            if action_info.get('type') == 'notify':47                users = action_info.get('users', [])48                users_column_key = action_info.get('users_column_key')49                msg = action_info.get('default_msg')50                NotifyAction(51                    context,52                    users,53                    msg,54                    NotifyAction.NOTIFY_TYPE_WORKFLOW,55                    users_column_key=users_column_key,56                    workflow_token=workflow_token,57                    workflow_name=workflow_name,58                    workflow_task_id=task_id59                ).do_action_with_row(converted_row)60            elif action_info.get('type') == 'send_email':61                msg = action_info.get('default_msg')62                subject = action_info.get('subject')63                send_to = action_info.get('send_to')64                copy_to = action_info.get('copy_to')65                account_id = int(action_info.get('account_id'))66                SendEmailAction(67                    context,68                    account_id,69                    subject,70                    msg,71                    send_to,72                    copy_to,73                    SendEmailAction.SEND_FROM_WORKFLOW74                ).do_action_with_row(converted_row)75            elif action_info.get('type') == 'send_wechat':76                account_id = int(action_info.get('account_id'))77                msg = action_info.get('default_msg')78                msg_type = action_info.get('msg_type', 'text')79                SendWechatAction(80                    context,81                    account_id,82                    msg,83                    msg_type84                ).do_action_with_row(converted_row)85            elif action_info.get('type') == 'send_dingtalk':86                account_id = int(action_info.get('account_id'))87                msg = action_info.get('default_msg')88                title = action_info.get('default_title')89                msg_type = action_info.get('msg_type', 'text')90                SendDingtalkAction(91                    context,92                    account_id,93                    msg,94                    msg_type,95                    title96                ).do_action_with_row(converted_row)97            elif action_info.get('type') == 'add_record':98                new_row = action_info.get('row')99                logger.debug('new_row: %s', new_row)100                if not new_row:101                    continue102                AddRowAction(103                    context,104                    new_row105                ).do_action_without_row()106            elif action_info.get('type') == 'update_record':107                updates = action_info.get('updates')108                logger.debug('updates: %s', updates)109                if not updates:110                    continue111                UpdateAction(112                    context,113                    updates114                ).do_action_with_row(converted_row)115            elif action_info.get('type') == 'lock_record':116                LockRecordAction(context).do_action_with_row(converted_row)117            elif action_info.get('type') == 'link_records':118                link_id = action_info.get('link_id')119                linked_table_id = action_info.get('linked_table_id')120                match_conditions = action_info.get('match_conditions')121                LinkRecordsAction(122                    context,123                    link_id,124                    linked_table_id,125                    match_conditions126                ).do_action_with_row(converted_row)127            elif action_info.get('type') == 'run_python_script':128                script_name = action_info.get('script_name')129                workspace_id = action_info.get('workspace_id')130                owner = action_info.get('owner')131                org_id = action_info.get('org_id')132                repo_id = action_info.get('repo_id')133                RunPythonScriptAction(134                    context,135                    script_name,136                    workspace_id,137                    owner,138                    org_id,139                    repo_id,140                    operate_from=RunPythonScriptAction.OPERATE_FROM_WORKFLOW,141                    operator=workflow_token142                ).do_action_with_row(converted_row)143            elif action_info.get('type') == 'add_record_to_other_table':144                row = action_info.get('row')145                dst_table_id = action_info.get('dst_table_id')146                logger.debug('row: %s dst_table_id: %s', row, dst_table_id)147                AddRecordToOtherTableAction(148                    context,149                    dst_table_id,150                    row151                ).do_action_without_row()152        except ActionInvalid as e:153            logger.error('task_id: %s node_id: %s action: %s invalid', task_id, node_id, action_info)154        except Exception as e:155            logger.exception(e)156            logger.error('workflow: %s, task: %s node: %s do action: %s error: %s', workflow_token, task_id, node_id, action_info, e)157class WorkflowActionsHandler(Thread):158    def __init__(self, config):159        Thread.__init__(self)160        self._finished = Event()161        self._db_session_class = init_db_session_class(config)162        self._redis_client = RedisClient(config)163    164    def run(self):165        logger.info('Starting handle workflow actions...')166        subscriber = self._redis_client.get_subscriber('workflow-actions')167        while not self._finished.is_set():168            try:169                message = subscriber.get_message()170                if message is not None:171                    sub_data = json.loads(message['data'])172                    session = self._db_session_class()173                    task_id = sub_data['task_id']174                    node_id = sub_data['node_id']175                    try:176                        do_workflow_actions(task_id, node_id, session)177                    except Exception as e:178                        logger.exception(e)179                        logger.error('task: %s node: %s do actions error: %s', task_id, node_id, e)180                    finally:181                        session.close()182                else:183                    time.sleep(0.5)184            except Exception as e:185                logger.error('Failed get workflow-actions message: %s', e)...

rl_utils.py

Source:rl_utils.py

...26        self.state.append(state)27        self.reward.append(reward)28    def store_oppo_hidden_prob(self, oppo_hidden_prob):29        self.oppo_hidden_prob.append(oppo_hidden_prob)30    def store_action_info(self, choose_action_return):31        self.action.append(choose_action_return[0])32        self.logp_a.append(choose_action_return[1])33        self.entropy.append(choose_action_return[2])34        self.value.append(choose_action_return[3])35        self.hidden_state.append(choose_action_return[6])36    def get_data(self, is_meta_mapg=False):37        data = dict({38            "state":np.stack(self.state),39            "action":np.stack(self.action),40            "logp_a":torch.stack(self.logp_a) if is_meta_mapg else np.stack(self.logp_a),41            "entropy":np.stack(self.entropy),42            "value":np.stack(self.value).squeeze(axis=1),43            "reward":np.stack(self.reward).reshape(-1, 1),44            "oppo_hidden_prob":np.stack(self.oppo_hidden_prob).squeeze(axis=1),45            "hidden_state":np.stack(self.hidden_state).squeeze(axis=1) if np.any(np.stack(self.hidden_state) != None) else np.stack(self.hidden_state),46        })47        if hasattr(self, "oppo_logp_a"):48            data["oppo_logp_a"] = np.stack(self.oppo_logp_a)49        return data50    def get_score(self):51        score = 052        for i in range(len(self.reward)):53            score += self.reward[i]54        return float(score)55def collect_trajectory(agents, env, args, global_step, is_prophetic=False, greedy=False):56    memories = [[], []]57    scores = [[], []]58    for _ in range(1, args.eps_per_epoch + 1):59        hidden_state = [agent.init_hidden_state(n_batch=1) for agent in agents]60        oppo_hidden_prob = np.array([None, None])61        state = env.reset()62        temp_memory = [Episode_Memory(), Episode_Memory()]63        while True:64            global_step += 165            actions = np.array([0, 0], dtype=int)66            for agent_idx, agent in enumerate(agents):67                if type(agent).__name__ == "MBAM":68                    action_info = agent.choose_action(state[agent_idx], hidden_state=hidden_state[agent_idx], oppo_hidden_prob=oppo_hidden_prob[1-agent_idx] if is_prophetic == True else None, greedy=greedy)69                elif type(agent).__name__ == "PPO":70                    action_info = agent.choose_action(state[agent_idx], hidden_state=hidden_state[agent_idx], oppo_hidden_prob=None, greedy=greedy)71                else:72                    raise TypeError73                if args.prophetic_onehot:74                    temp_a = action_info[0].item()75                    oppo_hidden_prob[agent_idx] = np.eye(action_info[5].shape[1])[temp_a].reshape(1, -1)76                    action_info = (action_info[0], action_info[1], action_info[2], action_info[3], action_info[4], oppo_hidden_prob[agent_idx], action_info[6])77                else:78                    oppo_hidden_prob[agent_idx] = action_info[5]79                temp_memory[agent_idx].store_action_info(action_info)80                temp_memory[1 - agent_idx].store_oppo_hidden_prob(action_info[5])81                hidden_state[agent_idx] = action_info[6]82                actions[agent_idx] = action_info[0].item()83            state_, reward, done, info = env.step(actions)84            for i in range(len(agents)):85                temp_memory[i].store_env_info(state[i], reward[i])86            if not is_prophetic:87                for agent_idx, agent in enumerate(agents):88                    if hasattr(agent, "observe_oppo_action"):89                        agent.observe_oppo_action(state=state[agent_idx], oppo_action=actions[1-agent_idx],90                                                  iteration=global_step, no_log=False)91            state = state_92            if done:93                for i in range(len(agents)):94                    temp_memory[i].store_final_state(state_[i], info)95                    memories[i].append(temp_memory[i])96                    scores[i].append(temp_memory[i].get_score())97                agents[1].logger.log_performance(tag=agents[1].name + "/steps", iteration=global_step,98                                                 Same_pick_sum=info["same_pick_sum"],99                                                 Coin_sum=info["coin_sum"],100                                                 Pick_ratio=info["same_pick_sum"]/info["coin_sum"])101                break102    scores = [sum(scores[i])/len(scores[i]) for i in range(len(agents))]103    return memories, scores, global_step104def collect_trajectory_reversed(agents, env, args, global_step, is_prophetic=False):105    memories = [[], []]106    scores = [[], []]107    for _ in range(1, args.eps_per_epoch + 1):108        hidden_state = [agent.init_hidden_state(n_batch=1) for agent in agents]109        oppo_hidden_prob = np.array([None, None])110        state = env.reset()111        temp_memory = [Episode_Memory(), Episode_Memory()]112        while True:113            global_step += 1114            actions = np.array([0, 0], dtype=int)115            for agent_idx, agent in list(enumerate(agents))[::-1]:116                if type(agent).__name__ == "MBAM":117                    action_info = agent.choose_action(state[agent_idx], hidden_state=hidden_state[agent_idx], oppo_hidden_prob=oppo_hidden_prob[1-agent_idx] if is_prophetic == True else None)118                elif type(agent).__name__ == "PPO":119                    action_info = agent.choose_action(state[agent_idx], hidden_state=hidden_state[agent_idx], oppo_hidden_prob=None)120                else:121                    raise TypeError122                if args.prophetic_onehot:123                    temp_a = action_info[0].item()124                    oppo_hidden_prob[agent_idx] = np.eye(action_info[5].shape[1])[temp_a].reshape(1, -1)125                    action_info = (action_info[0], action_info[1], action_info[2], action_info[3], action_info[4], oppo_hidden_prob[agent_idx], action_info[6])126                else:127                    oppo_hidden_prob[agent_idx] = action_info[5]128                temp_memory[agent_idx].store_action_info(action_info)129                temp_memory[1 - agent_idx].store_oppo_hidden_prob(action_info[5])130                hidden_state[agent_idx] = action_info[6]131                actions[agent_idx] = action_info[0].item()132            state_, reward, done, info = env.step(actions)133            for i in range(len(agents)):134                temp_memory[i].store_env_info(state[i], reward[i])135            if not is_prophetic:136                for agent_idx, agent in enumerate(agents):137                    if hasattr(agent, "observe_oppo_action"):138                        agent.observe_oppo_action(state=state[agent_idx], oppo_action=actions[1-agent_idx],139                                                  iteration=global_step, no_log=False)140            state = state_141            if done:142                for i in range(len(agents)):...

action.py

Source:action.py

1#  Copyright 2008-2015 Nokia Networks2#  Copyright 2016-     Robot Framework Foundation3#4#  Licensed under the Apache License, Version 2.0 (the "License");5#  you may not use this file except in compliance with the License.6#  You may obtain a copy of the License at7#8#      http://www.apache.org/licenses/LICENSE-2.09#10#  Unless required by applicable law or agreed to in writing, software11#  distributed under the License is distributed on an "AS IS" BASIS,12#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13#  See the License for the specific language governing permissions and14#  limitations under the License.15def ActionFactory(action_info):16    if action_info.is_separator():17        return _MenuSeparator(action_info)18    return Action(action_info)19class _Registrable(object):20    def __init__(self, action_info):21        self._registered_to = []22        self.action = None23        self.shortcut = None24        self.icon = None25        self._insertion_point = action_info.insertion_point26        self._enabled = True27        self._enabled_status_listeners = []28    disable = enable = lambda event: None29    def is_separator(self):30        return False31    def get_insertion_index(self, menu):32        return self._insertion_point.get_index(menu)33    def register(self, registerer):34        self._registered_to.append(registerer)35    def unregister(self):36        for registerer in self._registered_to:37            registerer.unregister(self)38        self._registered_to = []39    def has_action(self):40        return self.action is not None41    def has_shortcut(self):42        return bool(self.shortcut)43    def has_icon(self):44        return self.icon is not None45    def inform_changes_in_enabled_status(self, listener):46        self._enabled_status_listeners.append(listener)47class Action(_Registrable):48    """Acts based on user actions if action is enabled. Created from `ActionInfo`.49    If `ActionInfo` contains container, acts and allows to select related UI50    widgets (menu item, toolbar button) and shortcuts only if the focus is in the given 51    container or its children.52    Action can be set enabled or disabled which enables/disables also related UI53    widgets and shortcut.54    """55    def __init__(self, action_info):56        _Registrable.__init__(self, action_info)57        self.menu_name = action_info.menu_name58        self.name = action_info.name59        self.action = action_info.action60        self.container = action_info.container61        self.shortcut = action_info.shortcut62        self.icon = action_info.icon63        self.doc = action_info.doc64        # print("DEBUG: Action: %s::%s" % (self.menu_name,self.name))65    def get_shortcut(self):66        return self.shortcut.printable67    def act(self, event):68        if self.is_active():69            self.action(event)70    def disable(self):71        """Disables action and related menu item, toolbar button and shortcut"""72        self._enabled = False73        self._inform_changes_in_enabled_status()74    def enable(self):75        """Enables action and related menu item, toolbar button and shortcut"""76        self._enabled = True77        self._inform_changes_in_enabled_status()78    def _inform_changes_in_enabled_status(self):79        for listener in self._enabled_status_listeners:80            listener.enabled_status_changed(self)81    def is_active(self):82        if self._is_always_inactive() or not self._enabled:83            return False84        if self._is_always_active():85            return True86        return self._container_is_active()87    def _is_always_inactive(self):88        return self.action is None89    def _is_always_active(self):90        return self.container is None91    def _container_is_active(self):92        if not self.container.IsShownOnScreen():93            return False94        widget = self.container.FindFocus()95        while widget:96            if widget == self.container.Parent:97                return True98            widget = widget.GetParent()99        return False100class _MenuSeparator(_Registrable):101    def __init__(self, action_info):102        _Registrable.__init__(self, action_info)103        self.menu_name = action_info.menu_name104        self.name = '---'105    def is_separator(self):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.