Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use create_state_machine method in localstack
Best Python code snippet using localstack_python
test_stepfunctions.py
Source:test_stepfunctions.py 
...28def test_state_machine_creation_succeeds():29 client = boto3.client("stepfunctions", region_name=region)30 name = "example_step_function"31 #32 response = client.create_state_machine(33 name=name, definition=str(simple_definition), roleArn=_get_default_role()34 )35 #36 response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)37 response["creationDate"].should.be.a(datetime)38 response["stateMachineArn"].should.equal(39 "arn:aws:states:" + region + ":" + ACCOUNT_ID + ":stateMachine:" + name40 )41@mock_stepfunctions42def test_state_machine_creation_fails_with_invalid_names():43 client = boto3.client("stepfunctions", region_name=region)44 invalid_names = [45 "with space",46 "with<bracket",47 "with>bracket",48 "with{bracket",49 "with}bracket",50 "with[bracket",51 "with]bracket",52 "with?wildcard",53 "with*wildcard",54 'special"char',55 "special#char",56 "special%char",57 "special\\char",58 "special^char",59 "special|char",60 "special~char",61 "special`char",62 "special$char",63 "special&char",64 "special,char",65 "special;char",66 "special:char",67 "special/char",68 "uni\u0000code",69 "uni\u0001code",70 "uni\u0002code",71 "uni\u0003code",72 "uni\u0004code",73 "uni\u0005code",74 "uni\u0006code",75 "uni\u0007code",76 "uni\u0008code",77 "uni\u0009code",78 "uni\u000Acode",79 "uni\u000Bcode",80 "uni\u000Ccode",81 "uni\u000Dcode",82 "uni\u000Ecode",83 "uni\u000Fcode",84 "uni\u0010code",85 "uni\u0011code",86 "uni\u0012code",87 "uni\u0013code",88 "uni\u0014code",89 "uni\u0015code",90 "uni\u0016code",91 "uni\u0017code",92 "uni\u0018code",93 "uni\u0019code",94 "uni\u001Acode",95 "uni\u001Bcode",96 "uni\u001Ccode",97 "uni\u001Dcode",98 "uni\u001Ecode",99 "uni\u001Fcode",100 "uni\u007Fcode",101 "uni\u0080code",102 "uni\u0081code",103 "uni\u0082code",104 "uni\u0083code",105 "uni\u0084code",106 "uni\u0085code",107 "uni\u0086code",108 "uni\u0087code",109 "uni\u0088code",110 "uni\u0089code",111 "uni\u008Acode",112 "uni\u008Bcode",113 "uni\u008Ccode",114 "uni\u008Dcode",115 "uni\u008Ecode",116 "uni\u008Fcode",117 "uni\u0090code",118 "uni\u0091code",119 "uni\u0092code",120 "uni\u0093code",121 "uni\u0094code",122 "uni\u0095code",123 "uni\u0096code",124 "uni\u0097code",125 "uni\u0098code",126 "uni\u0099code",127 "uni\u009Acode",128 "uni\u009Bcode",129 "uni\u009Ccode",130 "uni\u009Dcode",131 "uni\u009Ecode",132 "uni\u009Fcode",133 ]134 #135 for invalid_name in invalid_names:136 with pytest.raises(ClientError):137 client.create_state_machine(138 name=invalid_name,139 definition=str(simple_definition),140 roleArn=_get_default_role(),141 )142@mock_stepfunctions143def test_state_machine_creation_requires_valid_role_arn():144 client = boto3.client("stepfunctions", region_name=region)145 name = "example_step_function"146 #147 with pytest.raises(ClientError):148 client.create_state_machine(149 name=name,150 definition=str(simple_definition),151 roleArn="arn:aws:iam::1234:role/unknown_role",152 )153@mock_stepfunctions154@mock_sts155def test_update_state_machine():156 client = boto3.client("stepfunctions", region_name=region)157 resp = client.create_state_machine(158 name="test", definition=str(simple_definition), roleArn=_get_default_role()159 )160 state_machine_arn = resp["stateMachineArn"]161 updated_role = _get_default_role() + "-updated"162 updated_definition = str(simple_definition).replace(163 "DefaultState", "DefaultStateUpdated"164 )165 resp = client.update_state_machine(166 stateMachineArn=state_machine_arn,167 definition=updated_definition,168 roleArn=updated_role,169 )170 resp["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)171 resp["updateDate"].should.be.a(datetime)172 desc = client.describe_state_machine(stateMachineArn=state_machine_arn)173 desc["definition"].should.equal(updated_definition)174 desc["roleArn"].should.equal(updated_role)175@mock_stepfunctions176def test_state_machine_list_returns_empty_list_by_default():177 client = boto3.client("stepfunctions", region_name=region)178 #179 list = client.list_state_machines()180 list["stateMachines"].should.be.empty181@mock_stepfunctions182@mock_sts183def test_state_machine_list_returns_created_state_machines():184 client = boto3.client("stepfunctions", region_name=region)185 #186 machine1 = client.create_state_machine(187 name="name1",188 definition=str(simple_definition),189 roleArn=_get_default_role(),190 tags=[{"key": "tag_key", "value": "tag_value"}],191 )192 machine2 = client.create_state_machine(193 name="name2", definition=str(simple_definition), roleArn=_get_default_role()194 )195 list = client.list_state_machines()196 #197 list["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)198 list["stateMachines"].should.have.length_of(2)199 list["stateMachines"][0]["creationDate"].should.be.a(datetime)200 list["stateMachines"][0]["creationDate"].should.equal(machine1["creationDate"])201 list["stateMachines"][0]["name"].should.equal("name1")202 list["stateMachines"][0]["stateMachineArn"].should.equal(203 machine1["stateMachineArn"]204 )205 list["stateMachines"][1]["creationDate"].should.be.a(datetime)206 list["stateMachines"][1]["creationDate"].should.equal(machine2["creationDate"])207 list["stateMachines"][1]["name"].should.equal("name2")208 list["stateMachines"][1]["stateMachineArn"].should.equal(209 machine2["stateMachineArn"]210 )211@mock_stepfunctions212def test_state_machine_list_pagination():213 client = boto3.client("stepfunctions", region_name=region)214 for i in range(25):215 machine_name = "StateMachine-{}".format(i)216 client.create_state_machine(217 name=machine_name,218 definition=str(simple_definition),219 roleArn=_get_default_role(),220 )221 resp = client.list_state_machines()222 resp.should_not.have.key("nextToken")223 resp["stateMachines"].should.have.length_of(25)224 paginator = client.get_paginator("list_state_machines")225 page_iterator = paginator.paginate(maxResults=5)226 for page in page_iterator:227 page["stateMachines"].should.have.length_of(5)228 page["stateMachines"][-1]["name"].should.contain("24")229@mock_stepfunctions230@mock_sts231def test_state_machine_creation_is_idempotent_by_name():232 client = boto3.client("stepfunctions", region_name=region)233 #234 client.create_state_machine(235 name="name", definition=str(simple_definition), roleArn=_get_default_role()236 )237 sm_list = client.list_state_machines()238 sm_list["stateMachines"].should.have.length_of(1)239 #240 client.create_state_machine(241 name="name", definition=str(simple_definition), roleArn=_get_default_role()242 )243 sm_list = client.list_state_machines()244 sm_list["stateMachines"].should.have.length_of(1)245 #246 client.create_state_machine(247 name="diff_name", definition=str(simple_definition), roleArn=_get_default_role()248 )249 sm_list = client.list_state_machines()250 sm_list["stateMachines"].should.have.length_of(2)251@mock_stepfunctions252@mock_sts253def test_state_machine_creation_can_be_described():254 client = boto3.client("stepfunctions", region_name=region)255 #256 sm = client.create_state_machine(257 name="name", definition=str(simple_definition), roleArn=_get_default_role()258 )259 desc = client.describe_state_machine(stateMachineArn=sm["stateMachineArn"])260 desc["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)261 desc["creationDate"].should.equal(sm["creationDate"])262 desc["definition"].should.equal(str(simple_definition))263 desc["name"].should.equal("name")264 desc["roleArn"].should.equal(_get_default_role())265 desc["stateMachineArn"].should.equal(sm["stateMachineArn"])266 desc["status"].should.equal("ACTIVE")267@mock_stepfunctions268@mock_sts269def test_state_machine_throws_error_when_describing_unknown_machine():270 client = boto3.client("stepfunctions", region_name=region)271 #272 with pytest.raises(ClientError):273 unknown_state_machine = (274 "arn:aws:states:"275 + region276 + ":"277 + _get_account_id()278 + ":stateMachine:unknown"279 )280 client.describe_state_machine(stateMachineArn=unknown_state_machine)281@mock_stepfunctions282@mock_sts283def test_state_machine_throws_error_when_describing_bad_arn():284 client = boto3.client("stepfunctions", region_name=region)285 #286 with pytest.raises(ClientError):287 client.describe_state_machine(stateMachineArn="bad")288@mock_stepfunctions289@mock_sts290def test_state_machine_throws_error_when_describing_machine_in_different_account():291 client = boto3.client("stepfunctions", region_name=region)292 #293 with pytest.raises(ClientError):294 unknown_state_machine = (295 "arn:aws:states:" + region + ":000000000000:stateMachine:unknown"296 )297 client.describe_state_machine(stateMachineArn=unknown_state_machine)298@mock_stepfunctions299@mock_sts300def test_state_machine_can_be_deleted():301 client = boto3.client("stepfunctions", region_name=region)302 sm = client.create_state_machine(303 name="name", definition=str(simple_definition), roleArn=_get_default_role()304 )305 #306 response = client.delete_state_machine(stateMachineArn=sm["stateMachineArn"])307 response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)308 #309 sm_list = client.list_state_machines()310 sm_list["stateMachines"].should.have.length_of(0)311@mock_stepfunctions312@mock_sts313def test_state_machine_can_deleted_nonexisting_machine():314 client = boto3.client("stepfunctions", region_name=region)315 #316 unknown_state_machine = (317 "arn:aws:states:" + region + ":" + ACCOUNT_ID + ":stateMachine:unknown"318 )319 response = client.delete_state_machine(stateMachineArn=unknown_state_machine)320 response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)321 #322 sm_list = client.list_state_machines()323 sm_list["stateMachines"].should.have.length_of(0)324@mock_stepfunctions325def test_state_machine_tagging_non_existent_resource_fails():326 client = boto3.client("stepfunctions", region_name=region)327 non_existent_arn = "arn:aws:states:{region}:{account}:stateMachine:non-existent".format(328 region=region, account=ACCOUNT_ID329 )330 with pytest.raises(ClientError) as ex:331 client.tag_resource(resourceArn=non_existent_arn, tags=[])332 ex.value.response["Error"]["Code"].should.equal("ResourceNotFound")333 ex.value.response["Error"]["Message"].should.contain(non_existent_arn)334@mock_stepfunctions335def test_state_machine_untagging_non_existent_resource_fails():336 client = boto3.client("stepfunctions", region_name=region)337 non_existent_arn = "arn:aws:states:{region}:{account}:stateMachine:non-existent".format(338 region=region, account=ACCOUNT_ID339 )340 with pytest.raises(ClientError) as ex:341 client.untag_resource(resourceArn=non_existent_arn, tagKeys=[])342 ex.value.response["Error"]["Code"].should.equal("ResourceNotFound")343 ex.value.response["Error"]["Message"].should.contain(non_existent_arn)344@mock_stepfunctions345@mock_sts346def test_state_machine_tagging():347 client = boto3.client("stepfunctions", region_name=region)348 tags = [349 {"key": "tag_key1", "value": "tag_value1"},350 {"key": "tag_key2", "value": "tag_value2"},351 ]352 machine = client.create_state_machine(353 name="test", definition=str(simple_definition), roleArn=_get_default_role(),354 )355 client.tag_resource(resourceArn=machine["stateMachineArn"], tags=tags)356 resp = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])357 resp["tags"].should.equal(tags)358 tags_update = [359 {"key": "tag_key1", "value": "tag_value1_new"},360 {"key": "tag_key3", "value": "tag_value3"},361 ]362 client.tag_resource(resourceArn=machine["stateMachineArn"], tags=tags_update)363 resp = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])364 tags_expected = [365 tags_update[0],366 tags[1],367 tags_update[1],368 ]369 resp["tags"].should.equal(tags_expected)370@mock_stepfunctions371@mock_sts372def test_state_machine_untagging():373 client = boto3.client("stepfunctions", region_name=region)374 tags = [375 {"key": "tag_key1", "value": "tag_value1"},376 {"key": "tag_key2", "value": "tag_value2"},377 {"key": "tag_key3", "value": "tag_value3"},378 ]379 machine = client.create_state_machine(380 name="test",381 definition=str(simple_definition),382 roleArn=_get_default_role(),383 tags=tags,384 )385 resp = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])386 resp["tags"].should.equal(tags)387 tags_to_delete = ["tag_key1", "tag_key2"]388 client.untag_resource(389 resourceArn=machine["stateMachineArn"], tagKeys=tags_to_delete390 )391 resp = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])392 expected_tags = [tag for tag in tags if tag["key"] not in tags_to_delete]393 resp["tags"].should.equal(expected_tags)394@mock_stepfunctions395@mock_sts396def test_state_machine_list_tags_for_created_machine():397 client = boto3.client("stepfunctions", region_name=region)398 #399 machine = client.create_state_machine(400 name="name1",401 definition=str(simple_definition),402 roleArn=_get_default_role(),403 tags=[{"key": "tag_key", "value": "tag_value"}],404 )405 response = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])406 tags = response["tags"]407 tags.should.have.length_of(1)408 tags[0].should.equal({"key": "tag_key", "value": "tag_value"})409@mock_stepfunctions410@mock_sts411def test_state_machine_list_tags_for_machine_without_tags():412 client = boto3.client("stepfunctions", region_name=region)413 #414 machine = client.create_state_machine(415 name="name1", definition=str(simple_definition), roleArn=_get_default_role()416 )417 response = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])418 tags = response["tags"]419 tags.should.have.length_of(0)420@mock_stepfunctions421@mock_sts422def test_state_machine_list_tags_for_nonexisting_machine():423 client = boto3.client("stepfunctions", region_name=region)424 #425 non_existing_state_machine = (426 "arn:aws:states:" + region + ":" + _get_account_id() + ":stateMachine:unknown"427 )428 response = client.list_tags_for_resource(resourceArn=non_existing_state_machine)429 tags = response["tags"]430 tags.should.have.length_of(0)431@mock_stepfunctions432@mock_sts433def test_state_machine_start_execution():434 client = boto3.client("stepfunctions", region_name=region)435 #436 sm = client.create_state_machine(437 name="name", definition=str(simple_definition), roleArn=_get_default_role()438 )439 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])440 #441 execution["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)442 uuid_regex = "[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}"443 expected_exec_name = (444 "arn:aws:states:"445 + region446 + ":"447 + _get_account_id()448 + ":execution:name:"449 + uuid_regex450 )451 execution["executionArn"].should.match(expected_exec_name)452 execution["startDate"].should.be.a(datetime)453@mock_stepfunctions454@mock_sts455def test_state_machine_start_execution_bad_arn_raises_exception():456 client = boto3.client("stepfunctions", region_name=region)457 #458 with pytest.raises(ClientError):459 client.start_execution(stateMachineArn="bad")460@mock_stepfunctions461@mock_sts462def test_state_machine_start_execution_with_custom_name():463 client = boto3.client("stepfunctions", region_name=region)464 #465 sm = client.create_state_machine(466 name="name", definition=str(simple_definition), roleArn=_get_default_role()467 )468 execution = client.start_execution(469 stateMachineArn=sm["stateMachineArn"], name="execution_name"470 )471 #472 execution["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)473 expected_exec_name = (474 "arn:aws:states:"475 + region476 + ":"477 + _get_account_id()478 + ":execution:name:execution_name"479 )480 execution["executionArn"].should.equal(expected_exec_name)481 execution["startDate"].should.be.a(datetime)482@mock_stepfunctions483@mock_sts484def test_state_machine_start_execution_fails_on_duplicate_execution_name():485 client = boto3.client("stepfunctions", region_name=region)486 #487 sm = client.create_state_machine(488 name="name", definition=str(simple_definition), roleArn=_get_default_role()489 )490 execution_one = client.start_execution(491 stateMachineArn=sm["stateMachineArn"], name="execution_name"492 )493 #494 with pytest.raises(ClientError) as ex:495 _ = client.start_execution(496 stateMachineArn=sm["stateMachineArn"], name="execution_name"497 )498 ex.value.response["Error"]["Message"].should.equal(499 "Execution Already Exists: '" + execution_one["executionArn"] + "'"500 )501@mock_stepfunctions502@mock_sts503def test_state_machine_start_execution_with_custom_input():504 client = boto3.client("stepfunctions", region_name=region)505 #506 sm = client.create_state_machine(507 name="name", definition=str(simple_definition), roleArn=_get_default_role()508 )509 execution_input = json.dumps({"input_key": "input_value"})510 execution = client.start_execution(511 stateMachineArn=sm["stateMachineArn"], input=execution_input512 )513 #514 execution["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)515 uuid_regex = "[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}"516 expected_exec_name = (517 "arn:aws:states:"518 + region519 + ":"520 + _get_account_id()521 + ":execution:name:"522 + uuid_regex523 )524 execution["executionArn"].should.match(expected_exec_name)525 execution["startDate"].should.be.a(datetime)526@mock_stepfunctions527@mock_sts528def test_state_machine_start_execution_with_invalid_input():529 client = boto3.client("stepfunctions", region_name=region)530 #531 sm = client.create_state_machine(532 name="name", definition=str(simple_definition), roleArn=_get_default_role()533 )534 with pytest.raises(ClientError):535 _ = client.start_execution(stateMachineArn=sm["stateMachineArn"], input="")536 with pytest.raises(ClientError):537 _ = client.start_execution(stateMachineArn=sm["stateMachineArn"], input="{")538@mock_stepfunctions539@mock_sts540def test_state_machine_list_executions():541 client = boto3.client("stepfunctions", region_name=region)542 #543 sm = client.create_state_machine(544 name="name", definition=str(simple_definition), roleArn=_get_default_role()545 )546 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])547 execution_arn = execution["executionArn"]548 execution_name = execution_arn[execution_arn.rindex(":") + 1 :]549 executions = client.list_executions(stateMachineArn=sm["stateMachineArn"])550 #551 executions["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)552 executions["executions"].should.have.length_of(1)553 executions["executions"][0]["executionArn"].should.equal(execution_arn)554 executions["executions"][0]["name"].should.equal(execution_name)555 executions["executions"][0]["startDate"].should.equal(execution["startDate"])556 executions["executions"][0]["stateMachineArn"].should.equal(sm["stateMachineArn"])557 executions["executions"][0]["status"].should.equal("RUNNING")558 executions["executions"][0].shouldnt.have("stopDate")559@mock_stepfunctions560def test_state_machine_list_executions_with_filter():561 client = boto3.client("stepfunctions", region_name=region)562 sm = client.create_state_machine(563 name="name", definition=str(simple_definition), roleArn=_get_default_role()564 )565 for i in range(20):566 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])567 if not i % 4:568 client.stop_execution(executionArn=execution["executionArn"])569 resp = client.list_executions(stateMachineArn=sm["stateMachineArn"])570 resp["executions"].should.have.length_of(20)571 resp = client.list_executions(572 stateMachineArn=sm["stateMachineArn"], statusFilter="ABORTED"573 )574 resp["executions"].should.have.length_of(5)575 all([e["status"] == "ABORTED" for e in resp["executions"]]).should.be.true576@mock_stepfunctions577def test_state_machine_list_executions_with_pagination():578 client = boto3.client("stepfunctions", region_name=region)579 sm = client.create_state_machine(580 name="name", definition=str(simple_definition), roleArn=_get_default_role()581 )582 for _ in range(100):583 client.start_execution(stateMachineArn=sm["stateMachineArn"])584 resp = client.list_executions(stateMachineArn=sm["stateMachineArn"])585 resp.should_not.have.key("nextToken")586 resp["executions"].should.have.length_of(100)587 paginator = client.get_paginator("list_executions")588 page_iterator = paginator.paginate(589 stateMachineArn=sm["stateMachineArn"], maxResults=25590 )591 for page in page_iterator:592 page["executions"].should.have.length_of(25)593 with pytest.raises(ClientError) as ex:594 resp = client.list_executions(595 stateMachineArn=sm["stateMachineArn"], maxResults=10596 )597 client.list_executions(598 stateMachineArn=sm["stateMachineArn"],599 maxResults=10,600 statusFilter="ABORTED",601 nextToken=resp["nextToken"],602 )603 ex.value.response["Error"]["Code"].should.equal("InvalidToken")604 ex.value.response["Error"]["Message"].should.contain(605 "Input inconsistent with page token"606 )607 with pytest.raises(ClientError) as ex:608 client.list_executions(609 stateMachineArn=sm["stateMachineArn"], nextToken="invalid"610 )611 ex.value.response["Error"]["Code"].should.equal("InvalidToken")612@mock_stepfunctions613@mock_sts614def test_state_machine_list_executions_when_none_exist():615 client = boto3.client("stepfunctions", region_name=region)616 #617 sm = client.create_state_machine(618 name="name", definition=str(simple_definition), roleArn=_get_default_role()619 )620 executions = client.list_executions(stateMachineArn=sm["stateMachineArn"])621 #622 executions["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)623 executions["executions"].should.have.length_of(0)624@mock_stepfunctions625@mock_sts626def test_state_machine_describe_execution_with_no_input():627 client = boto3.client("stepfunctions", region_name=region)628 #629 sm = client.create_state_machine(630 name="name", definition=str(simple_definition), roleArn=_get_default_role()631 )632 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])633 description = client.describe_execution(executionArn=execution["executionArn"])634 #635 description["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)636 description["executionArn"].should.equal(execution["executionArn"])637 description["input"].should.equal("{}")638 description["name"].shouldnt.be.empty639 description["startDate"].should.equal(execution["startDate"])640 description["stateMachineArn"].should.equal(sm["stateMachineArn"])641 description["status"].should.equal("RUNNING")642 description.shouldnt.have("stopDate")643@mock_stepfunctions644@mock_sts645def test_state_machine_describe_execution_with_custom_input():646 client = boto3.client("stepfunctions", region_name=region)647 #648 execution_input = json.dumps({"input_key": "input_val"})649 sm = client.create_state_machine(650 name="name", definition=str(simple_definition), roleArn=_get_default_role()651 )652 execution = client.start_execution(653 stateMachineArn=sm["stateMachineArn"], input=execution_input654 )655 description = client.describe_execution(executionArn=execution["executionArn"])656 #657 description["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)658 description["executionArn"].should.equal(execution["executionArn"])659 description["input"].should.equal(execution_input)660 description["name"].shouldnt.be.empty661 description["startDate"].should.equal(execution["startDate"])662 description["stateMachineArn"].should.equal(sm["stateMachineArn"])663 description["status"].should.equal("RUNNING")664 description.shouldnt.have("stopDate")665@mock_stepfunctions666@mock_sts667def test_execution_throws_error_when_describing_unknown_execution():668 client = boto3.client("stepfunctions", region_name=region)669 #670 with pytest.raises(ClientError):671 unknown_execution = (672 "arn:aws:states:" + region + ":" + _get_account_id() + ":execution:unknown"673 )674 client.describe_execution(executionArn=unknown_execution)675@mock_stepfunctions676@mock_sts677def test_state_machine_can_be_described_by_execution():678 client = boto3.client("stepfunctions", region_name=region)679 #680 sm = client.create_state_machine(681 name="name", definition=str(simple_definition), roleArn=_get_default_role()682 )683 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])684 desc = client.describe_state_machine_for_execution(685 executionArn=execution["executionArn"]686 )687 desc["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)688 desc["definition"].should.equal(str(simple_definition))689 desc["name"].should.equal("name")690 desc["roleArn"].should.equal(_get_default_role())691 desc["stateMachineArn"].should.equal(sm["stateMachineArn"])692@mock_stepfunctions693@mock_sts694def test_state_machine_throws_error_when_describing_unknown_execution():695 client = boto3.client("stepfunctions", region_name=region)696 #697 with pytest.raises(ClientError):698 unknown_execution = (699 "arn:aws:states:" + region + ":" + _get_account_id() + ":execution:unknown"700 )701 client.describe_state_machine_for_execution(executionArn=unknown_execution)702@mock_stepfunctions703@mock_sts704def test_state_machine_stop_execution():705 client = boto3.client("stepfunctions", region_name=region)706 #707 sm = client.create_state_machine(708 name="name", definition=str(simple_definition), roleArn=_get_default_role()709 )710 start = client.start_execution(stateMachineArn=sm["stateMachineArn"])711 stop = client.stop_execution(executionArn=start["executionArn"])712 #713 stop["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)714 stop["stopDate"].should.be.a(datetime)715@mock_stepfunctions716@mock_sts717def test_state_machine_stop_raises_error_when_unknown_execution():718 client = boto3.client("stepfunctions", region_name=region)719 client.create_state_machine(720 name="test-state-machine",721 definition=str(simple_definition),722 roleArn=_get_default_role(),723 )724 with pytest.raises(ClientError) as ex:725 unknown_execution = (726 "arn:aws:states:"727 + region728 + ":"729 + _get_account_id()730 + ":execution:test-state-machine:unknown"731 )732 client.stop_execution(executionArn=unknown_execution)733 ex.value.response["Error"]["Code"].should.equal("ExecutionDoesNotExist")734 ex.value.response["Error"]["Message"].should.contain("Execution Does Not Exist:")735@mock_stepfunctions736@mock_sts737def test_state_machine_describe_execution_after_stoppage():738 client = boto3.client("stepfunctions", region_name=region)739 sm = client.create_state_machine(740 name="name", definition=str(simple_definition), roleArn=_get_default_role()741 )742 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])743 client.stop_execution(executionArn=execution["executionArn"])744 description = client.describe_execution(executionArn=execution["executionArn"])745 #746 description["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)747 description["status"].should.equal("ABORTED")748 description["stopDate"].should.be.a(datetime)749@mock_stepfunctions750@mock_sts751def test_state_machine_get_execution_history_throws_error_with_unknown_execution():752 client = boto3.client("stepfunctions", region_name=region)753 client.create_state_machine(754 name="test-state-machine",755 definition=str(simple_definition),756 roleArn=_get_default_role(),757 )758 with pytest.raises(ClientError) as ex:759 unknown_execution = (760 "arn:aws:states:"761 + region762 + ":"763 + _get_account_id()764 + ":execution:test-state-machine:unknown"765 )766 client.get_execution_history(executionArn=unknown_execution)767 ex.value.response["Error"]["Code"].should.equal("ExecutionDoesNotExist")768 ex.value.response["Error"]["Message"].should.contain("Execution Does Not Exist:")769@mock_stepfunctions770@mock_sts771def test_state_machine_get_execution_history_contains_expected_success_events_when_started():772 expected_events = [773 {774 "timestamp": datetime(2020, 1, 1, 0, 0, 0, tzinfo=tzutc()),775 "type": "ExecutionStarted",776 "id": 1,777 "previousEventId": 0,778 "executionStartedEventDetails": {779 "input": "{}",780 "inputDetails": {"truncated": False},781 "roleArn": _get_default_role(),782 },783 },784 {785 "timestamp": datetime(2020, 1, 1, 0, 0, 10, tzinfo=tzutc()),786 "type": "PassStateEntered",787 "id": 2,788 "previousEventId": 0,789 "stateEnteredEventDetails": {790 "name": "A State",791 "input": "{}",792 "inputDetails": {"truncated": False},793 },794 },795 {796 "timestamp": datetime(2020, 1, 1, 0, 0, 10, tzinfo=tzutc()),797 "type": "PassStateExited",798 "id": 3,799 "previousEventId": 2,800 "stateExitedEventDetails": {801 "name": "A State",802 "output": "An output",803 "outputDetails": {"truncated": False},804 },805 },806 {807 "timestamp": datetime(2020, 1, 1, 0, 0, 20, tzinfo=tzutc()),808 "type": "ExecutionSucceeded",809 "id": 4,810 "previousEventId": 3,811 "executionSucceededEventDetails": {812 "output": "An output",813 "outputDetails": {"truncated": False},814 },815 },816 ]817 client = boto3.client("stepfunctions", region_name=region)818 sm = client.create_state_machine(819 name="test-state-machine",820 definition=simple_definition,821 roleArn=_get_default_role(),822 )823 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])824 execution_history = client.get_execution_history(825 executionArn=execution["executionArn"]826 )827 execution_history["events"].should.have.length_of(4)828 execution_history["events"].should.equal(expected_events)829@mock_stepfunctions830@mock_sts831@mock.patch.dict(os.environ, {"SF_EXECUTION_HISTORY_TYPE": "FAILURE"})832def test_state_machine_get_execution_history_contains_expected_failure_events_when_started():833 if os.environ.get("TEST_SERVER_MODE", "false").lower() == "true":834 raise SkipTest("Cant pass environment variable in server mode")835 expected_events = [836 {837 "timestamp": datetime(2020, 1, 1, 0, 0, 0, tzinfo=tzutc()),838 "type": "ExecutionStarted",839 "id": 1,840 "previousEventId": 0,841 "executionStartedEventDetails": {842 "input": "{}",843 "inputDetails": {"truncated": False},844 "roleArn": _get_default_role(),845 },846 },847 {848 "timestamp": datetime(2020, 1, 1, 0, 0, 10, tzinfo=tzutc()),849 "type": "FailStateEntered",850 "id": 2,851 "previousEventId": 0,852 "stateEnteredEventDetails": {853 "name": "A State",854 "input": "{}",855 "inputDetails": {"truncated": False},856 },857 },858 {859 "timestamp": datetime(2020, 1, 1, 0, 0, 10, tzinfo=tzutc()),860 "type": "ExecutionFailed",861 "id": 3,862 "previousEventId": 2,863 "executionFailedEventDetails": {864 "error": "AnError",865 "cause": "An error occurred!",866 },867 },868 ]869 client = boto3.client("stepfunctions", region_name=region)870 sm = client.create_state_machine(871 name="test-state-machine",872 definition=simple_definition,873 roleArn=_get_default_role(),874 )875 execution = client.start_execution(stateMachineArn=sm["stateMachineArn"])876 execution_history = client.get_execution_history(877 executionArn=execution["executionArn"]878 )879 execution_history["events"].should.have.length_of(3)880 execution_history["events"].should.equal(expected_events)881def _get_account_id():882 global account_id883 if account_id:884 return account_id...
integration_test_utils.py
Source:integration_test_utils.py
...46 This saves ~500ms per test.47 Given how many tests we're running, this adds up FAST.48 """49 return boto3.client("stepfunctions")50def create_state_machine(state_machine_definition: str):51 sfn = _sfn_client()52 response = sfn.create_state_machine(53 name=_state_machine_name(),54 roleArn=_role_arn(),55 definition=state_machine_definition,56 tags=[57 dict(key="rhodes", value="integration test"),58 dict(key="creation time", value=datetime.now(timezone.utc).isoformat()),59 ],60 )61 return response["stateMachineArn"]62def delete_state_machine(state_machine_arn: str):63 sfn = _sfn_client()64 sfn.delete_state_machine(stateMachineArn=state_machine_arn)65def try_to_create_and_delete_state_machine(state_machine_definition: str):66 state_machine_arn = create_state_machine(state_machine_definition)67 delete_state_machine(state_machine_arn)68def build_and_try_single_step_state_machine(step):69 workflow = StateMachine()70 workflow.start_with(step).end()71 definition = json.dumps(workflow.to_dict())...
states_test.py
Source:states_test.py
...10 pass11class MyOtherOtherState(State):12 pass13@pytest.fixture14def create_state_machine():15 def _create_state_machine():16 state_data = {17 'starting_state': MyState,18 'states': [19 {20 'state': MyState,21 'transitions': {22 'go_to_other_state': MyOtherState,23 'go_to_other_other_state': MyOtherOtherState24 }25 },26 {27 'state': MyOtherState,28 'transitions': {29 'go_back': MyState30 }31 },32 {33 'state': MyOtherOtherState,34 'transitions': {}35 }36 ]37 }38 return MyStateMachine(state_data)39 return _create_state_machine40class TestStateMachine:41 def test_starts_on_starting_state(self, create_state_machine):42 state_machine = create_state_machine()43 assert isinstance(state_machine.current_state, MyState)44 def test_transitions_to_other_states(self, create_state_machine):45 state_machine = create_state_machine()46 state_machine.current_state.go_to_other_state()47 assert isinstance(state_machine.current_state, MyOtherState)48 def test_allows_branching_transitions(self, create_state_machine):49 state_machine = create_state_machine()50 state_machine.current_state.go_to_other_state()51 state_machine.current_state.go_back()52 state_machine.current_state.go_to_other_other_state()...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
