From 0e5106a38bc4de79f6870e362342080e3728f28f Mon Sep 17 00:00:00 2001 From: Amin Ben Romdhane Date: Fri, 16 Feb 2024 16:23:16 +0100 Subject: [PATCH] Added validate bbfdm.AddObj & bbfdm.DelObj events --- gitlab-ci/functional-test.sh | 5 +- .../python-test-cases/json/add_del_event.json | 30 ++++++ .../python/validate_add_del_event.py | 100 ++++++++++++++++++ 3 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 test/python-test-cases/json/add_del_event.json create mode 100755 test/python-test-cases/python/validate_add_del_event.py diff --git a/gitlab-ci/functional-test.sh b/gitlab-ci/functional-test.sh index 5ef12563..bd54a519 100755 --- a/gitlab-ci/functional-test.sh +++ b/gitlab-ci/functional-test.sh @@ -18,9 +18,12 @@ 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 'bbfdm.event' event callback ./test/python-test-cases/python/validate_event_callback.py test/python-test-cases/json/event_callback.json +# Test 'bbfdm.AddObj' & 'bbfdm.DelObj' event +./test/python-test-cases/python/validate_add_del_event.py test/python-test-cases/json/add_del_event.json + supervisorctl stop all supervisorctl status diff --git a/test/python-test-cases/json/add_del_event.json b/test/python-test-cases/json/add_del_event.json new file mode 100644 index 00000000..edf74f3d --- /dev/null +++ b/test/python-test-cases/json/add_del_event.json @@ -0,0 +1,30 @@ +{ + "events": [ + { + "description": "Validate & event using internal object defined in the core Data Model: Device.DeviceInfo.FirmwareImage.", + "config_path": "/tmp/fwbank.data", + "objects_to_look": [ + "Device.DeviceInfo.FirmwareImage.1", + "Device.DeviceInfo.FirmwareImage.2" + ] + }, + { + "description": "Validate & event using external object JSON Plugin: Device.X_IOPSYS_EU_Dropbear.", + "config_path": "/etc/config/dropbear", + "objects_to_look": [ + "Device.X_IOPSYS_EU_Dropbear.1", + "Device.SSH.Server.1" + ] + }, + { + "description": "Validate & event using external object JSON Plugin: Device.X_IOPSYS_EU_IGMP.", + "config_path": "/etc/config/mcast", + "objects_to_look": [ + "Device.X_IOPSYS_EU_IGMP.Proxy.1", + "Device.X_IOPSYS_EU_IGMP.Proxy.1.Interface.1", + "Device.X_IOPSYS_EU_IGMP.Proxy.1.Interface.2", + "Device.X_IOPSYS_EU_IGMP.Proxy.1.Filter.1" + ] + } + ] +} \ No newline at end of file diff --git a/test/python-test-cases/python/validate_add_del_event.py b/test/python-test-cases/python/validate_add_del_event.py new file mode 100755 index 00000000..be98753f --- /dev/null +++ b/test/python-test-cases/python/validate_add_del_event.py @@ -0,0 +1,100 @@ +#!/usr/bin/python3 + +import os +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, config_path, objects_to_look): + self.description = description + self.config_path = config_path + self.objects_to_look = objects_to_look + +def callback(event, data): + """ + Callback function to handle UBUS events. + """ + global UBUS_DISCONNECT + # Check if the objects to look for are in the instances list + if set(args.objects_to_look).issubset(set(data['instances'])): + # Mark UBUS as disconnected + UBUS_DISCONNECT = 1 + # Disconnect from UBUS + ubus.disconnect() + +def run_global_test(args, event_name, path_1, path_2): + """ + Run a global test. + """ + global UBUS_DISCONNECT + + # Connect to UBUS + ubus.connect() + UBUS_DISCONNECT = 0 + + # Listen for UBUS events + ubus.listen((event_name, callback)) + + # Force change in schema by moving dependency file + os.rename(path_1, path_2) + + # Run UBUS loop with a timeout + ubus.loop(timeout=35000) + + # If UBUS is still connected, disconnect and mark test as failed + if UBUS_DISCONNECT == 0: + ubus.disconnect() + return -1 + + return 0 + +def run_test(args): + """ + Run a single test. + """ + print("Running: " + args.description) + + # Test deletion of object + res = run_global_test(args, "bbfdm.DelObj", args.config_path, "/tmp/bbfdm_test_1") + if res != 0: + print("FAIL: " + args.description) + return + + # Test addition of object + res = run_global_test(args, "bbfdm.AddObj", "/tmp/bbfdm_test_1", args.config_path) + if res != 0: + print("FAIL: " + args.description) + return + + print("PASS: " + args.description) + +if __name__ == "__main__": + # Check for correct command line arguments + if len(sys.argv) != 2: + print("Usage: {} ".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)