Test MessageSync

can/io/player.py

@@ -8,7 +8,7 @@
 
 from __future__ import absolute_import
 
-import time
+from time import time, sleep
 import logging
 
 from .generic import BaseIOHandler
@@ -73,8 +73,8 @@ class MessageSync(object):
     def __init__(self, messages, timestamps=True, gap=0.0001, skip=60):
         """Creates an new **MessageSync** instance.
 
-        :param messages: An iterable of :class:`can.Message` instances.
-        :param bool timestamps: Use the messages' timestamps.
+        :param Iterable[can.Message] messages: An iterable of :class:`can.Message` instances.
+        :param bool timestamps: Use the messages' timestamps. If False, uses the *gap* parameter as the time between messages.
         :param float gap: Minimum time between sent messages in seconds
         :param float skip: Skip periods of inactivity greater than this (in seconds).
         """
@@ -84,26 +84,25 @@ def __init__(self, messages, timestamps=True, gap=0.0001, skip=60):
         self.skip = skip
 
     def __iter__(self):
-        log.debug("Iterating over messages at real speed")
-
-        playback_start_time = time.time()
+        playback_start_time = time()
         recorded_start_time = None
 
-        for m in self.raw_messages:
-            if recorded_start_time is None:
-                recorded_start_time = m.timestamp
+        for message in self.raw_messages:
 
+            # Work out the correct wait time
             if self.timestamps:
-                # Work out the correct wait time
-                now = time.time()
+                if recorded_start_time is None:
+                    recorded_start_time = message.timestamp
+
+                now = time()
                 current_offset = now - playback_start_time
-                recorded_offset_from_start = m.timestamp - recorded_start_time
-                remaining_gap = recorded_offset_from_start - current_offset
+                recorded_offset_from_start = message.timestamp - recorded_start_time
+                remaining_gap = max(0.0, recorded_offset_from_start - current_offset)
 
                 sleep_period = max(self.gap, min(self.skip, remaining_gap))
             else:
                 sleep_period = self.gap
 
-            time.sleep(sleep_period)
+            sleep(sleep_period)
 
-            yield m
+            yield message

test/logformats_test.py

@@ -294,15 +294,6 @@ def _ensure_fsync(self, io_handler):
             io_handler.file.flush()
             os.fsync(io_handler.file.fileno())
 
