scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move event_breaker and event_flood to scripts_deprecated

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-10-14 12:47:09 -07:00
parent 7915ea197f
commit 471b988b88
4 changed files with 149 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
py-json/port_probe.py Normal file
View File

@@ -0,0 +1,43 @@
#!/usr/bin/env python3
from LANforge.lfcli_base import LFCliBase
class ProbePort(LFCliBase):
def __init__(self,
lfhost=None,
lfport='8080',
debug=False,
eid_str=None):
super().__init__(_lfjson_host=lfhost,
_lfjson_port=lfport,
_debug=debug)
hunks = eid_str.split(".")
self.eid_str = eid_str
self.probepath = "/probe/1/%s/%s" % (hunks[-2], hunks[-1])
self.response = None
self.signals = None
def refreshProbe(self):
self.json_post(self.probepath, {})
response = self.json_get(self.probepath)
self.response = response
text = self.response['probe-results'][0][self.eid_str]['probe results'].split('\n')
signals = [x.strip('\t').split('\t') for x in text if 'signal' in x]
keys = [x[0].strip(' ').strip(':') for x in signals]
values = [x[1].strip('dBm').strip(' ') for x in signals]
self.signals = dict(zip(keys, values))
def getSignalAvgCombined(self):
return self.signals['signal avg'].split(' ')[0]
def getSignalAvgPerChain(self):
return ' '.join(self.signals['signal avg'].split(' ')[1:])
def getSignalCombined(self):
return self.signals['signal'].split(' ')[0]
def getSignalPerChain(self):
return ' '.join(self.signals['signal'].split(' ')[1:])
def getBeaconSignalAvg(self):
return ' '.join(self.signals['beacon signal avg'])

46
py-scripts/scripts_deprecated/event_break_flood.py Executable file
View File

@@ -0,0 +1,46 @@
#!/usr/bin/python3
import threading
from event_flood import EventFlood
from event_breaker import EventBreaker
import importlib
import sys
import os
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
def event_break(args):
event_breaker = EventBreaker(host=args.mgr,
port=args.mgr_port,
duration='30s')
event_breaker.create()
event_breaker.run()
event_breaker.cleanup()
def event_flooding(args):
event_flood = EventFlood(host=args.mgr,
port=args.mgr_port,
duration='30s',
pause_ms=30)
event_flood.create()
event_flood.run()
event_flood.cleanup()
def main():
parser = LFCliBase.create_bare_argparse()
args = parser.parse_args()
flood = threading.Thread(target=event_flooding(args))
breaker = threading.Thread(target=event_break(args))
flood.start()
breaker.start()
if __name__ == "__main__":
main()

32
py-scripts/event_breaker.py → py-scripts/scripts_deprecated/event_breaker.py
View File

