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..63115cf3cf 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. @@ -75,6 +78,50 @@ def session(self, request, session_creation_kwargs): with nidcpower.Session(**init_args, **session_creation_kwargs) as simulated_session: yield simulated_session + @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): def test_self_test(self, session): session.self_test() @@ -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..24979225a0 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_burst_pattern_pass_fail(self, multi_instrument_session): + test_files_folder = 'simple_pattern' + self.configure_session(multi_instrument_session, test_files_folder) + + multi_instrument_session.load_pattern(self.get_test_file_path(test_files_folder, 'pattern.digipat')) + + result = multi_instrument_session.burst_pattern(start_label='new_pattern', wait_until_done=True) + assert result == {0: True, 1: True, 2: True, 3: True} + + 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: @@ -216,15 +259,6 @@ def test_burst_pattern_burst_only(self, multi_instrument_session): result = multi_instrument_session.burst_pattern(start_label='new_pattern', wait_until_done=False) assert result is None - def test_burst_pattern_pass_fail(self, multi_instrument_session): - test_files_folder = 'simple_pattern' - self.configure_session(multi_instrument_session, test_files_folder) - - multi_instrument_session.load_pattern(self.get_test_file_path(test_files_folder, 'pattern.digipat')) - - result = multi_instrument_session.burst_pattern(start_label='new_pattern', wait_until_done=True) - assert result == {0: True, 1: True, 2: True, 3: True} - def test_source_waveform_parallel_broadcast(self, multi_instrument_session): '''Test methods for using source waveform with parallel sourcing and broadcast data mapping. @@ -247,18 +281,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 +674,12 @@ 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/grpc_server_config.json b/src/nidmm/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nidmm/system_tests/grpc_server_config.json rename to src/nidmm/system_tests/grpc_server_config_no_tls.json diff --git a/src/nidmm/system_tests/grpc_server_config_tls.json b/src/nidmm/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..4300dc473d --- /dev/null +++ b/src/nidmm/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31762, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index 55cdea3988..71ea72ea72 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -7,6 +7,7 @@ import grpc import hightime +import nitlsconfig import numpy import pytest @@ -16,13 +17,14 @@ import system_test_utilities # noqa: E402 -class SystemTests: +# Defines a subset of system tests to validate basic DMM 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 nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', **session_creation_kwargs) as simulated_session: yield simulated_session - # Basic usability tests 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. @@ -40,6 +42,8 @@ def test_multi_point_acquisition(self, session): measurements = session.read_multi_point(8) assert len(measurements) == 8 + +class SystemTests(BasicValidationTests): # Attribute tests def test_vi_string_attribute(self, session): assert session.instrument_model == 'NI PXIe-4082' @@ -312,7 +316,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 {} def test_fetch_waveform_into(self, session): @@ -327,17 +332,19 @@ def test_fetch_waveform_into(self, session): assert not math.isnan(sample) -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 = nidmm.GrpcSessionOptions(grpc_channel, '') return {'grpc_options': grpc_options} @@ -369,3 +376,44 @@ def test_attach_to_non_existent_session(self, grpc_channel): assert e.rpc_code == expected_grpc_error assert e.description == expected_error_message assert str(e) == f'{expected_grpc_error}: {expected_error_message}' + + +@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 = nidmm.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 = nidmm.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} 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..aaf16b989e 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,12 +31,46 @@ 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: yield simulated_session + 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] + frequency_array = [1000, 100900, 200800, 300700, 400600, 500500, 600400, 700300, 800200, 900100] + waveform_handle = session.create_freq_list(nifgen.Waveform.SQUARE, frequency_array, duration_array) + session.configure_freq_list(waveform_handle, 2.0, 0, 0) + session.trigger_mode = nifgen.TriggerMode.CONTINUOUS + session.output_enabled = True + assert session.func_waveform == nifgen.Waveform.SQUARE + assert session.func_amplitude == 2.0 + + 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 + session.configure_arb_waveform(session.create_waveform(waveform_data), 1.0, 0.0) + + +class SystemTests(BasicValidationTests): def test_self_test(self, session): # We should not get an assert if self_test passes session.self_test() @@ -112,31 +147,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] - frequency_array = [1000, 100900, 200800, 300700, 400600, 500500, 600400, 700300, 800200, 900100] - waveform_handle = session.create_freq_list(nifgen.Waveform.SQUARE, frequency_array, duration_array) - session.configure_freq_list(waveform_handle, 2.0, 0, 0) - session.trigger_mode = nifgen.TriggerMode.CONTINUOUS - session.output_enabled = True - assert session.func_waveform == nifgen.Waveform.SQUARE - assert session.func_amplitude == 2.0 - def test_clear_freq_list(self, session): session.clear_freq_list(-1) @@ -144,11 +154,6 @@ 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 - session.configure_arb_waveform(session.create_waveform(waveform_data), 1.0, 0.0) - def test_disable(self, session): channel = session.channels['0'] assert channel.output_enabled is True @@ -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..f096d696d2 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,37 @@ 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_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 + + def test_read_power_spectrum_with_data_array_size_passed_as_none(self, rfsa_device_session): + rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.SPECTRUM + rfsa_device_session.number_of_spectral_lines = 1024 + power_spectrum_data_array = np.zeros(512, dtype=np.float64) + spectrum_info = rfsa_device_session.read_power_spectrum_into(power_spectrum_data_array) + assert len(spectrum_info.samples) == rfsa_device_session.number_of_spectral_lines + assert np.asarray(spectrum_info.samples).dtype == np.float64 + + +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: @@ -373,15 +407,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 +449,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 @@ -528,14 +541,6 @@ def test_read_power_spectrum_check_view_with_larger_buffer(self, rfsa_device_ses assert np.asarray(spectrum_info.samples).dtype == np.float64 assert len(power_spectrum_data_array) == 1024 - def test_read_power_spectrum_with_data_array_size_passed_as_none(self, rfsa_device_session): - rfsa_device_session.acquisition_type = nirfsa.AcquisitionType.SPECTRUM - rfsa_device_session.number_of_spectral_lines = 1024 - power_spectrum_data_array = np.zeros(512, dtype=np.float64) - spectrum_info = rfsa_device_session.read_power_spectrum_into(power_spectrum_data_array) - assert len(spectrum_info.samples) == rfsa_device_session.number_of_spectral_lines - assert np.asarray(spectrum_info.samples).dtype == np.float64 - # Deembedding tests def test_set_get_deembedding_sparameters(self, rfsa_device_session): frequencies = np.array([1e9, 2e9, 3e9], dtype=np.float64) @@ -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..46a318eb21 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,28 @@ 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_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() + + 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) + rfsg_device_session.write_arb_waveform('mywaveform1', waveform_data, False) + waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform1') + assert waveform_exists is True + waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform2') + assert waveform_exists is False + + def test_wait_until_settled(self, rfsg_device_session): + rfsg_device_session.configure_rf(2e9, -5.0) + with rfsg_device_session.initiate(): + rfsg_device_session.wait_until_settled() + + +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: @@ -202,15 +227,6 @@ def test_configure_rf(self, rfsg_device_session): 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) - rfsg_device_session.write_arb_waveform('mywaveform1', waveform_data, False) - waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform1') - assert waveform_exists is True - waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform2') - assert waveform_exists is False - def test_write_arb_waveform_numpy_complex64(self, rfsg_device_session): rfsg_device_session.generation_mode = nirfsg.GenerationMode.ARB_WAVEFORM waveform_data = np.full(1600, 1 + 0j, dtype=np.complex64) @@ -448,12 +464,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) @@ -618,11 +628,6 @@ def test_read_and_download_waveform_from_file_tdms(self, rfsg_device_session): waveform_exists = rfsg_device_session.check_if_waveform_exists('mywaveform') assert waveform_exists is True - def test_wait_until_settled(self, rfsg_device_session): - rfsg_device_session.configure_rf(2e9, -5.0) - with rfsg_device_session.initiate(): - rfsg_device_session.wait_until_settled() - @pytest.mark.skipif(use_simulated_session is True, reason="Scripts not compiled on simulated device") def test_get_all_script_names(self, rfsg_device_session): rfsg_device_session.generation_mode = nirfsg.GenerationMode.SCRIPT @@ -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..6d73ec3225 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,65 +67,14 @@ def check_fetched_data( assert data[i].record == expected_records[i] -class SystemTests: - @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: - yield simulated_session - - @pytest.fixture(scope='function') - def single_instrument_session_5171(self, session_creation_kwargs): # High channel-count session for get_channel_names testing - with niscope.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:5171R (8CH); BoardType:PXIe', **session_creation_kwargs) as simulated_session: - yield simulated_session - +# 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 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 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 +131,61 @@ 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 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: + yield simulated_session + + @pytest.fixture(scope='function') + def single_instrument_session_5171(self, session_creation_kwargs): # High channel-count session for get_channel_names testing + 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_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 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.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']) @@ -545,7 +550,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 +623,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 +650,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..b0aabf1235 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,21 +26,14 @@ 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: yield simulated_session - @pytest.fixture(scope='function') - def session_2532(self, session_creation_kwargs): - with daqmx_sim_db_lock: - simulated_session = niswitch.Session('', '2532/1-Wire 4x128 Matrix', True, False, **session_creation_kwargs) - yield simulated_session - with daqmx_sim_db_lock: - simulated_session.close() - - # Basic Use Case Tests def test_relayclose(self, session): relay_name = 'kr0c0' assert session.get_relay_position(relay_name) == niswitch.RelayPosition.OPEN @@ -63,6 +57,20 @@ def test_channel_connection(self, session): session.disconnect_all() assert session.can_connect(channel1, channel2) == niswitch.PathCapability.PATH_AVAILABLE + def test_functions_connect_disconnect_multiple(self, session): + session.connect_multiple('c0->r0, c0->r1') # expect no errors + session.disconnect_multiple('c0->r0, c0->r1') # expect no errors + + +class SystemTests(BasicValidationTests): + @pytest.fixture(scope='function') + def session_2532(self, session_creation_kwargs): + with daqmx_sim_db_lock: + simulated_session = niswitch.Session('', '2532/1-Wire 4x128 Matrix', True, False, **session_creation_kwargs) + yield simulated_session + with daqmx_sim_db_lock: + simulated_session.close() + @pytest.mark.skip(reason="TODO(sbethur): Intermittent failures, GitHub issue #1622.") def test_continuous_software_scanning(self, session_2532): scan_list = 'r0->c0; r1->c1' @@ -157,10 +165,6 @@ def test_functions_get_path(self, session): session.disconnect(channel1, channel2) session.set_path(path) - def test_functions_connect_disconnect_multiple(self, session): - session.connect_multiple('c0->r0, c0->r1') # expect no errors - session.disconnect_multiple('c0->r0, c0->r1') # expect no errors - def test_functions_disable(self, session): channel1 = 'c0' channel2 = 'r0' @@ -188,20 +192,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} diff --git a/src/shared/system_test_utilities.py b/src/shared/system_test_utilities.py index dea3b2d1cd..d1b5b7b9d5 100644 --- a/src/shared/system_test_utilities.py +++ b/src/shared/system_test_utilities.py @@ -1,8 +1,10 @@ +import json import os import pathlib import pytest import re import subprocess +import sys import threading import time @@ -104,3 +106,131 @@ def impl_test_multi_threading_ivi_synchronized_wrapper_releases_lock(ivi_method_ t2.start() t2.join() assert not t2.is_alive() + + +def exchange_certificates( + server_host: str, + server_user: str | None = None, + client_host: str | None = None, + client_user: str | None = None, + verbosity: int = 2, +): + # gRPC tests only run on Windows, so this isn't necessary on Linux. + if os.name != "nt": + return + + # 26.5 versions of ni-grpc-device server installers do not properly create the trusted.d directory, + # which causes issues with the certificate exchange process. This has been fixed in the 26.8 version + # of the installer, but it has not yet been released. For now, we're creating it manually; this can + # be removed once nimibot system tests are updated to test against >= 26.8 versions of the drivers. + trusted_servers_path = pathlib.Path(r"C:/ProgramData/National Instruments/nitlsconfig/server.d/ni-grpc-device/trusted.d") + trusted_servers_path.mkdir(parents=True, exist_ok=True) + + # 26.5 versions of ni-grpc-device client configuration use a default certificate_mode of Disabled, + # which prevents client-side certificate generation from this script. In 26.8 and beyond, the default + # is Managed. We set it manually here; this can be removed once nimibot system tests are updated to + # test against >= 26.8 versions of the drivers. + client_config_path = ( + pathlib.Path(os.environ["LOCALAPPDATA"]) + / "National Instruments" / "nitlsconfig" / "client.d" / "ni-grpc-device.conf.yml" + ) + content = client_config_path.read_text() + content = re.sub(r"(?m)^certificate_mode:.*$", "certificate_mode: Managed", content) + client_config_path.write_text(content) + + script_path = r"C:/NITests/nitlsconfigtest/exchange_certificates.py" + if not pathlib.Path(script_path).is_file(): + raise FileNotFoundError(f"Certificate exchange script not found: {script_path}") + + server_host_arg = f"--server-host={server_host}" + server_user_arg = f"--server-user={server_user}" if server_user else "--local-server" + client_host_arg = f"--client-host={client_host}" if client_host else None + client_user_arg = f"--client-user={client_user}" if client_user else None + + verbosity = max(0, min(verbosity, 4)) + verbosity_arg = { + 0: "-qq", + 1: "-q", + 3: "-v", + 4: "-vv", + }.get(verbosity) + + command = [sys.executable, str(pathlib.Path(script_path)), server_host_arg, server_user_arg] + command.extend(arg for arg in (client_host_arg, client_user_arg, verbosity_arg) if arg is not None) + + # The script expects this environment variable to be set + env = os.environ.copy() + env.setdefault("USERNAME", "Administrator") + + subprocess.run(command, check=True, env=env) + + +def configure_tls_modes( + service: str, + server_host: str, + server_cert_mode: str | None = None, + server_client_mode: str | None = None, + client_cert_mode: str | None = None, + client_server_mode: str | None = None, +): + # gRPC tests only run on Windows, so this isn't necessary on Linux. + if os.name != "nt": + return + + script_path = r"C:/NITests/nitlsconfigtest/configure_tls_modes.py" + if not pathlib.Path(script_path).is_file(): + raise FileNotFoundError(f"Configure TLS modes script not found: {script_path}") + + service_arg = f"--service={service}" + server_host_arg = f"--server-host={server_host}" + server_user_arg = "--local-server" + server_cert_mode_arg = f"--server-certificate-mode={server_cert_mode}" if server_cert_mode else None + server_client_mode_arg = f"--server-client-mode={server_client_mode}" if server_client_mode else None + client_cert_mode_arg = f"--client-certificate-mode={client_cert_mode}" if client_cert_mode else None + client_server_mode_arg = f"--client-server-mode={client_server_mode}" if client_server_mode else None + + command = [sys.executable, str(pathlib.Path(script_path)), service_arg, server_host_arg, server_user_arg] + command.extend( + arg + for arg in ( + server_cert_mode_arg, + server_client_mode_arg, + client_cert_mode_arg, + client_server_mode_arg, + ) + if arg is not None + ) + + # The script expects this environment variable to be set + env = os.environ.copy() + env.setdefault("USERNAME", "Administrator") + + subprocess.run(command, check=True, env=env) + + +def configure_tls_modes_secure( + service: str, + server_host: str +): + configure_tls_modes( + service=service, + server_host=server_host, + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + + +def configure_tls_modes_insecure( + service: str, + server_host: str +): + configure_tls_modes( + service=service, + server_host=server_host, + server_cert_mode="Disabled", + server_client_mode="Disabled", + client_cert_mode="Disabled", + client_server_mode="Disabled" + )