Added flash checks

__make.d__/test.mk

@@ -9,7 +9,7 @@ release:
 	cp -v $(OUTDIR)/coverage.xml ../smart_brain/_testresults_
 
 clean_coverage:
-	rm -v testresults/coverage.xml
+	rm -vf testresults/coverage.xml
 
 single: test_single pdf view clean
 	@echo FINISHED...

simulation/devices.py

@@ -138,8 +138,19 @@ class shelly(base):
     def __init__(self, mqtt_client, topic):
         super().__init__(mqtt_client, topic, default_values={self.KEY_OUTPUT_0: False, self.KEY_OUTPUT_1: False,
                                                              self.KEY_INPUT_0: False, self.KEY_INPUT_1: False,
+                                                             self.KEY_INPUT_0: False, self.KEY_INPUT_1: False,
+                                                             self.KEY_LONGPUSH_0: False, self.KEY_LONGPUSH_1: False,
                                                              self.KEY_TEMPERATURE: 35.2, self.KEY_OVERTEMPERATURE: False})
         #
+        self.ch_names = {
+            self.KEY_INPUT_0: "in0",
+            self.KEY_INPUT_1: "in1",
+            self.KEY_LONGPUSH_0: "long0",
+            self.KEY_LONGPUSH_1: "long1",
+            self.KEY_OUTPUT_0: "out0",
+            self.KEY_OUTPUT_1: "out1",
+        }
+        #
         self.__input_0_func = None
         self.__input_1_func = None
         self.__output_0_auto_off__ = None
@@ -149,6 +160,10 @@ class shelly(base):
         # publish state changes
         self.add_callback(self.KEY_OUTPUT_0, None, self.__send__, True)
         self.add_callback(self.KEY_OUTPUT_1, None, self.__send__, True)
+        self.add_callback(self.KEY_INPUT_0, None, self.__send__, True)
+        self.add_callback(self.KEY_INPUT_1, None, self.__send__, True)
+        self.add_callback(self.KEY_LONGPUSH_0, None, self.__send__, True)
+        self.add_callback(self.KEY_LONGPUSH_1, None, self.__send__, True)
         self.add_callback(self.KEY_TEMPERATURE, None, self.__send__, True)
         self.add_callback(self.KEY_OVERTEMPERATURE, None, self.__send__, True)
         #
@@ -159,7 +174,7 @@ class shelly(base):
         self.add_callback(self.KEY_INPUT_0, self.__input_function__, None)
         self.add_callback(self.KEY_INPUT_1, self.__input_function__, None)
         #
-        self.__tx__((self.KEY_OUTPUT_0, self.KEY_OUTPUT_1))
+        self.__tx__(self.keys())    # Send state of all values
 
     def configure(self, input_0_func=None, input_1_func=None, output_0_auto_off=None):
         self.__input_0_func = input_0_func
@@ -169,6 +184,8 @@ class shelly(base):
     def __int_to_ext__(self, key, value):
         if key == self.KEY_OVERTEMPERATURE:
             return int(value)
+        elif key in self.BOOL_KEYS:
+            return b'1' if value is True else b'0'
         return super().__int_to_ext__(key, value)
 
     def __rx__(self, client, userdata, message):

smart_brain_test.py

@@ -13,11 +13,9 @@ import tests.help
 
 
 # TODO: Extend tests in simulation
-#         - Add cross room functions like follow input_1 -> floor light
+#         - Compare with nodered capabilities + add garden
-#         - Compare with nodered capabilities
+#         - Complete inter room functions
 #         - cleanup (e.g. bedlight dirk, christmas star, ...)
-#         - circ pump on -> flash
-#         - Longpress -> flash
 #         - All off
 #         - circulation pump -> timer
 #         - stairway on -> timer

tests/__init__.py

@@ -1,4 +1,4 @@
-from .rooms import preconditions, ffe, ffw, gfe, gfw, stw
+from .rooms import preconditions, ffe, ffw, gfe, gfw, stw, inter_room
 
 # TODO: Coverage, *Information (Chapter 1), ... ausblenden, wenn leer
 
@@ -10,4 +10,5 @@ def add_all_testcases(ts, mc, h):
     gfe(ts, mc, h)
     gfw(ts, mc, h)
     stw(ts, mc, h)
+    inter_room(ts, mc, h)
     # TODO: GARDEN

tests/common_testcases.py

@@ -2,21 +2,22 @@ import time
 
 from unittest.test import equivalency_chk
 
