From 566017e670d64aaa235423c388046c1f5351c06b Mon Sep 17 00:00:00 2001 From: AxelC Date: Thu, 17 Jun 2021 23:13:03 +0200 Subject: [PATCH] Use mock to test and avoid real sound --- bipbip/bipbip.py | 32 ++++++++++++-- bipbip/bipbip_soco.py | 21 ++++++--- bipbip/test_bipbip_execution.py | 78 +++++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 9 deletions(-) diff --git a/bipbip/bipbip.py b/bipbip/bipbip.py index 664dd1b..9c48b6a 100644 --- a/bipbip/bipbip.py +++ b/bipbip/bipbip.py @@ -12,7 +12,7 @@ class BipBip: The generic bipbip class """ - def __init__(self, parameters: dict): + def __init__(self, parameters: dict, multi_read_timeout: int or float = 3): """ :param parameters: Parameter dict, each BipBip can implement its own parameters Generic parameters are: @@ -23,11 +23,15 @@ class BipBip: action: type of action to be executed data: extra data related to the action user: allowed user for the card + :param multi_read_timeout: (optional) Value to consider a multi read, default 3s """ self.parameters = parameters + self.multi_read_timeout = multi_read_timeout self._execution_log = [] - def _parameter(self, param_name:str) -> str or None: + self.locked = False + + def _parameter(self, param_name: str) -> str or None: try: return self.parameters[param_name] except KeyError: @@ -106,4 +110,26 @@ class BipBip: self._execution_log.append(time.time()) logger.info("Execute the bipbip: %s", self.name) - # TODO manage the multi-read + if len(self.execution_log) > 1: + last_time_period = (self.execution_log[-1] - self.execution_log[-2]) + if last_time_period < self.multi_read_timeout: + self.locked = True + else: + self.locked = False + + self.start() + + def start(self): + """ + Action of the bipbip to be overwritten by the child class + :return: + """ + + #################################################### + # ## Cancel conditions sample + #################################################### + if self.locked: + logger.warning("Action canceled because of the %ds multi read protection.", self.multi_read_timeout) + return + + logger.info("ACTION") diff --git a/bipbip/bipbip_soco.py b/bipbip/bipbip_soco.py index 0350ac0..c668428 100644 --- a/bipbip/bipbip_soco.py +++ b/bipbip/bipbip_soco.py @@ -51,7 +51,7 @@ class BipBipSoCo(BipBip): The SoCo bipbip class """ - def __init__(self, parameters: dict): + def __init__(self, parameters: dict, multi_read_timeout: int or float = 3): """ :param parameters: Parameter dict, each BipBip can implement its own parameters Generic parameters are: @@ -62,14 +62,16 @@ class BipBipSoCo(BipBip): action: type of action to be executed data: extra data related to the action user: allowed user for the card + :param multi_read_timeout: (optional) Value to consider a multi read, default 3s """ - super().__init__(parameters) + super().__init__(parameters, multi_read_timeout) self.soco_mode = SoCoMode(self.mode) # TODO create a sonos singleton player_name = "Hugo" self.player = soco.discovery.by_name(player_name) + self.sharelink = ShareLinkPlugin(self.player) if self.player: logger.info("Player %s is available @ %s", player_name, self.player.ip_address) @@ -77,18 +79,26 @@ class BipBipSoCo(BipBip): logger.critical("No player found with name: %s!", player_name) return - def execute(self): + def start(self): """ Execute of the bipbip :return: """ - super().execute() + #################################################### + # ## Cancel conditions + #################################################### + if self.locked and not self.action == "sonos-cmd": + logger.warning("Action canceled because of the %ds multi read protection.", self.multi_read_timeout) + return if not self.player: logger.critical("No player valid player configured during init, execution aborted!") return + #################################################### + # ####### Action + #################################################### if self.soco_mode.clear_queue: self.player.clear_queue() logger.debug("Clear sonos queue before execution") @@ -109,8 +119,7 @@ class BipBipSoCo(BipBip): spotify_link = f"{self.action}:{self.data}" logger.debug("System will execute on sonos the spotify %s: %s", spotify_mode, self.data) - sharelink = ShareLinkPlugin(self.player) - sharelink.add_share_link_to_queue(spotify_link) + self.sharelink.add_share_link_to_queue(spotify_link) self.player.play() # Start the queue play from beginning diff --git a/bipbip/test_bipbip_execution.py b/bipbip/test_bipbip_execution.py index d6ba15f..bd4b043 100644 --- a/bipbip/test_bipbip_execution.py +++ b/bipbip/test_bipbip_execution.py @@ -1,15 +1,23 @@ import logging +from time import sleep from unittest import TestCase +from unittest.mock import MagicMock from bipbip_soco import BipBipSoCo logging.basicConfig(level=logging.INFO) # logging.config.fileConfig('logging_unittest.yaml') TODO understand logging config +REAL = False + class TestBipBip(TestCase): + @classmethod + def setUpClass(cls): + pass + def test_execute_soco_play(self): param_dict = dict( name='play', @@ -18,8 +26,15 @@ class TestBipBip(TestCase): ) bip_bip = BipBipSoCo(param_dict) + + if not REAL: + bip_bip.player.play = MagicMock() + bip_bip.execute() + if not REAL: + bip_bip.player.play.assert_called() + def test_execute_soco_pause(self): param_dict = dict( name='pause', @@ -70,3 +85,66 @@ class TestBipBip(TestCase): bip_bip = BipBipSoCo(param_dict) bip_bip.execute() + + def test_execute_soco_spotify_multi_read(self): + param_dict = dict( + name='Les petits poissons', + action='spotify:track', + data='35VKLRwEjuR5IuFyGqjMaf', + mode='ClearQueue', + ) + timeout = 1 + + bip_bip = BipBipSoCo(param_dict, multi_read_timeout=timeout) + if not REAL: + bip_bip.player.clear_queue = MagicMock() + bip_bip.sharelink.add_share_link_to_queue = MagicMock() + bip_bip.player.play = MagicMock() + + bip_bip.execute() + + if not REAL: + bip_bip.player.clear_queue.assert_called() + bip_bip.sharelink.add_share_link_to_queue.assert_called_with("spotify:track:35VKLRwEjuR5IuFyGqjMaf") + bip_bip.player.play.assert_called() + + sleep(0.1) + + if not REAL: + bip_bip.player.clear_queue.reset_mock() + bip_bip.sharelink.add_share_link_to_queue.reset_mock() + bip_bip.player.play.reset_mock() + + bip_bip.execute() + if not REAL: + bip_bip.player.clear_queue.assert_not_called() + bip_bip.sharelink.add_share_link_to_queue.assert_not_called() + bip_bip.player.play.assert_not_called() + + sleep(0.1) + + if not REAL: + bip_bip.player.clear_queue.reset_mock() + bip_bip.sharelink.add_share_link_to_queue.reset_mock() + bip_bip.player.play.reset_mock() + + bip_bip.execute() + + if not REAL: + bip_bip.player.clear_queue.assert_not_called() + bip_bip.sharelink.add_share_link_to_queue.assert_not_called() + bip_bip.player.play.assert_not_called() + + sleep(timeout) + + if not REAL: + bip_bip.player.clear_queue.reset_mock() + bip_bip.sharelink.add_share_link_to_queue.reset_mock() + bip_bip.player.play.reset_mock() + + bip_bip.execute() + + if not REAL: + bip_bip.player.clear_queue.assert_called() + bip_bip.sharelink.add_share_link_to_queue.assert_called_with("spotify:track:35VKLRwEjuR5IuFyGqjMaf") + bip_bip.player.play.assert_called()