Improve test
This commit is contained in:
axxcat
2023-02-02 22:04:45 +01:00
parent 51268a5465
commit b38e1d23d2
3 changed files with 80 additions and 34 deletions

View File

@@ -103,26 +103,41 @@ class BipBip:
"""
return self._execution_log
def check(self):
"""
Check of the bipbip status
:return:
"""
logger.info("Playing: %s", self.name)
def execute(self):
"""
Execute of the bipbip
:return:
"""
self._execution_log.append(time.time())
logger.info("Playing:")
execution_time = time.time()
# Iterating over all the track info
[logger.info("%s: %s", key, value) for key, value in self.player.get_current_track_info().items()]
logger.info("Execute the bipbip: %s", self.name)
if len(self.execution_log) > 1:
last_time_period = (self.execution_log[-1] - self.execution_log[-2])
if len(self.execution_log) > 0:
last_time_period = (execution_time - self.execution_log[-1])
if last_time_period < self.multi_read_timeout:
self.locked = True
else:
self.locked = False
####################################################
# ## Cancel conditions
####################################################
if self.locked and "-cmd" not in self.action:
logger.warning("Action canceled because of the %ds multi read protection.",
self.multi_read_timeout)
return
####################################################
# ## Execution
####################################################
self._execution_log.append(execution_time)
self.check()
logger.info("Execute the bipbip: %s", self.name)
self.start()
def start(self):
@@ -131,11 +146,5 @@ class BipBip:
: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")
logger.info("ACTION (%s) To be overwritten by player implementation",
self.action)

View File

@@ -4,7 +4,6 @@ Interacting with Sonos using SoCo library
"""
import logging
import time
import soco
from soco.plugins.sharelink import ShareLinkPlugin
@@ -34,7 +33,7 @@ class SoCoMode:
return "ClearQueue" in self.modes
@property
def multi_read_cancel(self): # TODO
def multi_read_cancel(self):
"""
multi_read_cancel
:return:
@@ -42,7 +41,7 @@ class SoCoMode:
return "MultiReadCancel" in self.modes
@property
def multi_read_db_random(self): # TODO
def multi_read_db_random(self):
"""
multi_read_db_random
:return:
@@ -50,7 +49,7 @@ class SoCoMode:
return "MultiReadDbRandom" in self.modes
@property
def multi_read_next(self): # TODO
def multi_read_next(self):
"""
multi_read_next
:return:
@@ -58,7 +57,7 @@ class SoCoMode:
return "MultiReadNext" in self.modes
@property
def multi_read_random(self): # TODO
def multi_read_random(self):
"""
multi_read_random
:return:
@@ -88,7 +87,7 @@ class BipBipSoCo(BipBip):
self.soco_mode = SoCoMode(self.mode)
# TODO create a sonos singleton
# TODO: create a sonos singleton
player_name = "TV"
self.player = soco.discovery.by_name(player_name)
@@ -98,6 +97,15 @@ class BipBipSoCo(BipBip):
logger.critical("No player found with name: %s!", player_name)
return
def check(self):
"""
Check of the bipbip status
:return:
"""
logger.info("Playing: %s", self.name)
# Iterating over all the track info
[logger.info("%s: %s", key, value) for key, value in self.player.get_current_track_info().items()]
def start(self):
"""
Execute of the bipbip
@@ -107,10 +115,6 @@ class BipBipSoCo(BipBip):
####################################################
# ## 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
@@ -120,7 +124,7 @@ class BipBipSoCo(BipBip):
####################################################
if self.player.is_playing_radio:
logger.debug("Stop radio before continue")
# Todo manage the radio kill if actually playing
# TODO: manage the radio kill if actually playing
if self.soco_mode.clear_queue:
self.player.stop()
@@ -143,8 +147,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)
ShareLinkPlugin(self.player).add_share_link_to_queue(spotify_link)
self.player.play()
# Start the queue play from beginning
@@ -152,7 +155,8 @@ class BipBipSoCo(BipBip):
elif self.action == "radio":
# radio action execution
logger.critical('TODO to be implemented') # TODO to be implemented
logger.critical('TODO to be implemented')
# TODO: to be implemented
else:
logger.critical('Action "%s" is not supported by BipBipSoCo!', self.action)

View File

@@ -1,6 +1,10 @@
"""
BipBip
Base class tests
"""
import logging
from time import sleep
from unittest import TestCase
from bipbip import BipBip
@@ -8,7 +12,13 @@ logging.basicConfig(level=logging.INFO)
class TestBipBip(TestCase):
"""
Test of main comportment
"""
def test_parameters(self):
"""
Test parameters config
"""
param_dict = dict(
name='name',
ids='ids',
@@ -29,6 +39,9 @@ class TestBipBip(TestCase):
self.assertEqual("user", bip_bip.user)
def test_parameters_failed(self):
"""
Test missing parameters
"""
param_dict = dict(
name='name',
)
@@ -38,12 +51,32 @@ class TestBipBip(TestCase):
self.assertEqual(None, bip_bip.ids)
def test_execute(self):
"""
Test action execution
"""
param_dict = dict(
name='name',
name='test card',
action='test action',
)
bip_bip = BipBip(param_dict, multi_read_timeout=0.1)
bip_bip.execute()
self.assertEqual(1, len(bip_bip.execution_log))
sleep(0.5)
bip_bip.execute()
self.assertEqual(2, len(bip_bip.execution_log))
def test_execute_canceled(self):
"""
Test cancelled execution
"""
param_dict = dict(
name='test card',
action='test action',
)
bip_bip = BipBip(param_dict)
bip_bip.execute()
self.assertEqual(1, len(bip_bip.execution_log))
bip_bip.execute()
self.assertEqual(2, len(bip_bip.execution_log))
self.assertEqual(1, len(bip_bip.execution_log))