Improve test coverage

tests/test_send_messages.py

@@ -1220,6 +1220,324 @@ class TestResendChannelMessage:
         assert "expired" in exc_info.value.detail.lower()
 
 
+class TestPathHashModeOverride:
+    """Test per-channel path_hash_mode_override apply/restore behavior."""
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_uses_phm_override(self, test_db):
+        """Override is applied before send and baseline restored after."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock(return_value=_make_radio_result())
+        chan_key = "f1" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#phm")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 2)
+
+        radio_manager.path_hash_mode = 0
+        radio_manager.path_hash_mode_supported = True
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+        ):
+            await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        assert mc.commands.set_path_hash_mode.await_args_list == [call(2), call(0)]
+        assert radio_manager.path_hash_mode == 0
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_skips_phm_when_matching_baseline(self, test_db):
+        """No set_path_hash_mode calls when override matches baseline."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock()
+        chan_key = "f2" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#same")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 1)
+
+        radio_manager.path_hash_mode = 1
+        radio_manager.path_hash_mode_supported = True
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+        ):
+            await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        mc.commands.set_path_hash_mode.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_skips_phm_when_unsupported(self, test_db):
+        """No set_path_hash_mode calls when radio doesn't support it."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock()
+        chan_key = "f3" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#nosupport")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 2)
+
+        radio_manager.path_hash_mode = 0
+        radio_manager.path_hash_mode_supported = False
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+        ):
+            await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        mc.commands.set_path_hash_mode.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_aborts_on_phm_apply_error(self, test_db):
+        """ERROR on apply aborts the send entirely."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock(
+            return_value=MagicMock(type=EventType.ERROR, payload="unsupported mode")
+        )
+        chan_key = "f4" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#fail")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 2)
+
+        radio_manager.path_hash_mode = 0
+        radio_manager.path_hash_mode_supported = True
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+            pytest.raises(HTTPException) as exc_info,
+        ):
+            await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        assert exc_info.value.status_code == 500
+        assert "path hash mode" in exc_info.value.detail.lower()
+        mc.commands.send_chan_msg.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_phm_restore_failure_broadcasts_error(self, test_db):
+        """Message sends OK but restore failure after 3 attempts broadcasts an error."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock(
+            side_effect=[
+                _make_radio_result(),  # apply succeeds
+                MagicMock(type=EventType.ERROR, payload="fail 1"),
+                MagicMock(type=EventType.ERROR, payload="fail 2"),
+                MagicMock(type=EventType.ERROR, payload="fail 3"),
+            ]
+        )
+        chan_key = "f5" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#restorefail")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 2)
+
+        radio_manager.path_hash_mode = 0
+        radio_manager.path_hash_mode_supported = True
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+            patch("app.routers.messages.broadcast_error") as mock_err,
+        ):
+            result = await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        assert result is not None  # message sent OK
+        mock_err.assert_called_once()
+        assert "path hash mode" in mock_err.call_args.args[0].lower()
+        # 1 apply + 3 restore attempts = 4 calls total
+        assert mc.commands.set_path_hash_mode.await_count == 4
+
+    @pytest.mark.asyncio
+    async def test_send_channel_msg_phm_restore_succeeds_on_second_attempt(self, test_db):
+        """Restore retries and succeeds on the second attempt — no error broadcast."""
+        mc = _make_mc(name="MyNode")
+        mc.commands.set_path_hash_mode = AsyncMock(
+            side_effect=[
+                _make_radio_result(),  # apply succeeds
+                MagicMock(type=EventType.ERROR, payload="transient"),  # restore attempt 1
+                _make_radio_result(),  # restore attempt 2 succeeds
+            ]
+        )
+        chan_key = "f6" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#retry")
+        await ChannelRepository.update_path_hash_mode_override(chan_key, 2)
+
+        radio_manager.path_hash_mode = 0
+        radio_manager.path_hash_mode_supported = True
+
+        with (
+            patch("app.routers.messages.radio_manager.require_connected", return_value=mc),
+            patch.object(radio_manager, "_meshcore", mc),
+            patch("app.routers.messages.broadcast_event"),
+            patch("app.routers.messages.broadcast_error") as mock_err,
+        ):
+            await send_channel_message(
+                SendChannelMessageRequest(channel_key=chan_key, text="hello")
+            )
+
+        mock_err.assert_not_called()
+        assert radio_manager.path_hash_mode == 0  # restored to baseline
+        # 1 apply + 2 restore attempts = 3 calls
+        assert mc.commands.set_path_hash_mode.await_count == 3
+
+
+class TestChannelEchoWatchdog:
+    """Test the auto-resend echo watchdog for channel messages."""
+
+    @pytest.fixture(autouse=True)
+    def _skip_watchdog_delay(self, monkeypatch):
+        monkeypatch.setattr(message_send_service, "ECHO_WATCHDOG_DELAY_SECONDS", 0)
+
+    @pytest.mark.asyncio
+    async def test_watchdog_skips_when_echo_already_received(self, test_db):
+        """Watchdog sees acked > 0 and returns without resending."""
+        chan_key = "e1" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#echo")
+
+        msg_id = await MessageRepository.create(
+            msg_type="CHAN",
+            text="MyNode: hello",
+            conversation_key=chan_key.upper(),
+            sender_timestamp=int(time.time()),
+            received_at=int(time.time()),
+            outgoing=True,
+        )
+        await MessageRepository.increment_ack_count(msg_id)
+
+        mc = _make_mc(name="MyNode")
+
+        with patch.object(radio_manager, "_meshcore", mc):
+            await message_send_service._channel_echo_watchdog(
+                message_id=msg_id,
+                radio_manager=radio_manager,
+                broadcast_fn=MagicMock(),
+                error_broadcast_fn=MagicMock(),
+            )
+
+        # No radio operation attempted
+        mc.commands.send_chan_msg.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_watchdog_skips_when_outside_resend_window(self, test_db):
+        """Watchdog skips resend when message is older than 30 seconds."""
+        chan_key = "e2" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#stale")
+
+        old_ts = int(time.time()) - 60
+        msg_id = await MessageRepository.create(
+            msg_type="CHAN",
+            text="MyNode: old",
+            conversation_key=chan_key.upper(),
+            sender_timestamp=old_ts,
+            received_at=old_ts,
+            outgoing=True,
+        )
+
+        mc = _make_mc(name="MyNode")
+
+        with patch.object(radio_manager, "_meshcore", mc):
+            await message_send_service._channel_echo_watchdog(
+                message_id=msg_id,
+                radio_manager=radio_manager,
+                broadcast_fn=MagicMock(),
+                error_broadcast_fn=MagicMock(),
+            )
+
+        mc.commands.send_chan_msg.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_watchdog_resends_when_no_echo(self, test_db):
+        """Watchdog resends byte-perfect when no echo has arrived."""
+        chan_key = "e3" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#resend")
+
+        now = int(time.time())
+        msg_id = await MessageRepository.create(
+            msg_type="CHAN",
+            text="MyNode: payload",
+            conversation_key=chan_key.upper(),
+            sender_timestamp=now,
+            received_at=now,
+            outgoing=True,
+        )
+
+        mc = _make_mc(name="MyNode")
+
+        with patch.object(radio_manager, "_meshcore", mc):
+            await message_send_service._channel_echo_watchdog(
+                message_id=msg_id,
+                radio_manager=radio_manager,
+                broadcast_fn=MagicMock(),
+                error_broadcast_fn=MagicMock(),
+            )
+
+        mc.commands.send_chan_msg.assert_awaited_once()
+        call_kwargs = mc.commands.send_chan_msg.await_args.kwargs
+        assert call_kwargs["msg"] == "payload"  # sender prefix stripped
+        assert call_kwargs["timestamp"] == now.to_bytes(4, "little")
+
+    @pytest.mark.asyncio
+    async def test_watchdog_handles_radio_busy_gracefully(self, test_db):
+        """RadioOperationBusyError is caught — no exception propagates."""
+
+        chan_key = "e4" * 16
+        await ChannelRepository.upsert(key=chan_key, name="#busy")
+
+        now = int(time.time())
+        msg_id = await MessageRepository.create(
+            msg_type="CHAN",
+            text="MyNode: busy test",
+            conversation_key=chan_key.upper(),
+            sender_timestamp=now,
+            received_at=now,
+            outgoing=True,
+        )
+
+        mc = _make_mc(name="MyNode")
+        radio_manager._meshcore = mc
+        # Lock the radio so the non-blocking acquire raises RadioOperationBusyError
+        if radio_manager._operation_lock is None:
+            radio_manager._operation_lock = asyncio.Lock()
+        await radio_manager._operation_lock.acquire()
+
+        try:
+            # Should not raise — RadioOperationBusyError is caught internally
+            await message_send_service._channel_echo_watchdog(
+                message_id=msg_id,
+                radio_manager=radio_manager,
+                broadcast_fn=MagicMock(),
+                error_broadcast_fn=MagicMock(),
+            )
+        finally:
+            radio_manager._operation_lock.release()
+
+        mc.commands.send_chan_msg.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_watchdog_skips_deleted_message(self, test_db):
+        """Watchdog exits cleanly if the message was deleted before it wakes."""
+        # Use a message_id that doesn't exist
+        mc = _make_mc(name="MyNode")
+        with patch.object(radio_manager, "_meshcore", mc):
+            await message_send_service._channel_echo_watchdog(
+                message_id=999999,
+                radio_manager=radio_manager,
+                broadcast_fn=MagicMock(),
+                error_broadcast_fn=MagicMock(),
+            )
+
+        mc.commands.send_chan_msg.assert_not_called()
+
+
 class TestRadioExceptionMidSend:
     """Test that radio exceptions during send don't leave orphaned DB state."""
 