-from .help import DELAY_SET_GET
+from .help import DELAY_SET_GET, DELAY_FLASH, set_precondition
+
+FLASH_TRIGGER_LONGPUSH = 'longpush'
+FLASH_TRIGGER_ACTIVATE = 'activate'
 
 
 def device_follow(tLogger, master, master_key, slave, slave_key, values, switch_on):
+    prepare_list = []
     # Prepare System
     if switch_on is not None:
-        switch_on.set(switch_on.KEY_OUTPUT_0, True)
+        prepare_list.append((switch_on, switch_on.KEY_OUTPUT_0, True, True))
-        time.sleep(DELAY_SET_GET)
-        tLogger.debug("Prepare: Switching on device")
-    # Prepare devices to last state
     start_state = values[-1]
-    master.set(master_key, start_state)
+    prepare_list.append((master, master_key, start_state, False))
-    slave.set(slave_key, start_state)
+    prepare_list.append((slave, slave_key, start_state, False))
-    time.sleep(DELAY_SET_GET)
+    set_precondition(tLogger, prepare_list)
-    tLogger.debug("Prepare: Setting devices to last state %s", repr(start_state))
+    #
     master_exp = master.get(master_key)
     slave_exp = slave.get(slave_key)
     equivalency_chk((master_exp, slave_exp), (start_state, start_state), tLogger, "Start state (master, slave)")
@@ -28,3 +29,33 @@ def device_follow(tLogger, master, master_key, slave, slave_key, values, switch_
         tLogger.debug("Setting state of %s to %s", master.get_name(master_key, master.topic), repr(value))
         result = slave.get(slave_key)
         equivalency_chk(result, value, tLogger, f"Value for {slave.get_name(slave_key, slave.topic)}")
+
+
+def device_flash(tLogger, device, device_flash_key, trigger_device, trigger_device_key, start_state):
+    #
+    # Precondition
+    #
+    time.sleep(DELAY_SET_GET + DELAY_FLASH)     # Wait for possibly flashing dvice to finish
+    set_precondition(tLogger, ((trigger_device, trigger_device_key, False, False), (device, device_flash_key, start_state, False), ))
+
+    #
+    # Trigger
+    #
+    trigger_device.set(trigger_device_key, True)
+
+    #
+    # Test
+    #
+    time.sleep(DELAY_SET_GET)
+    #
+    tLogger.debug("Waiting for first state change after %.1fs.", DELAY_SET_GET)
+    equivalency_chk(not start_state, device.get(device_flash_key), tLogger, f"Value for {device.get_name(device_flash_key, device.topic)}")
+    #
+    time.sleep(DELAY_FLASH)
+    tLogger.debug("Waiting for second state change after %.1fs.", DELAY_FLASH)
+    equivalency_chk(start_state, device.get(device_flash_key), tLogger, f"Value for {device.get_name(device_flash_key, device.topic)}")
+
+    # Reset Trigger
+    trigger_device.set(trigger_device_key, False)
+    time.sleep(DELAY_SET_GET)
+    tLogger.debug("Collect finalise logs.")

tests/help.py

@@ -2,21 +2,41 @@ import distro
 import getpass
 import jinja2
 import json
+import logging
 import os
 import platform
 import sys
+import time
 
 import report
 from report import testSession as testSessionBase
 from unittest import jsonlog
+from unittest.test import equivalency_chk
 
 from config import APP_NAME as ROOT_LOGGER_NAME
+logger = logging.getLogger(ROOT_LOGGER_NAME).getChild(__name__)
 
 DELAY_SET_GET = 0.15
+DELAY_FLASH = 0.75
 STATES_SW = (True, False)
-STATES_BR = list(range(0, 101, 20))
+STATES_BR = (0, 20, 40, 60, 80, 100)
-STATES_CT = list(range(0, 11, 2))
+STATES_CT = (0, 2, 4, 6, 8, 10)
-STATES_TEMP = list(range(15, 31, 5))
+STATES_TEMP = (15., 20., 25., 30.)
+
+
+def set_precondition(tLogger, list_of_device_key_value_wait_pairs):
+    expectation = []
+    for device, key, value, wait in list_of_device_key_value_wait_pairs:
+        expectation.append(value)
+        logger.info("Setting %s to %s", repr(device.get_name(key)), repr(value))
+        device.set(key, value)
+        if wait is True:
+            time.sleep(DELAY_SET_GET)
+    time.sleep(DELAY_SET_GET)
+    result = []
+    for device, key, value, wait in list_of_device_key_value_wait_pairs:
+        result.append(device.get(key))
+    equivalency_chk(result, expectation, tLogger, "Test prepare state")
 
 
 class testSession(testSessionBase):

tests/rooms.py

@@ -8,17 +8,14 @@ from simulation.rooms import house
 
 from .help import testSession
 from .help import STATES_SW, STATES_BR, STATES_CT, STATES_TEMP
-from tests.common_testcases import device_follow
+from tests.common_testcases import device_follow, device_flash, FLASH_TRIGGER_LONGPUSH, FLASH_TRIGGER_ACTIVATE
 
 
 def rand_tcel():
     return random.choice(5 * [report.TCEL_FULL] + 3 * [report.TCEL_SHORT] + 2 * [report.TCEL_SMOKE])
 
 
-def device_follow_sw(ts, master, slave, master_ch=0, slave_ch=0, single=False):
+def device_follow_sw_with_keys(ts, master, master_key, slave, slave_key, single=False):
-    master_key = getattr(master, "KEY_OUTPUT_%d" % master_ch)
-    slave_key = getattr(slave, "KEY_OUTPUT_%d" % slave_ch)
-    #
     master_name = master.get_name(master_key)
     slave_name = slave.get_name(slave_key)
     #
@@ -31,6 +28,10 @@ def device_follow_sw(ts, master, slave, master_ch=0, slave_ch=0, single=False):
     )
 
 
