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

How to use stepfunctions_client method in localstack

Best Python code snippet using localstack_python

test_stepfunctions_basics.py

Source:test_stepfunctions_basics.py

1# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.2# SPDX-License-Identifier: Apache-2.03"""4Unit tests for stepfunctions_basics.py5"""6import boto37from botocore.exceptions import ClientError8import pytest9from stepfunctions_basics import StepFunctionsStateMachine10@pytest.mark.parametrize('error_code', [None, 'TestException'])11def test_create(make_stubber, error_code):12    stepfunctions_client = boto3.client('stepfunctions')13    stepfunctions_stubber = make_stubber(stepfunctions_client)14    state_machine = StepFunctionsStateMachine(stepfunctions_client)15    name = 'test-name'16    definition = {'Comment': 'test-definition'}17    role_arn = 'test-role-arn'18    arn = 'test-arn'19    stepfunctions_stubber.stub_create_state_machine(20        name, definition, role_arn, arn, error_code=error_code)21    if error_code is None:22        got_arn = state_machine.create(name, definition, role_arn)23        assert got_arn == arn24        assert state_machine.state_machine_name == name25        assert state_machine.state_machine_arn == arn26    else:27        with pytest.raises(ClientError) as exc_info:28            state_machine.create(name, definition, role_arn)29        assert exc_info.value.response['Error']['Code'] == error_code30@pytest.mark.parametrize('role_arn,error_code', [31    (None, None), ('test-role-arn', None), (None, 'TestException')])32def test_update(make_stubber, role_arn, error_code):33    stepfunctions_client = boto3.client('stepfunctions')34    stepfunctions_stubber = make_stubber(stepfunctions_client)35    state_machine = StepFunctionsStateMachine(stepfunctions_client)36    state_machine.state_machine_arn = 'test-arn'37    definition = {'Comment': 'test-definition'}38    stepfunctions_stubber.stub_update_state_machine(39        state_machine.state_machine_arn, definition, role_arn, error_code=error_code)40    if error_code is None:41        state_machine.update(definition, role_arn)42    else:43        with pytest.raises(ClientError) as exc_info:44            state_machine.update(definition, role_arn)45        assert exc_info.value.response['Error']['Code'] == error_code46@pytest.mark.parametrize('error_code', [None, 'TestException'])47def test_delete(make_stubber, error_code):48    stepfunctions_client = boto3.client('stepfunctions')49    stepfunctions_stubber = make_stubber(stepfunctions_client)50    state_machine = StepFunctionsStateMachine(stepfunctions_client)51    state_machine.state_machine_arn = 'test-state_machine_arn'52    stepfunctions_stubber.stub_delete_state_machine(53        state_machine.state_machine_arn, error_code=error_code)54    if error_code is None:55        state_machine.delete()56    else:57        with pytest.raises(ClientError) as exc_info:58            state_machine.delete()59        assert exc_info.value.response['Error']['Code'] == error_code60@pytest.mark.parametrize('found,error_code', [61    (True, None), (False, None), (True, 'TestException')])62def test_find(make_stubber, found, error_code):63    stepfunctions_client = boto3.client('stepfunctions')64    stepfunctions_stubber = make_stubber(stepfunctions_client)65    state_machine = StepFunctionsStateMachine(stepfunctions_client)66    state_machine_name = 'test-state_machine_name'67    state_machine_arn = 'test-arn'68    machine_data = [('wrong-name', 'wrong-arn')]69    if found:70        machine_data.append((state_machine_name, state_machine_arn))71    state_machines = [72        {'name': name, 'stateMachineArn': arn} for name, arn in machine_data]73    stepfunctions_stubber.stub_list_state_machines(74        state_machines, error_code=error_code)75    if error_code is None:76        got_state_machine_arn = state_machine.find(state_machine_name)77        if found:78            assert got_state_machine_arn == state_machine_arn79        else:80            assert got_state_machine_arn is None81    else:82        with pytest.raises(ClientError) as exc_info:83            state_machine.find(state_machine_name)84        assert exc_info.value.response['Error']['Code'] == error_code85@pytest.mark.parametrize('error_code', [None, 'TestException'])86def test_describe(make_stubber, error_code):87    stepfunctions_client = boto3.client('stepfunctions')88    stepfunctions_stubber = make_stubber(stepfunctions_client)89    state_machine = StepFunctionsStateMachine(stepfunctions_client)90    state_machine.state_machine_arn = 'test-state_machine_arn'91    name = 'test-name'92    definition = 'test-definition'93    role_arn = 'test-role_arn'94    stepfunctions_stubber.stub_describe_state_machine(95        state_machine.state_machine_arn, name, definition, role_arn,96        error_code=error_code)97    if error_code is None:98        got_response = state_machine.describe()99        assert got_response['name'] == name100        assert got_response['definition'] == definition101        assert got_response['roleArn'] == role_arn102        assert got_response['stateMachineArn'] == state_machine.state_machine_arn103    else:104        with pytest.raises(ClientError) as exc_info:105            state_machine.describe()106        assert exc_info.value.response['Error']['Code'] == error_code107@pytest.mark.parametrize('run_input,error_code', [108    ({'test-key': 'test-value'}, None),109    (None, None),110    ({'test-key': 'test-value'}, 'TestException')])111def test_start_run(make_stubber, run_input, error_code):112    stepfunctions_client = boto3.client('stepfunctions')113    stepfunctions_stubber = make_stubber(stepfunctions_client)114    state_machine = StepFunctionsStateMachine(stepfunctions_client)115    state_machine.state_machine_arn = 'test-arn'116    run_name = 'test-run_name'117    run_arn = 'test-run_arn'118    stepfunctions_stubber.stub_start_execution(119        state_machine.state_machine_arn, run_name, run_arn, run_input,120        error_code=error_code)121    if error_code is None:122        got_run_arn = state_machine.start_run(run_name, run_input)123        assert got_run_arn == run_arn124    else:125        with pytest.raises(ClientError) as exc_info:126            state_machine.start_run(run_name, run_input)127        assert exc_info.value.response['Error']['Code'] == error_code128@pytest.mark.parametrize('run_status,error_code', [129    ('test-run_status', None), (None, None), ('test-run_status', 'TestException')])130def test_list_runs(make_stubber, run_status, error_code):131    stepfunctions_client = boto3.client('stepfunctions')132    stepfunctions_stubber = make_stubber(stepfunctions_client)133    state_machine = StepFunctionsStateMachine(stepfunctions_client)134    state_machine.state_machine_arn = 'test-arn'135    runs = [{'name': name, 'executionArn': arn} for name, arn in136            [('run-name-1', 'run-arn-1'), ('run-name-2', 'run-arn-2')]]137    stepfunctions_stubber.stub_list_executions(138        state_machine.state_machine_arn, runs, run_status, error_code=error_code)139    if error_code is None:140        got_runs = state_machine.list_runs(run_status)141        assert [{'name': run['name'], 'executionArn': run['executionArn']}142                for run in got_runs] == runs143    else:144        with pytest.raises(ClientError) as exc_info:145            state_machine.list_runs(run_status)146        assert exc_info.value.response['Error']['Code'] == error_code147@pytest.mark.parametrize('error_code', [None, 'TestException'])148def test_stop_run(make_stubber, error_code):149    stepfunctions_client = boto3.client('stepfunctions')150    stepfunctions_stubber = make_stubber(stepfunctions_client)151    state_machine = StepFunctionsStateMachine(stepfunctions_client)152    run_arn = 'test-run_arn'153    cause = 'test cause'154    stepfunctions_stubber.stub_stop_execution(run_arn, cause, error_code=error_code)155    if error_code is None:156        state_machine.stop_run(run_arn, cause)157    else:158        with pytest.raises(ClientError) as exc_info:159            state_machine.stop_run(run_arn, cause)...

