scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-10-30 18:27:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

scenario.py: sleep 30 seconds if /events/ flag is not present

Browse Source 
regression_test: work with new scenario.py

Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-12-03 15:23:22 -08:00
parent fed17368dc
commit 9b31c41749
2 changed files with 44 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
py-scripts/regression_test.sh
View File

@@ -74,15 +74,15 @@ if [[ ${#MGR} -eq 0 ]]; then # Allow the user to change the radio they test agai
MGR="localhost" MGR="localhost"
fi fi
SCENARIO_CHECK="$(python3 -c "import requests; print(requests.get('http://${MGR}:8080/events/').status_code)")" #SCENARIO_CHECK="$(python3 -c "import requests; print(requests.get('http://${MGR}:8080/events/').status_code)")"
if [[ ${SCENARIO_CHECK} -eq 200 ]]; then #if [[ ${SCENARIO_CHECK} -eq 200 ]]; then
pass # pass
else #else
echo "${SCENARIO_CHECK}" # echo "${SCENARIO_CHECK}"
echo "Your LANforge Manager is out of date. Regression test requires LANforge version 5.4.4 or higher in order to run" # echo "Your LANforge Manager is out of date. Regression test requires LANforge version 5.4.4 or higher in order to run"
echo "Please upgrade your LANforge using instructions found at https://www.candelatech.com/downloads.php#releases" # echo "Please upgrade your LANforge using instructions found at https://www.candelatech.com/downloads.php#releases"
exit 1 # exit 1
fi #fi
if [ -d "/home/lanforge/lanforge_env" ] if [ -d "/home/lanforge/lanforge_env" ]
then then

51
py-scripts/scenario.py
View File

@@ -4,6 +4,7 @@ import os
import importlib import importlib
import argparse import argparse
import time import time
import requests
if sys.version_info[0] != 3: if sys.version_info[0] != 3:
print("This script requires Python 3") print("This script requires Python 3")
@@ -21,10 +22,8 @@ Realm = realm.Realm
def get_events(event_log, value): def get_events(event_log, value):
results = [] results = []
for event in event_log: for event in event_log:
try: if event.values():
results.append(list(event.values())[0][value]) results.append(list(event.values())[0][value])
except:
pass
return results return results
@@ -54,6 +53,7 @@ class LoadScenario(Realm):
stop=None, stop=None,
quiesce=None, quiesce=None,
timeout=120, timeout=120,
code=None,
debug=False): debug=False):
super().__init__(lfclient_host=mgr, super().__init__(lfclient_host=mgr,
lfclient_port=mgr_port, lfclient_port=mgr_port,
@@ -67,9 +67,13 @@ class LoadScenario(Realm):
self.stop = stop self.stop = stop
self.quiesce = quiesce self.quiesce = quiesce
self.timeout = timeout self.timeout = timeout
self.code = code
self.starting_events = None
starting_events = self.json_get('/events/since=time/1h') def start_test(self):
self.starting_events = self.json_get('/events/since=time/1h')
def load_scenario(self):
if self.scenario is not None: if self.scenario is not None:
data = { data = {
"name": self.scenario, "name": self.scenario,
@@ -93,12 +97,14 @@ class LoadScenario(Realm):
print("Quiescing test group %s..." % self.quiesce) print("Quiescing test group %s..." % self.quiesce)
self.json_post("/cli-json/quiesce_group", {"name": self.quiesce}) self.json_post("/cli-json/quiesce_group", {"name": self.quiesce})
def check_if_complete(self):
completed = False completed = False
timer = 0 timer = 0
while not completed: while not completed:
current_events = self.json_get('/events/since=time/1h') current_events = self.json_get('/events/since=time/1h')
new_events = find_new_events(starting_events, current_events) new_events = find_new_events(self.starting_events, current_events)
target_events = [event for event in get_events(new_events, 'event description') if event.startswith('LOAD COMPLETED')] target_events = [event for event in get_events(new_events, 'event description') if
event.startswith('LOAD COMPLETED')]
if 'LOAD-DB: Load attempt has been completed.' in get_events(new_events, 'event description'): if 'LOAD-DB: Load attempt has been completed.' in get_events(new_events, 'event description'):
completed = True completed = True
print('Scenario %s fully loaded after %s seconds' % (self.scenario, timer)) print('Scenario %s fully loaded after %s seconds' % (self.scenario, timer))
@@ -153,16 +159,29 @@ def main():
parser.add_argument('--timeout', help='Stop trying to load scenario after this many seconds', default=120) parser.add_argument('--timeout', help='Stop trying to load scenario after this many seconds', default=120)
args = parser.parse_args() args = parser.parse_args()
LoadScenario(mgr=args.mgr, code = requests.get('http://%s:8080/events' % args.mgr)
scenario=args.load, scenario = LoadScenario(mgr=args.mgr,
action=args.action, scenario=args.load,
clean_dut=args.clean_dut, action=args.action,
clean_chambers=args.clean_chambers, clean_dut=args.clean_dut,
start=args.start, clean_chambers=args.clean_chambers,
stop=args.stop, start=args.start,
quiesce=args.quiesce, stop=args.stop,
timeout=args.timeout, quiesce=args.quiesce,
debug=args.debug) timeout=args.timeout,
code=code,
debug=args.debug)
if code == 200:
scenario.start_test()
scenario.load_scenario()
if code != 200:
print('sleeping 30 seconds, please upgrade your LANforge for a better experience, more information at https://www.candelatech.com/downloads.php#releases')
time.sleep(30)
if code == 200:
scenario.check_if_complete()
# scenario_loader.load_scenario() # scenario_loader.load_scenario()