+def device_follow_sw(ts, master, slave, single=False):
+    device_follow_sw_with_keys(ts, master, master.KEY_OUTPUT_0, slave, slave.KEY_OUTPUT_0)
+
+
 def device_follow_br(ts, master, slave, switch_on=None, single=False):
     master_name = master.get_name(master.KEY_BRIGHTNESS)
     slave_name = slave.get_name(slave.KEY_BRIGHTNESS)
@@ -70,6 +71,25 @@ def device_follow_temp(ts, master, slave, single=False):
     )
 
 
+def device_flash_by_vers(ts, flash_device, trigger_device, start_state, trigger_vers, single=True):
+    flash_device_flash_key = flash_device.KEY_OUTPUT_0
+    if trigger_vers == FLASH_TRIGGER_LONGPUSH:
+        trigger_device_key = trigger_device.KEY_LONGPUSH_0
+    elif trigger_vers == FLASH_TRIGGER_ACTIVATE:
+        trigger_device_key = trigger_device.KEY_OUTPUT_0
+    else:
+        raise AttributeError("Unknown trigger_vers in device_flash: %s", repr(trigger_vers))
+
+    trigger_name = trigger_device.get_name(trigger_device_key)
+    flash_name = flash_device.get_name(flash_device_flash_key) + " from %s" % repr(start_state)
+    #
+    testcase_id = trigger_name + " -> " + flash_name
+    ts.testCase(
+        testcase_id, report.TCEL_SINGLE if single else report.TCEL_SHORT, device_flash,
+        flash_device, flash_device_flash_key, trigger_device, trigger_device_key, start_state,
+    )
+
+
 def preconditions(ts: testSession, mc: mqtt_client, h: house):
     def precond_test(tLogger):
         mc.send("videv/all/oof", True)
@@ -162,6 +182,9 @@ def ffe(ts: testSession, mc: mqtt_client, h: house):
     # circulation_pump
     device_follow_sw(ts, room.videv_circulation_pump, room.switch_circulation_pump)
     device_follow_sw(ts, room.switch_circulation_pump, room.videv_circulation_pump)
+    # circulation pump feedback (flash test)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_circulation_pump, True, FLASH_TRIGGER_ACTIVATE, single=True)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_circulation_pump, False, FLASH_TRIGGER_ACTIVATE, single=True)
     # Brightness and Colortemperature ####################
     # TODO: device_follow_br(ts, room.videv_main_light, room.light_main_light, room.switch_main_light)
     # TODO: device_follow_br(ts, room.light_main_light, room.videv_main_light, room.switch_main_light)
@@ -178,6 +201,9 @@ def ffe(ts: testSession, mc: mqtt_client, h: house):
     # main_light videv<-->shelly
     device_follow_sw(ts, room.videv_main_light, room.switch_main_light)
     device_follow_sw(ts, room.switch_main_light, room.videv_main_light)
+    # area all_off (flash test)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, True, FLASH_TRIGGER_LONGPUSH)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, False, FLASH_TRIGGER_LONGPUSH)
 
 
 def ffw(ts: testSession, mc: mqtt_client, h: house):
@@ -247,6 +273,9 @@ def ffw(ts: testSession, mc: mqtt_client, h: house):
     # main_light videv<-->shelly
     device_follow_sw(ts, room.videv_main_light, room.switch_main_light)
     device_follow_sw(ts, room.switch_main_light, room.videv_main_light)
