diff --git a/tests/test_send_messages.py b/tests/test_send_messages.py index 8136d4b..3b9e4f2 100644 --- a/tests/test_send_messages.py +++ b/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.""" diff --git a/tests/test_sqs_fanout.py b/tests/test_sqs_fanout.py new file mode 100644 index 0000000..b5c7f3e --- /dev/null +++ b/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"