@@ -9,14 +9,12 @@ import os
import importlib
from datetime import datetime
import pprint
import argparse
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
@@ -33,7 +31,7 @@ class EventBreaker(Realm):
super().__init__(host, port)
self.counter = 0
self.test_duration = duration
if (self.test_duration is None):
if self.test_duration is None:
raise ValueError("run wants numeric run_duration_sec")
def create(self):
@@ -42,25 +40,18 @@ class EventBreaker(Realm):
def run(self):
now = datetime.now()
now_ms = 0
end_time = self.parse_time(self.test_duration) + now
client_time_ms = 0
prev_client_time_ms = 0
start_loop_time_ms = 0
loop_time_ms = 0
prev_loop_time_ms = 0
num_events = 0
prev_num_events = 0
bad_events = []
while datetime.now() < end_time:
bad_events = []
start_loop_time_ms = int(self.get_milliseconds(datetime.now()))
print('\r♦ ', end='')
#prev_loop_time_ms = loop_time_ms
# loop_time_ms = self.get_milliseconds(datetime.now())
prev_client_time_ms = client_time_ms
response = self.json_get("/events/all")
#pprint.pprint(response)
if self.debug:
pprint.pprint(response)
if "events" not in response:
pprint.pprint(response)
@@ -72,7 +63,8 @@ class EventBreaker(Realm):
print("%s events Δ%s" % (num_events, (num_events - prev_num_events)))
if "candela.lanforge.HttpEvents" in response:
client_time_ms = float(response["candela.lanforge.HttpEvents"]["duration"])
# print(" client_time %d"%client_time_ms)
if self.debug:
print(" client_time %d" % client_time_ms)
if abs(prev_client_time_ms - client_time_ms) > 30:
print(" client time %d ms Δ%d" % (client_time_ms, (prev_client_time_ms - client_time_ms)),
@@ -84,14 +76,16 @@ class EventBreaker(Realm):
if record[k] is None:
print(' ☠no %s☠' % k, end='')
continue
# pprint.pprint( record[k])
if self.debug:
pprint.pprint(record[k])
if "NA" == record[k]["event"] \
or "NA" == record[k]["name"] \
or "NA" == record[k]["type"] \
or "NA" == record[k]["priority"]:
bad_events.append(int(k))
pprint.pprint(record[k])
# print( " ☠id[%s]☠"%k, end='')
if self.debug:
print(" ☠id[%s]☠" % k, end='')
if len(bad_events) > 0:
pprint.pprint(events[event_index])
print(" ☠id[%s]☠" % bad_events, end='')
@@ -110,13 +104,14 @@ class EventBreaker(Realm):
def cleanup(self):
pass
def main():
parser = LFCliBase.create_bare_argparse(
prog='event_breaker.py',
formatter_class=argparse.RawTextHelpFormatter)
description="""event breaker is meant to be used in conjunction with event flood
Please use the event_break_flood.py script""")
parser.add_argument("--test_duration", help='test duration', default="30s")
# if optional_args is not None:
args = parser.parse_args()
event_breaker = EventBreaker(host=args.mgr,
@@ -129,5 +124,6 @@ def main():
event_breaker.run()
event_breaker.cleanup()
if __name__ == "__main__":
main()

36
py-scripts/event_flood.py → py-scripts/scripts_deprecated/event_flood.py
View File

@@ -7,16 +7,15 @@ Please concurrently use with event_breaker.py.
import sys
import os
import importlib
import argparse
from datetime import datetime
from time import sleep
import pprint
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
@@ -24,7 +23,7 @@ realm = importlib.import_module("py-json.realm")
Realm = realm.Realm
class EventBreaker(Realm):
class EventFlood(Realm):
def __init__(self, host, port,
duration=None,
pause_ms=None,
@@ -35,7 +34,7 @@ class EventBreaker(Realm):
self.counter = 0
self.test_duration = duration
self.pause_ms = pause_ms
if (self.test_duration is None):
if self.test_duration is None:
raise ValueError("run wants numeric run_duration_sec")
def create(self):
@@ -44,25 +43,16 @@ class EventBreaker(Realm):
def run(self):
last_second_ms = 0
start_time = datetime.now()
now_ms = 0
end_time = self.parse_time(self.test_duration) + start_time
client_time_ms = 0
prev_client_time_ms = 0
start_loop_time_ms = 0
loop_time_ms = 0
prev_loop_time_ms = 0
num_events = 0
prev_num_events = 0
while datetime.now() < end_time:
sleep(self.pause_ms / 1000)
start_loop_time_ms = int(self.get_milliseconds(datetime.now()))
print('\r♦ ', end='')
#prev_loop_time_ms = loop_time_ms
# loop_time_ms = self.get_milliseconds(datetime.now())
prev_client_time_ms = client_time_ms
response_list = []
response = self.json_post("/cli-json/add_event",
self.json_post("/cli-json/add_event",
{
"event_id": "new",
"details": "event_flood %d" % start_loop_time_ms,
@@ -70,38 +60,41 @@ class EventBreaker(Realm):
"name": "custom"
},
response_json_list_=response_list)
# pprint.pprint(response_list)
if self.debug:
pprint.pprint(response_list)
prev_client_time_ms = client_time_ms
prev_loop_time_ms = loop_time_ms
now = int(self.get_milliseconds(datetime.now()))
loop_time_ms = now - start_loop_time_ms
client_time_ms = response_list[0]["LAST"]["duration"]
if (client_time_ms != prev_client_time_ms):
if client_time_ms != prev_client_time_ms:
print(" client %d ms %d" % (client_time_ms,
(prev_client_time_ms - client_time_ms)),
end='')
if (loop_time_ms != prev_loop_time_ms):
if loop_time_ms != prev_loop_time_ms:
print(" loop %d ms %d " % (loop_time_ms,
(prev_loop_time_ms - loop_time_ms)),
end='')
if (last_second_ms + 1000) < now:
last_second_ms = now
print("")
def cleanup(self):
pass
def main():
parser = LFCliBase.create_bare_argparse(
prog='event_breaker.py',
formatter_class=argparse.RawTextHelpFormatter)
description="""event flood is meant to be used in conjunction with event breaker
Please use the event_break_flood.py script""")
parser.add_argument("--test_duration", help='test duration', default="30s")
parser.add_argument("--pause_ms", help='interval between submitting events', default="30")
# if optional_args is not None:
args = parser.parse_args()
event_breaker = EventBreaker(host=args.mgr,
event_breaker = EventFlood(host=args.mgr,
port=args.mgr_port,
duration=args.test_duration,
pause_ms=int(args.pause_ms),
@@ -112,5 +105,6 @@ def main():
event_breaker.run()
event_breaker.cleanup()
if __name__ == "__main__":
main()