Skip to content

Refactor tests for stream writers into smaller files #826

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
Sep 10, 2025
Merged

Refactor tests for stream writers into smaller files #826

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a90c4e7
Split out stream writer tests into separate files
mikeprosserni Sep 9, 2025
4b39a0f
remove test_stream_writers.py
mikeprosserni Sep 10, 2025
96d5e6e
move utils out of conftest
mikeprosserni Sep 10, 2025
9cc7dd6
poetry run nps fix
mikeprosserni Sep 10, 2025
f9bd535
poke test_analog_multi_channel_reader.py to try to fix the CI
mikeprosserni Sep 10, 2025
0e5d505
poking again to see what the CI does
mikeprosserni Sep 10, 2025
deb610f
undo poking
mikeprosserni Sep 10, 2025
2a8ff11
Merge remote-tracking branch 'origin/master' into users/mprosser/stre…
mikeprosserni Sep 10, 2025
d8042a3
fix merge
mikeprosserni Sep 10, 2025
fd596ac
import nidaqmx.system
mikeprosserni Sep 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
387 changes: 381 additions & 6 deletions tests/component/conftest.py
View file
Open in desktop

Large diffs are not rendered by default.

30 changes: 15 additions & 15 deletions tests/component/stream_readers/test_digital_multi_channel_reader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,10 @@
from tests.component.conftest import (
_bool_array_to_int,
_get_expected_data_for_line,
_get_expected_digital_data,
_get_digital_data,
_get_expected_digital_port_data_port_major,
_get_expected_digital_port_data_sample_major,
_get_num_lines_in_task,
_get_num_di_lines_in_task,
_get_waveform_data,
_is_timestamp_close_to_now,
_read_and_copy,
Expand All @@ -33,13 +33,13 @@ def test___digital_multi_channel_reader___read_one_sample_one_line___returns_val
di_single_line_task: nidaqmx.Task,
) -> None:
reader = DigitalMultiChannelReader(di_single_line_task.in_stream)
num_lines = _get_num_lines_in_task(di_single_line_task)
num_lines = _get_num_di_lines_in_task(di_single_line_task)
samples_to_read = 256
sample = numpy.full(num_lines, False, dtype=numpy.bool_)

data = [_read_and_copy(reader.read_one_sample_one_line, sample) for _ in range(samples_to_read)]

assert [_bool_array_to_int(sample) for sample in data] == _get_expected_digital_data(
assert [_bool_array_to_int(sample) for sample in data] == _get_digital_data(
num_lines, samples_to_read
)

Expand All @@ -48,7 +48,7 @@ def test___digital_multi_channel_reader___read_one_sample_one_line_with_wrong_dt
di_single_line_task: nidaqmx.Task,
) -> None:
reader = DigitalMultiChannelReader(di_single_line_task.in_stream)
num_lines = _get_num_lines_in_task(di_single_line_task)
num_lines = _get_num_di_lines_in_task(di_single_line_task)
data = numpy.full(num_lines, math.inf, dtype=numpy.float64)

with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info:
Expand All @@ -69,7 +69,7 @@ def test___digital_multi_channel_reader___read_one_sample_multi_line___returns_v
_read_and_copy(reader.read_one_sample_multi_line, sample) for _ in range(samples_to_read)
]

assert [_bool_array_to_int(sample[:, 0]) for sample in data] == _get_expected_digital_data(
assert [_bool_array_to_int(sample[:, 0]) for sample in data] == _get_digital_data(
num_channels, samples_to_read
)

Expand Down Expand Up @@ -328,7 +328,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms___returns_va
) -> None:
reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10
waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]

Expand All @@ -354,7 +354,7 @@ def test___digital_multi_channel_different_lines_reader___read_waveforms___retur
) -> None:
reader = DigitalMultiChannelReader(di_multi_chan_diff_lines_timing_task.in_stream)
num_channels = di_multi_chan_diff_lines_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_diff_lines_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_diff_lines_timing_task)
samples_to_read = 10
waveforms = [
DigitalWaveform(samples_to_read, 1),
Expand Down Expand Up @@ -406,7 +406,7 @@ def test___digital_multi_channel_lines_and_port_reader___read_waveforms___return
) -> None:
reader = DigitalMultiChannelReader(di_multi_chan_lines_and_port_task.in_stream)
num_channels = di_multi_chan_lines_and_port_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_lines_and_port_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_lines_and_port_task)
samples_to_read = 10
waveforms = [
DigitalWaveform(samples_to_read, 1),
Expand Down Expand Up @@ -492,7 +492,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_no_args___re
) -> None:
reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
waveforms = [DigitalWaveform(50) for _ in range(num_channels)]

samples_read = reader.read_waveforms(waveforms)
Expand All @@ -516,7 +516,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_in_place___p
) -> None:
reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10

waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]
Expand Down Expand Up @@ -614,7 +614,7 @@ def test___digital_multi_channel_multi_line_reader_with_timing_flag___read_wavef
in_stream.waveform_attribute_mode = WaveformAttributeMode.TIMING
reader = DigitalMultiChannelReader(in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10
waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]

Expand Down Expand Up @@ -642,7 +642,7 @@ def test___digital_multi_channel_multi_line_reader_with_extended_properties_flag
in_stream.waveform_attribute_mode = WaveformAttributeMode.EXTENDED_PROPERTIES
reader = DigitalMultiChannelReader(in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10
waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]

Expand Down Expand Up @@ -670,7 +670,7 @@ def test___digital_multi_channel_multi_line_reader_with_both_flags___read_wavefo
)
reader = DigitalMultiChannelReader(in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10
waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]

Expand Down Expand Up @@ -698,7 +698,7 @@ def test___digital_multi_channel_multi_line_reader_with_none_flag___read_wavefor
in_stream.waveform_attribute_mode = WaveformAttributeMode.NONE
reader = DigitalMultiChannelReader(in_stream)
num_channels = di_multi_chan_multi_line_timing_task.number_of_channels
num_lines = _get_num_lines_in_task(di_multi_chan_multi_line_timing_task)
num_lines = _get_num_di_lines_in_task(di_multi_chan_multi_line_timing_task)
samples_to_read = 10
waveforms = [DigitalWaveform(samples_to_read) for _ in range(num_channels)]

Expand Down
Loading
Loading