From 69d739fcf13a3f17e9320c551444772f9694c6c9 Mon Sep 17 00:00:00 2001 From: Justin Tervala Date: Fri, 29 Sep 2017 15:27:06 -0400 Subject: [PATCH] Revert "add implementation and unit tests for a trigger step" This reverts commit 7dccf119be47d5950d31e8f5ef0ddb8458a9edc8. --- core/step.py | 228 +++++++++++++++++++--- core/stepaction.py | 265 -------------------------- core/steptrigger.py | 149 --------------- tests/test_step_action.py | 374 ------------------------------------- tests/test_step_trigger.py | 171 ----------------- 5 files changed, 207 insertions(+), 980 deletions(-) delete mode 100644 core/stepaction.py delete mode 100644 core/steptrigger.py delete mode 100644 tests/test_step_action.py delete mode 100644 tests/test_step_trigger.py diff --git a/core/step.py b/core/step.py index 1e0248d07..d95de28c3 100644 --- a/core/step.py +++ b/core/step.py @@ -1,14 +1,23 @@ +import json from collections import namedtuple import logging -from core.stepaction import StepAction -from core.steptrigger import StepTrigger - +from core import contextdecorator +import core.config.config +from core.decorators import ActionResult +from core.executionelement import ExecutionElement +from core.helpers import (get_app_action_api, InvalidElementConstructed, InvalidInput, + dereference_step_routing, format_exception_message) +from core.nextstep import NextStep +from core.widgetsignals import get_widget_signal +from apps import get_app_action +from core.validator import validate_app_action_parameters +import uuid logger = logging.getLogger(__name__) _Widget = namedtuple('Widget', ['app', 'widget']) -class Step(StepAction, StepTrigger): +class Step(ExecutionElement): def __init__(self, name='', action='', @@ -16,17 +25,14 @@ def __init__(self, device='', inputs=None, next_steps=None, - flags=None, - status='Success', position=None, widgets=None, risk=0, uid=None, templated=False, raw_json=None): - """Initializes a new Step object. A Workflow has many steps that it executes. A Step object can be either a Step - Action or Step Trigger object. - + """Initializes a new Step object. A Workflow has many steps that it executes. + Args: name (str, optional): The name of the Step object. Defaults to an empty string. action (str, optional): The name of the action associated with a Step. Defaults to an empty string. @@ -45,25 +51,205 @@ def __init__(self, Created from uuid.uuid4().hex in Python raw_json (dict, optional): JSON representation of this object. Used for Jinja templating """ - super(Step, self).__init__(name, action, app, device, inputs, next_steps, flags, status, position, widgets, - risk, uid, templated, raw_json) + ExecutionElement.__init__(self, name, uid) + if action == '' or app == '': + raise InvalidElementConstructed('Either both action and app or xml must be ' + 'specified in step constructor') + self.action = action + self.app = app + self.run, self.input_api = get_app_action_api(self.app, self.action) + get_app_action(self.app, self.run) + inputs = inputs if inputs is not None else {} + self.templated = templated + if not self.templated: + self.input = validate_app_action_parameters(self.input_api, inputs, self.app, self.action) + else: + self.input = inputs + self.device = device + self.risk = risk + self.conditionals = next_steps if next_steps is not None else [] + self.position = position if position is not None else {} + self.widgets = [_Widget(widget_app, widget_name) + for (widget_app, widget_name) in widgets] if widgets is not None else [] + + self.output = None + self.next_up = None + self.raw_json = raw_json if raw_json is not None else {} + self.execution_uid = 'default' + self.results_sock = None + self.execution_uid = None + + def _update_json(self, updated_json): + self.action = updated_json['action'] + self.app = updated_json['app'] + self.device = updated_json['device'] if 'device' in updated_json else '' + self.risk = updated_json['risk'] if 'risk' in updated_json else 0 + inputs = {arg['name']: arg['value'] for arg in updated_json['inputs']} if 'inputs' in updated_json else {} + if inputs is not None: + if not self.templated: + self.input = validate_app_action_parameters(self.input_api, inputs, self.app, self.action) + else: + self.input = inputs + else: + self.input = validate_app_action_parameters(self.input_api, {}, self.app, self.action) + self.conditionals = [NextStep.from_json(cond_json) for cond_json in updated_json['next']] + + @contextdecorator.context + def render_step(self, **kwargs): + """Uses JINJA templating to render a Step object. + + Args: + kwargs (dict[str]): Arguments to use in the JINJA templating. + """ + if self.templated: + from jinja2 import Environment + env = Environment().from_string(json.dumps(self.raw_json)).render(core.config.config.JINJA_GLOBALS, **kwargs) + self._update_json(updated_json=json.loads(env)) + + def set_input(self, new_input): + """Updates the input for a Step object. + + Args: + new_input (dict): The new inputs for the Step object. + """ + self.input = validate_app_action_parameters(self.input_api, new_input, self.app, self.action) + + def __send_callback(self, callback_name, data={}): + data['sender'] = {} + data['sender']['name'] = self.name + data['sender']['app'] = self.app + data['sender']['action'] = self.action + data['sender']['inputs'] = self.input + data['callback_name'] = callback_name + data['sender']['id'] = self.name + data['sender']['execution_uid'] = self.execution_uid + data['sender']['uid'] = self.uid + if self.results_sock: + self.results_sock.send_json(data) + + def execute(self, instance, accumulator): + """Executes a Step by calling the associated app function. + + Args: + instance (App): The instance of an App object to be used to execute the associated function. + accumulator (dict): Dict containing the results of the previous steps + + Returns: + The result of the executed function. + """ + self.execution_uid = uuid.uuid4().hex + self.__send_callback('Step Started') + try: + args = dereference_step_routing(self.input, accumulator, 'In step {0}'.format(self.name)) + args = validate_app_action_parameters(self.input_api, args, self.app, self.action) + action = get_app_action(self.app, self.run) + result = action(instance, **args) + self.__send_callback('Function Execution Success', + {'name': self.name, 'data': {'result': result.as_json()}}) + except InvalidInput as e: + formatted_error = format_exception_message(e) + logger.error('Error calling step {0}. Error: {1}'.format(self.name, formatted_error)) + self.__send_callback('Step Input Invalid') + self.output = ActionResult('error: {0}'.format(formatted_error), 'InvalidInput') + raise + except Exception as e: + formatted_error = format_exception_message(e) + logger.error('Error calling step {0}. Error: {1}'.format(self.name, formatted_error)) + self.output = ActionResult('error: {0}'.format(formatted_error), 'UnhandledException') + raise + else: + self.output = result + for widget in self.widgets: + get_widget_signal(widget.app, widget.widget).send(self, data=json.dumps({"result": result.as_json()})) + logger.debug('Step {0}-{1} (uid {2}) executed successfully'.format(self.app, self.action, self.uid)) + return result + + def get_next_step(self, accumulator): + """Gets the NextStep object to be executed after the current Step. + + Args: + accumulator (dict): A record of teh previously-executed steps. Of form {step_name: result} + + Returns: + The NextStep object to be executed. + """ + + for next_step in self.conditionals: + next_step.results_sock = self.results_sock + next_step = next_step(self.output, accumulator) + if next_step is not None: + self.next_up = next_step + self.__send_callback('Conditionals Executed') + return next_step def __repr__(self): output = {'uid': self.uid, 'name': self.name, 'action': self.action, + 'app': self.app, + 'device': self.device, + 'risk': str(self.risk), + 'input': self.input, + 'next': [next_step for next_step in self.conditionals], + 'nextUp': self.next_up, 'position': self.position, 'widget': str([{'app': widget.app, 'name': widget.widget} for widget in self.widgets])} - for attr, value in self.__dict__.items(): - if attr == 'risk': - output[attr] = str(value) - elif attr == 'next_step': - output[attr] = [next_step for next_step in self.conditionals] - elif attr == 'flags': - output[attr] = [flag.as_json() for flag in self.flags] - else: - output[attr] = value + if self.output: + output["output"] = self.output.as_json() + return str(output) + def as_json(self): + """Gets the JSON representation of a Step object. + + Returns: + The JSON representation of a Step object. + """ + output = {"uid": self.uid, + "name": str(self.name), + "action": str(self.action), + "app": str(self.app), + "device": str(self.device), + "risk": self.risk, + "inputs": [{'name': arg_name, 'value': arg_value} for arg_name, arg_value in self.input.items()], + 'widgets': [{'app': widget.app, 'name': widget.widget} for widget in self.widgets], + "position": self.position, + "next": [next_step.as_json() for next_step in self.conditionals if next_step.name is not None]} if self.output: output["output"] = self.output.as_json() - return str(output) \ No newline at end of file + return output + + @staticmethod + def from_json(json_in, position): + """Forms a Step object from the provided JSON object. + + Args: + json_in (dict): The JSON object to convert from. + position (dict): position of the step node of the form {'x': , 'y': } + + Returns: + The Step object parsed from the JSON object. + """ + device = json_in['device'] if ('device' in json_in + and json_in['device'] is not None + and json_in['device'] != 'None') else '' + risk = json_in['risk'] if 'risk' in json_in else 0 + widgets = [] + uid = json_in['uid'] if 'uid' in json_in else uuid.uuid4().hex + if 'widgets' in json_in: + widgets = [(widget['app'], widget['name']) + for widget in json_in['widgets'] if ('app' in widget and 'name' in widget)] + conditionals = [] + if 'next' in json_in: + conditionals = [NextStep.from_json(next_step) for next_step in json_in['next'] if next_step] + return Step(name=json_in['name'], + action=json_in['action'], + app=json_in['app'], + device=device, + risk=risk, + inputs={arg['name']: arg['value'] for arg in json_in['inputs']}, + next_steps=conditionals, + position={key: value for key, value in position.items()}, + widgets=widgets, + uid=uid, + templated=json_in['templated'] if 'templated' in json_in else False, + raw_json=json_in) diff --git a/core/stepaction.py b/core/stepaction.py deleted file mode 100644 index bfa3bd29e..000000000 --- a/core/stepaction.py +++ /dev/null @@ -1,265 +0,0 @@ -import json -from collections import namedtuple -import logging -from core import contextdecorator -import core.config.config -from core.decorators import ActionResult -from core.executionelement import ExecutionElement -from core.helpers import (get_app_action_api, InvalidElementConstructed, InvalidInput, - dereference_step_routing, format_exception_message) -from core.nextstep import NextStep -from core.widgetsignals import get_widget_signal -from apps import get_app_action -from core.validator import validate_app_action_parameters -import uuid -logger = logging.getLogger(__name__) - -_Widget = namedtuple('Widget', ['app', 'widget']) - - -class StepAction(ExecutionElement): - def __init__(self, - name='', - action='', - app='', - device='', - inputs=None, - next_steps=None, - flags=None, - status='', - position=None, - widgets=None, - risk=0, - uid=None, - templated=False, - raw_json=None): - """Initializes a new Step Action object. A Workflow has many steps that it executes. - - Args: - name (str, optional): The name of the Step Action object. Defaults to an empty string. - action (str, optional): The name of the action associated with a Step. Defaults to an empty string. - app (str, optional): The name of the app associated with the Step. Defaults to an empty string. - device (str, optional): The name of the device associated with the app associated with the Step. Defaults - to an empty string. - inputs (dict, optional): A dictionary of Argument objects that are input to the step execution. Defaults - to None. - next_steps (list[NextStep], optional): A list of NextStep Action objects for the Step Action object. - Defaults to None. - position (dict, optional): A dictionary with the x and y coordinates of the Step Action object. This is used - for UI display purposes. Defaults to None. - widgets (list[tuple(str, str)], optional): A list of widget tuples, which holds the app and the - corresponding widget. Defaults to None. - risk (int, optional): The risk associated with the Step. Defaults to 0. - uid (str, optional): A universally unique identifier for this object. - Created from uuid.uuid4().hex in Python - raw_json (dict, optional): JSON representation of this object. Used for Jinja templating - """ - ExecutionElement.__init__(self, name, uid) - if action == '' or app == '': - raise InvalidElementConstructed('Either both action and app or xml must be ' - 'specified in step constructor') - if flags is not None and status is not '': - raise InvalidElementConstructed('Flags and status must be specified in either a next step or ' - 'trigger step constructor and not an action step constructor') - self.action = action - self.app = app - self.run, self.input_api = get_app_action_api(self.app, self.action) - get_app_action(self.app, self.run) - inputs = inputs if inputs is not None else {} - self.templated = templated - if not self.templated: - self.input = validate_app_action_parameters(self.input_api, inputs, self.app, self.action) - else: - self.input = inputs - self.device = device - self.risk = risk - self.conditionals = next_steps if next_steps is not None else [] - self.position = position if position is not None else {} - self.widgets = [_Widget(widget_app, widget_name) - for (widget_app, widget_name) in widgets] if widgets is not None else [] - - self.output = None - self.next_up = None - self.raw_json = raw_json if raw_json is not None else {} - self.execution_uid = 'default' - self.results_sock = None - self.execution_uid = None - - def _update_json(self, updated_json): - self.action = updated_json['action'] - self.app = updated_json['app'] - self.device = updated_json['device'] if 'device' in updated_json else '' - self.risk = updated_json['risk'] if 'risk' in updated_json else 0 - inputs = {arg['name']: arg['value'] for arg in updated_json['inputs']} if 'inputs' in updated_json else {} - if inputs is not None: - if not self.templated: - self.input = validate_app_action_parameters(self.input_api, inputs, self.app, self.action) - else: - self.input = inputs - else: - self.input = validate_app_action_parameters(self.input_api, {}, self.app, self.action) - self.conditionals = [NextStep.from_json(cond_json) for cond_json in updated_json['next']] - - @contextdecorator.context - def render_step(self, **kwargs): - """Uses JINJA templating to render a Step Action object. - - Args: - kwargs (dict[str]): Arguments to use in the JINJA templating. - """ - if self.templated: - from jinja2 import Environment - env = Environment().from_string(json.dumps(self.raw_json)).render(core.config.config.JINJA_GLOBALS, **kwargs) - self._update_json(updated_json=json.loads(env)) - - def set_input(self, new_input): - """Updates the input for a Step Action object. - - Args: - new_input (dict): The new inputs for the Step Action object. - """ - self.input = validate_app_action_parameters(self.input_api, new_input, self.app, self.action) - - def __send_callback(self, callback_name, data={}): - data['sender'] = {} - data['sender']['name'] = self.name - data['sender']['app'] = self.app - data['sender']['action'] = self.action - data['sender']['inputs'] = self.input - data['callback_name'] = callback_name - data['sender']['id'] = self.name - data['sender']['execution_uid'] = self.execution_uid - data['sender']['uid'] = self.uid - if self.results_sock: - self.results_sock.send_json(data) - - def execute(self, instance, accumulator): - """Executes a Step by calling the associated app function. - - Args: - instance (App): The instance of an App object to be used to execute the associated function. - accumulator (dict): Dict containing the results of the previous steps - - Returns: - The result of the executed function. - """ - self.execution_uid = uuid.uuid4().hex - self.__send_callback('Step Started') - try: - args = dereference_step_routing(self.input, accumulator, 'In step {0}'.format(self.name)) - args = validate_app_action_parameters(self.input_api, args, self.app, self.action) - action = get_app_action(self.app, self.run) - result = action(instance, **args) - self.__send_callback('Function Execution Success', - {'name': self.name, 'data': {'result': result.as_json()}}) - except InvalidInput as e: - formatted_error = format_exception_message(e) - logger.error('Error calling step {0}. Error: {1}'.format(self.name, formatted_error)) - self.__send_callback('Step Input Invalid') - self.output = ActionResult('error: {0}'.format(formatted_error), 'InvalidInput') - raise - except Exception as e: - formatted_error = format_exception_message(e) - logger.error('Error calling step {0}. Error: {1}'.format(self.name, formatted_error)) - self.output = ActionResult('error: {0}'.format(formatted_error), 'UnhandledException') - raise - else: - self.output = result - for widget in self.widgets: - get_widget_signal(widget.app, widget.widget).send(self, data=json.dumps({"result": result.as_json()})) - logger.debug('Step {0}-{1} (uid {2}) executed successfully'.format(self.app, self.action, self.uid)) - return result - - def get_next_step(self, accumulator): - """Gets the NextStep object to be executed after the current Step. - - Args: - accumulator (dict): A record of teh previously-executed steps. Of form {step_name: result} - - Returns: - The NextStep object to be executed. - """ - - for next_step in self.conditionals: - next_step.results_sock = self.results_sock - next_step = next_step(self.output, accumulator) - if next_step is not None: - self.next_up = next_step - self.__send_callback('Conditionals Executed') - return next_step - - def __repr__(self): - output = {'uid': self.uid, - 'name': self.name, - 'action': self.action, - 'app': self.app, - 'device': self.device, - 'risk': str(self.risk), - 'input': self.input, - 'next': [next_step for next_step in self.conditionals], - 'nextUp': self.next_up, - 'position': self.position, - 'widget': str([{'app': widget.app, 'name': widget.widget} for widget in self.widgets])} - if self.output: - output["output"] = self.output.as_json() - return str(output) - - def as_json(self): - """Gets the JSON representation of a Step Action object. - - Returns: - The JSON representation of a Step Action object. - """ - output = {"uid": self.uid, - "name": str(self.name), - "action": str(self.action), - "app": str(self.app), - "device": str(self.device), - "risk": self.risk, - "inputs": [{'name': arg_name, 'value': arg_value} for arg_name, arg_value in self.input.items()], - 'widgets': [{'app': widget.app, 'name': widget.widget} for widget in self.widgets], - "position": self.position, - "next": [next_step.as_json() for next_step in self.conditionals if next_step.name is not None]} - if self.output: - output["output"] = self.output.as_json() - return output - - @staticmethod - def from_json(json_in, position): - """Forms a Step Action object from the provided JSON object. - - Args: - json_in (dict): The JSON object to convert from. - position (dict): position of the step node of the form {'x': , 'y': } - - Returns: - The Step Action object parsed from the JSON object. - """ - device = json_in['device'] if ('device' in json_in - and json_in['device'] is not None - and json_in['device'] != 'None') else '' - risk = json_in['risk'] if 'risk' in json_in else 0 - widgets = [] - uid = json_in['uid'] if 'uid' in json_in else uuid.uuid4().hex - if 'widgets' in json_in: - widgets = [(widget['app'], widget['name']) - for widget in json_in['widgets'] if ('app' in widget and 'name' in widget)] - conditionals = [] - if 'next' in json_in: - conditionals = [NextStep.from_json(next_step) for next_step in json_in['next'] if next_step] - return StepAction(name=json_in['name'], - action=json_in['action'], - app=json_in['app'], - device=device, - risk=risk, - inputs={arg['name']: arg['value'] for arg in json_in['inputs']}, - next_steps=conditionals, - position={key: value for key, value in position.items()}, - widgets=widgets, - uid=uid, - templated=json_in['templated'] if 'templated' in json_in else False, - raw_json=json_in) - - def __dir__(self): - return ['name', 'action', 'app', 'device', 'inputs', 'next_steps', 'position', 'widgets', 'risk', 'uid', - 'templated', 'raw_json'] \ No newline at end of file diff --git a/core/steptrigger.py b/core/steptrigger.py deleted file mode 100644 index 311a1e1ef..000000000 --- a/core/steptrigger.py +++ /dev/null @@ -1,149 +0,0 @@ -from collections import namedtuple -import logging -from core.executionelement import ExecutionElement -from core.helpers import (InvalidElementConstructed) -from core.flag import Flag -import uuid -logger = logging.getLogger(__name__) - -_Widget = namedtuple('Widget', ['app', 'widget']) - - -class StepTrigger(ExecutionElement): - def __init__(self, - name='', - action='', - app='', - device='', - inputs=None, - next_steps=None, - flags=None, - status='Success', - position=None, - widgets=None, - risk=0, - uid=None, - templated=False, - raw_json=None): - """Initializes a new Step Trigger object. A Workflow has many steps that it executes. - - Args: - name (str, optional): The name of the Step Trigger object. Defaults to an empty string. - action (str, optional): The name of the action associated with a Step Trigger. Defaults to an empty string. - flags (list[Flag], optional): A list of Flag objects for the Step Trigger object. Defaults to None. - position (dict, optional): A dictionary with the x and y coordinates of the Step Trigger object. This is - used for UI display purposes. Defaults to None. - widgets (list[tuple(str, str)], optional): A list of widget tuples, which holds the app and the - corresponding widget. Defaults to None. - risk (int, optional): The risk associated with the Step. Defaults to 0. - uid (str, optional): A universally unique identifier for this object. - Created from uuid.uuid4().hex in Python - raw_json (dict, optional): JSON representation of this object. Used for Jinja templating - """ - ExecutionElement.__init__(self, name, uid) - if action == '': - raise InvalidElementConstructed('Either action or xml must be specified ' - 'in step trigger constructor') - if app is not '' and device is not '' and inputs is not None and next_steps is not None and risk is not 0: - raise InvalidElementConstructed('App, device, inputs, next steps, and risk must be specified in ' - 'an action step constructor and not a trigger step constructor') - self.action = action - self.templated = templated - self.flags = flags if flags is not None else [] - self.status = status - self.position = position if position is not None else {} - self.widgets = [_Widget(widget_app, widget_name) - for (widget_app, widget_name) in widgets] if widgets is not None else [] - - self.output = None - self.next_up = None - self.raw_json = raw_json if raw_json is not None else {} - self.execution_uid = 'default' - self.results_sock = None - self.execution_uid = None - - def __call__(self, data_in, accumulator): - for flag in self.flags: - flag.results_sock = self.results_sock - if data_in is not None and data_in.status == self.status: - if all(flag(data_in=data_in.result, accumulator=accumulator) for flag in self.flags): - self.__send_callback("Trigger Step Taken") - logger.debug('TriggerStep is valid for input {0}'.format(data_in)) - - return self.name - else: - logger.debug('TriggerStep is not valid for input {0}'.format(data_in)) - self.__send_callback("Trigger Step Not Taken") - return None - else: - return None - - def __send_callback(self, callback_name, data={}): - data['sender'] = {} - data['sender']['name'] = self.name - data['sender']['action'] = self.action - data['callback_name'] = callback_name - data['sender']['id'] = self.name - data['sender']['execution_uid'] = self.execution_uid - data['sender']['uid'] = self.uid - if self.results_sock: - self.results_sock.send_json(data) - - def __repr__(self): - output = {'uid': self.uid, - 'name': self.name, - 'action': self.action, - 'flags': [flag.as_json() for flag in self.flags], - 'status': self.status, - 'position': self.position, - 'widget': str([{'app': widget.app, 'name': widget.widget} for widget in self.widgets])} - if self.output: - output["output"] = self.output.as_json() - return str(output) - - def as_json(self): - """Gets the JSON representation of a Step Trigger object. - - Returns: - The JSON representation of a Step Trigger object. - """ - output = {"uid": self.uid, - "name": str(self.name), - "action": str(self.action), - "flags": [flag.as_json() for flag in self.flags], - "status": self.status, - "widgets": [{'app': widget.app, 'name': widget.widget} for widget in self.widgets], - "position": self.position} - if self.output: - output["output"] = self.output.as_json() - return output - - @staticmethod - def from_json(json_in, position): - """Forms a Step Trigger object from the provided JSON object. - - Args: - json_in (dict): The JSON object to convert from. - position (dict): position of the step node of the form {'x': , 'y': } - - Returns: - The Step Trigger object parsed from the JSON object. - """ - widgets = [] - uid = json_in['uid'] if 'uid' in json_in else uuid.uuid4().hex - if 'widgets' in json_in: - widgets = [(widget['app'], widget['name']) - for widget in json_in['widgets'] if ('app' in widget and 'name' in widget)] - flags = [] - if 'flags' in json_in: - flags = [Flag.from_json(flag) for flag in json_in['flags'] if flag] - status = json_in['status'] if 'status' in json_in else 'Success' - return StepTrigger(name=json_in['name'], - action=json_in['action'], - flags=flags, - status=status, - position={key: value for key, value in position.items()}, - widgets=widgets, - uid=uid, - templated=json_in['templated'] if 'templated' in json_in else False, - raw_json=json_in) diff --git a/tests/test_step_action.py b/tests/test_step_action.py deleted file mode 100644 index b17c24c0f..000000000 --- a/tests/test_step_action.py +++ /dev/null @@ -1,374 +0,0 @@ -import unittest -import socket -from core.flag import Flag -from core.stepaction import StepAction, _Widget -from core.nextstep import NextStep -from core.decorators import ActionResult -import core.config.config -import core.config.paths -from tests.config import test_apps_path, function_api_path -from core.instance import Instance -from core.helpers import (import_all_apps, UnknownApp, UnknownAppAction, InvalidInput, import_all_flags, - import_all_filters) -import uuid - - -class TestStepAction(unittest.TestCase): - @classmethod - def setUpClass(cls): - import_all_apps(path=test_apps_path) - core.config.config.load_app_apis(apps_path=test_apps_path) - core.config.config.flags = import_all_flags('tests.util.flagsfilters') - core.config.config.filters = import_all_filters('tests.util.flagsfilters') - core.config.config.load_flagfilter_apis(path=function_api_path) - - def setUp(self): - self.uid = uuid.uuid4().hex - self.basic_json = {'app': 'HelloWorld', - 'action': 'helloWorld', - 'device': '', - 'name': '', - 'next': [], - 'position': {}, - 'inputs': [], - 'widgets': [], - 'risk': 0, - 'uid': self.uid} - self.basic_input_json = {'app': 'HelloWorld', - 'action': 'helloWorld', - 'name': '', - 'next': [], - 'position': {}, - 'inputs': [], - 'uid': self.uid} - - def __compare_init(self, elem, name, action, app, device, inputs, next_steps, - widgets, risk=0., position=None, uid=None): - self.assertEqual(elem.name, name) - self.assertEqual(elem.action, action) - self.assertEqual(elem.app, app) - self.assertEqual(elem.device, device) - self.assertDictEqual({key: input_element for key, input_element in elem.input.items()}, inputs) - self.assertListEqual([conditional.as_json() for conditional in elem.conditionals], next_steps) - self.assertEqual(elem.risk, risk) - widgets = [_Widget(app, widget) for (app, widget) in widgets] - self.assertEqual(len(elem.widgets), len(widgets)) - for widget in elem.widgets: - self.assertIn(widget, widgets) - position = position if position is not None else {} - self.assertDictEqual(elem.position, position) - self.assertIsNone(elem.output) - self.assertFalse(elem.templated) - if uid is None: - self.assertIsNotNone(elem.uid) - else: - self.assertEqual(elem.uid, uid) - - def test_init_app_and_action_only(self): - step = StepAction(app='HelloWorld', action='helloWorld') - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_init_with_uid(self): - uid = uuid.uuid4().hex - step = StepAction(app='HelloWorld', action='helloWorld', uid=uid) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], uid=uid) - - def test_init_app_and_action_name_different_than_method_name(self): - step = StepAction(app='HelloWorld', action='Hello World') - self.__compare_init(step, '', 'Hello World', 'HelloWorld', '', {}, [], []) - - def test_init_invalid_app(self): - with self.assertRaises(UnknownApp): - StepAction(app='InvalidApp', action='helloWorld') - - def test_init_invalid_action(self): - with self.assertRaises(UnknownAppAction): - StepAction(app='HelloWorld', action='invalid') - - def test_init_with_inputs_no_conversion(self): - step = StepAction(app='HelloWorld', action='returnPlusOne', inputs={'number': -5.6}) - self.__compare_init(step, '', 'returnPlusOne', 'HelloWorld', '', {'number': -5.6}, [], []) - - def test_init_with_inputs_with_conversion(self): - step = StepAction(app='HelloWorld', action='returnPlusOne', inputs={'number': '-5.6'}) - self.__compare_init(step, '', 'returnPlusOne', 'HelloWorld', '', {'number': -5.6}, [], []) - - def test_init_with_invalid_input_name(self): - with self.assertRaises(InvalidInput): - StepAction(app='HelloWorld', action='returnPlusOne', inputs={'invalid': '-5.6'}) - - def test_init_with_invalid_input_type(self): - with self.assertRaises(InvalidInput): - StepAction(app='HelloWorld', action='returnPlusOne', inputs={'number': 'invalid'}) - - def test_init_with_name(self): - step = StepAction(app='HelloWorld', action='helloWorld', name='name') - self.__compare_init(step, 'name', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_init_with_device(self): - step = StepAction(app='HelloWorld', action='helloWorld', device='dev') - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', 'dev', {}, [], []) - - def test_init_with_risk(self): - step = StepAction(app='HelloWorld', action='helloWorld', risk=42.3) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], risk=42.3) - - def test_init_with_widgets(self): - widgets = [('aaa', 'bbb'), ('ccc', 'ddd'), ('eee', 'fff')] - step = StepAction(app='HelloWorld', action='helloWorld', widgets=widgets) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], widgets) - - def test_init_with_position(self): - step = StepAction(app='HelloWorld', action='helloWorld', position={'x': -12.3, 'y': 485}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], position={'x': -12.3, 'y': 485}) - - def test_init_with_next_steps(self): - next_steps = [NextStep(), NextStep(name='name'), NextStep(name='name2')] - step = StepAction(app='HelloWorld', action='helloWorld', next_steps=next_steps) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [step.as_json() for step in next_steps], []) - - def test_as_json(self): - step = StepAction(app='HelloWorld', action='helloWorld', uid=self.uid) - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_name(self): - step = StepAction(app='HelloWorld', action='helloWorld', name='name', uid=self.uid) - self.basic_json['name'] = 'name' - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_device(self): - step = StepAction(app='HelloWorld', action='helloWorld', device='device', uid=self.uid) - self.basic_json['device'] = 'device' - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_risk(self): - step = StepAction(app='HelloWorld', action='helloWorld', risk=120.6, uid=self.uid) - self.basic_json['risk'] = 120.6 - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_inputs(self): - step = StepAction(app='HelloWorld', action='returnPlusOne', inputs={'number': '-5.6'}, uid=self.uid) - self.basic_json['action'] = 'returnPlusOne' - self.basic_json['inputs'] = [{'name': 'number', 'value': -5.6}] - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_next_steps(self): - next_steps = [NextStep(), NextStep(name='name'), NextStep(name='name2')] - step = StepAction(app='HelloWorld', action='helloWorld', next_steps=next_steps, uid=self.uid) - self.basic_json['next'] = [next_step.as_json() for next_step in next_steps] - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_position(self): - step = StepAction(app='HelloWorld', action='helloWorld', position={'x': -12.3, 'y': 485}, uid=self.uid) - self.basic_json['position'] = {'x': -12.3, 'y': 485} - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_widgets(self): - widgets = [('aaa', 'bbb'), ('ccc', 'ddd'), ('eee', 'fff')] - step = StepAction(app='HelloWorld', action='helloWorld', widgets=widgets, uid=self.uid) - self.basic_json['widgets'] = [{'app': widget[0], 'name': widget[1]} for widget in widgets] - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_after_executed(self): - step = StepAction(app='HelloWorld', action='helloWorld', uid=self.uid) - instance = Instance.create(app_name='HelloWorld', device_name='device1') - step.execute(instance.instance, {}) - step_json = step.as_json() - self.assertDictEqual(step_json['output'], ActionResult({'message': 'HELLO WORLD'}, 'Success').as_json()) - - def test_from_json_app_and_action_only(self): - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_from_json_with_uid(self): - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], uid=self.uid) - - def test_from_json_invalid_app(self): - self.basic_input_json['app'] = 'Invalid' - with self.assertRaises(UnknownApp): - StepAction.from_json(self.basic_input_json, {}) - - def test_from_json_invalid_action(self): - self.basic_input_json['action'] = 'invalid' - with self.assertRaises(UnknownAppAction): - StepAction.from_json(self.basic_input_json, {}) - - def test_from_json_with_name(self): - self.basic_input_json['name'] = 'name1' - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, 'name1', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_from_json_with_risk(self): - self.basic_input_json['risk'] = 132.3 - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], risk=132.3) - - def test_from_json_with_device(self): - self.basic_input_json['device'] = 'device1' - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', 'device1', {}, [], []) - - def test_from_json_with_device_is_none(self): - self.basic_input_json['device'] = None - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_from_json_with_device_is_none_string(self): - self.basic_input_json['device'] = 'None' - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], []) - - def test_from_json_with_widgets(self): - widget_json = [{'name': 'widget_name', 'app': 'app1'}, {'name': 'w2', 'app': 'app2'}] - widget_tuples = [('app1', 'widget_name'), ('app2', 'w2')] - self.basic_input_json['widgets'] = widget_json - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], widget_tuples) - - def test_from_json_with_inputs(self): - self.basic_input_json['action'] = 'Add Three' - self.basic_input_json['inputs'] = [{'name': 'num1', 'value': '-5.6'}, {'name': 'num2', 'value': '4.3'}, - {'name': 'num3', 'value': '-10.265'}] - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'Add Three', 'HelloWorld', '', - {'num1': -5.6, 'num2': 4.3, 'num3': -10.265}, [], []) - - def test_from_json_with_inputs_invalid_name(self): - self.basic_input_json['action'] = 'Add Three' - self.basic_input_json['inputs'] = [{'name': 'num1', 'value': '-5.6'}, {'name': 'invalid', 'value': '4.3'}, - {'name': 'num3', 'value': '-10.265'}] - with self.assertRaises(InvalidInput): - StepAction.from_json(self.basic_input_json, {}) - - def test_from_json_with_inputs_invalid_format(self): - self.basic_input_json['action'] = 'Add Three' - self.basic_input_json['inputs'] = [{'name': 'num1', 'value': '-5.6'}, {'name': 'num2', 'value': '4.3'}, - {'name': 'num3', 'value': 'invalid'}] - with self.assertRaises(InvalidInput): - StepAction.from_json(self.basic_input_json, {}) - - def test_from_json_with_step_routing(self): - self.basic_input_json['action'] = 'Add Three' - self.basic_input_json['inputs'] = [{'name': 'num1', 'value': '-5.6'}, {'name': 'num2', 'value': '@step1'}, - {'name': 'num3', 'value': '@step2'}] - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'Add Three', 'HelloWorld', '', - {'num1': -5.6, 'num2': '@step1', 'num3': '@step2'}, [], []) - - def test_from_json_with_position(self): - step = StepAction.from_json(self.basic_input_json, {'x': 125.3, 'y': 198.7}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, [], [], position={'x': 125.3, 'y': 198.7}) - - def test_from_json_with_next_steps(self): - next_steps = [NextStep(), NextStep(name='name'), NextStep(name='name2')] - next_steps_json = [next_step.as_json() for next_step in next_steps] - self.basic_input_json['next'] = next_steps_json - step = StepAction.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', 'HelloWorld', '', {}, next_steps_json, []) - - def test_execute_no_args(self): - step = StepAction(app='HelloWorld', action='helloWorld') - instance = Instance.create(app_name='HelloWorld', device_name='device1') - self.assertEqual(step.execute(instance.instance, {}), ActionResult({'message': 'HELLO WORLD'}, 'Success')) - self.assertEqual(step.output, ActionResult({'message': 'HELLO WORLD'}, 'Success')) - - def test_execute_with_args(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '-5.6', 'num2': '4.3', 'num3': '10.2'}) - instance = Instance.create(app_name='HelloWorld', device_name='device1') - result = step.execute(instance.instance, {}) - self.assertAlmostEqual(result.result, 8.9) - self.assertEqual(result.status, 'Success') - self.assertEqual(step.output, result) - - def test_execute_with_accumulator_with_conversion(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '@1', 'num2': '@step2', 'num3': '10.2'}) - accumulator = {'1': '-5.6', 'step2': '4.3'} - instance = Instance.create(app_name='HelloWorld', device_name='device1') - result = step.execute(instance.instance, accumulator) - self.assertAlmostEqual(result.result, 8.9) - self.assertEqual(result.status, 'Success') - self.assertEqual(step.output, result) - - def test_execute_with_accumulator_with_extra_steps(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '@1', 'num2': '@step2', 'num3': '10.2'}) - accumulator = {'1': '-5.6', 'step2': '4.3', '3': '45'} - instance = Instance.create(app_name='HelloWorld', device_name='device1') - result = step.execute(instance.instance, accumulator) - self.assertAlmostEqual(result.result, 8.9) - self.assertEqual(result.status, 'Success') - self.assertEqual(step.output, result) - - def test_execute_with_accumulator_missing_step(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '@1', 'num2': '@step2', 'num3': '10.2'}) - accumulator = {'1': '-5.6', 'missing': '4.3', '3': '45'} - instance = Instance.create(app_name='HelloWorld', device_name='device1') - with self.assertRaises(InvalidInput): - step.execute(instance.instance, accumulator) - - def test_execute_with_complex_inputs(self): - step = StepAction(app='HelloWorld', action='Json Sample', - inputs={'json_in': {'a': '-5.6', 'b': {'a': '4.3', 'b': 5.3}, 'c': ['1', '2', '3'], - 'd': [{'a': '', 'b': 3}, {'a': '', 'b': -1.5}, {'a': '', 'b': -0.5}]}}) - instance = Instance.create(app_name='HelloWorld', device_name='device1') - result = step.execute(instance.instance, {}) - self.assertAlmostEqual(result.result, 11.0) - self.assertEqual(result.status, 'Success') - self.assertEqual(step.output, result) - - def test_execute_action_which_raises_exception(self): - from tests.apps.HelloWorld.exceptions import CustomException - step = StepAction(app='HelloWorld', action='Buggy') - instance = Instance.create(app_name='HelloWorld', device_name='device1') - with self.assertRaises(CustomException): - step.execute(instance.instance, {}) - - def test_execute_event(self): - step = StepAction(app='HelloWorld', action='Sample Event', inputs={'arg1': 1}) - instance = Instance.create(app_name='HelloWorld', device_name='device1') - - import time - from tests.apps.HelloWorld.events import event1 - import threading - - def sender(): - time.sleep(0.1) - event1.trigger(3) - - thread = threading.Thread(target=sender) - start = time.time() - thread.start() - result = step.execute(instance.instance, {}) - end = time.time() - thread.join() - self.assertEqual(result, ActionResult(4, 'Success')) - self.assertGreater((end-start), 0.1) - - def test_get_next_step_no_next_steps(self): - step = StepAction(app='HelloWorld', action='helloWorld') - step.output = 'something' - self.assertIsNone(step.get_next_step({})) - - def test_get_next_step(self): - flag1 = [Flag(action='mod1_flag2', args={'arg1': '3'}), Flag(action='mod1_flag2', args={'arg1': '-1'})] - next_steps = [NextStep(flags=flag1, name='name1'), NextStep(name='name2')] - step = StepAction(app='HelloWorld', action='helloWorld', next_steps=next_steps) - step.output = ActionResult(2, 'Success') - self.assertEqual(step.get_next_step({}), 'name2') - step.output = ActionResult(1, 'Success') - self.assertEqual(step.get_next_step({}), 'name1') - - def test_set_input_valid(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '-5.6', 'num2': '4.3', 'num3': '10.2'}) - step.set_input({'num1': '-5.62', 'num2': '5', 'num3': '42.42'}) - self.assertDictEqual(step.input, {'num1': -5.62, 'num2': 5., 'num3': 42.42}) - - def test_set_input_invalid_name(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '-5.6', 'num2': '4.3', 'num3': '10.2'}) - with self.assertRaises(InvalidInput): - step.set_input({'num1': '-5.62', 'invalid': '5', 'num3': '42.42'}) - - def test_set_input_invalid_format(self): - step = StepAction(app='HelloWorld', action='Add Three', inputs={'num1': '-5.6', 'num2': '4.3', 'num3': '10.2'}) - with self.assertRaises(InvalidInput): - step.set_input({'num1': '-5.62', 'num2': '5', 'num3': 'invalid'}) diff --git a/tests/test_step_trigger.py b/tests/test_step_trigger.py deleted file mode 100644 index 3e512014a..000000000 --- a/tests/test_step_trigger.py +++ /dev/null @@ -1,171 +0,0 @@ -import unittest -import socket -from core.flag import Flag -from core.steptrigger import StepTrigger, _Widget -from core.decorators import ActionResult -import core.config.config -import core.config.paths -from tests.config import test_apps_path, function_api_path -from core.instance import Instance -from core.helpers import (import_all_apps, UnknownApp, UnknownAppAction, InvalidInput, import_all_flags, - import_all_filters, InvalidElementConstructed) -import uuid - - -class TestStepTrigger(unittest.TestCase): - @classmethod - def setUpClass(cls): - import_all_apps(path=test_apps_path) - core.config.config.load_app_apis(apps_path=test_apps_path) - core.config.config.flags = import_all_flags('tests.util.flagsfilters') - core.config.config.filters = import_all_filters('tests.util.flagsfilters') - core.config.config.load_flagfilter_apis(path=function_api_path) - - def setUp(self): - self.uid = uuid.uuid4().hex - self.basic_json = {'action': 'helloWorld', - 'name': '', - 'flags': [], - 'status': 'Success', - 'position': {}, - 'widgets': [], - 'uid': self.uid} - self.basic_input_json = {'action': 'helloWorld', - 'name': '', - 'flags': [], - 'status': 'Success', - 'position': {}, - 'uid': self.uid} - - def __compare_init(self, elem, name, action, flags, widgets, status='Success', position=None, uid=None): - self.assertEqual(elem.name, name) - self.assertEqual(elem.action, action) - self.assertListEqual([flag.action for flag in elem.flags], [flag['action'] for flag in flags]) - self.assertEqual(elem.status, status) - widgets = [_Widget(app, widget) for (app, widget) in widgets] - self.assertEqual(len(elem.widgets), len(widgets)) - for widget in elem.widgets: - self.assertIn(widget, widgets) - position = position if position is not None else {} - self.assertDictEqual(elem.position, position) - self.assertIsNone(elem.output) - self.assertFalse(elem.templated) - if uid is None: - self.assertIsNotNone(elem.uid) - else: - self.assertEqual(elem.uid, uid) - - def test_init_action_only(self): - step = StepTrigger(action='helloWorld') - self.__compare_init(step, '', 'helloWorld', [], []) - - def test_init_with_name(self): - step = StepTrigger(action='helloWorld', name='name') - self.__compare_init(step, 'name', 'helloWorld', [], []) - - def test_init_with_empty_flags(self): - step = StepTrigger(action='helloWorld', name='name', flags=[]) - self.__compare_init(step, 'name', 'helloWorld', [], []) - - def test_init_with_flags(self): - flags = [Flag(action='Top Flag'), Flag(action='mod1_flag1')] - expected_flag_json = [{'action': 'Top Flag', 'args': [], 'filters': []}, - {'action': 'mod1_flag1', 'args': [], 'filters': []}] - step = StepTrigger(action='helloWorld', name='name', flags=flags) - self.__compare_init(step, 'name', 'helloWorld', expected_flag_json, []) - - def test_init_with_status(self): - step = StepTrigger(action='helloWorld', status='test_status') - self.__compare_init(step, '', 'helloWorld', [], [], status='test_status') - - def test_init_with_position(self): - step = StepTrigger(action='helloWorld', position={'x': -12.3, 'y': 485}) - self.__compare_init(step, '', 'helloWorld', [], [], position={'x': -12.3, 'y': 485}) - - def test_init_with_widgets(self): - widgets = [('aaa', 'bbb'), ('ccc', 'ddd'), ('eee', 'fff')] - step = StepTrigger(action='helloWorld', widgets=widgets) - self.__compare_init(step, '', 'helloWorld', [], widgets) - - def test_init_with_uid(self): - uid = uuid.uuid4().hex - step = StepTrigger(action='helloWorld', uid=uid) - self.__compare_init(step, '', 'helloWorld', [], [], uid=uid) - - def test_as_json(self): - step = StepTrigger(action='helloWorld', uid=self.uid) - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_name(self): - step = StepTrigger(action='helloWorld', name='name', uid=self.uid) - self.basic_json['name'] = 'name' - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_flags(self): - flags = [Flag(action='Top Flag'), Flag(action='mod1_flag1')] - step = StepTrigger(action='helloWorld', flags=flags, uid=self.uid) - self.basic_json['flags'] = [flag.as_json() for flag in flags] - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_status(self): - step = StepTrigger(action='helloWorld', status='test_status', uid=self.uid) - self.basic_json['status'] = 'test_status' - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_position(self): - step = StepTrigger(action='helloWorld', position={'x': -12.3, 'y': 485}, uid=self.uid) - self.basic_json['position'] = {'x': -12.3, 'y': 485} - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_as_json_with_widgets(self): - widgets = [('aaa', 'bbb'), ('ccc', 'ddd'), ('eee', 'fff')] - step = StepTrigger(action='helloWorld', widgets=widgets, uid=self.uid) - self.basic_json['widgets'] = [{'app': widget[0], 'name': widget[1]} for widget in widgets] - self.assertDictEqual(step.as_json(), self.basic_json) - - def test_from_json_with_name(self): - self.basic_input_json['name'] = 'name' - step = StepTrigger.from_json(self.basic_input_json, {}) - self.__compare_init(step, 'name', 'helloWorld', [], []) - - def test_from_json_with_flags(self): - flag_json = [{'action': 'Top Flag', 'args': [], 'filters': []}, - {'action': 'mod1_flag1', 'args': [], 'filters': []}] - self.basic_input_json['flags'] = flag_json - step = StepTrigger.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', flag_json, []) - - def test_from_json_with_status(self): - self.basic_input_json['status'] = 'test_status' - step = StepTrigger.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', [], [], status='test_status') - - def test_from_json_with_position(self): - step = StepTrigger.from_json(self.basic_input_json, {'x': 125.3, 'y': 198.7}) - self.__compare_init(step, '', 'helloWorld', [], [], position={'x': 125.3, 'y': 198.7}) - - def test_from_json_with_widgets(self): - widget_json = [{'name': 'widget_name', 'app': 'app1'}, {'name': 'w2', 'app': 'app2'}] - widget_tuples = [('app1', 'widget_name'), ('app2', 'w2')] - self.basic_input_json['widgets'] = widget_json - step = StepTrigger.from_json(self.basic_input_json, {}) - self.__compare_init(step, '', 'helloWorld', [], widget_tuples) - - def test_call(self): - flags1 = [Flag(action='regMatch', args={'regex': '(.*)'})] - flags2 = [Flag(action='regMatch', args={'regex': '(.*)'}), - Flag(action='regMatch', args={'regex': 'a'})] - - inputs = [('name1', [], ActionResult('aaaa', 'Success'), True), - ('name2', flags1, ActionResult('anyString', 'Success'), True), - ('name3', flags2, ActionResult('anyString', 'Success'), True), - ('name4', flags2, ActionResult('bbbb', 'Success'), False), - ('name4', flags2, ActionResult('aaaa', 'Custom'), False)] - - for name, flags, input_str, expect_name in inputs: - step = StepTrigger(action='helloWorld', flags=flags) - if expect_name: - expected_name = step.name - self.assertEqual(step(input_str, {}), expected_name) - else: - self.assertIsNone(step(input_str, {}))