from types import SimpleNamespace import pytest from banban.dao.pending_voice_message import PendingVoiceMessageDAO from banban.service.pending_voice_message import ( PendingVoiceMessageService, PendingVoicePlaybackItem, ) from handlers.mqtt_handler import TalkingQMQTTService class FakeResult: def __init__(self, rows=None): self._rows = rows or [] def mappings(self): return self def all(self): return self._rows class FakeSession: def __init__(self, rows=None): self.rows = rows or [] self.calls = [] self.commits = 0 self.rollbacks = 0 self.closed = False async def execute(self, statement, params=None): self.calls.append((str(statement), params or {})) return FakeResult(self.rows) async def commit(self): self.commits += 1 async def rollback(self): self.rollbacks += 1 async def close(self): self.closed = True @pytest.mark.asyncio async def test_upsert_pending_keeps_original_media_file_key_on_duplicate(): session = FakeSession() dao = PendingVoiceMessageDAO(session) await dao.upsert_pending( target_device_id="TalkingQ_XQSN00001001", sender_device_id="TalkingQ_XQSN00001002", im_message_id=123, media_file_key="audio/original.mp3", audio_url="http://example.test/audio.mp3", source="device_peer_voice", ) sql, params = session.calls[0] update_sql = sql.split("ON DUPLICATE KEY UPDATE", 1)[1] assert "ON DUPLICATE KEY UPDATE" in sql assert "media_file_key" not in update_sql assert params["target_device_id"] == "TalkingQ_XQSN00001001" assert params["im_message_id"] == 123 assert params["media_file_key"] == "audio/original.mp3" assert params["status"] == "pending" @pytest.mark.asyncio async def test_mark_delivered_only_advances_pending_rows(): session = FakeSession() dao = PendingVoiceMessageDAO(session) await dao.mark_delivered([7, 9]) sql, params = session.calls[0] assert "SET status = :status" in sql assert "delivery_count = delivery_count + 1" in sql assert "AND status = :pending_status" in sql assert params == { "id_0": 7, "id_1": 9, "status": "delivered", "pending_status": "pending", } @pytest.mark.asyncio async def test_get_playback_items_uses_fallback_url_when_signed_url_fails(monkeypatch): service = PendingVoiceMessageService() session = FakeSession( rows=[ { "id": 11, "target_device_id": "TalkingQ_XQSN00001001", "sender_device_id": "TalkingQ_XQSN00001002", "im_message_id": 123, "media_file_key": "cos/audio.mp3", "audio_url": "http://cached.example/audio.mp3", "source": "device_peer_voice", "status": "pending", "delivery_count": 0, "created_at": None, } ] ) async def fake_get_session(): return session class FailingAudioCache: async def get_device_audio_url(self, media_file_key, device_id=None): raise RuntimeError("cos unavailable") monkeypatch.setattr(service, "get_session", fake_get_session) monkeypatch.setattr( "banban.service.pending_voice_message.device_audio_cache_service", FailingAudioCache(), ) items = await service.get_playback_items(target_device_id="TalkingQ_XQSN00001001") assert items == [ PendingVoicePlaybackItem( pending_id=11, audio_url="http://cached.example/audio.mp3", im_message_id=123, media_file_key="cos/audio.mp3", source="device_peer_voice", ) ] assert session.closed is True @pytest.mark.asyncio async def test_nfc_owner_listen_marks_pending_delivered_only_after_publish(monkeypatch): service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) published = [] marked = [] cleared = [] class FakeCardService: async def get_card_by_uuid(self, nfc_uuid): return SimpleNamespace(device_id="TalkingQ_XQSN00001001") async def check_card_ownership(self, nfc_uuid, device_id): return True class FakePendingVoiceService: async def get_playback_items(self, target_device_id): return [ PendingVoicePlaybackItem( pending_id=11, audio_url="http://cached.example/audio.mp3", im_message_id=123, media_file_key="cos/audio.mp3", source="device_peer_voice", ) ] async def mark_delivered(self, pending_ids): marked.extend(pending_ids) class FakeOfflineAudioCache: async def clear_audio_urls(self, device_id): cleared.append(device_id) async def fake_publish(topic, payload): published.append((topic, payload)) return True monkeypatch.setattr("handlers.mqtt_handler.card_service", FakeCardService()) monkeypatch.setattr("handlers.mqtt_handler.pending_voice_message_service", FakePendingVoiceService()) monkeypatch.setattr("handlers.mqtt_handler.offline_audio_cache", FakeOfflineAudioCache()) monkeypatch.setattr(service, "_publish", fake_publish) await service._send_nfc_listen_response("TalkingQ_XQSN00001001", "53DA2B6DA20001") assert published == [ ( "device/TalkingQ_XQSN00001001/event_resp", { "msg_id": "005", "type": 0, "params": {"url_1": "http://cached.example/audio.mp3"}, }, ) ] assert marked == [11] assert cleared == ["TalkingQ_XQSN00001001"] @pytest.mark.asyncio async def test_nfc_owner_listen_keeps_pending_when_publish_fails(monkeypatch): service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) marked = [] class FakeCardService: async def get_card_by_uuid(self, nfc_uuid): return SimpleNamespace(device_id="TalkingQ_XQSN00001001") async def check_card_ownership(self, nfc_uuid, device_id): return True class FakePendingVoiceService: async def get_playback_items(self, target_device_id): return [ PendingVoicePlaybackItem( pending_id=11, audio_url="http://cached.example/audio.mp3", im_message_id=123, media_file_key="cos/audio.mp3", source="device_peer_voice", ) ] async def mark_delivered(self, pending_ids): marked.extend(pending_ids) async def fake_publish(topic, payload): return False monkeypatch.setattr("handlers.mqtt_handler.card_service", FakeCardService()) monkeypatch.setattr("handlers.mqtt_handler.pending_voice_message_service", FakePendingVoiceService()) monkeypatch.setattr(service, "_publish", fake_publish) await service._send_nfc_listen_response("TalkingQ_XQSN00001001", "53DA2B6DA20001") assert marked == []