aws_stepfunctions.py

Source:aws_stepfunctions.py

1import json2import botocore3import skwadon.main as sic_main4import skwadon.lib as sic_lib5import skwadon.common_action as common_action6import skwadon.aws as sic_aws7class StateMachineListHandler(common_action.ListHandler):8    def __init__(self, session):9        self.session = session10        self.stepfunctions_client = None11    def init_client(self):12        if self.stepfunctions_client == None:13            self.stepfunctions_client = self.session.client("stepfunctions")14    def list(self):15        self.init_client()16        account_id = sic_aws.fetch_account_id(self.session)17        region_name = sic_aws.fetch_region_name(self.session)18        result = []19        res = self.stepfunctions_client.list_state_machines()20        while True:21            for elem in res['stateMachines']:22                name = elem["name"]23                arn = calc_statemachine_arn2(account_id, region_name, name)24                if elem["stateMachineArn"] == arn:25                    result.append(name)26            if not "nextToken" in res:27                break28            res = self.stepfunctions_client.list_state_machines(nextToken = res["nextToken"])29        return result30    def child_handler(self, name):31        self.init_client()32        return common_action.NamespaceHandler(33            "conf", ["conf", "definition"], {34            "conf": StateMachineConfHandler(self.session, self.stepfunctions_client, name),35            "definition": StateMachineDefinitionHandler(self.session, self.stepfunctions_client, name),36        })37class StateMachineBasicHandler(common_action.ResourceHandler):38    def __init__(self, session, stepfunctions_client, machine_name):39        self.session = session40        self.stepfunctions_client = stepfunctions_client41        self.machine_name = machine_name42        self.arn = None43        self.info = None44    def _describe(self):45        if self.arn != None:46            return self.info47        if self.arn == None:48            arn = self._calc_statemachine_arn()49        try:50            res = self.stepfunctions_client.describe_state_machine(stateMachineArn = arn)51            curr_data = sic_lib.dict_key_capitalize(res)52            curr_data = sic_lib.pickup(curr_data, self._properties)53            self.info = curr_data54            return self.info55        except botocore.exceptions.ClientError as e:56            if e.response["Error"]["Code"] == "StateMachineDoesNotExist":57                return None58            else:59                raise60    def _create(self, confirmation_flag, src_data):61        update_data = {}62        update_data["Name"] = self.machine_name63        update_data["Definition"] = self._defaultDefinition()64        for key in self._properties_for_create:65            if key in src_data:66                update_data[key] = src_data[key]67        update_data = sic_lib.dict_key_to_lower(update_data)68        sic_main.exec_put(confirmation_flag,69            f"stepfunctions_client.create_state_machine(name = {self.machine_name}, ...)",70            lambda:71                self.stepfunctions_client.create_state_machine(**update_data)72        )73        if confirmation_flag:74            self.info = None75    def _update(self, confirmation_flag, src_data):76        curr_data = self._describe()77        if curr_data is None:78            curr_data = {}79        update_data = {}80        dirty = False81        for key in self._properties_for_update:82            if key in src_data:83                update_data[key] = src_data[key]84                if key not in curr_data or update_data[key] != curr_data[key]:85                    dirty = True86            elif key in curr_data:87                update_data[key] = curr_data[key]88        if dirty:89            arn = self._calc_statemachine_arn()90            update_data = sic_lib.dict_key_to_lower(update_data)91            sic_main.exec_put(confirmation_flag,92                f"stepfunctions_client.update_state_machine(stateMachineArn = {arn}, ...)",93                lambda:94                    self.stepfunctions_client.update_state_machine(stateMachineArn = arn, **update_data)95            )96            if confirmation_flag:97                self.info = None98    def _calc_statemachine_arn(self):99        return calc_statemachine_arn(self.session, self.machine_name)100    def _defaultDefinition(self):101        return json.dumps({102            "StartAt": "__skwadon_dummy",103            "States": {104                "__skwadon_dummy": {105                    "Type": "Pass",106                    "End": True,107                },108            },109        })110    _properties = [111        "RoleArn",112        "Type",113        "LoggingConfiguration",114        "TracingConfiguration",115        "Definition"116    ]117    _properties_for_create = [118        "RoleArn",119        "Type",120        "LoggingConfiguration",121        "TracingConfiguration",122    ]123    _properties_for_update = [124        "RoleArn",125        "LoggingConfiguration",126        "TracingConfiguration",127        "Definition"128    ]129class StateMachineConfHandler(StateMachineBasicHandler):130    def __init__(self, session, stepfunctions_client, machine_name):131        StateMachineBasicHandler.__init__(self, session, stepfunctions_client, machine_name)132    def describe(self):133        info = self._describe()134        if info == None:135            return None136        curr_data = sic_lib.pickup(info, self.properties)137        return curr_data138    def create(self, confirmation_flag, src_data):139        self._create(confirmation_flag, src_data)140    def update(self, confirmation_flag, src_data, curr_data):141        update_data = sic_lib.pickupAndCompareForUpdate(src_data, curr_data, self.properties_for_update)142        if update_data != None:143            self._update(confirmation_flag, update_data)144    properties = [145        "RoleArn",146        "Type",147        "LoggingConfiguration",148        "TracingConfiguration",149    ]150    properties_for_update = [151        "RoleArn",152        "LoggingConfiguration",153        "TracingConfiguration",154    ]155class StateMachineDefinitionHandler(StateMachineBasicHandler):156    def __init__(self, session, stepfunctions_client, machine_name):157        StateMachineBasicHandler.__init__(self, session, stepfunctions_client, machine_name)158    def describe(self):159        info = self._describe()160        if info == None:161            return None162        result = self._definition_str_to_dict(info["Definition"])163        if "__skwadon_dummy" in result["States"]:164            del result["States"]["__skwadon_dummy"]165        return result166    def put(self, confirmation_flag, src_data):167        update_data = {168            "Definition": self._definition_dict_to_str(src_data),169        }170        self._update(confirmation_flag, update_data)171    def _definition_str_to_dict(self, definition: str):172        return json.loads(definition)173    def _definition_dict_to_str(self, definition: dict):174        return json.dumps(definition)175def calc_statemachine_arn(session, machine_name):176    account_id = sic_aws.fetch_account_id(session)177    region_name = sic_aws.fetch_region_name(session)178    return calc_statemachine_arn2(account_id, region_name, machine_name)179def calc_statemachine_arn2(account_id, region_name, machine_name):...

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.