Implement test timing, to allow testing of how much each test takes.

Can be used to compare timings on different hardware, as well as checking if the timings apply to the ones specified in EEP.
Currently only run with timings, when environment variable `WITH_TIMINGS` is set, as these tests can take some time to run...

run_tests_with_timing.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+WITH_TIMINGS=1 nosetests -s -q

tests/decorators.py

@@ -0,0 +1,38 @@
+# -*- encoding: utf-8 -*-
+from __future__ import print_function, unicode_literals, division
+import time
+import functools
+from os import environ
+
+
+def timing(rounds=1, limit=None):
+    '''
+    Wrapper to implement simple timing of tests.
+    Allows running multiple rounds to calculate average time.
+    Limit (in milliseconds) can be set to assert, if (average) duration is too high.
+    '''
+    def decorator(method):
+        @functools.wraps(method)
+        def f():
+            if rounds == 1:
+                start = time.time()
+                method()
+                duration = time.time() - start
+            else:
+                start = time.time()
+                for i in range(rounds):
+                    method()
+                duration = (time.time() - start) / rounds
+            # Use milliseconds for duration counter
+            duration = duration * 1e3
+
+            print('Test "%s.%s" took %.06f ms.' % (method.__module__, method.__name__, duration))
+            if limit is not None:
+                assert limit > duration, 'Timing failure: %.06f > %.06f' % (duration, limit)
+
+        # Run tests with timings, only if WITH_TIMINGS environment variable is set.
+        # This is because tests with multiple rounds can take long to process.
+        if environ.get('WITH_TIMINGS', None) is '1':
+            return method
+        return f
+    return decorator

tests/test_communicator.py

@@ -4,8 +4,10 @@
 from enocean.communicators.communicator import Communicator
 from enocean.protocol.packet import Packet
 from enocean.protocol.constants import PACKET
+from decorators import timing
 
 
+@timing(100)
 def test_buffer():
     ''' Test buffer parsing for Communicator '''
     data = bytearray([
@@ -25,6 +27,7 @@ def test_buffer():
     assert com.receive.qsize() == 1
 
 
+@timing(100)
 def test_send():
     ''' Test sending packets to Communicator '''
     com = Communicator()

tests/test_eep.py

@@ -3,8 +3,10 @@
 
 from enocean.protocol.packet import Packet
 from enocean.protocol.constants import RORG
+from decorators import timing
 
 
+@timing(100)
 def test_temperature():
     ''' Tests RADIO message for EEP -profile 0xA5 0x02 0x05 '''
     status, buf, packet = Packet.parse_msg(bytearray([
@@ -30,6 +32,7 @@ def test_temperature():
     assert packet.sender_hex == '01:81:B7:44'
 
 
+@timing(100)
 def test_magnetic_switch():
     ''' Tests RADIO message for EEP -profile 0xD5 0x00 0x01 '''
     status, buf, packet = Packet.parse_msg(bytearray([
@@ -62,6 +65,7 @@ def test_magnetic_switch():
     assert packet.repeater_count == 0
 
 
+@timing(100)
 def test_switch():
     status, buf, packet = Packet.parse_msg(bytearray([
         0x55,
@@ -99,6 +103,7 @@ def test_switch():
     assert packet.repeater_count == 0
 
 
+@timing(100)
 def test_eep_parsing():
     status, buf, packet = Packet.parse_msg(bytearray([
         0x55,
@@ -116,6 +121,7 @@ def test_eep_parsing():
     assert packet.repeater_count == 0
 
 
+@timing(100)
 def test_eep_remaining():
     # Magnetic switch -example
     status, buf, packet = Packet.parse_msg(bytearray([
@@ -143,6 +149,7 @@ def test_eep_remaining():
     assert packet.parse_eep(0x02, 0x05) == ['TMP']
 
 
+@timing(100)
 def test_eep_direction():
     status, buf, packet = Packet.parse_msg(bytearray([
         0x55,
@@ -158,6 +165,7 @@ def test_eep_direction():
     assert packet.parsed['SP']['value'] == 50
 
 
+@timing(100)
 def test_vld():
     status, buf, p = Packet.parse_msg(bytearray([
         0x55,

tests/test_packet.py

@@ -3,8 +3,10 @@
 
 from enocean.protocol.packet import Packet
 from enocean.protocol.constants import PACKET, PARSE_RESULT
+from decorators import timing
 
 
+@timing(100)
 def test_packet_examples():
     ''' Tests examples found at EnOceanSerialProtocol3.pdf / 74 '''
     telegram_examples = {
@@ -114,6 +116,7 @@ def test_packet_examples():
         assert pack.repeater_count == 0
 
 
+@timing(100)
 def test_packet_fails():
     '''
     Tests designed to fail.

tests/test_packet_creation.py

@@ -4,8 +4,10 @@
 
 from enocean.protocol.packet import Packet, RadioPacket
 from enocean.protocol.constants import PACKET, RORG
+from decorators import timing
 
 
+@timing(100)
 def test_packet_assembly():
     PACKET_CONTENT_1 = bytearray([
         0x55,
@@ -97,6 +99,7 @@ def test_packet_assembly():
 
 
 # Corresponds to the tests done in test_eep
+@timing(100)
 def test_temperature():
     TEMPERATURE = bytearray([
         0x55,
@@ -127,6 +130,7 @@ def test_temperature():
 
 
 # Corresponds to the tests done in test_eep
+@timing(100)
 def test_magnetic_switch():
     MAGNETIC_SWITCH = bytearray([
         0x55,
@@ -187,6 +191,7 @@ def test_magnetic_switch():
     assert packet.learn is True
 
 
+@timing(100)
 def test_switch():
     SWITCH = bytearray([
         0x55,
@@ -229,17 +234,20 @@ def test_switch():
     assert list(packet_serialized) == list(SWITCH)
 
 
+@timing(100)
 @raises(ValueError)
 def test_illegal_eep_enum1():
     RadioPacket.create(rorg=RORG.RPS, func=0x02, type=0x02, sender=[0x00, 0x29, 0x89, 0x79], EBO='inexisting')
 
 
 @raises(ValueError)
+@timing(100)
 def test_illegal_eep_enum2():
     RadioPacket.create(rorg=RORG.RPS, func=0x02, type=0x02, sender=[0x00, 0x29, 0x89, 0x79], EBO=2)
 
 
 # Corresponds to the tests done in test_eep
+@timing(100)
 def test_packets_with_destination():
     TEMPERATURE = bytearray([
         0x55,

tests/test_teachin.py

@@ -3,9 +3,11 @@
 
 from enocean.protocol.packet import Packet
 from enocean.protocol.constants import RORG
+from decorators import timing
 
 
-def test_ute():
+@timing(rounds=100, limit=0.2)
+def test_ute_in():
     # ['0xd4', '0xa0', '0xff', '0x3e', '0x0', '0x1', '0x1', '0xd2', '0x1', '0x94', '0xe3', '0xb9', '0x0'] ['0x1', '0xff', '0xff', '0xff', '0xff', '0x40', '0x0']
     status, buf, p = Packet.parse_msg(bytearray([
         0x55,