+    # area all_off (flash test)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, True, FLASH_TRIGGER_LONGPUSH)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, False, FLASH_TRIGGER_LONGPUSH)
 
 
 def gfe(ts: testSession, mc: mqtt_client, h: house):
@@ -274,18 +303,18 @@ def gfw(ts: testSession, mc: mqtt_client, h: house):
     device_follow_sw(ts, room.videv_pc_dock, room.switch_pc_dock)
     device_follow_sw(ts, room.switch_pc_dock, room.videv_pc_dock)
     # powerplug videv <-> channel
-    device_follow_sw(ts, room.videv_amplifier, room.switch_powerplug_4, slave_ch=0)
+    device_follow_sw_with_keys(ts, room.videv_amplifier, room.videv_amplifier.KEY_OUTPUT_0, room.switch_powerplug_4, room.KEY_POWERPLUG_AMPLIFIER)
-    device_follow_sw(ts, room.switch_powerplug_4, room.videv_amplifier, master_ch=0)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_AMPLIFIER, room.videv_amplifier, room.videv_amplifier.KEY_OUTPUT_0)
-    device_follow_sw(ts, room.videv_phono, room.switch_powerplug_4, slave_ch=1)
+    device_follow_sw_with_keys(ts, room.videv_phono, room.videv_phono.KEY_OUTPUT_0, room.switch_powerplug_4, room.KEY_POWERPLUG_PHONO)
-    device_follow_sw(ts, room.switch_powerplug_4, room.videv_phono, master_ch=1)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_PHONO, room.videv_phono, room.videv_phono.KEY_OUTPUT_0)
-    device_follow_sw(ts, room.videv_cd_player, room.switch_powerplug_4, slave_ch=2)
+    device_follow_sw_with_keys(ts, room.videv_cd_player, room.videv_cd_player.KEY_OUTPUT_0, room.switch_powerplug_4, room.KEY_POWERPLUG_CD_PLAYER)
-    device_follow_sw(ts, room.switch_powerplug_4, room.videv_cd_player, master_ch=2)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_CD_PLAYER, room.videv_cd_player, room.videv_cd_player.KEY_OUTPUT_0)
-    device_follow_sw(ts, room.videv_bluetooth, room.switch_powerplug_4, slave_ch=3)
+    device_follow_sw_with_keys(ts, room.videv_bluetooth, room.videv_bluetooth.KEY_OUTPUT_0, room.switch_powerplug_4, room.KEY_POWERPLUG_BT)
-    device_follow_sw(ts, room.switch_powerplug_4, room.videv_bluetooth, master_ch=3)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_BT, room.videv_bluetooth, room.videv_bluetooth.KEY_OUTPUT_0)
     # powerplug follow
-    device_follow_sw(ts, room.switch_powerplug_4, room.switch_powerplug_4, master_ch=1, slave_ch=0)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_BT, room.switch_powerplug_4, room.KEY_POWERPLUG_AMPLIFIER)
-    device_follow_sw(ts, room.switch_powerplug_4, room.switch_powerplug_4, master_ch=2, slave_ch=0)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_CD_PLAYER, room.switch_powerplug_4, room.KEY_POWERPLUG_AMPLIFIER)
-    device_follow_sw(ts, room.switch_powerplug_4, room.switch_powerplug_4, master_ch=3, slave_ch=0)
+    device_follow_sw_with_keys(ts, room.switch_powerplug_4, room.KEY_POWERPLUG_PHONO, room.switch_powerplug_4, room.KEY_POWERPLUG_AMPLIFIER)
     # Brightness and Colortemperature ####################
     device_follow_br(ts, room.videv_main_light, room.light_main_light, room.switch_main_light)
     device_follow_br(ts, room.light_main_light, room.videv_main_light, room.switch_main_light)
@@ -337,3 +366,15 @@ def stw(ts: testSession, mc: mqtt_client, h: house):
     # main_light videv<-->shelly
     device_follow_sw(ts, room.videv_main_light, room.switch_main_light)
     device_follow_sw(ts, room.switch_main_light, room.videv_main_light)
+    # area all_off (flash test)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, True, FLASH_TRIGGER_LONGPUSH)
+    device_flash_by_vers(ts, room.switch_main_light, room.switch_main_light, False, FLASH_TRIGGER_LONGPUSH)
+
+
+def inter_room(ts: testSession, mc: mqtt_client, h: house):
+    #
+    # gfw.dirk.swith
+    #
+    input = h.gfw.dirk.switch_main_light
+    output = h.gfw.floor.switch_main_light
+    device_follow_sw_with_keys(ts, input, input.KEY_INPUT_1, output, output.KEY_OUTPUT_0)