-
Notifications
You must be signed in to change notification settings - Fork 0
/
fey.py
131 lines (110 loc) · 5.33 KB
/
fey.py
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
import json
import requests
import os
import sys
from datetime import datetime
# ANSI color codes for terminal output
GREEN = '\033[92m'
RED = '\033[91m'
RESET = '\033[0m'
# Constants
REQUESTS_DIR = "requests"
LOGS_DIR = "requests/logs"
ENCODING = "utf-8"
def colorize(text, color):
return f"{color}{text}{RESET}"
def execute_requests_from_json(json_filename):
try:
with open(f'{REQUESTS_DIR}/{json_filename}.json', 'r') as file:
request_data = json.load(file)
name = request_data['name']
root = request_data['root']
print(f"Executing requests for: {name}")
headers = request_data.get('headers', {})
routes = request_data.get('routes', [])
# Check if the root URL responds
try:
root_response = requests.head(root)
if not root_response.ok:
pass
except requests.ConnectionError:
print(
f"Error: Failed to connect to {root}. Is the server running?")
sys.exit(1)
try:
for route_info in routes:
path = route_info['path']
method = route_info.get('method', 'GET')
body = route_info.get('body', None)
url = root + path
try:
if method == 'GET':
response = requests.get(url, headers=headers)
elif method == 'POST':
response = requests.post(
url, json=body, headers=headers)
elif method == 'UPDATE':
response = requests.put(
url, json=body, headers=headers)
elif method == 'DELETE':
response = requests.delete(
url, json=body, headers=headers)
# Add support for other HTTP methods as needed
status_code = response.status_code
if 200 <= status_code < 300:
color_status = colorize(status_code, GREEN)
else:
color_status = colorize(status_code, RED)
print(
f"Request to {url} ({method}): {color_status}")
date = datetime.now().strftime("%Y-%m-%d")
if response.ok:
# make directory if it doesn't exist for requests
if not os.path.exists(LOGS_DIR):
os.makedirs(LOGS_DIR)
# write the response to a file if there's content
if response.text:
RECORD_TEXT = f"{LOGS_DIR}/{date}-out-{method.lower()}{path.replace('/', '-')}.txt"
with open(RECORD_TEXT, 'w', encoding=ENCODING) as f:
f.write(response.text)
elif response.json():
RECORD_JSON = f"{LOGS_DIR}/{date}-out-{method.lower()}{path.replace('/', '-')}.json"
with open(RECORD_JSON, 'w', encoding=ENCODING) as f:
json.dump(response.json(), f)
else:
print("No content")
else:
# make directory if it doesn't exist for logs
if not os.path.exists(LOGS_DIR):
os.makedirs(LOGS_DIR)
# write the response to a file if there's content
if response.text:
LOG_TEXT = f"{LOGS_DIR}/{date}-error-{method.lower()}{path.replace('/', '-')}.txt"
with open(LOG_TEXT, 'w', encoding=ENCODING) as f:
f.write(response.text)
elif response.json():
LOG_JSON = f"{LOGS_DIR}/{date}-error-{method.lower()}{path.replace('/', '-')}.json"
with open(LOG_JSON, 'w', encoding=ENCODING) as f:
json.dump(response.json(), f)
else:
print("No content")
except requests.ConnectionError:
print(
f"Error: Failed to connect to {url}. Is the server running?")
except requests.Timeout:
print(f"Error: Request to {url} timed out")
except requests.ConnectionError:
print(
f"Error: Failed to connect to {root}. Is the server running?")
except FileNotFoundError:
print(
f"Error: File '{json_filename}.json' not found in 'requests' directory")
if __name__ == "__main__":
if len(sys.argv) < 2:
for filename in os.listdir(REQUESTS_DIR):
filename = filename.replace(".json", "")
execute_requests_from_json(filename)
sys.exit(0)
# specific filename
JSON_FILENAME = sys.argv[1]
execute_requests_from_json(JSON_FILENAME)