Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use signal_workflow_execution method in localstack

Best Python code snippet using localstack_python

test_swf.py

Source:test_swf.py

...113              ({'domain':'d', 'workflow_id':'id', 'signal_name':'my_signal', 'input':{'f':'b'}},114              {'domain':'d', 'workflowId':'id', 'signalName':'my_signal', 'input':'{"f": "b"}'}),115              ({'domain':'d', 'workflow_id':'id', 'signal_name':'my_signal', 'run_id':'rid'},116              {'domain':'d', 'workflowId':'id', 'signalName':'my_signal', 'runId':'rid'})])117    def test_signal_workflow_execution(self, args, expected, mocker):118        client_mock = type("ClientMock", (object,), {"signal_workflow_execution":Mock()})119        mocker.patch('floto.api.Swf.client', new_callable=PropertyMock, return_value=client_mock())120        swf = floto.api.Swf()121        swf.signal_workflow_execution(**args)122        swf.client.signal_workflow_execution.assert_called_once_with(**expected)123    @pytest.mark.parametrize('args,expected',124            [({'domain':'d', 'workflow_id':'id'},125              {'domain':'d', 'workflowId':'id'}),126              127              ({'domain':'d', 'workflow_id':'id', 'run_id':'my_run_id'},128               {'domain':'d', 'workflowId':'id', 'runId':'my_run_id'})])129    def test_terminate_workflow_exeuction(self, args, expected, mocker):130        client_mock = type("ClientMock", (object,), {"terminate_workflow_execution":Mock()})131        mocker.patch('floto.api.Swf.client', new_callable=PropertyMock, return_value=client_mock())132        swf = floto.api.Swf()133        swf.terminate_workflow_execution(**args)134        swf.client.terminate_workflow_execution.assert_called_once_with(**expected)135    @pytest.mark.parametrize('args',[({'domain':'d'}), ({'workflow_id':'id'}), ({})])...

test_workflow_stub.py

Source:test_workflow_stub.py

1import json2from unittest.mock import Mock, MagicMock3import pytest4from cadence.cadence_types import WorkflowExecution, SignalWorkflowExecutionRequest5from cadence.workflow import WorkflowClient, signal_method, WorkflowStub, SignalMethod6from cadence.workflowservice import WorkflowService7@pytest.fixture8def workflow_service():9    m = Mock(spec=WorkflowService)10    m.signal_workflow_execution = MagicMock(return_value=(object(), None))11    return m12@pytest.fixture13def workflow_client(workflow_service):14    return WorkflowClient(service=workflow_service, domain="domain", options=None)15@pytest.fixture16def workflow_execution():17    return WorkflowExecution(workflow_id="the-workflow-id", run_id="the-run-id")18class DummyWorkflow:19    @signal_method()20    def the_signal_method(self, name, age):21        pass22def test_new_workflow_stub_signal(workflow_client):23    stub = workflow_client.new_workflow_stub(DummyWorkflow)24    assert isinstance(stub, WorkflowStub)25    assert hasattr(stub, "the_signal_method")26    assert hasattr(stub.the_signal_method, "_signal_method")27    assert isinstance(stub.the_signal_method._signal_method, SignalMethod)28def test_exec_signal(workflow_client, workflow_service, workflow_execution):29    stub = workflow_client.new_workflow_stub(DummyWorkflow)30    stub._execution = workflow_execution31    stub.the_signal_method("bob", 25)32    workflow_service.signal_workflow_execution.assert_called_once()33    args, kwargs = workflow_service.signal_workflow_execution.call_args_list[0]34    request: SignalWorkflowExecutionRequest = args[0]35    assert request.signal_name == "DummyWorkflow::the_signal_method"36    assert request.input == json.dumps(["bob", 25])...

signal.py

Source:signal.py

1from caravan.commands import run_swf_command2from caravan.commands.base import BaseCommand3class Command(BaseCommand):4    description = 'Signal a workflow execution'5    def setup_arguments(self, parser):6        parser.add_argument('-d', '--domain', required=True)7        parser.add_argument('-i', '--id', required=True)8        parser.add_argument('--run-id')9        parser.add_argument('-s', '--signal', required=True)10        parser.add_argument(11            '--input',12            help='Raw input data for this signal',13            metavar='"<Some data>"')14    def run(self):15        run_swf_command(16            'signal_workflow_execution',17            domain=self.args.domain,18            workflowId=self.args.id,19            signalName=self.args.signal,20            runId=self.args.run_id,21            input=self.args.input,22            )...

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.