259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""
|
|
Unit tests for turnupd protocol parser and knob conversion helpers.
|
|
No external dependencies — these are pure-function tests.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from turnup.turnupd import (
|
|
KNOB_MAX,
|
|
NUM_KNOBS,
|
|
VOLUME_MAX,
|
|
handle_knob,
|
|
knob_to_norm,
|
|
knob_to_volume,
|
|
parse_messages,
|
|
)
|
|
|
|
|
|
# ── knob_to_norm ──────────────────────────────────────────────────────────────
|
|
|
|
class TestKnobToNorm:
|
|
def test_min_value(self):
|
|
assert knob_to_norm(0) == pytest.approx(0.0)
|
|
|
|
def test_max_value(self):
|
|
assert knob_to_norm(KNOB_MAX) == pytest.approx(1.0)
|
|
|
|
def test_midpoint(self):
|
|
result = knob_to_norm(KNOB_MAX // 2)
|
|
assert 0.49 < result < 0.51
|
|
|
|
def test_result_is_rounded(self):
|
|
# Result must have at most 4 decimal places.
|
|
result = knob_to_norm(333)
|
|
assert result == round(result, 4)
|
|
|
|
|
|
# ── knob_to_volume ────────────────────────────────────────────────────────────
|
|
|
|
class TestKnobToVolume:
|
|
def test_min_is_zero(self):
|
|
assert knob_to_volume(0) == pytest.approx(0.0)
|
|
|
|
def test_max_is_volume_max(self):
|
|
assert knob_to_volume(KNOB_MAX) == pytest.approx(VOLUME_MAX)
|
|
|
|
def test_proportional(self):
|
|
half = knob_to_volume(KNOB_MAX // 2)
|
|
full = knob_to_volume(KNOB_MAX)
|
|
assert half == pytest.approx(full / 2, rel=0.01)
|
|
|
|
|
|
# ── parse_messages ────────────────────────────────────────────────────────────
|
|
|
|
class TestParseMessages:
|
|
# ── heartbeat ─────────────────────────────────────────────────────────────
|
|
|
|
def test_heartbeat(self):
|
|
buf = bytearray([0xFE, 0x02, 0xFF])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == [{"type": "heartbeat"}]
|
|
assert remainder == bytearray()
|
|
|
|
def test_heartbeat_leaves_trailing_bytes(self):
|
|
# Trailing non-0xFE bytes are consumed (no partial frame to preserve).
|
|
buf = bytearray([0xFE, 0x02, 0xFF, 0x01, 0x02])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert len(msgs) == 1
|
|
assert remainder == bytearray()
|
|
|
|
# ── button ────────────────────────────────────────────────────────────────
|
|
|
|
def test_button_press(self):
|
|
buf = bytearray([0xFE, 0x06, 0x03, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs == [{"type": "button", "action": "press", "id": 3}]
|
|
|
|
def test_button_release(self):
|
|
buf = bytearray([0xFE, 0x07, 0x02, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs == [{"type": "button", "action": "release", "id": 2}]
|
|
|
|
def test_all_button_ids(self):
|
|
for btn_id in range(5):
|
|
buf = bytearray([0xFE, 0x06, btn_id, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs[0]["id"] == btn_id
|
|
|
|
# ── knob ──────────────────────────────────────────────────────────────────
|
|
|
|
def test_knob_message(self):
|
|
# value = 0x03F4 = 1012
|
|
buf = bytearray([0xFE, 0x03, 0x01, 0x03, 0xF4, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs == [{"type": "knob", "id": 1, "value": 1012}]
|
|
|
|
def test_knob_value_zero(self):
|
|
buf = bytearray([0xFE, 0x03, 0x00, 0x00, 0x00, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs[0]["value"] == 0
|
|
|
|
def test_knob_value_max(self):
|
|
buf = bytearray([0xFE, 0x03, 0x04, 0x03, 0xF4, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs[0]["value"] == KNOB_MAX
|
|
|
|
# ── multi-message ─────────────────────────────────────────────────────────
|
|
|
|
def test_multiple_messages_in_one_buffer(self):
|
|
heartbeat = bytearray([0xFE, 0x02, 0xFF])
|
|
button = bytearray([0xFE, 0x06, 0x00, 0xFF])
|
|
knob = bytearray([0xFE, 0x03, 0x02, 0x01, 0xF4, 0xFF])
|
|
msgs, remainder = parse_messages(heartbeat + button + knob)
|
|
assert len(msgs) == 3
|
|
assert msgs[0]["type"] == "heartbeat"
|
|
assert msgs[1]["type"] == "button"
|
|
assert msgs[2]["type"] == "knob"
|
|
assert remainder == bytearray()
|
|
|
|
def test_garbage_bytes_skipped(self):
|
|
buf = bytearray([0x00, 0x11, 0xFE, 0x02, 0xFF])
|
|
msgs, _ = parse_messages(buf)
|
|
assert msgs == [{"type": "heartbeat"}]
|
|
|
|
def test_incomplete_frame_stays_in_remainder(self):
|
|
# A partial knob frame at the end of the buffer must be returned in
|
|
# the remainder so the next serial read can complete it. Previously
|
|
# the parser consumed the leading 0xFE as garbage, silently dropping
|
|
# the message and causing LED colors to miss updates (flicker).
|
|
buf = bytearray([0xFE, 0x03, 0x01, 0x03])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == []
|
|
assert remainder == bytearray([0xFE, 0x03, 0x01, 0x03])
|
|
|
|
def test_partial_frame_completed_on_next_read(self):
|
|
# Simulate a knob message split across two ser.read() calls.
|
|
# First read: 4 of 6 bytes.
|
|
partial = bytearray([0xFE, 0x03, 0x01, 0x03])
|
|
msgs1, remainder1 = parse_messages(partial)
|
|
assert msgs1 == []
|
|
assert remainder1 == partial # preserved, not discarded
|
|
# Second read supplies the remaining 2 bytes.
|
|
msgs2, remainder2 = parse_messages(remainder1 + bytearray([0xF4, 0xFF]))
|
|
assert msgs2 == [{"type": "knob", "id": 1, "value": 1012}]
|
|
assert remainder2 == bytearray()
|
|
|
|
def test_partial_heartbeat_stays_in_remainder(self):
|
|
# Lone 0xFE 0x02 (missing 0xFF terminator) must be preserved.
|
|
buf = bytearray([0xFE, 0x02])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == []
|
|
assert remainder == bytearray([0xFE, 0x02])
|
|
|
|
def test_partial_button_stays_in_remainder(self):
|
|
# FE 06 id — missing 0xFF terminator.
|
|
buf = bytearray([0xFE, 0x06, 0x02])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == []
|
|
assert remainder == bytearray([0xFE, 0x06, 0x02])
|
|
|
|
def test_lone_start_byte_stays_in_remainder(self):
|
|
# A single 0xFE with no following bytes must be preserved.
|
|
buf = bytearray([0xFE])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == []
|
|
assert remainder == bytearray([0xFE])
|
|
|
|
def test_unknown_frame_type_skipped(self):
|
|
# 0xFE followed by an unknown type byte (and enough bytes to decide)
|
|
# should be skipped, not stall the parser.
|
|
buf = bytearray([0xFE, 0x99, 0x00, 0x00, 0x00, 0x00, 0xFE, 0x02, 0xFF])
|
|
msgs, remainder = parse_messages(buf)
|
|
assert msgs == [{"type": "heartbeat"}]
|
|
assert remainder == bytearray()
|
|
|
|
def test_empty_buffer(self):
|
|
msgs, remainder = parse_messages(bytearray())
|
|
assert msgs == []
|
|
assert remainder == bytearray()
|
|
|
|
|
|
# ── handle_knob ───────────────────────────────────────────────────────────────
|
|
|
|
def _make_knob_fixtures(knob_id: int = 0, action: str = "sink_volume"):
|
|
"""Return (config, pulse_mock, ser_mock, knob_norms, last_led_colors, last_knob_event)."""
|
|
config = {
|
|
"knobs": {
|
|
str(knob_id): {
|
|
"action": action,
|
|
"target": "default",
|
|
"led": {"low_color": "#000000", "high_color": "#ffffff"},
|
|
}
|
|
}
|
|
}
|
|
pulse = MagicMock()
|
|
ser = MagicMock()
|
|
knob_norms = [0.0] * NUM_KNOBS
|
|
last_led_colors = [(0, 0, 0)] * NUM_KNOBS
|
|
last_knob_event = [0.0]
|
|
return config, pulse, ser, knob_norms, last_led_colors, last_knob_event
|
|
|
|
|
|
class TestHandleKnobLEDDedup:
|
|
"""Fix 1 — duplicate LED packets must be suppressed."""
|
|
|
|
def test_sends_leds_when_color_changes(self):
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
# Turn knob 0 to max; color should change from (0,0,0).
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
assert ser.write.called
|
|
|
|
def test_skips_leds_when_color_unchanged(self):
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
# First call sets the color.
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
write_count_after_first = ser.write.call_count
|
|
# Second call with same value — color is identical, no new write.
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
assert ser.write.call_count == write_count_after_first
|
|
|
|
def test_last_led_colors_updated_after_send(self):
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
# last_led_colors must not remain all-zero after a successful send.
|
|
assert llc != [(0, 0, 0)] * NUM_KNOBS
|
|
|
|
def test_color_change_after_stable_triggers_send(self):
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
# Reach a stable state at max value.
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
count_stable = ser.write.call_count
|
|
# Now change to zero — should trigger another send.
|
|
handle_knob(0, 0, config, pulse, ser, knob_norms, llc, lke)
|
|
assert ser.write.call_count == count_stable + 1
|
|
|
|
|
|
class TestHandleKnobLastKnobEvent:
|
|
"""Fix 2 — last_knob_event[0] must be updated on every call."""
|
|
|
|
def test_last_knob_event_updated(self):
|
|
import time
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
before = time.monotonic()
|
|
handle_knob(0, 500, config, pulse, ser, knob_norms, llc, lke)
|
|
after = time.monotonic()
|
|
assert before <= lke[0] <= after
|
|
|
|
def test_last_knob_event_updated_even_when_led_unchanged(self):
|
|
import time
|
|
config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures()
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
first_ts = lke[0]
|
|
import time as _t; _t.sleep(0.01)
|
|
handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke)
|
|
# Timestamp must advance even though color did not change.
|
|
assert lke[0] > first_ts
|