test: add tests and refactor wecker.py state machine for testability
This commit is contained in:
@@ -17,5 +17,6 @@ dev = [
|
||||
"httpx>=0.28.1",
|
||||
"pytest>=9.0.3",
|
||||
"pytest-asyncio>=1.3.0",
|
||||
"pytest-mock>=3.15.1",
|
||||
"ruff>=0.15.12",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import pytest
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Mock RPi.GPIO and pygame before importing wecker
|
||||
sys.modules['RPi'] = MagicMock()
|
||||
sys.modules['RPi.GPIO'] = MagicMock()
|
||||
sys.modules['pygame'] = MagicMock()
|
||||
|
||||
import wecker # noqa: E402
|
||||
|
||||
@pytest.fixture
|
||||
def mock_gpio():
|
||||
with patch.object(wecker, 'GPIO') as mock:
|
||||
mock.LOW = 0
|
||||
mock.HIGH = 1
|
||||
yield mock
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pygame():
|
||||
with patch.object(wecker, 'pygame') as mock:
|
||||
yield mock
|
||||
|
||||
def test_set_led(mock_gpio):
|
||||
wecker.set_led(True)
|
||||
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.LOW)
|
||||
|
||||
wecker.set_led(False)
|
||||
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
|
||||
|
||||
@patch('wecker.time.sleep')
|
||||
def test_blink_led(mock_sleep, mock_gpio):
|
||||
wecker.blink_led(2)
|
||||
# 2 blinks = 4 sleep calls, 2 set_led(True), 2 set_led(False)
|
||||
assert mock_sleep.call_count == 4
|
||||
# GPIO output called 4 times total (on, off, on, off)
|
||||
assert mock_gpio.output.call_count == 4
|
||||
|
||||
@patch('wecker.time.time')
|
||||
def test_run_alarm_start_to_wait(mock_time, mock_gpio, mock_pygame):
|
||||
# Mock pygame.mixer.get_init() to return True so music plays
|
||||
mock_pygame.mixer.get_init.return_value = True
|
||||
|
||||
# Just run it in test mode, it should execute the loop once and exit
|
||||
wecker.run_alarm(test_mode=True)
|
||||
assert mock_pygame.mixer.music.play.called
|
||||
|
||||
def test_state_machine_evaluation(mock_gpio, mock_pygame):
|
||||
clock = wecker.AlarmClock()
|
||||
|
||||
# Transition to ringing -> wait before blink
|
||||
clock.update(100.0, True) # button press
|
||||
assert clock.state == wecker.STATE_WAIT_BEFORE_BLINK
|
||||
|
||||
# Wait 3 seconds -> blinking
|
||||
with patch('wecker.blink_led'):
|
||||
clock.update(103.1, False)
|
||||
assert clock.state == wecker.STATE_BLINKING
|
||||
|
||||
with patch('wecker.blink_led'):
|
||||
clock.update(103.2, False)
|
||||
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
|
||||
assert clock.target_blinks >= 1
|
||||
|
||||
# Set the user presses to be correct
|
||||
clock.target_blinks = 3
|
||||
|
||||
# Press 1
|
||||
clock.update(104.0, True)
|
||||
clock.update(104.1, False)
|
||||
# Press 2
|
||||
clock.update(104.5, True)
|
||||
clock.update(104.6, False)
|
||||
# Press 3
|
||||
clock.update(105.0, True)
|
||||
clock.update(105.1, False)
|
||||
|
||||
assert clock.user_presses == 3
|
||||
|
||||
# Wait 3 seconds to evaluate
|
||||
clock.update(108.2, False) # Triggers state change
|
||||
keep_running = clock.update(108.3, False) # Triggers evaluation
|
||||
|
||||
# It should evaluate, see it's correct, and return False (stop running)
|
||||
assert clock.state == wecker.STATE_EVALUATING
|
||||
assert not keep_running
|
||||
|
||||
def test_state_machine_incorrect(mock_gpio, mock_pygame):
|
||||
clock = wecker.AlarmClock()
|
||||
clock.state = wecker.STATE_WAIT_FOR_INPUT
|
||||
clock.target_blinks = 3
|
||||
|
||||
# Only press once
|
||||
clock.update(100.0, True)
|
||||
clock.update(100.1, False)
|
||||
|
||||
# Wait to evaluate
|
||||
clock.last_interaction_time = 100.1
|
||||
clock.update(103.2, False) # triggers eval state
|
||||
keep_running = clock.update(103.3, False) # evals to incorrect
|
||||
|
||||
# Evaluated incorrectly, should wait before retry
|
||||
assert keep_running
|
||||
assert clock.state == wecker.STATE_WAIT_BEFORE_RETRY
|
||||
|
||||
@@ -280,6 +280,18 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest-mock"
|
||||
version = "3.15.1"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "pytest" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "python-crontab"
|
||||
version = "3.3.0"
|
||||
@@ -423,6 +435,7 @@ dev = [
|
||||
{ name = "httpx" },
|
||||
{ name = "pytest" },
|
||||
{ name = "pytest-asyncio" },
|
||||
{ name = "pytest-mock" },
|
||||
{ name = "ruff" },
|
||||
]
|
||||
|
||||
@@ -440,5 +453,6 @@ dev = [
|
||||
{ name = "httpx", specifier = ">=0.28.1" },
|
||||
{ name = "pytest", specifier = ">=9.0.3" },
|
||||
{ name = "pytest-asyncio", specifier = ">=1.3.0" },
|
||||
{ name = "pytest-mock", specifier = ">=3.15.1" },
|
||||
{ name = "ruff", specifier = ">=0.15.12" },
|
||||
]
|
||||
|
||||
@@ -32,6 +32,8 @@ try:
|
||||
pygame.mixer.music.set_volume(1.0)
|
||||
except Exception as e:
|
||||
logging.error(f"Error loading audio file: {e}")
|
||||
# Don't exit in test mode
|
||||
if 'pytest' not in sys.modules:
|
||||
sys.exit(1)
|
||||
|
||||
# State Machine states for the flow
|
||||
@@ -50,6 +52,7 @@ button_was_pressed = False
|
||||
|
||||
def set_led(on):
|
||||
"""Turns the LED on or off (LOW = ON, HIGH = OFF)"""
|
||||
if 'GPIO' in globals() and hasattr(GPIO, 'output'):
|
||||
if on:
|
||||
GPIO.output(LED_PIN, GPIO.LOW)
|
||||
else:
|
||||
@@ -63,89 +66,112 @@ def blink_led(times):
|
||||
set_led(False)
|
||||
time.sleep(0.3) # LED off for 300ms
|
||||
|
||||
class AlarmClock:
|
||||
def __init__(self):
|
||||
self.state = STATE_RINGING
|
||||
self.target_blinks = 0
|
||||
self.user_presses = 0
|
||||
self.last_interaction_time = 0
|
||||
self.button_was_pressed = False
|
||||
|
||||
def update(self, now, is_pressed):
|
||||
button_just_pressed = False
|
||||
if is_pressed and not self.button_was_pressed:
|
||||
self.button_was_pressed = True
|
||||
button_just_pressed = True
|
||||
elif not is_pressed and self.button_was_pressed:
|
||||
self.button_was_pressed = False
|
||||
|
||||
if self.state == STATE_RINGING:
|
||||
if button_just_pressed:
|
||||
logging.info("Alarm button pressed! Puzzle started. Waiting 3 seconds...")
|
||||
self.state = STATE_WAIT_BEFORE_BLINK
|
||||
self.last_interaction_time = now
|
||||
|
||||
elif self.state == STATE_WAIT_BEFORE_BLINK:
|
||||
if now - self.last_interaction_time >= 3.0:
|
||||
self.target_blinks = random.randint(1, 7)
|
||||
logging.info(f"Blinking {self.target_blinks} times...")
|
||||
self.state = STATE_BLINKING
|
||||
|
||||
elif self.state == STATE_BLINKING:
|
||||
# Move the blink_led out of the update loop for testability,
|
||||
# or just call it directly. Here we call it.
|
||||
blink_led(self.target_blinks)
|
||||
logging.info("Blinking finished. Waiting for input...")
|
||||
self.state = STATE_WAIT_FOR_INPUT
|
||||
self.user_presses = 0
|
||||
self.last_interaction_time = now # Use the current update time!
|
||||
if 'GPIO' in globals() and hasattr(GPIO, 'input'):
|
||||
self.button_was_pressed = (GPIO.input(BUTTON_PIN) == GPIO.LOW)
|
||||
|
||||
|
||||
elif self.state == STATE_WAIT_FOR_INPUT:
|
||||
set_led(is_pressed)
|
||||
|
||||
if button_just_pressed:
|
||||
self.user_presses += 1
|
||||
logging.info(f"Button pressed: {self.user_presses} times")
|
||||
|
||||
if is_pressed:
|
||||
self.last_interaction_time = now
|
||||
|
||||
if not is_pressed and (now - self.last_interaction_time >= 3.0):
|
||||
self.state = STATE_EVALUATING
|
||||
|
||||
elif self.state == STATE_EVALUATING:
|
||||
logging.info(f"Evaluation: Target={self.target_blinks}, Entered={self.user_presses}")
|
||||
if self.user_presses == self.target_blinks:
|
||||
logging.info("Puzzle solved correctly! Alarm clock is stopping.")
|
||||
if 'pygame' in globals() and hasattr(pygame, 'mixer') and pygame.mixer.get_init():
|
||||
pygame.mixer.music.stop()
|
||||
set_led(False)
|
||||
return False # Indicate we should stop running
|
||||
else:
|
||||
logging.info("Incorrect input! Waiting 3 seconds before retrying...")
|
||||
self.state = STATE_WAIT_BEFORE_RETRY
|
||||
self.last_interaction_time = time.time()
|
||||
set_led(False)
|
||||
|
||||
elif self.state == STATE_WAIT_BEFORE_RETRY:
|
||||
if now - self.last_interaction_time >= 5.0:
|
||||
self.target_blinks = random.randint(1, 7)
|
||||
logging.info(f"New attempt! Blinking {self.target_blinks} times...")
|
||||
self.state = STATE_BLINKING
|
||||
|
||||
return True # Keep running
|
||||
|
||||
def run_alarm(test_mode=False):
|
||||
logging.info("Alarm clock started. Music is playing in an endless loop.")
|
||||
# -1 means the song is played in an endless loop
|
||||
if 'pygame' in globals() and hasattr(pygame, 'mixer') and pygame.mixer.get_init():
|
||||
pygame.mixer.music.play(-1)
|
||||
|
||||
set_led(False) # LED off at start
|
||||
|
||||
clock = AlarmClock()
|
||||
|
||||
try:
|
||||
while True:
|
||||
now = time.time()
|
||||
|
||||
# Read button state (LOW = pressed, due to Pull-Up)
|
||||
if 'GPIO' in globals() and hasattr(GPIO, 'input'):
|
||||
is_pressed = (GPIO.input(BUTTON_PIN) == GPIO.LOW)
|
||||
|
||||
button_just_pressed = False
|
||||
if is_pressed and not button_was_pressed:
|
||||
button_was_pressed = True
|
||||
button_just_pressed = True
|
||||
elif not is_pressed and button_was_pressed:
|
||||
button_was_pressed = False
|
||||
|
||||
# --- State Machine ---
|
||||
|
||||
if state == STATE_RINGING:
|
||||
if button_just_pressed:
|
||||
logging.info("Alarm button pressed! Puzzle started. Waiting 3 seconds...")
|
||||
state = STATE_WAIT_BEFORE_BLINK
|
||||
last_interaction_time = now
|
||||
|
||||
elif state == STATE_WAIT_BEFORE_BLINK:
|
||||
if now - last_interaction_time >= 3.0:
|
||||
target_blinks = random.randint(1, 7)
|
||||
logging.info(f"Blinking {target_blinks} times...")
|
||||
state = STATE_BLINKING
|
||||
|
||||
elif state == STATE_BLINKING:
|
||||
blink_led(target_blinks)
|
||||
logging.info("Blinking finished. Waiting for input...")
|
||||
state = STATE_WAIT_FOR_INPUT
|
||||
user_presses = 0
|
||||
last_interaction_time = time.time()
|
||||
# Update the button state after the blocking blink
|
||||
button_was_pressed = (GPIO.input(BUTTON_PIN) == GPIO.LOW)
|
||||
|
||||
elif state == STATE_WAIT_FOR_INPUT:
|
||||
# LED lights up as confirmation when the button is pressed
|
||||
set_led(is_pressed)
|
||||
|
||||
if button_just_pressed:
|
||||
user_presses += 1
|
||||
logging.info(f"Button pressed: {user_presses} times")
|
||||
|
||||
if is_pressed:
|
||||
# As long as the button is held down, we keep resetting the timeout
|
||||
last_interaction_time = now
|
||||
|
||||
# If the button was released and 3 seconds have passed without action -> Evaluation
|
||||
if not is_pressed and (now - last_interaction_time >= 3.0):
|
||||
state = STATE_EVALUATING
|
||||
|
||||
elif state == STATE_EVALUATING:
|
||||
logging.info(f"Evaluation: Target={target_blinks}, Entered={user_presses}")
|
||||
if user_presses == target_blinks:
|
||||
logging.info("Puzzle solved correctly! Alarm clock is stopping.")
|
||||
pygame.mixer.music.stop()
|
||||
set_led(False)
|
||||
break # Exits the while loop and thus the program
|
||||
else:
|
||||
logging.info("Incorrect input! Waiting 3 seconds before retrying...")
|
||||
state = STATE_WAIT_BEFORE_RETRY
|
||||
last_interaction_time = time.time()
|
||||
set_led(False)
|
||||
is_pressed = False
|
||||
|
||||
elif state == STATE_WAIT_BEFORE_RETRY:
|
||||
if now - last_interaction_time >= 5.0:
|
||||
# Puzzle is repeated with a new random blink count
|
||||
target_blinks = random.randint(1, 7)
|
||||
logging.info(f"New attempt! Blinking {target_blinks} times...")
|
||||
state = STATE_BLINKING
|
||||
keep_running = clock.update(now, is_pressed)
|
||||
|
||||
time.sleep(0.02) # Short polling interval to save CPU and ensure good responsiveness (20ms)
|
||||
if not keep_running or test_mode:
|
||||
break
|
||||
|
||||
time.sleep(0.02)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logging.info("Manually aborted (CTRL+C).")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
run_alarm()
|
||||
finally:
|
||||
pygame.mixer.quit()
|
||||
GPIO.cleanup()
|
||||
|
||||
Reference in New Issue
Block a user