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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
import json
import moto.server as server
from moto import mock_aws
@mock_aws
def test_textract_start_text_detection():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.StartDocumentTextDetection"}
request_body = {
"DocumentLocation": {
"S3Object": {"Bucket": "bucket", "Name": "name", "Version": "version"}
}
}
resp = test_client.post("/", headers=headers, json=request_body)
data = json.loads(resp.data.decode("utf-8"))
assert resp.status_code == 200
assert isinstance(data["JobId"], str)
@mock_aws
def test_detect_document_text():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.DetectDocumentText"}
request_body = {
"DocumentLocation": {
"S3Object": {"Bucket": "bucket", "Name": "name", "Version": "version"}
}
}
resp = test_client.post("/", headers=headers, json=request_body)
data = json.loads(resp.data.decode("utf-8"))
assert resp.status_code == 200
assert isinstance(data["Blocks"], list)
@mock_aws
def test_textract_start_text_detection_without_document_location():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.StartDocumentTextDetection"}
resp = test_client.post("/", headers=headers, json={})
data = json.loads(resp.data.decode("utf-8"))
assert resp.status_code == 400
assert data["__type"] == "InvalidParameterException"
assert (
data["message"]
== "An input parameter violated a constraint. For example, in synchronous operations, an InvalidParameterException exception occurs when neither of the S3Object or Bytes values are supplied in the Document request parameter. Validate your parameter before calling the API operation again."
)
@mock_aws
def test_textract_get_text_detection():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.StartDocumentTextDetection"}
request_body = {
"DocumentLocation": {
"S3Object": {"Bucket": "bucket", "Name": "name", "Version": "version"}
}
}
resp = test_client.post("/", headers=headers, json=request_body)
start_job_data = json.loads(resp.data.decode("utf-8"))
headers = {"X-Amz-Target": "X-Amz-Target=Textract.GetDocumentTextDetection"}
request_body = {
"JobId": start_job_data["JobId"],
}
resp = test_client.post("/", headers=headers, json=request_body)
assert resp.status_code == 200
data = json.loads(resp.data.decode("utf-8"))
assert data["JobStatus"] == "SUCCEEDED"
@mock_aws
def test_textract_get_text_detection_without_job_id():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.GetDocumentTextDetection"}
request_body = {
"JobId": "invalid_job_id",
}
resp = test_client.post("/", headers=headers, json=request_body)
assert resp.status_code == 400
data = json.loads(resp.data.decode("utf-8"))
assert data["__type"] == "InvalidJobIdException"
assert data["message"] == "An invalid job identifier was passed."
@mock_aws
def test_textract_start_document_analysis():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.StartDocumentAnalysis"}
request_body = {
"DocumentLocation": {
"S3Object": {"Bucket": "bucket", "Name": "name", "Version": "version"}
},
"FeatureTypes": ["TABLES", "FORMS"],
}
resp = test_client.post("/", headers=headers, json=request_body)
data = json.loads(resp.data.decode("utf-8"))
assert resp.status_code == 200
assert isinstance(data["JobId"], str)
@mock_aws
def test_textract_get_document_analysis():
backend = server.create_backend_app("textract")
test_client = backend.test_client()
headers = {"X-Amz-Target": "X-Amz-Target=Textract.StartDocumentAnalysis"}
request_body = {
"DocumentLocation": {
"S3Object": {"Bucket": "bucket", "Name": "name", "Version": "version"}
},
"FeatureTypes": ["TABLES", "FORMS"],
}
resp = test_client.post("/", headers=headers, json=request_body)
start_job_data = json.loads(resp.data.decode("utf-8"))
headers = {"X-Amz-Target": "X-Amz-Target=Textract.GetDocumentAnalysis"}
request_body = {
"JobId": start_job_data["JobId"],
}
resp = test_client.post("/", headers=headers, json=request_body)
assert resp.status_code == 200
data = json.loads(resp.data.decode("utf-8"))
assert data["JobStatus"] == "SUCCEEDED"
|
