Component tests for Global Chans, Task Done, and Watchdog (#548)

* Add fixtures for added simulated devices

* Add configuration for new simulated devices

* Additional component tests

* Change to 9189 for more slots

* Fix linting errors

* Update tests/max_config/nidaqmxMaxConfig.ini

Co-authored-by: Zach Hindes <[email protected]>

* Review comments

* More linting fixes

* Review comments

* Fix lint issues

* Missing change

* Missing pytest.fixtures statement

* Update tests/component/system/storage/test_persisted_channel.py

Co-authored-by: Brad Keryan <[email protected]>

* Update tests/component/test_watchdog.py

Co-authored-by: Brad Keryan <[email protected]>

* Update tests/component/test_watchdog.py

Co-authored-by: Brad Keryan <[email protected]>

* Update tests/component/test_watchdog.py

Co-authored-by: Brad Keryan <[email protected]>

* Update tests/component/test_watchdog.py

Co-authored-by: Brad Keryan <[email protected]>

* Fix lint issues

---------

Co-authored-by: Zach Hindes <[email protected]>
Co-authored-by: Brad Keryan <[email protected]>

tests/component/system/storage/test_persisted_channel.py

@@ -1,6 +1,18 @@
+import pytest
+
+import nidaqmx
 from nidaqmx.system.storage import PersistedChannel
 
 
+@pytest.fixture
+def ai_task(task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device) -> nidaqmx.Task:
+    """Gets an AI task."""
+    task.ai_channels.add_ai_voltage_chan(
+        f"{sim_6363_device.name}/ai0:3", name_to_assign_to_channel="TestChannel"
+    )
+    return task
+
+
 def test___persisted_channels_with_same_name___compare___equal(init_kwargs):
     persisted_channel1 = PersistedChannel("Channel1", **init_kwargs)
     persisted_channel2 = PersistedChannel("Channel1", **init_kwargs)
@@ -29,3 +41,26 @@ def test___persisted_channels_with_different_names___hash___not_equal(init_kwarg
     persisted_channel2 = PersistedChannel("Channel2", **init_kwargs)
 
     assert hash(persisted_channel1) != hash(persisted_channel2)
+
+
+def test___save_persisted_channel___saves_persisted_channel_with_author(
+    ai_task: nidaqmx.Task, init_kwargs: dict
+) -> None:
+    persisted_channel_name = "PersistedChannelSaveTest"
+    persisted_channel_author = "test___save_persisted_channel___no_error"
+    # We first need to check if the channel exists and delete it if it does
+    # We test by trying to access the author property
+    persisted_channel = PersistedChannel(persisted_channel_name, **init_kwargs)
+    try:
+        _ = persisted_channel.author
+        persisted_channel.delete()
+    except Exception:
+        pass
+    # Now we need to create a channel associated with a task, then we can save it
+    channel = ai_task.ai_channels[0]
+
+    channel.save(save_as=persisted_channel_name, author=persisted_channel_author)
+
+    persisted_channel = PersistedChannel(persisted_channel_name, **init_kwargs)
+    assert persisted_channel.author == persisted_channel_author
+    persisted_channel.delete()

tests/component/task/test_triggers.py

@@ -19,9 +19,9 @@ def ai_voltage_task(task, sim_time_aware_9215_device) -> Task:
 
 
 @pytest.fixture()
-def ci_count_edges_task(task, sim_9185_device) -> Task:
-    chan = task.ci_channels.add_ci_count_edges_chan(f"{sim_9185_device.name}/_ctr0")
-    chan.ci_count_edges_term = f"/{sim_9185_device.name}/te0/SampleClock"
+def ci_count_edges_task(task, sim_9189_device) -> Task:
+    chan = task.ci_channels.add_ci_count_edges_chan(f"{sim_9189_device.name}/_ctr0")
+    chan.ci_count_edges_term = f"/{sim_9189_device.name}/te0/SampleClock"
     chan.ci_count_edges_active_edge = Edge.RISING
     return task
 

tests/component/test_task.py

@@ -4,6 +4,7 @@
 from nidaqmx.constants import ShuntCalSelect, ShuntCalSource, ShuntElementLocation
 from nidaqmx.error_codes import DAQmxErrors
 from nidaqmx.errors import RpcError
+from nidaqmx.system.storage import PersistedChannel
 
 
 @pytest.fixture
@@ -26,6 +27,15 @@ def ai_thermocouple_task(task: nidaqmx.Task, device) -> nidaqmx.Task:
     return task
 
 
+@pytest.fixture
+def ai_on_demand_task(task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device) -> nidaqmx.Task:
+    """Gets an AI task."""
+    task.ai_channels.add_ai_voltage_chan(
+        f"{sim_6363_device.name}/ai0:3", name_to_assign_to_channel="MyChannel"
+    )
+    return task
+
+
 @pytest.mark.library_only(reason="Default gRPC initialization behavior is auto (create or attach)")
 def test___task___create_task_with_same_name___raises_duplicate_task(init_kwargs):
     task1 = nidaqmx.Task("MyTask1", **init_kwargs)
@@ -194,3 +204,20 @@ def test___default_arguments___perform_thrmcpl_lead_offset_nulling_cal___no_erro
     ai_thermocouple_task: nidaqmx.Task,
 ) -> None:
     ai_thermocouple_task.perform_thrmcpl_lead_offset_nulling_cal()
+
+
+def test___on_demand_task_is_done___is_task_done___returns_true(ai_on_demand_task: nidaqmx.Task):
+    assert ai_on_demand_task.is_task_done()
+    ai_on_demand_task.start()
+    ai_on_demand_task.wait_until_done()
+
+    assert ai_on_demand_task.is_task_done()
+
+
+def test___task___add_global_channels___adds_to_channel_names(task: nidaqmx.Task):
+    persisted_channel = PersistedChannel("VoltageTesterChannel")
+    persisted_channel2 = PersistedChannel("VoltageTesterChannel2")
+
+    task.add_global_channels([persisted_channel, persisted_channel2])
+
+    assert task.channel_names == [persisted_channel.name, persisted_channel2.name]

tests/component/test_watchdog.py

@@ -0,0 +1,83 @@
+from nidaqmx import Task
+from nidaqmx.constants import WatchdogAOExpirState, WatchdogCOExpirState
+from nidaqmx.system import Device
+from nidaqmx.system.watchdog import AOExpirationState, COExpirationState
+
+
+def test___watchdog_task___cfg_watchdog_ao_expir_states___no_error(
+    generate_watchdog_task: Task, sim_9189_device: Device, sim_9263_device: Device
+):
+    watchdog_task = generate_watchdog_task(f"{sim_9189_device.name}", timeout=0.8)
+    expir_states = [
+        AOExpirationState(
+            physical_channel=sim_9263_device.ao_physical_chans[0].name,
+            expiration_state=0.0,
+            output_type=WatchdogAOExpirState.VOLTAGE,
+        ),
+        AOExpirationState(
+            physical_channel=sim_9263_device.ao_physical_chans[1].name,
+            expiration_state=0.0,
+            output_type=WatchdogAOExpirState.VOLTAGE,
+        ),
+    ]
+
+    watchdog_task.cfg_watchdog_ao_expir_states(expir_states)
+    watchdog_task.start()
+
+    assert not watchdog_task.expired
+    assert watchdog_task.timeout == 0.8
+    assert (
+        watchdog_task.expiration_states[sim_9263_device.ao_physical_chans[1].name].ao_state == 0.0
+    )
+    assert (
+        watchdog_task.expiration_states[sim_9263_device.ao_physical_chans[1].name].ao_output_type
+        == WatchdogAOExpirState.VOLTAGE
+    )
+
+
+def test___watchdog_task___cfg_watchdog_co_expir_states___no_error(
+    generate_watchdog_task: Task, sim_9189_device: Device, sim_9401_device: Device
+):
+    watchdog_task = generate_watchdog_task(f"{sim_9189_device.name}", timeout=0.8)
+    expir_states = [
+        COExpirationState(
+            physical_channel=sim_9401_device.co_physical_chans[0].name,
+            expiration_state=WatchdogCOExpirState.LOW,
+        ),
+        COExpirationState(
+            physical_channel=sim_9401_device.co_physical_chans[1].name,
+            expiration_state=WatchdogCOExpirState.LOW,
+        ),
+    ]
+
+    watchdog_task.cfg_watchdog_co_expir_states(expir_states)
+    watchdog_task.start()
+
+    assert not watchdog_task.expired
+    assert watchdog_task.timeout == 0.8
+    assert (
+        watchdog_task.expiration_states[sim_9401_device.co_physical_chans[1].name].co_state
+        == WatchdogCOExpirState.LOW
+    )
+
+
+def test___watchdog_task___clear_expiration___no_error(
+    generate_watchdog_task: Task, sim_9189_device: Device, sim_9263_device: Device
+):
+    watchdog_task = generate_watchdog_task(f"{sim_9189_device.name}", timeout=0.8)
+    expir_states = [
+        AOExpirationState(
+            physical_channel=sim_9263_device.ao_physical_chans[0].name,
+            expiration_state=0.0,
+            output_type=WatchdogAOExpirState.VOLTAGE,
+        ),
+        AOExpirationState(
+            physical_channel=sim_9263_device.ao_physical_chans[1].name,
+            expiration_state=0.0,
+            output_type=WatchdogAOExpirState.VOLTAGE,
+        ),
+    ]
+    watchdog_task.cfg_watchdog_ao_expir_states(expir_states)
+    watchdog_task.start()
+
+    watchdog_task.clear_expiration()

tests/conftest.py

@@ -193,21 +193,33 @@ def sim_6363_device(system: nidaqmx.system.System) -> nidaqmx.system.Device:
 
 
 @pytest.fixture(scope="function")
-def sim_9185_device(system: nidaqmx.system.System) -> nidaqmx.system.Device:
+def sim_9189_device(system: nidaqmx.system.System) -> nidaqmx.system.Device:
     """Gets simulated 9185 device information."""
-    return _device_by_product_type("cDAQ-9185", DeviceType.SIMULATED, system)
+    return _device_by_product_type("cDAQ-9189", DeviceType.SIMULATED, system)
 
 
 @pytest.fixture(scope="function")
-def sim_time_aware_9215_device(sim_9185_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
+def sim_time_aware_9215_device(sim_9189_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
     """Gets device information for a simulated 9215 device within a 9185."""
-    return _cdaq_module_by_product_type("NI 9215", sim_9185_device)
+    return _cdaq_module_by_product_type("NI 9215", sim_9189_device)
 
 
 @pytest.fixture(scope="function")
-def sim_9775_device(sim_9185_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
+def sim_9263_device(sim_9189_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
+    """Gets device information for a simulated 9263 device within a 9185."""
+    return _cdaq_module_by_product_type("NI 9263", sim_9189_device)
+
+
+@pytest.fixture(scope="function")
+def sim_9401_device(sim_9189_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
+    """Gets device information for a simulated 9401 device within a 9185."""
+    return _cdaq_module_by_product_type("NI 9401", sim_9189_device)
+
+
+@pytest.fixture(scope="function")
+def sim_9775_device(sim_9189_device: nidaqmx.system.Device) -> nidaqmx.system.Device:
     """Gets device information for a simulated 9775 device within a 9185."""
-    return _cdaq_module_by_product_type("NI 9775", sim_9185_device)
+    return _cdaq_module_by_product_type("NI 9775", sim_9189_device)
 
 
 @pytest.fixture(scope="function")

tests/max_config/linux/nidaqmxMaxConfig.ini

@@ -233,7 +233,7 @@ CompactDAQ.ChassisDevName = tsChassisTester
 CompactDAQ.SlotNum = 3
 
 [DAQmxCDAQChassis cdaqChassisTester]
-ProductType = cDAQ-9185
+ProductType = cDAQ-9189
 DevSerialNum = 0x0
 DevIsSimulated = 1
 BusType = TCP/IP
@@ -256,3 +256,17 @@ DevSerialNum = 0x0
 DevIsSimulated = 1
 CompactDAQ.ChassisDevName = cdaqChassisTester
 CompactDAQ.SlotNum = 2
+
+[DAQmxCDAQModule cdaqTesterMod4]
+ProductType = NI 9401
+DevSerialNum = 0x0
+DevIsSimulated = 1
+CompactDAQ.ChassisDevName = cdaqChassisTester
+CompactDAQ.SlotNum = 4
+
+[DAQmxCDAQModule cdaqTesterMod5]
+ProductType = NI 9263
+DevSerialNum = 0x0
+DevIsSimulated = 1
+CompactDAQ.ChassisDevName = cdaqChassisTester
+CompactDAQ.SlotNum = 5

tests/max_config/nidaqmxMaxConfig.ini

@@ -240,7 +240,7 @@ CompactDAQ.ChassisDevName = tsChassisTester
 CompactDAQ.SlotNum = 3
 
 [DAQmxCDAQChassis cdaqChassisTester]
-ProductType = cDAQ-9185
+ProductType = cDAQ-9189
 DevSerialNum = 0x0
 DevIsSimulated = 1
 BusType = TCP/IP
@@ -270,3 +270,17 @@ DevSerialNum = 0x0
 DevIsSimulated = 1
 CompactDAQ.ChassisDevName = cdaqChassisTester
 CompactDAQ.SlotNum = 3
+
+[DAQmxCDAQModule cdaqTesterMod4]
+ProductType = NI 9401
+DevSerialNum = 0x0
+DevIsSimulated = 1
+CompactDAQ.ChassisDevName = cdaqChassisTester
+CompactDAQ.SlotNum = 4
+
+[DAQmxCDAQModule cdaqTesterMod5]
+ProductType = NI 9263
+DevSerialNum = 0x0
+DevIsSimulated = 1
+CompactDAQ.ChassisDevName = cdaqChassisTester
+CompactDAQ.SlotNum = 5