tests: Verify waveform sessions

Author: epage

tests/test_signals.py

@@ -86,3 +86,57 @@ def test_singlepoint_loopback(nixnet_in_interface, nixnet_out_interface):
             actual_signals = list(input_session.signals.read())
             for expected, (_, actual) in zip(expected_signals, actual_signals):
                 assert pytest.approx(expected, rel=1) == actual
+
+
+def generate_ramp(min, max, rate, length):
+    """Generate ramp for test data
+
+    Args:
+        min(float): minimum value
+        max(float): maximum value
+        rate(float): rate data is transmitted (Hz)
+        length(float): Duration of ramp (secs)
+    """
+    samples = int(rate * length)
+    step = (max - min) / samples
+    return [
+        min + step * i
+        for i in range(samples)
+    ]
+
+
+@pytest.mark.integration
+def test_waveform_loopback(nixnet_in_interface, nixnet_out_interface):
+    database_name = 'NIXNET_example'
+    cluster_name = 'CAN_Cluster'
+    signal_names = 'CANEventSignal1'
+
+    with nixnet.SignalInWaveformSession(
+            nixnet_in_interface,
+            database_name,
+            cluster_name,
+            signal_names) as input_session:
+        with nixnet.SignalOutWaveformSession(
+                nixnet_out_interface,
+                database_name,
+                cluster_name,
+                signal_names) as output_session:
+            output_session.signals.resamp_rate = 1000
+            input_session.signals.resamp_rate = output_session.signals.resamp_rate / 2
+            # Start the input session manually to make sure that the first
+            # frame value sent before the initial read will be received.
+            input_session.start()
+
+            signal_ramp = generate_ramp(4, 11, output_session.signal_ramp.resamp_rate, 1)
+            output_session.signals.write([signal_ramp])
+
+            # Wait 1 s and then read the received values.
+            # They should be the same as the ones sent.
+            time.sleep(1)
+
+            t0, dt, waveforms = input_session.signals.read()
+            print(t0)
+            assert pytest.approx(dt, 1 / input_session.signals.resamp_rate)
+            assert len(waveforms) == 1
+            for expected, actual in zip(signal_ramp, waveforms[0]):
+                assert pytest.approx(expected, rel=0.1) == actual