-    def assertMessagesEqual(self, messages_1, messages_2):
-        """
-        Checks the order and content of the individual messages.
-        """
-        self.assertEqual(len(messages_1), len(messages_2))
-
-        for message_1, message_2 in zip(messages_1, messages_2):
-            self.assertMessageEqual(message_1, message_2)
-
     def assertIncludesComments(self, filename):
         """
         Ensures that all comments are literally contained in the given file.

test/message_helper.py

@@ -47,3 +47,13 @@ def assertMessageEqual(self, message_1, message_2):
                 print("           message 2: {!r}".format(message_2))
                 self.fail("messages are unequal with allowed timestamp delta {} even when ignoring channels" \
                           .format(self.allowed_timestamp_delta))
+
+    def assertMessagesEqual(self, messages_1, messages_2):
+        """
+        Checks the order and content of the individual messages pairwise.
+        Raises an error if the lengths of the sequences are not equal.
+        """
+        self.assertEqual(len(messages_1), len(messages_2), "the number of messages differs")
+
+        for message_1, message_2 in zip(messages_1, messages_2):
+            self.assertMessageEqual(message_1, message_2)

test/simplecyclic_test.py

@@ -9,6 +9,7 @@
 
 from time import sleep
 import unittest
+import gc
 
 import can
 
@@ -24,34 +25,38 @@ def __init__(self, *args, **kwargs):
 
     @unittest.skipIf(IS_CI, "the timing sensitive behaviour cannot be reproduced reliably on a CI server")
     def test_cycle_time(self):
-        msg = can.Message(is_extended_id=False, arbitration_id=0x123, data=[0,1,2,3,4,5,6,7])
-        bus1 = can.interface.Bus(bustype='virtual')
-        bus2 = can.interface.Bus(bustype='virtual')
-        task = bus1.send_periodic(msg, 0.01, 1)
-        self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC)
-
-        sleep(2)
-        size = bus2.queue.qsize()
-        # About 100 messages should have been transmitted
-        self.assertTrue(80 <= size <= 120,
-                        '100 +/- 20 messages should have been transmitted. But queue contained {}'.format(size))
-        last_msg = bus2.recv()
-        next_last_msg = bus2.recv()
-        # Check consecutive messages are spaced properly in time and have
-        # the same id/data
-        self.assertMessageEqual(last_msg, next_last_msg)
-
-
-        # Check the message id/data sent is the same as message received
-        # Set timestamp and channel to match recv'd because we don't care
-        # and they are not initialized by the can.Message constructor.
-        msg.timestamp = last_msg.timestamp
-        msg.channel = last_msg.channel
-        self.assertMessageEqual(msg, last_msg)
-
-        bus1.shutdown()
-        bus2.shutdown()
-
+        msg = can.Message(is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7])
+
+        with can.interface.Bus(bustype='virtual') as bus1:
+            with can.interface.Bus(bustype='virtual') as bus2:
+
+                # disabling the garbage collector makes the time readings more reliable
+                gc.disable()
+
+                task = bus1.send_periodic(msg, 0.01, 1)
+                self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC)
+
+                sleep(2)
+                size = bus2.queue.qsize()
+                # About 100 messages should have been transmitted
+                self.assertTrue(80 <= size <= 120,
+                                '100 +/- 20 messages should have been transmitted. But queue contained {}'.format(size))
+                last_msg = bus2.recv()
+                next_last_msg = bus2.recv()
+
+                # we need to reenable the garbage collector again
+                gc.enable()
+
+                # Check consecutive messages are spaced properly in time and have
+                # the same id/data
+                self.assertMessageEqual(last_msg, next_last_msg)
+
+                # Check the message id/data sent is the same as message received
+                # Set timestamp and channel to match recv'd because we don't care
+                # and they are not initialized by the can.Message constructor.
+                msg.timestamp = last_msg.timestamp
+                msg.channel = last_msg.channel
+                self.assertMessageEqual(msg, last_msg)
 
     def test_removing_bus_tasks(self):
         bus = can.interface.Bus(bustype='virtual')

test/test_message_sync.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+"""
+This module tests :class:`can.MessageSync`.
+"""
+
+from __future__ import absolute_import
+
+from copy import copy
+from time import time
+import gc
+
+import unittest
+import pytest
+
+from can import MessageSync, Message
+
+from .config import IS_CI, IS_APPVEYOR, IS_TRAVIS, IS_OSX
+from .message_helper import ComparingMessagesTestCase
+from .data.example_data import TEST_MESSAGES_BASE
+
+
+TEST_FEWER_MESSAGES = TEST_MESSAGES_BASE[::2]
+
+
+def inc(value):
+    """Makes the test boundaries give some more space when run on the CI server."""
+    if IS_CI:
+        return value * 1.5
+    else:
+        return value
+
+
+@unittest.skipIf(IS_APPVEYOR or (IS_TRAVIS and IS_OSX),
+                 "this environment's timings are too unpredictable")
+class TestMessageSync(unittest.TestCase, ComparingMessagesTestCase):
+
+    def __init__(self, *args, **kwargs):
+        unittest.TestCase.__init__(self, *args, **kwargs)
+        ComparingMessagesTestCase.__init__(self)
+
+    def setup_method(self, _):
+        # disabling the garbage collector makes the time readings more reliable
+        gc.disable()
+
+    def teardown_method(self, _):
+        # we need to reenable the garbage collector again
+        gc.enable()
+
+    @pytest.mark.timeout(inc(0.2))
+    def test_general(self):
+        messages = [
+            Message(timestamp=50.0),
+            Message(timestamp=50.0),
+            Message(timestamp=50.0 + 0.05),
+            Message(timestamp=50.0 + 0.05 + 0.08),
+            Message(timestamp=50.0) # back in time
+        ]
+        sync = MessageSync(messages, gap=0.0)
+
+        start = time()
+        collected = []
+        timings = []
+        for message in sync:
+            collected.append(message)
+            now = time()
+            timings.append(now - start)
+            start = now
+
+        self.assertMessagesEqual(messages, collected)
+        self.assertEqual(len(timings), len(messages), "programming error in test code")
+
+        self.assertTrue(0.0 <= timings[0] < inc(0.005), str(timings[0]))
+        self.assertTrue(0.0 <= timings[1] < inc(0.005), str(timings[1]))
+        self.assertTrue(0.045 <= timings[2] < inc(0.055), str(timings[2]))
+        self.assertTrue(0.075 <= timings[3] < inc(0.085), str(timings[3]))
+        self.assertTrue(0.0 <= timings[4] < inc(0.005), str(timings[4]))
+
+    @pytest.mark.timeout(inc(0.1) * len(TEST_FEWER_MESSAGES)) # very conservative
+    def test_skip(self):
+        messages = copy(TEST_FEWER_MESSAGES)
+        sync = MessageSync(messages, skip=0.005, gap=0.0)
+
+        before = time()
+        collected = list(sync)
+        after = time()
+        took = after - before
+
+        # the handling of the messages itself also takes some time:
+        # ~0.001 s/message on a ThinkPad T560 laptop (Ubuntu 18.04, i5-6200U)
+        assert 0 < took < inc(len(messages) * (0.005 + 0.003)), "took: {}s".format(took)
+
+        self.assertMessagesEqual(messages, collected)
+
+
+if not IS_APPVEYOR: # this environment's timings are too unpredictable
+
+    @pytest.mark.timeout(inc(0.3))
+    @pytest.mark.parametrize("timestamp_1,timestamp_2", [
+        (0.0, 0.0),
+        (0.0, 0.01),
+        (0.01, 0.0),
+    ])
+    def test_gap(timestamp_1, timestamp_2):
+        """This method is alone so it can be parameterized."""
+        messages = [
+            Message(arbitration_id=0x1, timestamp=timestamp_1),
+            Message(arbitration_id=0x2, timestamp=timestamp_2)
+        ]
+        sync = MessageSync(messages, gap=0.1)
+
+        gc.disable()
+        before = time()
+        collected = list(sync)
+        after = time()
+        gc.enable()
+        took = after - before
+
+        assert 0.1 <= took < inc(0.3)
+        assert messages == collected
+
+
+if __name__ == '__main__':
+    unittest.main()