90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
import pytest
|
|
|
|
from handlers.mqtt_handler import TalkingQMQTTService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_desired_volume_republishes_when_reported_volume_differs(monkeypatch):
|
|
service = TalkingQMQTTService({"device_prefix": "TalkingQ"})
|
|
sent_commands = []
|
|
|
|
class FakeDeviceVolumeManager:
|
|
async def get_configured_volume(self, device_id):
|
|
assert device_id == "TalkingQ_XQSN00001001"
|
|
return 70
|
|
|
|
async def fake_send_volume_command(device_id, level):
|
|
sent_commands.append((device_id, level))
|
|
return "002"
|
|
|
|
monkeypatch.setattr("handlers.mqtt_handler.device_volume_manager", FakeDeviceVolumeManager())
|
|
monkeypatch.setattr(service, "send_volume_command", fake_send_volume_command)
|
|
|
|
await service._sync_desired_volume("TalkingQ_XQSN00001001", 30)
|
|
|
|
assert sent_commands == [("TalkingQ_XQSN00001001", 70)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_desired_volume_skips_when_reported_volume_matches(monkeypatch):
|
|
service = TalkingQMQTTService({"device_prefix": "TalkingQ"})
|
|
sent_commands = []
|
|
|
|
class FakeDeviceVolumeManager:
|
|
async def get_configured_volume(self, device_id):
|
|
return 70
|
|
|
|
async def fake_send_volume_command(device_id, level):
|
|
sent_commands.append((device_id, level))
|
|
return "002"
|
|
|
|
monkeypatch.setattr("handlers.mqtt_handler.device_volume_manager", FakeDeviceVolumeManager())
|
|
monkeypatch.setattr(service, "send_volume_command", fake_send_volume_command)
|
|
|
|
await service._sync_desired_volume("TalkingQ_XQSN00001001", "70")
|
|
|
|
assert sent_commands == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_device_service_persists_desired_volume_before_mqtt(monkeypatch):
|
|
from banban.service.device import DeviceService
|
|
|
|
service = DeviceService()
|
|
calls = []
|
|
|
|
async def fake_ensure_device_access(*, device_id, user_id):
|
|
calls.append(("access", device_id, user_id))
|
|
|
|
class FakeDeviceVolumeManager:
|
|
async def set_volume(self, device_id, level):
|
|
calls.append(("persist", device_id, level))
|
|
return True
|
|
|
|
class FakeMQTTService:
|
|
async def send_volume_command(self, device_id, level):
|
|
calls.append(("mqtt", device_id, level))
|
|
return "002"
|
|
|
|
class FakeTalkingQMQTTService:
|
|
@classmethod
|
|
async def get_instance(cls):
|
|
return FakeMQTTService()
|
|
|
|
monkeypatch.setattr(service, "ensure_device_access", fake_ensure_device_access)
|
|
monkeypatch.setattr("banban.service.device.device_volume_manager", FakeDeviceVolumeManager())
|
|
monkeypatch.setattr("handlers.mqtt_handler.TalkingQMQTTService", FakeTalkingQMQTTService)
|
|
|
|
msg_id = await service.set_device_volume(
|
|
device_id="TalkingQ_XQSN00001001",
|
|
user_id=1,
|
|
level=70,
|
|
)
|
|
|
|
assert msg_id == "002"
|
|
assert calls == [
|
|
("access", "TalkingQ_XQSN00001001", 1),
|
|
("persist", "TalkingQ_XQSN00001001", 70),
|
|
("mqtt", "TalkingQ_XQSN00001001", 70),
|
|
]
|