tests/test_sqs_fanout.py

@@ -0,0 +1,127 @@
+"""Tests for the SQS fanout module helper functions.
+
+Covers region inference from queue URLs, FIFO deduplication ID fallback chains,
+and message group ID construction — the non-trivial logic in app/fanout/sqs.py.
+"""
+
+import hashlib
+
+from app.fanout.sqs import (
+    _build_message_deduplication_id,
+    _build_message_group_id,
+    _infer_region_from_queue_url,
+    _is_fifo_queue,
+)
+
+
+class TestInferRegionFromQueueUrl:
+    """URL parsing for AWS region extraction."""
+
+    def test_standard_us_east_1(self):
+        url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"
+        assert _infer_region_from_queue_url(url) == "us-east-1"
+
+    def test_standard_eu_west_2(self):
+        url = "https://sqs.eu-west-2.amazonaws.com/123456789012/my-queue"
+        assert _infer_region_from_queue_url(url) == "eu-west-2"
+
+    def test_china_region(self):
+        url = "https://sqs.cn-north-1.amazonaws.com.cn/123456789012/my-queue"
+        assert _infer_region_from_queue_url(url) == "cn-north-1"
+
+    def test_non_sqs_hostname_returns_none(self):
+        url = "https://s3.us-east-1.amazonaws.com/bucket/key"
+        assert _infer_region_from_queue_url(url) is None
+
+    def test_localstack_endpoint_returns_none(self):
+        url = "http://localhost:4566/000000000000/my-queue"
+        assert _infer_region_from_queue_url(url) is None
+
+    def test_empty_url_returns_none(self):
+        assert _infer_region_from_queue_url("") is None
+
+    def test_non_amazonaws_domain_returns_none(self):
+        url = "https://sqs.us-east-1.example.com/123/queue"
+        assert _infer_region_from_queue_url(url) is None
+
+    def test_fifo_queue_url_still_parses_region(self):
+        url = "https://sqs.ap-southeast-1.amazonaws.com/123456789012/my-queue.fifo"
+        assert _infer_region_from_queue_url(url) == "ap-southeast-1"
+
+
+class TestIsFifoQueue:
+    def test_fifo_suffix(self):
+        assert _is_fifo_queue("https://sqs.us-east-1.amazonaws.com/123/queue.fifo") is True
+
+    def test_standard_queue(self):
+        assert _is_fifo_queue("https://sqs.us-east-1.amazonaws.com/123/queue") is False
+
+    def test_trailing_slash_stripped(self):
+        assert _is_fifo_queue("https://sqs.us-east-1.amazonaws.com/123/queue.fifo/") is True
+
+
+class TestBuildMessageGroupId:
+    """FIFO message group ID selection."""
+
+    def test_message_event_with_conversation_key(self):
+        data = {"conversation_key": "abc123", "text": "hello"}
+        assert _build_message_group_id(data, event_type="message") == "message-abc123"
+
+    def test_message_event_without_conversation_key_falls_back(self):
+        data = {"text": "hello"}
+        assert _build_message_group_id(data, event_type="message") == "message-default"
+
+    def test_raw_packet_event_always_returns_raw_packets(self):
+        data = {"id": 1, "payload": "deadbeef"}
+        assert _build_message_group_id(data, event_type="raw_packet") == "raw-packets"
+
+    def test_message_event_with_empty_conversation_key_falls_back(self):
+        data = {"conversation_key": "  ", "text": "hello"}
+        assert _build_message_group_id(data, event_type="message") == "message-default"
+
+
+class TestBuildMessageDeduplicationId:
+    """FIFO deduplication ID fallback chain."""
+
+    def test_message_with_int_id(self):
+        data = {"id": 42}
+        result = _build_message_deduplication_id(data, event_type="message", body="{}")
+        assert result == "message-42"
+
+    def test_message_with_string_id_falls_back_to_hash(self):
+        body = '{"event_type":"message","data":{"id":"not-an-int"}}'
+        data = {"id": "not-an-int"}
+        result = _build_message_deduplication_id(data, event_type="message", body=body)
+        assert result == hashlib.sha256(body.encode()).hexdigest()
+
+    def test_message_without_id_falls_back_to_hash(self):
+        body = '{"event_type":"message","data":{}}'
+        data = {}
+        result = _build_message_deduplication_id(data, event_type="message", body=body)
+        assert result == hashlib.sha256(body.encode()).hexdigest()
+
+    def test_raw_with_observation_id(self):
+        data = {"observation_id": "obs-123", "id": 7}
+        result = _build_message_deduplication_id(data, event_type="raw_packet", body="{}")
+        assert result == "raw-obs-123"
+
+    def test_raw_with_empty_observation_id_falls_to_packet_id(self):
+        data = {"observation_id": "  ", "id": 7}
+        result = _build_message_deduplication_id(data, event_type="raw_packet", body="{}")
+        assert result == "raw-7"
+
+    def test_raw_with_no_observation_id_uses_packet_id(self):
+        data = {"id": 99}
+        result = _build_message_deduplication_id(data, event_type="raw_packet", body="{}")
+        assert result == "raw-99"
+
+    def test_raw_with_no_ids_falls_back_to_hash(self):
+        body = '{"event_type":"raw_packet","data":{}}'
+        data = {}
+        result = _build_message_deduplication_id(data, event_type="raw_packet", body=body)
+        assert result == hashlib.sha256(body.encode()).hexdigest()
+
+    def test_raw_with_non_string_observation_id_falls_to_packet_id(self):
+        data = {"observation_id": 123, "id": 5}
+        result = _build_message_deduplication_id(data, event_type="raw_packet", body="{}")
+        assert result == "raw-5"