updated test files add_syspath

common/tests/functional/test_jamming/add_syspath.py

@@ -1,14 +1,8 @@
 import os
 import sys
 
-# Get the path of the jamming_avoidance scripts current script
-current_dir = os.path.dirname(os.path.abspath(__file__))
+# Path to the jamming_detection directory
+module_path = '/opt/mesh_com/modules/sc-mesh-secure-deployment/src/2_0/features/jamming'
 
-# Construct the path to the parent director
-parent_dir = os.path.dirname(current_dir)
-
-# Construct the path to the jamming_detection directory
-jamming_detection_dir = os.path.join(parent_dir, "/opt/mesh_com/modules/sc-mesh-secure-deployment/src/2_0/features/jamming")
-
-# Add the jamming_detection directory to sys.path
-sys.path.append(jamming_detection_dir)
+# Append the path to the module directory to the system path
+sys.path.append(module_path)

common/tests/functional/test_jamming/test_channel_quality_est.py

@@ -2,6 +2,7 @@
 import pytest
 from unittest.mock import patch
 
+import add_syspath
 from channel_quality_est import ChannelQualityEstimator
 
 

common/tests/functional/test_jamming/test_jamming_client_fsm.py

@@ -8,6 +8,7 @@
 
 import pytest
 
+import add_syspath
 from jamming_client_fsm import JammingDetectionClient, ClientEvent, ClientState, ClientFSM
 from options import Options
 
@@ -192,7 +193,7 @@ def test_receive_message_unknown_action_id(self, mocker):
             receive_thread.join()
 
             # Assert that logger.error was called with the expected message
-            error_mock.assert_called_with("Error in received message: int is not allowed for map key when strict_map_key=True")
+            error_mock.assert_called_with('Error in received message: int is not allowed for map key')
 
         # Assert that no state transition occurred
         assert client.fsm.state == ClientState.IDLE

common/tests/functional/test_jamming/test_jamming_setup.py

@@ -4,7 +4,7 @@
 
 import add_syspath
 import util
-from jamming_setup import is_interface_up, switch_frequency, check_osf_interface, start_jamming_scripts, start_client, start_server, main
+from jamming_setup import is_interface_up, switch_frequency, check_client_osf_connectivity, check_osf_interface, start_jamming_scripts, start_client, start_server, validate_configuration, main
 from options import Options
 
 
@@ -25,19 +25,10 @@ def test_switch_frequency_success(self, mocker):
         mocker.patch('util.map_channel_to_freq', return_value=5180)
 
         # Mock the run_command function to not raise an exception
-        mocker.patch('util.run_command', side_effect=lambda command, error_message: (True, None))
-        mocker.patch('util.is_process_running', return_value=False)
-        mocker.patch('os.path.exists', return_value=False)
-        mocker.patch('util.read_file', return_value=True)
-        mocker.patch('util.write_file', return_value=True)
-        mocker.patch('util.kill_process_by_pid', return_value=True)
-        mocker.patch('util.get_pid_by_process_name', return_value=True)
         mocker.patch('util.switch_frequency', return_value=True)
 
-        # Create a mock for re.sub within the scope of switch_frequency
-        with mock.patch('re.sub', return_value='conf'):
-            # Run function under test, no exceptions should be returned
-            switch_frequency(args)
+        # Run function under test, no exceptions should be returned
+        switch_frequency(args)
 
     #  IPv6 address of tun0 interface is retrieved successfully
     def test_get_ipv6_address_success(self, mocker):
@@ -50,13 +41,13 @@ def test_get_ipv6_address_success(self, mocker):
         # Assert that the returned IPv6 address is correct
         assert ipv6_address == '2001:db8::1'
 
-    #  IPv6 connectivity with remote server is successful
-    def test_check_osf_interface_success(self, mocker):
+    #  IPv6 client connectivity with remote server is successful
+    def test_check_client_osf_connectivity_success(self, mocker):
         # Mock the subprocess.check_output function to not raise an exception
         mocker.patch('subprocess.check_output')
 
         # Call the function under test
-        check_osf_interface(Options())
+        check_client_osf_connectivity(Options())
 
         # Assert that the subprocess.check_output function was called with the correct command
         subprocess.check_output.assert_called_with(['ping6', '-c', '1', Options().jamming_osf_orchestrator], text=True)
@@ -77,57 +68,34 @@ def test_start_jamming_scripts_success(self, mocker):
         mock_run_command.assert_called_with(["python", "jamming_client_fsm.py"], 'Failed to run jamming_client_fsm file')
 
 
-    #  Mesh frequency cannot be switched to starting frequency
-    def test_switch_frequency_failure(self, mocker):
-        # Mock the get_mesh_freq function to return a different frequency than the starting frequency
-        mocker.patch('util.get_mesh_freq', return_value=5180)
-        mocker.patch('util.map_channel_to_freq', return_value=5200)
-
-        # Mock the run_command function to not raise an exception
-        mocker.patch('util.run_command', side_effect=lambda command, error_message: (True, None))
-        mocker.patch('util.is_process_running', return_value=False)
-        mocker.patch('os.path.exists', return_value=False)
-        mocker.patch('util.read_file', return_value=True)
-        mocker.patch('util.write_file', return_value=True)
-
-        # Create a mock for re.sub within the scope of switch_frequency and call switch_frequency function
-        with mock.patch('re.sub', return_value='conf'):
-            # Call the function under test and assert that it raises a ValueError
-            with pytest.raises(SystemExit):
-                switch_frequency(Options())
-
-    #  IPv6 address of tun0 interface is not retrieved successfully
-    def test_get_ipv6_address_failure(self, mocker):
-        # Mock the get_ipv6_addr function to return None
-        mocker.patch('util.get_ipv6_addr', return_value=None)
-
-        # Mock all other functions to set value error
-        mocker.patch('util.get_mesh_freq', return_value=True)
-        mocker.patch('jamming_setup.switch_frequency', return_value=True)
-        mocker.patch('util.map_channel_to_freq', return_value=True)
-        mocker.patch('jamming_setup.check_osf_interface', return_value=True)
-        mocker.patch('jamming_setup.start_jamming_scripts', return_value=True)
-
-        # Call the main function and assert that it raises a ValueError
-        with pytest.raises(ValueError):
-            main()
-
     #  IPv6 connectivity with remote server fails
