jbr_monitor_bssids.py: fixes loading logic

- querying for load was happening before load was requested, which was clearly dumb
- incorporates show_events before load, and after every 5 seconds
- sys.settrace() call is experimental, is going to be removed

Signed-off-by: Jed Reynolds <jed@bitratchet.com>
This commit is contained in:
Jed Reynolds
2021-10-28 21:05:56 -07:00
parent d8b474139a
commit c71e93ed75

View File

@@ -47,7 +47,7 @@ import argparse
import time import time
from http.client import HTTPResponse from http.client import HTTPResponse
from typing import Optional from typing import Optional
from pprint import pprint, pformat from pprint import pprint
path_hunks = os.path.abspath(__file__).split('/') path_hunks = os.path.abspath(__file__).split('/')
while( path_hunks[-1] != 'lanforge-scripts'): while( path_hunks[-1] != 'lanforge-scripts'):
@@ -155,21 +155,20 @@ class BssidMonitor(Realm):
self.lf_query : LFJsonQuery = self.lf_session.get_query() self.lf_query : LFJsonQuery = self.lf_session.get_query()
def build(self): def build(self):
err_warn_list = []
# query for the last response # get last event id
event_response = self.lf_query.events_last_events(event_count=1, last_event_id = self.before_load_action()
debug=self.debug,
errors_warnings=err_warn_list)
last_event_id = event_response["id"]
if not self.wait_for_load_to_finish(since_id=last_event_id):
exit(1)
# load a database # load a database
sys.settrace(lanforge_api.trace)
response: HTTPResponse = self.lf_command.post_load(name="BLANK", response: HTTPResponse = self.lf_command.post_load(name="BLANK",
action="overwrite", action="overwrite",
clean_dut="NA", clean_dut="NA",
clean_chambers="NA", clean_chambers="NA",
debug=self.debug) debug=self.debug)
sys.settrace(None)
if not self.wait_for_load_to_finish(since_id=last_event_id):
exit(1)
if not response: if not response:
raise ConnectionError("lf_command::post_load returned no response") raise ConnectionError("lf_command::post_load returned no response")
@@ -179,7 +178,29 @@ class BssidMonitor(Realm):
for bssid in self.bssid_list: for bssid in self.bssid_list:
print("build: bssid: %s" % bssid) print("build: bssid: %s" % bssid)
def before_load_action(self):
"""
Use this
:return: last event ID in event list
"""
err_warn_list = []
self.lf_command.post_show_events(p_type='all',
shelf=1,
card='all',
port='all',
endp='all')
time.sleep(0.1)
event_response = self.lf_query.events_last_events(event_count=1,
debug=self.debug,
errors_warnings=err_warn_list)
return event_response["id"]
def wait_for_load_to_finish(self, since_id:int=None): def wait_for_load_to_finish(self, since_id:int=None):
"""
TODO: make this a standard method outside this module
:param since_id:
:return:
"""
completed = False completed = False
timer = 0 timer = 0
timeout = 60 timeout = 60
@@ -189,7 +210,12 @@ class BssidMonitor(Realm):
for event_tup in new_events: for event_tup in new_events:
for event_id in event_tup.keys(): for event_id in event_tup.keys():
event_record = event_tup[event_id] event_record = event_tup[event_id]
# pprint(event_record) if self.debug:
pprint("\n wait_for_load_to_finish: {since} -> {id}: {descr}\n".format(
since=since_id,
id=event_id,
descr=event_record['event description']
))
if event_record['event description'].startswith('LOAD COMPLETED'): if event_record['event description'].startswith('LOAD COMPLETED'):
completed = True completed = True
self.lf_query.logger.warning('Scenario loaded after %s seconds' % timer) self.lf_query.logger.warning('Scenario loaded after %s seconds' % timer)
@@ -197,6 +223,12 @@ class BssidMonitor(Realm):
if completed: if completed:
break break
else: else:
if (timer % 5) == 0:
self.lf_command.post_show_events(p_type='all',
shelf=1,
card='all',
port='all',
endp='all')
timer += 1 timer += 1
time.sleep(1) time.sleep(1)
if timer > timeout: if timer > timeout: