1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
import json
from unittest import SkipTest
import boto3
from moto import mock_aws, settings
from . import verify_execution_result
@mock_aws(config={"stepfunctions": {"execute_state_machine": True}})
def test_state_machine_with_map_reader_csv():
if settings.TEST_SERVER_MODE:
raise SkipTest("Don't need to test this in ServerMode")
# Prepare CSV
s3 = boto3.client("s3", "us-east-1")
bucket_name = "mybucket"
s3.create_bucket(Bucket=bucket_name)
key = "file.csv"
csv_file = (
"Value1,Value2,Value3\n"
"Value4,Value5,Value6\n"
",,,\n"
"true,1,'HelloWorld'\n"
"Null,None,\n"
" \n"
)
s3.put_object(Bucket=bucket_name, Key=key, Body=csv_file)
tmpl_name = "map_item_reader"
exec_input = {"Bucket": bucket_name, "Key": key}
def _verify_result(client, execution, execution_arn):
output = json.loads(execution["output"])
assert {"H1": "Value1", "H2": "Value2", "H3": "Value3"} in output
assert {"H1": "Value4", "H2": "Value5", "H3": "Value6"} in output
assert {"H1": "", "H2": "", "H3": ""} in output
assert {"H1": "true", "H2": "1", "H3": "'HelloWorld'"} in output
assert {"H1": "Null", "H2": "None", "H3": ""} in output
assert {"H1": " ", "H2": "", "H3": ""} in output
verify_execution_result(
_verify_result, "SUCCEEDED", tmpl_name, exec_input=json.dumps(exec_input)
)
@mock_aws(config={"stepfunctions": {"execute_state_machine": True}})
def test_state_machine_with_map_reader_csv_with_header():
if settings.TEST_SERVER_MODE:
raise SkipTest("Don't need to test this in ServerMode")
# Prepare CSV
s3 = boto3.client("s3", "us-east-1")
bucket_name = "mybucket"
s3.create_bucket(Bucket=bucket_name)
key = "file.csv"
csv_file = (
"Col1,Col2,Col3\n"
"Value1,Value2,Value3\n"
"Value4,Value5,Value6\n"
",,,\n"
"true,1,'HelloWorld'\n"
"Null,None,\n"
" \n"
)
s3.put_object(Bucket=bucket_name, Key=key, Body=csv_file)
tmpl_name = "map_item_reader_with_header"
exec_input = {"Bucket": bucket_name, "Key": key}
def _verify_result(client, execution, execution_arn):
output = json.loads(execution["output"])
assert {"Col1": "Value1", "Col2": "Value2", "Col3": "Value3"} in output
assert {"Col1": "", "Col2": "", "Col3": ""} in output
runs = client.list_map_runs(executionArn=execution_arn)["mapRuns"]
assert len(runs) == 1
assert runs[0]["executionArn"] == execution_arn
run = client.describe_map_run(mapRunArn=runs[0]["mapRunArn"])
assert run["itemCounts"]["total"] == 6
client.update_map_run(mapRunArn=runs[0]["mapRunArn"], maxConcurrency=4)
verify_execution_result(
_verify_result, "SUCCEEDED", tmpl_name, exec_input=json.dumps(exec_input)
)
|
