-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathAPI_Testing.py
More file actions
43 lines (34 loc) · 1.51 KB
/
API_Testing.py
File metadata and controls
43 lines (34 loc) · 1.51 KB
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
import json
import requests
import random
import uuid
# Updated to use local Temporal API endpoint instead of AWS API Gateway
# Original AWS endpoint (commented out):
# url = "https://<random_id>.execute-api.us-east-2.amazonaws.com/v1/execution"
url = "http://localhost:8000/v1/execution"
with open("names.txt", "r", encoding="utf-8") as f:
names = f.readlines()[0].replace('"', '').strip().split(', ')
results_dict = {}
for name in names:
num1, num2 = random.randrange(0,100), random.randrange(200,300)
num_sum = num1 + num2
exe_num = str(uuid.uuid1())
data = '{' + f'\\"personName\\": \\"{name}\\", \\"baseNumber\\": \\"{num1}\\", \\"additionalNumber\\": \\"{num2}\\"' + '}'
body = '{' + f'"input": "{data}", "name": "{exe_num}", "stateMachineArn":"arn:aws:states:us-east-2:<AWS_Account_ID>:stateMachine:state-machine-test"' + '}'
response = requests.post(url=url, data=body)
# Updated response parsing for Temporal API
# AWS returns: executionArn, startDate, etc.
# Temporal returns: workflow_id, run_id, status
response_data = response.json()
results_dict[name] = {
"base": num1,
"add": num2,
"sum": num_sum,
"uuid": exe_num,
"result": response_data,
"workflow_id": response_data.get("workflow_id", "N/A"),
"run_id": response_data.get("run_id", "N/A"),
"status": response_data.get("status", "unknown")
}
with open("results.json", "w", encoding="utf-8") as f:
f.writelines(json.dumps(results_dict, indent=4))