Added event test

gitlab-ci/functional-test.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+echo "$0 preparation script"
+pwd
+
+source ./gitlab-ci/shared.sh
+
+echo "Starting services..."
+cp ./gitlab-ci/bbfdm_services.conf /etc/supervisor/conf.d/
+
+supervisorctl reread
+supervisorctl update
+sleep 10
+
+supervisorctl status all
+exec_cmd ubus wait_for bbfdm
+
+# Test ubus bbfdm 'notify_event' method
+./test/python-test-cases/python/validate_ubus_notify_event_method.py test/python-test-cases/json/ubus_notify_event_method.json
+
+# Test ubus 'bbfdm.event' event
+./test/python-test-cases/python/validate_event_callback.py test/python-test-cases/json/event_callback.json
+
+supervisorctl stop all
+supervisorctl status
+
+echo "Functional Test :: PASS"

test/files/tmp/bbfdm/input.json

@@ -1,6 +1,7 @@
 {
 	"daemon": {
 		"config": {
+		    "loglevel": 2
 		},
 		"input": {
 			"type": "DotSo",

test/python-test-cases/json/event_callback.json

@@ -0,0 +1,70 @@
+{
+    "events": [
+        {
+            "description": "Validate <bbfdm.event> event using internal event defined in the core Data Model: Device.WiFi.DataElements.AssociationEvent.Associated!",
+            "event_name": "wifi.dataelements.Associated",
+            "inputs": {
+                "eventTime": "2020-10-24T01:09:34+01:00",
+                "wfa-dataelements:AssociationEvent": {
+                    "AssocData": {
+                        "BSSID": "00:22:07:6b:17:52",
+                        "MACAddress": "54:b8:02:80:25:e8",
+                        "StatusCode": 12,
+                        "HTCapabilities": "AA==\n",
+                        "VHTCapabilities": "AAAAAAAA\n",
+                        "HECapabilities": "AA==\n"
+                    }
+                }
+            },
+            "output": {
+                "name": "Device.WiFi.DataElements.AssociationEvent.Associated!",
+                "input": {
+                    "TimeStamp": "2020-10-24T01:09:34+01:00",
+                    "BSSID": "00:22:07:6b:17:52",
+                    "MACAddress": "54:b8:02:80:25:e8",
+                    "StatusCode": "12",
+                    "HTCapabilities": "AA==\n",
+                    "VHTCapabilities": "AAAAAAAA\n",
+                    "HECapabilities": "AA==\n"
+                }
+            }
+        },
+        {
+            "description": "Validate <bbfdm.event> event using internal event defined in the core Data Model: Device.WiFi.DataElements.DisassociationEvent.Disassociated!",
+            "event_name": "wifi.dataelements.Disassociated",
+            "inputs": {
+                "eventTime": "2020-10-24T01:10:20+01:00",
+                "wfa-dataelements:DisassociationEvent": {
+                    "DisassocData": {
+                        "BSSID": "00:22:07:6b:17:51",
+                        "MACAddress": "d2:a2:d1:17:27:f7",
+                        "ReasonCode": 12,
+                        "BytesSent": 181625,
+                        "BytesReceived": 11497,
+                        "PacketsSent": 1951,
+                        "PacketsReceived": 105,
+                        "ErrorsSent": 107,
+                        "ErrorsReceived": 78,
+                        "RetransCount": 9999
+                    }
+                }
+            },
+            "output": {
+                "name": "Device.WiFi.DataElements.DisassociationEvent.Disassociated!",
+                "input": {
+                    "TimeStamp": "2020-10-24T01:10:20+01:00",
+                    "BSSID": "00:22:07:6b:17:51",
+                    "MACAddress": "d2:a2:d1:17:27:f7",
+                    "ReasonCode": "12",
+                    "BytesSent": "181625",
+                    "BytesReceived": "11497",
+                    "PacketsSent": "1951",
+                    "PacketsReceived": "105",
+                    "ErrorsSent": "107",
+                    "ErrorsReceived": "78",
+                    "RetransCount": "9999"
+                }
+            }
+        }
+    ]
+}

test/python-test-cases/json/ubus_notify_event_method.json

@@ -0,0 +1,23 @@
+{
+    "events": [
+        {
+            "description": "Validate ubus bbfdm <notify_event> method",
+            "service": "bbfdm",
+            "method": "notify_event",
+            "input": {
+                "name": "Device.LocalAgent.TransferComplete!",
+                "input": {
+                    "param1": "val1",
+                    "param2": "val2"
+                }
+            },
+            "output": {
+                "name": "Device.LocalAgent.TransferComplete!",
+                "input": {
+                    "param1": "val1",
+                    "param2": "val2"
+                }
+            }
+        }
+    ]
+}

test/python-test-cases/python/validate_event_callback.py

@@ -0,0 +1,80 @@
+#!/usr/bin/python3
+
+import sys
+import ubus
+import json
+
+# Global variable to track UBUS disconnection
+UBUS_DISCONNECT = 0
+
+class TestArguments:
+    """
+    Class to hold test arguments.
+    """
+    def __init__(self, description, event_name, inputs, output):
+        self.description = description
+        self.event_name = event_name
+        self.inputs = inputs
+        self.output = output
+
+def callback(event, data):
+    """
+    Callback function to handle UBUS events.
+    """
+    global UBUS_DISCONNECT
+    if data == args.output:
+        print("PASS: " + args.description)
+    else:
+        print("FAIL: " + args.description)
+    ubus.disconnect()
+    UBUS_DISCONNECT = 1
+
+def run_test(args):
+    """
+    Run a single test.
+    """
+    global UBUS_DISCONNECT
+
+    print("Running: " + args.description)
+
+    # Connect to UBUS
+    ubus.connect()
+    UBUS_DISCONNECT = 0
+
+    # Listen for UBUS events
+    ubus.listen(("bbfdm.event", callback))
+
+    # Send UBUS event with inputs
+    ubus.send(args.event_name, args.inputs)
+
+    # Run UBUS loop with a timeout
+    ubus.loop(timeout=5000)
+    
+    # If UBUS is still connected, disconnect and mark test as failed
+    if UBUS_DISCONNECT == 0:
+        ubus.disconnect()
+        print("FAIL: " + args.description)
+
+if __name__ == "__main__":
+    # Check for correct command line arguments
+    if len(sys.argv) != 2:
+        print("Usage: {} <test_arguments_json>".format(sys.argv[0]))
+        sys.exit(1)
+
+    test_arguments_file = sys.argv[1]
+
+    try:
+        # Load test arguments from JSON file
+        with open(test_arguments_file, 'r') as f:
+            test_arguments_data = json.load(f)
+            args_list = [TestArguments(**event) for event in test_arguments_data["events"]]
+    except FileNotFoundError:
+        print("File not found:", test_arguments_file)
+        sys.exit(1)
+    except json.JSONDecodeError as e:
+        print("Error parsing JSON:", e)
+        sys.exit(1)
+
+    # Run tests for each set of arguments
+    for args in args_list:
+        run_test(args)

test/python-test-cases/python/validate_ubus_notify_event_method.py

@@ -0,0 +1,69 @@
+#!/usr/bin/python3
+
+import sys
+import ubus
+import json
+
+class TestArguments:
+    """
+    Class to hold test arguments.
+    """
+    def __init__(self, description, service, method, input, output):
+        self.description = description
+        self.service = service
+        self.method = method
+        self.input = input
+        self.output = output
+
+def callback(event, data):
+    """
+    Callback function to handle UBUS events.
+    """
+    if data == args.output:
+        print("PASS: " + args.description)
+    else:
+        print("FAIL: " + args.description)
+    ubus.disconnect()
+    exit(0)
+
+def run_test(args):
+    """
+    Run a single test.
+    """
+
+    print("Running: " + args.description)
+
+    ubus.connect()
+
+    ubus.listen(("bbfdm.event", callback))
+
+    ubus.call(args.service, args.method, args.input)
+
+    # Run UBUS loop with a timeout
+    ubus.loop(timeout=5000)
+
+    ubus.disconnect()
+    print("FAIL: " + args.description)
+
+if __name__ == "__main__":
+    # Check for correct command line arguments
+    if len(sys.argv) != 2:
+        print("Usage: {} <test_arguments_json>".format(sys.argv[0]))
+        sys.exit(1)
+
+    test_arguments_file = sys.argv[1]
+
+    try:
+        with open(test_arguments_file, 'r') as f:
+            test_arguments_data = json.load(f)
+            args_list = [TestArguments(**event) for event in test_arguments_data["events"]]
+    except FileNotFoundError:
+        print("File not found:", test_arguments_file)
+        sys.exit(1)
+    except json.JSONDecodeError as e:
+        print("Error parsing JSON:", e)
+        sys.exit(1)
+
+    # Run tests for each set of arguments
+    for args in args_list:
+        run_test(args)

test/python/validate_schema_notify.py

@@ -12,7 +12,7 @@ child = pexpect.spawn('ubus monitor')
 os.rename("/etc/config/dropbear", "/etc/config/dropbear_1")
 
 try:
-    ret = child.expect('notify', timeout=65)
+    ret = child.expect('notify', timeout=40)
 except:
     print("FAIL: Schema updater notification")
 
@@ -26,6 +26,18 @@ if ret == 0:
 os.rename("/etc/config/dropbear_1", "/etc/config/dropbear")
 
 if ret == 0:
+    try:
+        ret = child.expect('notify', timeout=40)
+    except:
+        print("FAIL: Schema updater notification")
+        
+if ret == 0:
+    try:
+        ret = child.expect('bbfdm.AddObj')
+    except:
+        print("FAIL: Schema updater notification")
+
+if ret == 0:        
     print("PASS: Schema updater notification")
 
 exit(ret)