-    def test_check_osf_interface_failure(self, mocker):
+    def test_check_client_osf_connectivity_failure(self, mocker):
         # Mock the subprocess.check_output function to raise a subprocess.CalledProcessError
         mocker.patch('subprocess.check_output', side_effect=subprocess.CalledProcessError(1, 'ping6'))
 
         # Call the function under test and assert that it raises a SystemExit
         with pytest.raises(SystemExit):
-            check_osf_interface(Options())
+            check_client_osf_connectivity(Options())
 
     #  is_interface_up function returns True if interface is up
     def test_is_interface_up_returns_true_if_interface_is_up(self, mocker):
         # Mock the netifaces.interfaces function to return a list of available interfaces
-        mocker.patch('netifaces.interfaces', return_value=['eth0', 'wlan0', 'tun0'])
+        mocker.patch('netifaces.interfaces', return_value=['eth0', 'wlp1s0', 'tun0'])
 
         # Call the function under test
-        result = is_interface_up('wlan0')
+        result = is_interface_up('wlp1s0')
 
         # Assert that the result is True
         assert result is True
+
+    #  Validate configuration fails when channels5 list includes invalid channels
+    def test_validate_configuration_fails_channels5_list_includes_invalid_channels(self, mocker):
+        args = Options()
+        args.channels5 = [36, 40, 44, 48, 149, 153, 157, 1600]
+        assert validate_configuration(args) == False
+
+    #  Validate configuration succeeds when channels5 list includes valid channels
+    def test_validate_configuration_succeeds_channels5_list_includes_valid_channels(self, mocker):
+        args = Options()
+        args.channels5 = [36, 40, 44, 48, 149, 153, 157, 161]
+        assert validate_configuration(args) == True

common/tests/functional/test_jamming/test_options.py

@@ -9,17 +9,6 @@ def test_options_object_created_successfully(self):
         options = Options()
         assert isinstance(options, Options)
 
-    #  Validate configuration fails when mesh interface channel width is not set to 20 MHz
-    def test_validate_configuration_fails_mesh_interface_channel_width_not_set_to_20MHz(self, mocker):
-        options = Options()
-        options.validate_configuration = mocker.Mock(return_value=False)
-        assert options.validate_configuration() == False
-
-    #  Validate configuration fails when channels5 list includes invalid channels
-    def test_validate_configuration_fails_channels5_list_includes_invalid_channels(self, mocker):
-        options = Options()
-        options.channels5 = [36, 40, 44, 48, 149, 153, 157, 162]
-        assert options.validate_configuration() == False
 
     #  Options object attributes are set correctly
     def test_options_object_attributes_set_correctly(self):
@@ -57,14 +46,3 @@ def test_options_object_attributes_set_correctly(self):
         assert options.periodic_target_freq_broadcast == 10
         assert options.spectrum_data_expiry_time == 5
 
-    #  Validate configuration succeeds when mesh interface channel width is set to 20 MHz
-    def test_validate_configuration_succeeds_mesh_interface_channel_width_set_to_20MHz(self, mocker):
-        options = Options()
-        options.validate_configuration = mocker.Mock(return_value=True)
-        assert options.validate_configuration() == True
-
-    #  Validate configuration succeeds when channels5 list includes valid channels
-    def test_validate_configuration_succeeds_channels5_list_includes_valid_channels(self, mocker):
-        options = Options()
-        options.channels5 = [36, 40, 44, 48, 149, 153, 157, 161]
-        assert options.validate_configuration() == True

common/tests/functional/test_jamming/test_preprocessor.py

@@ -67,10 +67,10 @@ def test_resize_with_valid_input(self):
     def test_normalize_with_valid_input(self):
         # Set expected data
         expected_data = {
-            'freq1': [5180], 'max_magnitude': [-3.00653], 'rssi': [-0.237583], 'relpwr_db': [4.563367],
-            'base_pwr_db': [-30.556816], 'avgpwr_db': [-9.808239], 'total_gain_db': [-4.744545], 'snr': [-1.930179],
-            'cnr': [-9.305355], 'pn': [-10.448828], 'ssi': [-2.702977], 'pd': [-26.652357], 'sinr': [-2.702977],
-            'sir': [3.455223], 'mr': [-2.898512], 'pr': [239.60759]
+            'freq1': [5180], 'max_magnitude': [-2.699096], 'rssi': [0.386689], 'relpwr_db': [5.723355],
+            'base_pwr_db': [-54.127222], 'avgpwr_db': [-7.553099], 'total_gain_db': [-8.580278], 'snr': [-4.467233],
+            'cnr': [-19.34198], 'pn': [-22.509996], 'ssi': [-4.161893], 'pd': [-33.130156], 'sinr': [-4.161893],
+            'sir': [7.230196], 'mr': [-2.481305], 'pr': [187.510565]
         }
         expected_result = pd.DataFrame(expected_data)