Added websocket generic listener for port 8081, added a test script for it and updated test_ipv4_connection.py to add websocket events

py-json/ws_generic_monitor.py

@@ -0,0 +1,25 @@
+"""
+pip install websocket_client
+
+https://pypi.org/project/websocket_client/
+
+WS_Listener has three arguments in general : lfclient_host, _scriptname, _callback
+1.  Enter the LF Client Host address on which you want to monitor events (by Default is localhost)
+2.  Enter the _scriptname that should be present in the event triggered by your script,
+    refer add_event() in lfcli_base.add_event()
+    _scriptname can be any string that you want to monitor in your websocket message
+3.  Enter the Callback function that you wanna see your messages in everytime your event will trigger up.
+    refer py-scripts/ws_generic_monitor_test.py to see an example
+
+"""
+
+class WS_Listener():
+    def __init__(self, lfclient_host="localhost", _scriptname=None, _callback=None):
+        import websocket
+        self.scriptname = _scriptname
+        websocket.enableTrace(True)
+        self.ws = websocket.WebSocketApp("ws://"+lfclient_host+":8081", on_message=_callback)
+        self.ws.run_forever()
+
+
+

py-scripts/test_ipv4_connection.py

@@ -142,6 +142,8 @@ python3 ./test_ipv4_connection.py --upstream_port eth1 \\
         num_stations_converted = int(args.num_stations)
         num_sta = num_stations_converted
 
+    ws_event = LFCliBase("192.168.200.15", 8080)
+    
     station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=num_sta-1, padding_number_=10000)
     ip_test = IPv4Test(lfjson_host, lfjson_port, ssid=args.ssid, password=args.passwd,
                        security=args.security, sta_list=station_list, radio=args.radio)
@@ -150,17 +152,19 @@ python3 ./test_ipv4_connection.py --upstream_port eth1 \\
     ip_test.build()
     if not ip_test.passes():
         print(ip_test.get_fail_message())
+        ws_event.add_event(name="test_ipv4_connection.py", message=ip_test.get_fail_message())
         exit(1)
     ip_test.start(station_list, False, False)
     ip_test.stop()
     if not ip_test.passes():
         print(ip_test.get_fail_message())
+        ws_event.add_event(name="test_ipv4_connection.py", message=ip_test.get_fail_message())
         exit(1)
     time.sleep(30)
     ip_test.cleanup(station_list)
     if ip_test.passes():
         print("Full test passed, all stations associated and got IP")
-
+        ws_event.add_event(name="test_ipv4_connection.py", message="Full test passed, all stations associated and got IP")
 
 if __name__ == "__main__":
     main()

py-scripts/ws_generic_monitor_test.py

@@ -0,0 +1,28 @@
+"""
+This example is to demonstrate ws_generic_monitor to monitor events triggered by scripts,
+This script when running, will monitor the events triggered by test_ipv4_connection.py
+
+"""
+
+
+import sys
+import json
+if 'py-json' not in sys.path:
+    sys.path.append('../py-json')
+from ws_generic_monitor import WS_Listener
+
+reference = "test_ipv4_connection.py"
+def main():
+    WS_Listener(lfclient_host="192.168.200.15", _scriptname=reference, _callback=TestRun)
+
+
+def TestRun(ws, message):
+    if (str(message).__contains__(reference)):
+        #print(message)
+        temp = json.loads(message)
+        event_message = str(temp['name']) + "/" + str(temp['details']) + "/" + str(temp['timestamp'])
+        print(event_message)
+
+if __name__ == "__main__":
+    main()
+