Manual test script
Background ¶
MCT needs to migrate from VMware to OCP4. Consequently, the test timer has to be migrated from the VMware agent to the Kubed agent. However, this timer migration has a significant impact and cannot be fully tested prior to the migration.
For instance, some tests rely on network accessibility. We can inform customers to open ACL to grant access to Kubed, but there may be oversights. If the migration occurs, it could lead to false alert incidents.
We could conduct manual tests first for all cases planned for migration, but the quantity is too large to be done manually. Thus, we believe we can write a script to simulate this batch of manual test operations.
Script Design ¶
Language: Python
Simulate MCT page operations for batch operations.
Manual test zone on page ¶
Check the http request and use it to perform manual tests in the script.
Input ¶
- Username (the person conducting the manual test, who can then check the result on the manual test history page).
- zoneid list (the zones to be manually tested; you can use SQL to filter the target zoneid list).
Output ¶
- The zoneid and ManualSessionId for each request.
- All ManualSessionIds combined to facilitate input for the manual test check script.
Check manual test result on page ¶
Check the http request and use it to check the manual test results in the script.
You may need to execute this script more than once as manual test execution takes some time.
Input ¶
- ManualSessionId list (obtainable from the manual test script output).
Output ¶
- Only output manual tests with a result other than "pass".
- For each non-pass manual test, check its check's past 1 hour status with "Current Hour Status" API used in search page
- Key information for each request, such as the check name, test result, and detail status link.
- All check name and detail status links for non-passing tests combined.
- All check name and detail status links for non-passing tests combined.(except for always failed check)
- Currently, "always failed check" means that all statuses of the check have not passed in the past 1 hour.
Set up script env ¶
It is recommended to run with a Python 3.9 virtual environment.
lewan@LEWAN-M-6FV6 manual-test-script % python3.9 -m venv venv_py39
lewan@LEWAN-M-6FV6 manual-test-script % source venv_py39/bin/activate
(venv_py39) lewan@LEWAN-M-6FV6 manual-test-script % pip install --upgrade requests
Script ¶
Replace {ci-token} with a valid token, e.g., an MCT internal token or a token from an MCT portal request.
Manual test script ¶
import argparse
import requests
import json
def main():
"""
Main function to parse arguments, make API requests, and print results.
"""
parser = argparse.ArgumentParser(description="Make API requests for manual testing with dynamic zone IDs.")
parser.add_argument("-cec", "--cec", required=True, help="Username for the API request (e.g., 'lewan' or 'test').")
parser.add_argument("-zoneids", "--zoneids", required=True, help="Comma-separated list of zone IDs (e.g., '1221961,1221962').")
args = parser.parse_args()
username_input = args.cec
zone_ids_str = args.zoneids
try:
# Parse the comma-separated zone IDs into a list of integers
zone_ids = [int(zid.strip()) for zid in zone_ids_str.split(',') if zid.strip()]
except ValueError:
print("Error: Zone IDs must be a comma-separated list of integers.")
return
if not zone_ids:
print("Error: No valid zone IDs provided after parsing.")
return
api_url = "https://mctapi.prod.webex.com/mpi/test/manual/123"
all_manual_session_ids = []
# Base headers for the API request
base_headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7",
"authorization": "{ci-token}",
"content-type": "application/json; charset=UTF-8",
}
# Base payload for the API request. zoneId will be updated dynamically.
base_payload = {
"clusterId": "78791",
"mtype": 2,
"serverIp": "",
"domainId": 0,
"zoneId": 0, # This will be replaced for each request
"serverId": 0,
"serverType": 0,
"protocol": 0,
"agent": "wjfkgen-p-1",
"identifier": "",
"vGroupId": 1100100102
}
# Set the 'username' header based on the input 'cec'
headers = base_headers.copy()
headers["username"] = "test" if username_input.lower() == "test" else username_input
# Iterate through each zone ID and make a request
for i, zone_id in enumerate(zone_ids):
print(f"request {i + 1}:")
# Create a copy of the base payload and update the zoneId
payload = base_payload.copy()
payload["zoneId"] = zone_id
try:
# Make the POST request
response = requests.post(api_url, headers=headers, json=payload)
response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
response_data = response.json()
# Check if the response indicates success and contains ManualSessionId
if response_data.get("errorCode") == "OKOKOK" and "ManualSessionId" in response_data:
manual_session_id = response_data["ManualSessionId"]
print(f'"zoneId": {zone_id}')
print(f'ManualSessionId": "{manual_session_id}"')
all_manual_session_ids.append(manual_session_id)
else:
print(f" API response error for zoneId {zone_id}: {response_data.get('errorCode', 'Unknown Error')}")
print(f" Full response: {json.dumps(response_data, indent=2)}")
except requests.exceptions.HTTPError as err:
print(f" HTTP error occurred for zoneId {zone_id}: {err}")
print(f" Response content: {err.response.text}")
except requests.exceptions.ConnectionError as err:
print(f" Connection error occurred for zoneId {zone_id}: {err}")
except requests.exceptions.Timeout as err:
print(f" Timeout error occurred for zoneId {zone_id}: {err}")
except requests.exceptions.RequestException as err:
print(f" An unexpected request error occurred for zoneId {zone_id}: {err}")
except json.JSONDecodeError:
print(f" Failed to decode JSON response for zoneId {zone_id}. Response content: {response.text}")
except Exception as e:
print(f" An unexpected error occurred for zoneId {zone_id}: {e}")
print() # Add a newline for better readability between requests
# Print all collected ManualSessionIds at the end
if all_manual_session_ids:
print(f'ManualSessionIds: "{", ".join(all_manual_session_ids)}"')
else:
print("No ManualSessionIds were successfully retrieved.")
if __name__ == "__main__":
main()
Manual test result check script ¶
import argparse
import requests
import json
import datetime
import urllib.parse
def main():
"""
Main function to parse arguments, make API requests to get manual test results,
and print relevant information, including historical status checks.
"""
parser = argparse.ArgumentParser(description="Retrieve and display manual test results based on manual IDs.")
parser.add_argument("-manualids", "--manualids", required=True,
help="Comma-separated list of manual IDs (e.g., '135752171,135752421').")
args = parser.parse_args()
manual_ids_str = args.manualids
try:
# Parse the comma-separated manual IDs into a list of integers
manual_ids = [int(mid.strip()) for mid in manual_ids_str.split(',') if mid.strip()]
except ValueError:
print("Error: Manual IDs must be a comma-separated list of integers.")
return
if not manual_ids:
print("Error: No valid manual IDs provided after parsing.")
return
base_manual_api_url = "https://mctapi.prod.webex.com/mpi/test/manual-id/"
base_detail_status_url = "https://mct.webex.com/detail-status/"
base_historical_status_api_url = "https://mctapi.prod.webex.com/mpi/clusterlist/queryZoneServerDetail"
all_failed_detail_links_summary = [] # Stores (name, link) for all non-passing items
new_or_intermittent_failures_summary = [] # Stores (name, link) for items where "all status not pass in last 1h: no"
# Headers for the API requests
headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7",
"authorization": "{ci-token}", # Include if required by the API
"content-type": "application/json; charset=UTF-8",
}
for i, manual_id in enumerate(manual_ids):
print(f"request {i + 1}:")
manual_api_url = f"{base_manual_api_url}{manual_id}/0"
found_failed_items_in_request = False
try:
response = requests.get(manual_api_url, headers=headers)
response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
response_data = response.json()
# Correctly extract the list of items from the 'manualList' key
items = response_data.get('manualList', [])
if not items:
print(" No items found in 'manualList' in the response for this manual ID.")
print()
continue
check_count = 0
for item in items:
type_id = item.get("typeId")
status = item.get("status")
result = item.get("result")
server_id = item.get("serverId") # Get serverId for historical check
item_name = item.get("name", "N/A")
# Condition for non-passing items: "typeId": is 3, "status": is not 1, "result": is not "Pass"
# Using OR for status and result because if either indicates failure, it's a non-pass.
if type_id == 3 and (status != 1 or result != "Pass"):
found_failed_items_in_request = True
check_count += 1
print(f"check {check_count}:")
item_manual_id = item.get("manualId")
detail_link = f"{base_detail_status_url}{item_manual_id}?isManual=true"
print(f"details status link: {detail_link}")
print(f'"manualId": {item_manual_id}')
print(f'"serverId": "{server_id}"')
print(f'"type": "{item.get("type", "N/A")}",')
print(f'"name": "{item_name}",')
print(f'"status": {status}')
print(f'"result": "{result}"')
# Add to the list of all failed checks for final summary
all_failed_detail_links_summary.append({"name": item_name, "link": detail_link})
# --- Historical Status Check ---
if server_id:
# Calculate UTC+0 hour for the startTime parameter
now_utc = datetime.datetime.now(datetime.timezone.utc)
# Format to YYYY-MM-DD HH:00:00
start_time_utc_str = now_utc.strftime("%Y-%m-%d %H:00:00")
# URL encode the time string
encoded_start_time = urllib.parse.quote(start_time_utc_str)
historical_status_url = (
f"{base_historical_status_api_url}?"
f"serverIds={server_id}&"
f"startTime={encoded_start_time}&"
f"timeDuration=1&" # Query for the last 1 hour
f"timeZone=0" # UTC timezone
)
try:
hist_response = requests.get(historical_status_url, headers=headers)
hist_response.raise_for_status()
hist_data = hist_response.json()
is_consistently_failing = True # Assume true until a 'Pass' status is found
server_status_items = hist_data.get('serverStatusItems', [])
if not server_status_items:
# No data for this server in the last hour, so it's not consistently failing (i.e., "no")
is_consistently_failing = False
else:
display1_statuses = []
# The response structure shows display1 within the first item of serverStatusItems
if server_status_items and 'display1' in server_status_items[0]:
display1_statuses = server_status_items[0].get('display1', [])
if not display1_statuses:
# No status entries in display1, so not consistently failing (i.e., "no")
is_consistently_failing = False
else:
for d_item in display1_statuses:
# If any status is 0 (Pass) or statusSuccess is true, then it's NOT consistently failing
# Assuming serverStatus 0 means success
if d_item.get("serverStatus") == 0 or d_item.get("statusSuccess") is True:
is_consistently_failing = False
break
if is_consistently_failing:
print("all status not pass in last 1h: yes")
else:
print("all status not pass in last 1h: no")
# Add to the list for new/intermittent failures
new_or_intermittent_failures_summary.append({"name": item_name, "link": detail_link})
except requests.exceptions.RequestException as hist_err:
print(f" Error fetching historical status for serverId {server_id}: {hist_err}")
print("all status not pass in last 1h: N/A (Error fetching historical data)")
except json.JSONDecodeError:
print(f" Failed to decode JSON from historical status for serverId {server_id}. Response: {hist_response.text}")
print("all status not pass in last 1h: N/A (Invalid historical data response)")
else:
print(" No serverId found for historical check.")
print() # Newline after historical check result for readability
if not found_failed_items_in_request:
print("all pass!")
print() # Add a newline for better readability between requests
except requests.exceptions.HTTPError as err:
print(f" HTTP error occurred for manualId {manual_id}: {err}")
print(f" Response content: {err.response.text}")
except requests.exceptions.ConnectionError as err:
print(f" Connection error occurred for manualId {manual_id}: {err}")
except requests.exceptions.Timeout as err:
print(f" Timeout error occurred for manualId {manual_id}: {err}")
except requests.exceptions.RequestException as err:
print(f" An unexpected request error occurred for manualId {manual_id}: {err}")
except json.JSONDecodeError:
print(f" Failed to decode JSON response for manualId {manual_id}. Response content: {response.text}")
except Exception as e:
print(f" An unexpected error occurred for manualId {manual_id}: {e}")
print() # Add a newline for better readability between requests
# Print all collected details status links for all failed checks
if all_failed_detail_links_summary:
print("all fail check:")
for item_summary in all_failed_detail_links_summary:
print(f"check name: {item_summary['name']}")
print(f"details status link: {item_summary['link']}")
print() # Newline for separation
# Print details for items where "all status not pass in last 1h: no"
print("all fail check: （all status not pass in last 1h: no）")
if new_or_intermittent_failures_summary:
for item_summary in new_or_intermittent_failures_summary:
print(f"check name: {item_summary['name']}")
print(f"details status link: {item_summary['link']}")
else:
print("no case")
if __name__ == "__main__":
main()
Case ¶
Specify the manual test user as lewan and the test target zoneid list as 1221961, 2184411:
(venv_py39) lewan@LEWAN-M-6FV6 manual-test-script % python3.9 manual-test-zoneids.py -cec "lewan" -zoneids "1221961, 2184411"
request 1:
"zoneId": 1221961
ManualSessionId": "135752171"
request 2:
"zoneId": 2184411
ManualSessionId": "135752421"
ManualSessionIds: "135752171, 135752421"
(venv_py39) lewan@LEWAN-M-6FV6 manual-test-script %
Use the manual test script output ManualSessionIds: "135752171, 135752421" as the input for the manual test result script to check the manual test result.
(venv_py39) lewan@LEWAN-M-6FV6 manual-test-script % python3.9 manual-test-result-manualids.py -manualids "135752171, 136062421"
request 1:
all pass!
request 2:
check 1:
details status link: https://mct.webex.com/detail-status/2112527601?isManual=true
"manualId": 2112527601
"serverId": "38319171"
"type": "URL Watch Server",
"name": "mttx2mul003.webex.com()",
"status": 2
"result": "Not Test"
all status not pass in last 1h: yes
all fail check:
check name: mttx2mul003.webex.com()
details status link: https://mct.webex.com/detail-status/2112527601?isManual=true
all fail check: （all status not pass in last 1h: no）
no case # which means mttx2mul003.webex.com check is always failed (all not pass in last hour), so ignore it here
(venv_py39) lewan@LEWAN-M-6FV6 manual-test-script %
Search output
- If you find "Init" or "Running", please wait for the test to finish.
- If you find "No result", please check this check. For example, conduct a direct manual test on the page.
API case ¶
These API are used in script, you can check them to better understand and modify the script.
Manual test with zoneid and cec ¶
curl --location 'https://mctapi.prod.webex.com/mpi/test/manual/123' \
--header 'accept: application/json, text/plain, */*' \
--header 'accept-language: en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7' \
--header 'authorization: {ci-token}' \
--header 'content-type: application/json; charset=UTF-8' \
--header 'username: lewan' \
--data '{
"clusterId": "78791",
"mtype": 2,
"serverIp": "",
"domainId": 0,
"zoneId": 1221961,
"serverId": 0,
"serverType": 0,
"protocol": 0,
"agent": "wjfkgen-p-1",
"identifier": "",
"vGroupId": 1100100102
}'
response:
{
"errorCode": "OKOKOK",
"localIp": "172.16.12.194",
"status": "OKOKOK",
"ManualSessionId": "135742671"
}
Check manual test result ¶
curl --location 'https://mctapi.prod.webex.com/mpi/test/manual-id/135742671/0' \
--header 'accept: application/json, text/plain, */*' \
--header 'accept-language: en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7' \
--header 'authorization: {ci-token}' \
--header 'content-type: application/json; charset=UTF-8' \
response:
{
"errorCode": "OKOKOK",
"localIp": "172.16.12.194",
"manualList": [
{
"id": 0,
"manualId": 0,
"serverId": null,
"zoneType": 0,
"name": "WebACD_Cluster_AS(plugin)",
"type": "javaUrlStation",
"typeId": 2,
"result": "0/1",
"tagID": "zone-WebACD_Cluster_AS(plugin)",
"className": "child-of-null",
"startTime": null,
"endTime": null,
"status": 1,
"serverType": 0,
"agent": null,
"agentId": 0,
"userName": null,
"sourceId": 1100100102,
"networkEnv": null
},
{
"id": 0,
"manualId": 2111681401,
"serverId": "4875681",
"zoneType": 7101,
"name": "WebACD_Cluster_AS_Fail=Impact(WebACD_Cluster_AS_Fail=Impact)",
"type": "jmeter",
"typeId": 3,
"result": "Pass",
"tagID": "server-WebACD_Cluster_AS_Fail=Impact",
"className": null,
"startTime": null,
"endTime": null,
"status": 1,
"serverType": 9401,
"agent": null,
"agentId": 0,
"userName": null,
"sourceId": 1100100102,
"networkEnv": null
}
],
"totalCount": 0,
"init": 0,
"running": 0,
"error": 0,
"finish": 0,
"testName": "Zone - WebACD_Cluster_AS(plugin)",
"testStartTime": "2025-07-22 08:09:50",
"testEndTime": "2025-07-22 08:09:58",
"testStatus": "Pass",
"progress": "100%",
"failServerRate": "0/1",
"result": "success",
"clusterId": 78791,
"agentId": 0,
"sourceId": 1100100102
}
Query last hour statuses for a check ¶
curl --location 'https://mctapi.prod.webex.com/mpi/clusterlist/queryZoneServerDetail?serverIds=38601496&startTime=2025-07-23%2009:00:00&timeDuration=1&timeZone=0' \
--header 'accept: application/json, text/plain, */*' \
--header 'accept-language: en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7' \
--header 'authorization: {ci-token}' \
--header 'content-type: application/json; charset=UTF-8' \
response:
{
"errorCode": "OKOKOK",
"localIp": "172.16.12.194",
"serverStatusItems": [
{
"id": 38601496,
"name": null,
"type": 10001,
"typeName": null,
"status": 0,
"childs": 0,
"userId": 0,
"hostname": "http://kubed-as.webex.com",
"ip": "http://kubed-as.webex.com",
"display1": [
{
"zoneId": 2153201,
"serverId": 38601496,
"statusId": 5741175505,
"serverStatus": 0,
"timeshow": 1753261327000,
"displaytime": "2025-07-23 09:02:07",
"psttime": "2025-07-23 02:02:07",
"gmttime": "2025-07-23 09:02:07",
"hourpos": 0,
"protocol": 0,
"image": null,
"icon": 1,
"sourceId": 1100007903,
"agentName": "INTRL",
"domainId": 767991,
"serverType": 10001,
"pollingFlag": 1,
"resultType": 0,
"dbtime": "2025-07-23 09:02:07",
"lastFail": false,
"networkEnv": null,
"orgName": null,
"maintainStatus": false,
"forceMaintainStatus": false,
"statusFail": false,
"statusSuccess": true,
"statusIgnored": false,
"gastatus": true
},
{
"zoneId": 2153201,
"serverId": 38601496,
"statusId": 5741191830,
"serverStatus": 0,
"timeshow": 1753261601000,
"displaytime": "2025-07-23 09:06:41",
"psttime": "2025-07-23 02:06:41",
"gmttime": "2025-07-23 09:06:41",
"hourpos": 0,
"protocol": 0,
"image": null,
"icon": 1,
"sourceId": 1100007903,
"agentName": "INTRL",
"domainId": 767991,
"serverType": 10001,
"pollingFlag": 1,
"resultType": 0,
"dbtime": "2025-07-23 09:06:41",
"lastFail": false,
"networkEnv": null,
"orgName": null,
"maintainStatus": false,
"forceMaintainStatus": false,
"statusFail": false,
"statusSuccess": true,
"statusIgnored": false,
"gastatus": true
}
],
"display2": [],
"display3": [],
"display4": [],
"lastPollError": false,
"fourHourError": false,
"lastHourError": false,
"maintainReason": "",
"flagPST": true,
"zoneType": 7101,
"zoneId": 2153201,
"resultType": 0,
"sourceId": 1100007903
}
],
"serverIds": "38601496",
"failedZoneIds": []
}