diff --git a/src/nidcpower/system_tests/grpc_server_config.json b/src/nidcpower/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nidcpower/system_tests/grpc_server_config.json rename to src/nidcpower/system_tests/grpc_server_config_no_tls.json diff --git a/src/nidcpower/system_tests/grpc_server_config_tls.json b/src/nidcpower/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..b3a341dabc --- /dev/null +++ b/src/nidcpower/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31760, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nidcpower/system_tests/test_system_nidcpower.py b/src/nidcpower/system_tests/test_system_nidcpower.py index 4b6ce4c61b..8a5cd94a4e 100644 --- a/src/nidcpower/system_tests/test_system_nidcpower.py +++ b/src/nidcpower/system_tests/test_system_nidcpower.py @@ -5,6 +5,7 @@ import grpc import hightime +import nitlsconfig import pytest import nidcpower @@ -37,7 +38,9 @@ def pytest_generate_tests(metafunc): metafunc.parametrize('session', [True], indirect=True) -class SystemTests: +# Defines a subset of system tests to validate basic NI-DCPower functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def session(self, request, session_creation_kwargs): """Creates an NI-DCPower Session. @@ -78,6 +81,50 @@ def session(self, request, session_creation_kwargs): def test_self_test(self, session): session.self_test() + @pytest.mark.channels('0') + def test_measure(self, session): + session.source_mode = nidcpower.SourceMode.SINGLE_POINT + session.output_function = nidcpower.OutputFunction.DC_VOLTAGE + session.voltage_level_range = 6 + session.voltage_level = 2 + with session.initiate(): + reading = session.measure(nidcpower.MeasurementTypes.VOLTAGE) + assert session.query_in_compliance() is False + assert reading == 2 + + @pytest.mark.channels('0') + def test_fetch_multiple(self, session): + session.source_mode = nidcpower.SourceMode.SINGLE_POINT + session.configure_aperture_time(0, nidcpower.ApertureTimeUnits.SECONDS) + session.voltage_level = 1 + count = 10 + session.measure_when = nidcpower.MeasureWhen.AUTOMATICALLY_AFTER_SOURCE_COMPLETE + with session.initiate(): + measurements = session.fetch_multiple(count) + assert len(measurements) == count + assert isinstance(measurements[1].voltage, float) + assert isinstance(measurements[1].current, float) + assert measurements[1].in_compliance in [True, False] + assert measurements[1].voltage == 1.0 + assert measurements[1].current == 0.00001 + + def test_measure_multiple(self, session): + with session.initiate(): + # session is open to all 12 channels on the device + measurements = session.measure_multiple() + assert len(measurements) == 12 + assert measurements[1].in_compliance is None + assert measurements[1].voltage == 0.0 + assert measurements[1].current == 0.00001 + # now a subset of the channels + measurements = session.channels[range(4)].measure_multiple() + assert len(measurements) == 4 + assert measurements[1].in_compliance is None + assert measurements[1].voltage == 0.0 + assert measurements[1].current == 0.00001 + + +class SystemTests(BasicValidationTests): # Workaround for driver runtime bug. See issue #1798 for details. @pytest.mark.legacy_session_only def test_self_cal(self, session): @@ -161,17 +208,6 @@ def test_disable(self, session): session.disable() assert channel.output_enabled is False - @pytest.mark.channels('0') - def test_measure(self, session): - session.source_mode = nidcpower.SourceMode.SINGLE_POINT - session.output_function = nidcpower.OutputFunction.DC_VOLTAGE - session.voltage_level_range = 6 - session.voltage_level = 2 - with session.initiate(): - reading = session.measure(nidcpower.MeasurementTypes.VOLTAGE) - assert session.query_in_compliance() is False - assert reading == 2 - @pytest.mark.channels('0') def test_query_output_state(self, session): with session.initiate(): @@ -192,37 +228,6 @@ def test_config_aperture_time(self, session): aperture_time_in_range = abs(aperture_time - expected_aperture_time) <= max(1e-09 * max(abs(aperture_time), abs(expected_aperture_time)), 0.0) # https://stackoverflow.com/questions/5595425/what-is-the-best-way-to-compare-floats-for-almost-equality-in-python assert aperture_time_in_range is True - @pytest.mark.channels('0') - def test_fetch_multiple(self, session): - session.source_mode = nidcpower.SourceMode.SINGLE_POINT - session.configure_aperture_time(0, nidcpower.ApertureTimeUnits.SECONDS) - session.voltage_level = 1 - count = 10 - session.measure_when = nidcpower.MeasureWhen.AUTOMATICALLY_AFTER_SOURCE_COMPLETE - with session.initiate(): - measurements = session.fetch_multiple(count) - assert len(measurements) == count - assert isinstance(measurements[1].voltage, float) - assert isinstance(measurements[1].current, float) - assert measurements[1].in_compliance in [True, False] - assert measurements[1].voltage == 1.0 - assert measurements[1].current == 0.00001 - - def test_measure_multiple(self, session): - with session.initiate(): - # session is open to all 12 channels on the device - measurements = session.measure_multiple() - assert len(measurements) == 12 - assert measurements[1].in_compliance is None - assert measurements[1].voltage == 0.0 - assert measurements[1].current == 0.00001 - # now a subset of the channels - measurements = session.channels[range(4)].measure_multiple() - assert len(measurements) == 4 - assert measurements[1].in_compliance is None - assert measurements[1].voltage == 0.0 - assert measurements[1].current == 0.00001 - @pytest.mark.parametrize( 'resource_name,channels,independent_channels,measurement_channels,expected_measured_channel', [ @@ -1074,7 +1079,8 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} @pytest.mark.resource_name("4190/0") @@ -1096,17 +1102,19 @@ def test_lcr_compensation_data(self, session): session.configure_lcr_compensation(compensation_data_bytes_from_file) -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nidcpower.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options} @@ -1121,3 +1129,44 @@ def test_configure_lcr_compensation(self, session): session.configure_lcr_compensation([]) assert exc_info.value.args[0] == 'configure_lcr_compensation is not supported over gRPC' assert str(exc_info.value) == 'configure_lcr_compensation is not supported over gRPC' + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidcpower.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidcpower.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} diff --git a/src/nidigital/system_tests/grpc_server_config.json b/src/nidigital/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nidigital/system_tests/grpc_server_config.json rename to src/nidigital/system_tests/grpc_server_config_no_tls.json diff --git a/src/nidigital/system_tests/grpc_server_config_tls.json b/src/nidigital/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..09634ecc79 --- /dev/null +++ b/src/nidigital/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31761, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nidigital/system_tests/test_system_nidigital.py b/src/nidigital/system_tests/test_system_nidigital.py index ea477713f6..7709a2bb7c 100644 --- a/src/nidigital/system_tests/test_system_nidigital.py +++ b/src/nidigital/system_tests/test_system_nidigital.py @@ -6,6 +6,7 @@ import grpc import hightime +import nitlsconfig import numpy import pytest @@ -18,12 +19,54 @@ test_files_base_dir = os.path.join(os.path.dirname(__file__), 'test_files') -class SystemTests: +# Defines a subset of system tests to validate basic NI-Digital functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def multi_instrument_session(self, session_creation_kwargs): with nidigital.Session(resource_name=','.join(instruments), options='Simulate=1, DriverSetup=Model:6570', **session_creation_kwargs) as simulated_session: yield simulated_session + def configure_session(self, session, test_name): + session.load_pin_map(self.get_test_file_path(test_name, 'pin_map.pinmap')) + + session.load_specifications_levels_and_timing( + specifications_file_paths=self.get_test_file_path(test_name, 'specifications.specs'), + levels_file_paths=self.get_test_file_path(test_name, 'pin_levels.digilevels'), + timing_file_paths=self.get_test_file_path(test_name, 'timing.digitiming')) + session.apply_levels_and_timing(levels_sheet='pin_levels', timing_sheet='timing') + + def get_test_file_path(self, test_name, file_name): + return os.path.join(test_files_base_dir, test_name, file_name) + + def test_self_test(self, multi_instrument_session): + multi_instrument_session.self_test() + + def test_ppmu_source(self, multi_instrument_session): + test_name = 'simple_pattern' + self.configure_session(multi_instrument_session, test_name) + + multi_instrument_session.pins['site0/LO0', 'site1/HI0'].ppmu_source() + + def test_ppmu_measure(self, multi_instrument_session): + test_name = 'simple_pattern' + self.configure_session(multi_instrument_session, test_name) + + voltage_measurements = multi_instrument_session.pins['site0/LO0', 'site1/HI0'].ppmu_measure( + nidigital.PPMUMeasurementType.VOLTAGE) + + assert len(voltage_measurements) == 2 + + def test_read_static(self, multi_instrument_session): + test_name = 'simple_pattern' + self.configure_session(multi_instrument_session, test_name) + + pin_states = multi_instrument_session.pins['site0/LO0', 'site1/HI0'].read_static() + + assert pin_states == [nidigital.PinState.L] * 2 + + +class SystemTests(BasicValidationTests): @pytest.fixture(scope='function') def single_instrument_session(self, session_creation_kwargs): with nidigital.Session(resource_name=instruments[0], options='Simulate=1, DriverSetup=Model:6570', **session_creation_kwargs) as simulated_session: @@ -51,9 +94,6 @@ def test_reset_device(self, multi_instrument_session): multi_instrument_session.reset_device() assert multi_instrument_session.selected_function == nidigital.SelectedFunction.DISCONNECT - def test_self_test(self, multi_instrument_session): - multi_instrument_session.self_test() - def test_get_error(self, multi_instrument_session): try: multi_instrument_session.supported_instrument_models = '' @@ -247,18 +287,6 @@ def test_source_waveform_parallel_broadcast(self, multi_instrument_session): pass_fail = multi_instrument_session.burst_pattern(start_label='new_pattern') assert pass_fail == {0: True, 1: True} - def configure_session(self, session, test_name): - session.load_pin_map(self.get_test_file_path(test_name, 'pin_map.pinmap')) - - session.load_specifications_levels_and_timing( - specifications_file_paths=self.get_test_file_path(test_name, 'specifications.specs'), - levels_file_paths=self.get_test_file_path(test_name, 'pin_levels.digilevels'), - timing_file_paths=self.get_test_file_path(test_name, 'timing.digitiming')) - session.apply_levels_and_timing(levels_sheet='pin_levels', timing_sheet='timing') - - def get_test_file_path(self, test_name, file_name): - return os.path.join(test_files_base_dir, test_name, file_name) - @pytest.fixture(params=[array.array, numpy.array, list]) def source_waveform_type(self, request): return request.param @@ -652,29 +680,6 @@ def test_get_fail_count(self, multi_instrument_session): fail_count = multi_instrument_session.pins['site0/LO0', 'site0/HI1', 'site2/HI3'].get_fail_count() assert fail_count == [0] * 3 - def test_ppmu_measure(self, multi_instrument_session): - test_name = 'simple_pattern' - self.configure_session(multi_instrument_session, test_name) - - voltage_measurements = multi_instrument_session.pins['site0/LO0', 'site1/HI0'].ppmu_measure( - nidigital.PPMUMeasurementType.VOLTAGE) - - assert len(voltage_measurements) == 2 - - def test_ppmu_source(self, multi_instrument_session): - test_name = 'simple_pattern' - self.configure_session(multi_instrument_session, test_name) - - multi_instrument_session.pins['site0/LO0', 'site1/HI0'].ppmu_source() - - def test_read_static(self, multi_instrument_session): - test_name = 'simple_pattern' - self.configure_session(multi_instrument_session, test_name) - - pin_states = multi_instrument_session.pins['site0/LO0', 'site1/HI0'].read_static() - - assert pin_states == [nidigital.PinState.L] * 2 - def test_write_static(self, multi_instrument_session): test_name = 'simple_pattern' self.configure_session(multi_instrument_session, test_name) @@ -1327,7 +1332,8 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, multi_inst class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} def test_enable_match_fail_combination(self, multi_instrument_session): @@ -1347,16 +1353,59 @@ def test_enable_match_fail_combination(self, multi_instrument_session): multi_instrument_session.read_sequencer_flag(nidigital.SequencerFlag.FLAG0) -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidigital.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidigital.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nidigital.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options} diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index 71ea72ea72..9df9b1c710 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -25,6 +25,10 @@ def session(self, session_creation_kwargs): with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', **session_creation_kwargs) as simulated_session: yield simulated_session + def test_method_self_test(self, session): + # We should not get an assert if self_test passes + session.self_test() + def test_take_simple_measurement_works(self, session): session.configure_measurement_digits(nidmm.Function.DC_CURRENT, 1, 5.5) assert session.read() != 0 # Assumes DMM reading is not exactly zero to support non-connected modules and simulated modules. @@ -85,10 +89,6 @@ def test_method_configure_trigger(self, session): except nidmm.Error: assert True - def test_method_self_test(self, session): - # We should not get an assert if self_test passes - session.self_test() - def test_method_get_dev_temp(self, session): temperature = session.get_dev_temp('') assert 20 <= temperature <= 50 diff --git a/src/nifgen/system_tests/grpc_server_config.json b/src/nifgen/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nifgen/system_tests/grpc_server_config.json rename to src/nifgen/system_tests/grpc_server_config_no_tls.json diff --git a/src/nifgen/system_tests/grpc_server_config_tls.json b/src/nifgen/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..1271df4313 --- /dev/null +++ b/src/nifgen/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31763, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nifgen/system_tests/test_system_nifgen.py b/src/nifgen/system_tests/test_system_nifgen.py index e16b62b2e5..8b7e8f7670 100644 --- a/src/nifgen/system_tests/test_system_nifgen.py +++ b/src/nifgen/system_tests/test_system_nifgen.py @@ -6,6 +6,7 @@ import grpc import hightime +import nitlsconfig import numpy import pytest @@ -30,7 +31,9 @@ def get_test_file_path(file_name): 3.14159, ] -class SystemTests: +# Defines a subset of system tests to validate basic NI-FGEN functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def session(self, session_creation_kwargs): with nifgen.Session('', '0', False, 'Simulate=1, DriverSetup=Model:5433 (2CH);BoardType:PXIe', **session_creation_kwargs) as simulated_session: @@ -40,6 +43,26 @@ def test_self_test(self, session): # We should not get an assert if self_test passes session.self_test() + def test_standard_waveform(self, session): + session.output_mode = nifgen.OutputMode.FUNC + session.configure_standard_waveform(nifgen.Waveform.SINE, 2.0, 2000000, 1.0, 0.0) + expected_frequency = 2000000 + with session.initiate(): + assert session.func_amplitude == 2.0 + assert session.func_waveform == nifgen.Waveform.SINE + actual_frequency = session.func_frequency + in_range = abs(actual_frequency - expected_frequency) <= max(1e-09 * max(abs(actual_frequency), abs(expected_frequency)), 0.0) # https://stackoverflow.com/questions/5595425/what-is-the-best-way-to-compare-floats-for-almost-equality-in-python + assert in_range is True + assert session.func_dc_offset == 1.0 + assert session.func_start_phase == 0.0 + assert session.is_done() is False + + def test_create_waveform_from_list(self, session): + data = [0.1] * 10000 + assert type(session.create_waveform(data)) is int + + +class SystemTests(BasicValidationTests): def test_get_attribute_string(self, session): model = session.instrument_model assert model == 'NI PXIe-5433 (2CH)' @@ -112,20 +135,6 @@ def test_script_triggers_rep_cap(self, session): session.script_triggers[0].exported_script_trigger_output_terminal = requested_terminal_name assert requested_terminal_name == session.script_triggers[0].exported_script_trigger_output_terminal - def test_standard_waveform(self, session): - session.output_mode = nifgen.OutputMode.FUNC - session.configure_standard_waveform(nifgen.Waveform.SINE, 2.0, 2000000, 1.0, 0.0) - expected_frequency = 2000000 - with session.initiate(): - assert session.func_amplitude == 2.0 - assert session.func_waveform == nifgen.Waveform.SINE - actual_frequency = session.func_frequency - in_range = abs(actual_frequency - expected_frequency) <= max(1e-09 * max(abs(actual_frequency), abs(expected_frequency)), 0.0) # https://stackoverflow.com/questions/5595425/what-is-the-best-way-to-compare-floats-for-almost-equality-in-python - assert in_range is True - assert session.func_dc_offset == 1.0 - assert session.func_start_phase == 0.0 - assert session.is_done() is False - def test_frequency_list(self, session): session.output_mode = nifgen.OutputMode.FREQ_LIST duration_array = [0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01] @@ -140,10 +149,6 @@ def test_frequency_list(self, session): def test_clear_freq_list(self, session): session.clear_freq_list(-1) - def test_create_waveform_from_list(self, session): - data = [0.1] * 10000 - assert type(session.create_waveform(data)) is int - def test_configure_arb_waveform(self, session): waveform_data = [x * (1.0 / 256.0) for x in range(256)] session.output_mode = nifgen.OutputMode.ARB @@ -471,7 +476,8 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} # Test doesn't run over gRPC because numpy isn't supported by gRPC. @@ -513,16 +519,59 @@ def test_write_named_waveform_numpy_array_int16(self, session): session.write_waveform('foo', data) -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nifgen.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nifgen.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nifgen.GrpcSessionOptions(grpc_channel, '') return {'grpc_options': grpc_options} diff --git a/src/nirfsa/system_tests/grpc_server_config.json b/src/nirfsa/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nirfsa/system_tests/grpc_server_config.json rename to src/nirfsa/system_tests/grpc_server_config_no_tls.json diff --git a/src/nirfsa/system_tests/grpc_server_config_tls.json b/src/nirfsa/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..bf1a141d87 --- /dev/null +++ b/src/nirfsa/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31767, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nirfsa/system_tests/test_system_nirfsa.py b/src/nirfsa/system_tests/test_system_nirfsa.py index 7d7dea5d99..a474863ee8 100644 --- a/src/nirfsa/system_tests/test_system_nirfsa.py +++ b/src/nirfsa/system_tests/test_system_nirfsa.py @@ -1,6 +1,7 @@ import grpc import hightime import nirfsa +import nitlsconfig import numpy as np import os import pathlib @@ -22,7 +23,9 @@ def get_test_file_path(file_name): return os.path.join(test_files_base_dir, file_name) -class SystemTests: +# Defines a subset of system tests to validate basic NI-RFSA functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def rfsa_device_session(self, session_creation_kwargs): if use_simulated_session: @@ -32,6 +35,33 @@ def rfsa_device_session(self, session_creation_kwargs): with nirfsa.Session(real_hw_resource_name, id_query=False, reset_device=False, **session_creation_kwargs) as real_rfsa_device_session: yield real_rfsa_device_session + def test_self_test(self, rfsa_device_session): + # We should not get an assert if self_test passes + rfsa_device_session.self_test() + + def test_fetch_iq_single_record_with_samples_passed_as_none(self, rfsa_device_session): + rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ + rfsa_device_session.iq_rate = 1e6 + iq_data_array = np.zeros(64, dtype=np.complex128) + with rfsa_device_session.initiate(): + wfm_info = rfsa_device_session.fetch_iq_single_record_into(iq_data_array) + assert len(wfm_info.samples) == wfm_info.actual_samples + assert np.asarray(wfm_info.samples).dtype == np.complex128 + + def test_fetch_iq_multi_record_with_records_passed_as_none(self, rfsa_device_session): + rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ + rfsa_device_session.number_of_samples = 64 + iq_data_arrays = np.zeros((2, 64), dtype=np.complex128) + with rfsa_device_session.initiate(): + wfm_info = rfsa_device_session.fetch_iq_multi_record_into(iq_data_arrays, number_of_samples=rfsa_device_session.number_of_samples) + assert len(wfm_info) == rfsa_device_session.number_of_records + for i in range(len(wfm_info)): + if isinstance(wfm_info[i], nirfsa.WaveformInfo): + assert np.asarray(wfm_info[i].samples).dtype == np.complex128 + assert len(wfm_info[i].samples) == rfsa_device_session.number_of_samples + + +class SystemTests(BasicValidationTests): @pytest.fixture(scope='function') def simulated_5831_device_session(self, session_creation_kwargs): with nirfsa.Session("5831sim", id_query=False, reset_device=False, options="Simulate=1, DriverSetup=Model:5831", **session_creation_kwargs) as sim_5831_session: @@ -156,10 +186,6 @@ def test_reset_with_options(self, rfsa_device_session): assert e.code == -1074097772 assert 'de-embedding table cannot be found' in e.description - def test_self_test(self, rfsa_device_session): - # We should not get an assert if self_test passes - rfsa_device_session.self_test() - @pytest.mark.skipif(use_simulated_session is False, reason="Takes long time on real device") def test_self_cal_range(self, rfsa_device_session): steps_to_omit = nirfsa.SelfCalibrateRangeStepsToOmit.DIGITIZER_SELF_CAL | nirfsa.SelfCalibrateRangeStepsToOmit.LO_SELF_CAL @@ -373,15 +399,6 @@ def test_send_software_edge_trigger_configured_with_start_trigger(self, rfsa_dev assert rfsa_device_session.check_acquisition_status() is True # Fetch tests - def test_fetch_iq_single_record_with_samples_passed_as_none(self, rfsa_device_session): - rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ - rfsa_device_session.iq_rate = 1e6 - iq_data_array = np.zeros(64, dtype=np.complex128) - with rfsa_device_session.initiate(): - wfm_info = rfsa_device_session.fetch_iq_single_record_into(iq_data_array) - assert len(wfm_info.samples) == wfm_info.actual_samples - assert np.asarray(wfm_info.samples).dtype == np.complex128 - def test_fetch_iq_single_record_subset(self, rfsa_device_session): rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ rfsa_device_session.iq_rate = 1e6 @@ -424,18 +441,6 @@ def test_fetch_iq_single_record_complex_i16(self, rfsa_device_session): assert np.asarray(wfm_info.samples).dtype == np.int16 assert len(wfm_info.samples) == wfm_info.actual_samples - def test_fetch_iq_multi_record_with_records_passed_as_none(self, rfsa_device_session): - rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ - rfsa_device_session.number_of_samples = 64 - iq_data_arrays = np.zeros((2, 64), dtype=np.complex128) - with rfsa_device_session.initiate(): - wfm_info = rfsa_device_session.fetch_iq_multi_record_into(iq_data_arrays, number_of_samples=rfsa_device_session.number_of_samples) - assert len(wfm_info) == rfsa_device_session.number_of_records - for i in range(len(wfm_info)): - if isinstance(wfm_info[i], nirfsa.WaveformInfo): - assert np.asarray(wfm_info[i].samples).dtype == np.complex128 - assert len(wfm_info[i].samples) == rfsa_device_session.number_of_samples - def test_fetch_iq_multi_record_with_samples_passed_as_none(self, rfsa_device_session): rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.IQ rfsa_device_session.number_of_records = 2 @@ -607,21 +612,65 @@ def test_delete_all_deembedding_tables(self, rfsa_device_session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} @pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nirfsa.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nirfsa.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nirfsa.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options} diff --git a/src/nirfsg/system_tests/grpc_server_config.json b/src/nirfsg/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nirfsg/system_tests/grpc_server_config.json rename to src/nirfsg/system_tests/grpc_server_config_no_tls.json diff --git a/src/nirfsg/system_tests/grpc_server_config_tls.json b/src/nirfsg/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..5f803c8c95 --- /dev/null +++ b/src/nirfsg/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31766, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nirfsg/system_tests/test_system_nirfsg.py b/src/nirfsg/system_tests/test_system_nirfsg.py index 21c32c3ba2..13e295c2de 100644 --- a/src/nirfsg/system_tests/test_system_nirfsg.py +++ b/src/nirfsg/system_tests/test_system_nirfsg.py @@ -2,6 +2,7 @@ import grpc import hightime import nirfsg +import nitlsconfig import numpy as np import os import pathlib @@ -26,7 +27,9 @@ def get_test_file_path(file_name): sys.path.insert(0, str(pathlib.Path(__file__).parent.parent.parent / 'generated/nirfsg')) -class SystemTests: +# Defines a subset of system tests to validate basic NI-RFSG functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def rfsg_device_session(self, session_creation_kwargs): if use_simulated_session: @@ -36,6 +39,23 @@ def rfsg_device_session(self, session_creation_kwargs): with nirfsg.Session(real_hw_resource_name, **session_creation_kwargs) as real_rfsg_device_session: yield real_rfsg_device_session + def test_self_test(self, rfsg_device_session): + # We should not get an assert if self_test passes + rfsg_device_session.self_test() + + def test_configure_rf(self, rfsg_device_session): + rfsg_device_session.configure_rf(2e9, -5.0) + assert rfsg_device_session.power_level == -5.0 + assert rfsg_device_session.frequency == 2e9 + + def test_abort(self, rfsg_device_session): + rfsg_device_session.configure_rf(2e9, -5.0) + rfsg_device_session.initiate() + rfsg_device_session.check_generation_status() + rfsg_device_session.abort() + + +class SystemTests(BasicValidationTests): @pytest.fixture(scope='function') def simulated_5831_device_session(self, session_creation_kwargs): with nirfsg.Session("5831sim", options="Simulate=1, DriverSetup=Model:5831", **session_creation_kwargs) as sim_5831_session: @@ -128,10 +148,6 @@ def test_reset_with_options(self, rfsg_device_session): waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform') assert waveform_exists is False - def test_self_test(self, rfsg_device_session): - # We should not get an assert if self_test passes - rfsg_device_session.self_test() - @pytest.mark.skipif(use_simulated_session is False, reason="Takes long time in real device") def test_self_cal(self, rfsg_device_session): rfsg_device_session.self_cal() @@ -197,11 +213,6 @@ def test_los_rep_cap(self, simulated_5831_device_session): assert simulated_5831_device_session.los[2].lo_source == requested_lo_source # Configuration methods related tests - def test_configure_rf(self, rfsg_device_session): - rfsg_device_session.configure_rf(2e9, -5.0) - assert rfsg_device_session.power_level == -5.0 - assert rfsg_device_session.frequency == 2e9 - def test_write_arb_waveform_numpy_complex128(self, rfsg_device_session): rfsg_device_session.generation_mode = nirfsg.GenerationMode.ARB_WAVEFORM waveform_data = np.full(1000, 1 + 0j, dtype=np.complex128) @@ -448,12 +459,6 @@ def test_cw_generation_with_status(self, rfsg_device_session): is_done = rfsg_device_session.check_generation_status() assert is_done is False # is_done will never be True in CW mode - def test_abort(self, rfsg_device_session): - rfsg_device_session.configure_rf(2e9, -5.0) - rfsg_device_session.initiate() - rfsg_device_session.check_generation_status() - rfsg_device_session.abort() - @pytest.mark.skipif(use_simulated_session is True, reason="is_done is always True on simulated device") def test_abort_with_status(self, rfsg_device_session): rfsg_device_session.configure_rf(2e9, -5.0) @@ -647,7 +652,8 @@ def test_get_all_script_names(self, rfsg_device_session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} # grpc-device had a bug in get_all_named_waveform_names @@ -666,17 +672,60 @@ def test_get_all_named_waveform_names(self, rfsg_device_session): @pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nirfsg.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nirfsg.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="gRPC tests not supported on 32-bit Python") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nirfsg.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options} diff --git a/src/niscope/system_tests/grpc_server_config.json b/src/niscope/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/niscope/system_tests/grpc_server_config.json rename to src/niscope/system_tests/grpc_server_config_no_tls.json diff --git a/src/niscope/system_tests/grpc_server_config_tls.json b/src/niscope/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..731ceba470 --- /dev/null +++ b/src/niscope/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31764, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/niscope/system_tests/test_system_niscope.py b/src/niscope/system_tests/test_system_niscope.py index 39a990fddb..7f6fb7a48e 100644 --- a/src/niscope/system_tests/test_system_niscope.py +++ b/src/niscope/system_tests/test_system_niscope.py @@ -8,6 +8,7 @@ import fasteners import grpc import hightime +import nitlsconfig import numpy import pytest @@ -66,7 +67,9 @@ def check_fetched_data( assert data[i].record == expected_records[i] -class SystemTests: +# Defines a subset of system tests to validate basic NI-SCOPE functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def single_instrument_session(self, session_creation_kwargs): with niscope.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:5164; BoardType:PXIe', **session_creation_kwargs) as simulated_session: @@ -77,53 +80,19 @@ def single_instrument_session_5171(self, session_creation_kwargs): # High chann with niscope.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:5171R (8CH); BoardType:PXIe', **session_creation_kwargs) as simulated_session: yield simulated_session - @pytest.fixture(scope='function') - def multi_instrument_session(self, session_creation_kwargs): - with niscope.Session(','.join(instruments), False, True, 'Simulate=1, DriverSetup=Model:5164; BoardType:PXIe', **session_creation_kwargs) as simulated_session: - yield simulated_session - @pytest.fixture(scope='function') def multi_instrument_session_5171(self, session_creation_kwargs): # High channel-count session for get_channel_names testing with niscope.Session(','.join(instruments), False, True, 'Simulate=1, DriverSetup=Model:5171R (8CH); BoardType:PXIe', **session_creation_kwargs) as simulated_session: yield simulated_session @pytest.fixture(scope='function') - def session_5124(self, session_creation_kwargs): - with daqmx_sim_5124_lock: - with niscope.Session('5124', False, False, '', **session_creation_kwargs) as simulated_session: # 5124 is needed for video triggering - yield simulated_session - - @pytest.fixture(scope='function') - def session_5142(self, session_creation_kwargs): - with daqmx_sim_5142_lock: - with niscope.Session('5142', False, False, '', **session_creation_kwargs) as simulated_session: # 5142 is needed for OSP - yield simulated_session - - # Attribute tests - def test_vi_boolean_attribute(self, multi_instrument_session): - multi_instrument_session.allow_more_records_than_memory = False - default_option = multi_instrument_session.allow_more_records_than_memory - assert default_option is False - - def test_vi_string_attribute(self, multi_instrument_session): - trigger_source = f'/{instruments[1]}/NISCOPE_VAL_IMMEDIATE' - multi_instrument_session.acq_arm_source = trigger_source - assert trigger_source == multi_instrument_session.acq_arm_source - - # Basic usability tests - def test_get_channel_names_with_single_instrument_session(self, single_instrument_session_5171): - expected_string = [f'{x}' for x in range(8)] - # Sanity test few different types of input. No need for test to be exhaustive - # since all the various types are covered by converter unit tests. - channel_indices = ['0-1, 2, 3:4', 5, range(6, 7), slice(7, 8)] - assert single_instrument_session_5171.get_channel_names(indices=channel_indices) == expected_string + def multi_instrument_session(self, session_creation_kwargs): + with niscope.Session(','.join(instruments), False, True, 'Simulate=1, DriverSetup=Model:5164; BoardType:PXIe', **session_creation_kwargs) as simulated_session: + yield simulated_session - def test_get_channel_names_with_multi_instrument_session(self, multi_instrument_session_5171): - expected_string = [f'{instruments[0]}/{x}' for x in range(8)] + [f'{instruments[1]}/{x}' for x in range(4)] - # Sanity test few different types of input. No need for test to be exhaustive - # since all the various types are covered by converter unit tests. - channel_indices = ['0-1, 2, 3:4', 5, (6, 7), range(8, 10), slice(10, 12)] - assert multi_instrument_session_5171.get_channel_names(indices=channel_indices) == expected_string + def test_self_test(self, multi_instrument_session): + # We should not get an assert if self_test passes + multi_instrument_session.self_test() @pytest.mark.parametrize( "test_channels,test_channels_expanded", @@ -141,6 +110,20 @@ def test_read(self, multi_instrument_session, test_channels, test_channels_expan waveforms = multi_instrument_session.channels[test_channels].read(num_samples=test_record_length, num_records=test_num_records) check_fetched_data(waveforms, test_channels_expanded, test_record_length, test_num_records) + def test_get_channel_names_with_single_instrument_session(self, single_instrument_session_5171): + expected_string = [f'{x}' for x in range(8)] + # Sanity test few different types of input. No need for test to be exhaustive + # since all the various types are covered by converter unit tests. + channel_indices = ['0-1, 2, 3:4', 5, range(6, 7), slice(7, 8)] + assert single_instrument_session_5171.get_channel_names(indices=channel_indices) == expected_string + + def test_get_channel_names_with_multi_instrument_session(self, multi_instrument_session_5171): + expected_string = [f'{instruments[0]}/{x}' for x in range(8)] + [f'{instruments[1]}/{x}' for x in range(4)] + # Sanity test few different types of input. No need for test to be exhaustive + # since all the various types are covered by converter unit tests. + channel_indices = ['0-1, 2, 3:4', 5, (6, 7), range(8, 10), slice(10, 12)] + assert multi_instrument_session_5171.get_channel_names(indices=channel_indices) == expected_string + @pytest.mark.parametrize( "test_channels,test_channels_expanded", [ @@ -181,6 +164,31 @@ def test_fetch_defaults(self, multi_instrument_session): for i in range(len(waveforms)): assert len(waveforms[i].samples) == test_record_length + +class SystemTests(BasicValidationTests): + @pytest.fixture(scope='function') + def session_5124(self, session_creation_kwargs): + with daqmx_sim_5124_lock: + with niscope.Session('5124', False, False, '', **session_creation_kwargs) as simulated_session: # 5124 is needed for video triggering + yield simulated_session + + @pytest.fixture(scope='function') + def session_5142(self, session_creation_kwargs): + with daqmx_sim_5142_lock: + with niscope.Session('5142', False, False, '', **session_creation_kwargs) as simulated_session: # 5142 is needed for OSP + yield simulated_session + + # Attribute tests + def test_vi_boolean_attribute(self, multi_instrument_session): + multi_instrument_session.allow_more_records_than_memory = False + default_option = multi_instrument_session.allow_more_records_than_memory + assert default_option is False + + def test_vi_string_attribute(self, multi_instrument_session): + trigger_source = f'/{instruments[1]}/NISCOPE_VAL_IMMEDIATE' + multi_instrument_session.acq_arm_source = trigger_source + assert trigger_source == multi_instrument_session.acq_arm_source + @pytest.fixture(params=[(1000, 1000), (2000, 2000), (3000, 2000)], ids=["less_than_actual", "equal_to_actual", "greater_than_actual"]) def measurement_wfm_length(self, request): MeasWfmLength = collections.namedtuple('MeasurementWaveformLength', ['passed_in', 'expected']) @@ -354,10 +362,6 @@ def test_measurement_stats_str(self, multi_instrument_session): assert isinstance(measurement_stat[0].__str__(), str) assert isinstance(measurement_stat[0].__repr__(), str) - def test_self_test(self, multi_instrument_session): - # We should not get an assert if self_test passes - multi_instrument_session.self_test() - def test_reset(self, multi_instrument_session): default_fetch_relative_to = multi_instrument_session._fetch_relative_to assert default_fetch_relative_to == niscope.FetchRelativeTo.PRETRIGGER @@ -545,7 +549,8 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, multi_inst class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} # not supported by grpc due to numpy usage @@ -617,17 +622,19 @@ def test_reset_with_defaults(self, single_instrument_session): assert single_instrument_session.meas_time_histogram_high_time == hightime.timedelta(microseconds=500) -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = niscope.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options} @@ -642,3 +649,44 @@ def test_reset_with_defaults(self, single_instrument_session): single_instrument_session.reset_with_defaults() assert exc_info.value.args[0] == 'reset_with_defaults is not supported over gRPC' assert str(exc_info.value) == 'reset_with_defaults is not supported over gRPC' + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = niscope.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = niscope.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} diff --git a/src/niswitch/system_tests/grpc_server_config.json b/src/niswitch/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/niswitch/system_tests/grpc_server_config.json rename to src/niswitch/system_tests/grpc_server_config_no_tls.json diff --git a/src/niswitch/system_tests/grpc_server_config_tls.json b/src/niswitch/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..ba6f872b5c --- /dev/null +++ b/src/niswitch/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31765, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/niswitch/system_tests/test_system_niswitch.py b/src/niswitch/system_tests/test_system_niswitch.py index 4bcfa0aad2..a81a7969d2 100644 --- a/src/niswitch/system_tests/test_system_niswitch.py +++ b/src/niswitch/system_tests/test_system_niswitch.py @@ -6,6 +6,7 @@ import fasteners import grpc import hightime +import nitlsconfig import pytest import niswitch @@ -25,7 +26,9 @@ daqmx_sim_db_lock = fasteners.InterProcessLock(daqmx_sim_db_lock_file) -class SystemTests: +# Defines a subset of system tests to validate basic NI-SWITCH functionality. This is run as a part of the full SystemTests class, and +# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today). +class BasicValidationTests: @pytest.fixture(scope='function') def session(self, session_creation_kwargs): with niswitch.Session('', '2737/2-Wire 4x64 Matrix', True, True, **session_creation_kwargs) as simulated_session: @@ -39,7 +42,10 @@ def session_2532(self, session_creation_kwargs): with daqmx_sim_db_lock: simulated_session.close() - # Basic Use Case Tests + def test_functions_self_test(self, session): + # We should not get an assert if self_test passes + session.self_test() + def test_relayclose(self, session): relay_name = 'kr0c0' assert session.get_relay_position(relay_name) == niswitch.RelayPosition.OPEN @@ -85,15 +91,16 @@ def test_continuous_software_scanning(self, session_2532): except niswitch.Error as e: assert e.code == -1074126826 # Error : Max time exceeded. + +class SystemTests(BasicValidationTests): # Attribute Tests # No R/W non-IVI boolean attributes on all devices - ''' - def test_vi_boolean_attribute(session): - session.power_down_latching_relays_after_debounce = False - assert session.power_down_latching_relays_after_debounce is False - session.power_down_latching_relays_after_debounce = True - assert session.power_down_latching_relays_after_debounce is True - ''' + + # def test_vi_boolean_attribute(session): + # session.power_down_latching_relays_after_debounce = False + # assert session.power_down_latching_relays_after_debounce is False + # session.power_down_latching_relays_after_debounce = True + # assert session.power_down_latching_relays_after_debounce is True def test_vi_string_attribute(self, session): assert 'NI PXIe-2737' == session.instrument_model @@ -138,10 +145,6 @@ def test_functions_get_channel_name(self, session): channel_name = session.get_channel_name(1) assert channel_name == 'r0' - def test_functions_self_test(self, session): - # We should not get an assert if self_test passes - session.self_test() - def test_locks_are_reentrant(self, session): with session.lock(): with session.lock(): @@ -188,20 +191,64 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} -class TestGrpc(SystemTests): +class TestGrpcNoTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: channel = grpc.insecure_channel(f"localhost:{proc.server_port}") yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = niswitch.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcSecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost") + system_test_utilities.exchange_certificates("localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = niswitch.GrpcSessionOptions(grpc_channel, "") + return {'grpc_options': grpc_options} + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes") +class TestGrpcUnsecuredTLS(BasicValidationTests): + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost") + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = niswitch.GrpcSessionOptions(grpc_channel, "") return {'grpc_options': grpc_options}