forked from mongodb-developer/mongo-core-test
-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
executable file
·165 lines (138 loc) · 5.65 KB
/
run.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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/env python3
import os
import subprocess
import sys
import time
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, OperationFailure
# Constants
VERSIONS = ['5', '7']
LOG_DIRECTORY = "logs"
LOG_FILENAME = "failed.log"
SKIP_DIRECTORY = "skip"
COMMON_SKIP_FILE = "common.txt"
LOG_FILE_PATH = os.path.join(LOG_DIRECTORY, LOG_FILENAME)
TEST_DIRECTORY = os.path.join(os.getcwd(), 'mongo/jstests')
COMMON_LIST = os.path.join(SKIP_DIRECTORY, COMMON_SKIP_FILE)
# User Variables (Consider moving these to a config file or environment variables)
# Note: No username/password required for MaxScale test
MONGO_HOST = '127.0.0.1'
MONGO_PORT = 27017
MONGO_USERNAME = 'admin'
MONGO_PASSWORD = 'password'
AUTHENTICATION_DATABASE = 'admin'
AUTHENTICATION_MECHANISM = 'SCRAM-SHA-1'
USE_TLS = 'false'
LOAD_BALANCE = 'false'
if MONGO_USERNAME and MONGO_PASSWORD:
creds = f"{MONGO_USERNAME}:{MONGO_PASSWORD}@"
auth_part = f"?authMechanism={AUTHENTICATION_MECHANISM}&tls={USE_TLS}&loadBalanced={LOAD_BALANCE}"
else:
creds, auth_part = "", ""
# Check if MongoDB version is valid.
def is_valid_version(version):
return version in VERSIONS
# Ensure the existence of skip lists.
def check_list_existence(file_path, file_description):
if not os.path.exists(file_path):
print(f"\033[91m{file_description} not found\033[0m")
sys.exit(1)
# Validate MongoDB connectivity and credentials.
def validate_connection(uri):
try:
client = MongoClient(uri, serverSelectionTimeoutMS=5000)
client[AUTHENTICATION_DATABASE].command('ping')
print("Connected successfully to database")
except ConnectionFailure:
print("\033[91mFailed to connect to database\033[0m")
sys.exit(1)
except OperationFailure:
print("\033[91mAuthentication failed for database\033[0m")
sys.exit(1)
# Create MongoDB URI from user variables.
def create_mongo_uri():
return f"mongodb://{creds}{MONGO_HOST}:{MONGO_PORT}/{auth_part}"
# Build the mongo command with the necessary arguments.
def build_mongo_command(script_path):
mongo_command = [
'mongo',
f"mongodb://{creds}{MONGO_HOST}:{MONGO_PORT}/{auth_part}"
]
mongo_command.append(script_path)
return mongo_command
# Run an individual script using subprocess.
def run_script(script_path):
mongo_command = build_mongo_command(script_path)
#print(mongo_command)
try:
subprocess.run(mongo_command, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
return True
except subprocess.CalledProcessError:
return False
# Find all JS files that are not in the exclusion lists.
def find_js_files(directory, mongo_version):
exclusions = set()
with open(COMMON_LIST, 'r') as f:
exclusions.update([TEST_DIRECTORY + line.strip() for line in f])
# Add specific mongo version exclusions.
mongo_exclusions = f"{SKIP_DIRECTORY}/mongo{mongo_version}.txt"
check_list_existence(mongo_exclusions, mongo_exclusions)
with open(mongo_exclusions, 'r') as f:
exclusions.update([TEST_DIRECTORY + line.strip() for line in f])
js_files = []
for root, dirs, files in os.walk(directory):
for file in files:
if file.endswith('.js'):
full_path = os.path.join(root, file)
if full_path not in exclusions:
js_files.append(full_path)
if not js_files:
print(f"No JS files found in {directory}")
sys.exit(1)
return js_files
# Main function.
if __name__ == '__main__':
start_time = time.time()
# Ensure there is at least one argument provided and it's valid
if len(sys.argv) < 2 or not is_valid_version(sys.argv[1]):
print("You must choose a MongoDB server version to compare against (5 or 7)")
print(f"Example usage: python3 {sys.argv[0]} 5")
sys.exit(1)
version = sys.argv[1]
os.makedirs(LOG_DIRECTORY, exist_ok=True)
mongo_list = os.path.join(SKIP_DIRECTORY, f"mongo{version}.txt")
uri = create_mongo_uri()
validate_connection(uri)
script_files = sorted(find_js_files(TEST_DIRECTORY, version))
total_scripts = len(script_files)
success_count = 0
failed_scripts = []
for idx, script_path in enumerate(script_files, start=1):
filename = os.path.basename(script_path)
script_start = time.time() # Start time for the current script
success = run_script(script_path)
script_end = time.time() # End time for the current script
# Calculate elapsed time for this script
elapsed_time = script_end - script_start
if success:
success_count += 1
print(f"Passed test {idx}/{total_scripts}: {filename} (Time: {elapsed_time:.4f}s)")
else:
print("\033[91m" + f"Failed test {idx}/{total_scripts}: {filename} (Time: {elapsed_time:.4f}s)" + "\033[0m")
failed_scripts.append(script_path)
end_time = time.time()
total_time = end_time - start_time
success_percentage = (success_count / total_scripts) * 100 if total_scripts > 0 else 0
summary_log = (
f"\nTotal Tests Attempted: {total_scripts}\n"
f"Total Successful Executions: {success_count}\n"
f"Percentage Successful: {success_percentage:.2f}%\n"
f"Total Execution Time: {total_time:.2f} seconds\n"
)
print(summary_log)
if failed_scripts:
with open(LOG_FILE_PATH, 'w') as log_file:
for failed_script in failed_scripts:
relative_path = os.path.relpath(failed_script, TEST_DIRECTORY)
log_file.write(f"{relative_path}\n")
print(f"Failed scripts have been logged to: {LOG_FILE_PATH}")