feat(tests): enhance coverage for LXMF message handling, including image attachment stripping for strangers and sticker validation tests

tests/backend/test_meshchat_coverage.py

@@ -1,9 +1,12 @@
 import pytest
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 import asyncio
 import json
 import os
+import subprocess
+import LXMF
 from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend.lxmf_message_fields import LxmfAudioField
 
 
 @pytest.fixture
@@ -167,6 +170,7 @@ def test_get_config_dict_basic(mock_app):
         "message_outbound_bubble_color",
         "message_inbound_bubble_color",
         "message_failed_bubble_color",
+        "message_waiting_bubble_color",
     ]:
         getattr(mock_config, attr).get.return_value = None
 
@@ -289,18 +293,34 @@ async def test_send_config_to_websocket_clients(mock_app):
 @pytest.mark.asyncio
 async def test_on_lxmf_sending_state_updated(mock_app):
     mock_msg = MagicMock()
-    mock_app.db_upsert_lxmf_message = MagicMock()
+    mock_msg.progress = 0.75
+    mock_msg.rssi = None
+    mock_msg.snr = None
+    mock_msg.q = None
+    mock_msg.hash.hex.return_value = "ab" * 16
+    mock_msg.delivery_attempts = 2
+
+    ctx = mock_app.current_context
     mock_app.websocket_broadcast = MagicMock(return_value=asyncio.Future())
     mock_app.websocket_broadcast.return_value.set_result(None)
 
-    with patch(
-        "meshchatx.meshchat.convert_lxmf_message_to_dict", return_value={"h": "v"}
+    with (
+        patch(
+            "meshchatx.meshchat.convert_lxmf_message_to_dict", return_value={"h": "v"}
+        ),
+        patch(
+            "meshchatx.meshchat.convert_lxmf_state_to_string", return_value="delivered"
+        ),
+        patch("meshchatx.meshchat.AsyncUtils.run_async") as mock_run_async,
     ):
-        # Pass context explicitly to match expectation or fix expectation
-        ctx = mock_app.current_context
         mock_app.on_lxmf_sending_state_updated(mock_msg, context=ctx)
-        mock_app.db_upsert_lxmf_message.assert_called_once_with(mock_msg, context=ctx)
-        mock_app.websocket_broadcast.assert_called_once()
+
+        ctx.database.messages.update_lxmf_message_state.assert_called_once()
+        call_kwargs = ctx.database.messages.update_lxmf_message_state.call_args
+        assert call_kwargs.kwargs["message_hash"] == "ab" * 16
+        assert call_kwargs.kwargs["progress"] == 75.0
+        assert call_kwargs.kwargs["state"] == "delivered"
+        mock_run_async.assert_called_once()
 
 
 @pytest.mark.asyncio
@@ -344,3 +364,141 @@ def test_on_lxmf_sending_failed_no_propagation(mock_app):
     mock_app.on_lxmf_sending_state_updated.assert_called_once_with(
         mock_msg, context=None
     )
+
+
+def test_convert_webm_opus_to_ogg_already_ogg(mock_app):
+    ogg_data = b"OggS" + b"\x00" * 100
+    result = mock_app._convert_webm_opus_to_ogg(ogg_data)
+    assert result is ogg_data
+
+
+def test_convert_webm_opus_to_ogg_no_ffmpeg(mock_app):
+    webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100
+    with patch("shutil.which", return_value=None):
+        result = mock_app._convert_webm_opus_to_ogg(webm_data)
+    assert result is webm_data
+
+
+def test_convert_webm_opus_to_ogg_success(mock_app):
+    webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100
+    fake_ogg = b"OggS" + b"\xff" * 50
+    mock_result = MagicMock()
+    mock_result.returncode = 0
+    mock_result.stdout = fake_ogg
+
+    with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
+        with patch("subprocess.run", return_value=mock_result) as mock_run:
+            result = mock_app._convert_webm_opus_to_ogg(webm_data)
+
+    assert result == fake_ogg
+    mock_run.assert_called_once()
+    call_args = mock_run.call_args
+    assert call_args.kwargs["input"] == webm_data
+    assert "-f" in call_args[0][0]
+    assert "ogg" in call_args[0][0]
+
+
+def test_convert_webm_opus_to_ogg_ffmpeg_fails(mock_app):
+    webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100
+    mock_result = MagicMock()
+    mock_result.returncode = 1
+    mock_result.stdout = b""
+
+    with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
+        with patch("subprocess.run", return_value=mock_result):
+            result = mock_app._convert_webm_opus_to_ogg(webm_data)
+
+    assert result is webm_data
+
+
+def test_convert_webm_opus_to_ogg_exception(mock_app):
+    webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100
+
+    with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
+        with patch(
+            "subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=30),
+        ):
+            result = mock_app._convert_webm_opus_to_ogg(webm_data)
+
+    assert result is webm_data
+
+
+@pytest.fixture
+def sendable_app(mock_app):
+    """mock_app wired so send_message can run to completion."""
+    ctx = mock_app.current_context
+    ctx.config.auto_send_failed_messages_to_propagation_node.get.return_value = False
+    ctx.message_router.delivery_link_available.return_value = True
+    ctx.local_lxmf_destination = MagicMock()
+    ctx.forwarding_manager = None
+
+    mock_app._await_transport_path = AsyncMock()
+    mock_app.get_current_icon_hash = MagicMock(return_value=None)
+    mock_app.db_upsert_lxmf_message = MagicMock()
+    mock_app.websocket_broadcast = AsyncMock()
+    mock_app.handle_lxmf_message_progress = AsyncMock()
+    mock_app._convert_webm_opus_to_ogg = MagicMock(side_effect=lambda b: b)
+
+    return mock_app
+
+
+async def _run_send(app, destination_hash="aa" * 16, **kwargs):
+    fake_identity = MagicMock()
+    fake_destination = MagicMock()
+    fake_lxm = MagicMock(spec=LXMF.LXMessage)
+    fake_lxm.fields = {}
+    fake_lxm.include_ticket = False
+
+    with (
+        patch("meshchatx.meshchat.RNS.Identity.recall", return_value=fake_identity),
+        patch("meshchatx.meshchat.RNS.Destination", return_value=fake_destination),
+        patch("meshchatx.meshchat.LXMF.LXMessage", return_value=fake_lxm),
+        patch(
+            "meshchatx.meshchat.convert_lxmf_message_to_dict",
+            return_value={"hash": "x"},
+        ),
+        patch("meshchatx.meshchat.AsyncUtils.run_async"),
+    ):
+        await app.send_message(
+            destination_hash=destination_hash, content="hi", **kwargs
+        )
+
+    return fake_lxm
+
+
+@pytest.mark.asyncio
+async def test_send_message_sets_renderer_markdown(sendable_app):
+    lxm = await _run_send(sendable_app)
+    assert LXMF.FIELD_RENDERER in lxm.fields
+    assert lxm.fields[LXMF.FIELD_RENDERER] == LXMF.RENDERER_MARKDOWN
+
+
+@pytest.mark.asyncio
+async def test_send_message_include_ticket_for_contact(sendable_app):
+    sendable_app._is_contact = MagicMock(return_value=True)
+    lxm = await _run_send(sendable_app)
+    assert lxm.include_ticket is True
+
+
+@pytest.mark.asyncio
+async def test_send_message_no_ticket_for_stranger(sendable_app):
+    sendable_app._is_contact = MagicMock(return_value=False)
+    lxm = await _run_send(sendable_app)
+    assert lxm.include_ticket is False
+
+
+@pytest.mark.asyncio
+async def test_send_message_opus_audio_triggers_conversion(sendable_app):
+    audio = LxmfAudioField(audio_mode=LXMF.AM_OPUS_OGG, audio_bytes=b"\x1a\x45")
+    lxm = await _run_send(sendable_app, audio_field=audio)
+    sendable_app._convert_webm_opus_to_ogg.assert_called_once_with(b"\x1a\x45")
+    assert LXMF.FIELD_AUDIO in lxm.fields
+
+
+@pytest.mark.asyncio
+async def test_send_message_codec2_audio_skips_conversion(sendable_app):
+    audio = LxmfAudioField(audio_mode=LXMF.AM_CODEC2_1200, audio_bytes=b"\xcc")
+    lxm = await _run_send(sendable_app, audio_field=audio)
+    sendable_app._convert_webm_opus_to_ogg.assert_not_called()
+    assert lxm.fields[LXMF.FIELD_AUDIO] == [LXMF.AM_CODEC2_1200, b"\xcc"]

tests/backend/test_security_fuzzing.py

@@ -2424,6 +2424,80 @@ class TestStrangerAttachmentBlocking:
         mock_app.on_lxmf_delivery(mock_msg)
         mock_app.db_upsert_lxmf_message.assert_called_once()
 
+    def test_stranger_lxmf_image_stripped_when_enabled(self, mock_app):
+        """Image attachments (including sticker-shaped PNGs) use FIELD_IMAGE; strip like files."""
+        source_hash = os.urandom(16)
+        mock_msg = MagicMock()
+        mock_msg.source_hash = source_hash
+        mock_msg.destination_hash = os.urandom(16)
+        mock_msg.hash = os.urandom(16)
+        mock_msg.content = b"sticker msg"
+        mock_msg.title = b""
+        mock_msg.incoming = True
+        mock_msg.state = LXMF.LXMessage.DELIVERED
+        mock_msg.method = LXMF.LXMessage.DIRECT
+        mock_msg.progress = 1.0
+        mock_msg.timestamp = 123456789.0
+        mock_msg.rssi = -50
+        mock_msg.snr = 10
+        mock_msg.q = 100
+        mock_msg.delivery_attempts = 1
+        mock_msg.next_delivery_attempt = None
+
+        fields = {
+            LXMF.FIELD_IMAGE: [b"png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 64],
+        }
+        mock_msg.get_fields.return_value = fields
+        mock_msg.fields = fields
+
+        mock_app.config.block_attachments_from_strangers.get.return_value = True
+        mock_app.config.block_all_from_strangers.get.return_value = False
+        mock_app._is_contact = MagicMock(return_value=False)
+        mock_app.is_destination_blocked = MagicMock(return_value=False)
+        mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+        mock_app.on_lxmf_delivery(mock_msg)
+
+        mock_app.db_upsert_lxmf_message.assert_called_once()
+        call_kwargs = mock_app.db_upsert_lxmf_message.call_args
+        assert call_kwargs[1].get("attachments_stripped") is True or (
+            len(call_kwargs[0]) > 2 and call_kwargs[0][2] is True
+        )
+        assert LXMF.FIELD_IMAGE not in mock_msg.get_fields()
+
+    def test_contact_lxmf_image_not_stripped(self, mock_app):
+        source_hash = os.urandom(16)
+        mock_msg = MagicMock()
+        mock_msg.source_hash = source_hash
+        mock_msg.destination_hash = os.urandom(16)
+        mock_msg.hash = os.urandom(16)
+        mock_msg.content = b"img"
+        mock_msg.title = b""
+        mock_msg.incoming = True
+        mock_msg.state = LXMF.LXMessage.DELIVERED
+        mock_msg.method = LXMF.LXMessage.DIRECT
+        mock_msg.progress = 1.0
+        mock_msg.timestamp = 123456789.0
+        mock_msg.rssi = -50
+        mock_msg.snr = 10
+        mock_msg.q = 100
+        mock_msg.delivery_attempts = 1
+        mock_msg.next_delivery_attempt = None
+
+        fields = {LXMF.FIELD_IMAGE: [b"png", b"\x00" * 32]}
+        mock_msg.get_fields.return_value = fields
+        mock_msg.fields = fields
+
+        mock_app.config.block_attachments_from_strangers.get.return_value = True
+        mock_app.config.block_all_from_strangers.get.return_value = False
+        mock_app._is_contact = MagicMock(return_value=True)
+        mock_app.is_destination_blocked = MagicMock(return_value=False)
+        mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+        mock_app.on_lxmf_delivery(mock_msg)
+
+        assert LXMF.FIELD_IMAGE in mock_msg.get_fields()
+
 
 class TestBlockAllFromStrangers:
     """Tests for the block-everything-from-strangers feature."""

tests/backend/test_sticker_utils.py

@@ -0,0 +1,180 @@
+"""Unit tests for sticker validation and export/import document parsing."""
+
+import base64
+
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from meshchatx.src.backend import sticker_utils
+
+
+def test_normalize_image_type():
+    assert sticker_utils.normalize_image_type("PNG") == "png"
+    assert sticker_utils.normalize_image_type("image/jpeg") == "jpeg"
+    assert sticker_utils.normalize_image_type("jpg") == "jpeg"
+    assert sticker_utils.normalize_image_type("webp") == "webp"
+    assert sticker_utils.normalize_image_type("svg") is None
+    assert sticker_utils.normalize_image_type("") is None
+    assert sticker_utils.normalize_image_type(None) is None
+
+
+def test_validate_sticker_payload_ok():
+    raw = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
+    nt, h = sticker_utils.validate_sticker_payload(raw, "png")
+    assert nt == "png"
+    assert len(h) == 64
+
+
+def test_validate_sticker_payload_too_large():
+    raw = b"x" * (sticker_utils.MAX_STICKER_BYTES + 1)
+    with pytest.raises(ValueError, match="image_too_large"):
+        sticker_utils.validate_sticker_payload(raw, "png")
+
+
+def test_validate_sticker_payload_empty():
+    with pytest.raises(ValueError, match="empty_image"):
+        sticker_utils.validate_sticker_payload(b"", "png")
+
+
+def test_validate_sticker_payload_bad_type():
+    with pytest.raises(ValueError, match="invalid_image_type"):
+        sticker_utils.validate_sticker_payload(b"abc", "svg")
+
+
+def test_validate_sticker_payload_bad_magic():
+    with pytest.raises(ValueError, match="invalid_image_signature"):
+        sticker_utils.validate_sticker_payload(b"not-an-image-bytes", "png")
+
+
+def test_validate_sticker_payload_magic_type_mismatch():
+    png_bytes = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
+    with pytest.raises(ValueError, match="magic_type_mismatch"):
+        sticker_utils.validate_sticker_payload(png_bytes, "jpeg")
+
+
+def test_detect_image_format_from_magic():
+    assert sticker_utils.detect_image_format_from_magic(b"\x89PNG\r\n\x1a\n") == "png"
+    assert (
+        sticker_utils.detect_image_format_from_magic(b"\xff\xd8\xff\xe0\x00\x10")
+        == "jpeg"
+    )
+    assert (
+        sticker_utils.detect_image_format_from_magic(b"GIF89a" + b"\x00" * 4) == "gif"
+    )
+    assert sticker_utils.detect_image_format_from_magic(b"BM" + b"\x00" * 20) == "bmp"
+    webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 8
+    assert sticker_utils.detect_image_format_from_magic(webp) == "webp"
+    assert sticker_utils.detect_image_format_from_magic(b"") is None
+    assert sticker_utils.detect_image_format_from_magic(b"short") is None
+
+
+def test_validate_sticker_payload_jpg_alias_matches_jpeg_magic():
+    raw = b"\xff\xd8\xff\xe0" + b"\x00" * 64
+    nt, _ = sticker_utils.validate_sticker_payload(raw, "jpg")
+    assert nt == "jpeg"
+
+
+def test_sanitize_sticker_name():
+    assert sticker_utils.sanitize_sticker_name("  hello  ") == "hello"
+    assert sticker_utils.sanitize_sticker_name("") is None
+    assert sticker_utils.sanitize_sticker_name("x" * 200) is not None
+    assert len(sticker_utils.sanitize_sticker_name("x" * 200) or "") == 128
+
+
+def test_validate_export_document_ok():
+    png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
+    b64 = base64.b64encode(png).decode("ascii")
+    doc = {
+        "format": "meshchatx-stickers",
+        "version": 1,
+        "stickers": [
+            {"name": "a", "image_type": "png", "image_bytes": b64},
+        ],
+    }
+    items = sticker_utils.validate_export_document(doc)
+    assert len(items) == 1
+    assert items[0]["image_bytes_b64"] == b64
+
+
+def test_validate_export_document_wrong_format():
+    with pytest.raises(ValueError, match="invalid_format"):
+        sticker_utils.validate_export_document(
+            {"format": "other", "version": 1, "stickers": []}
+        )
+
+
+def test_validate_export_document_bad_version():
+    with pytest.raises(ValueError, match="unsupported_version"):
+        sticker_utils.validate_export_document(
+            {"format": "meshchatx-stickers", "version": 99, "stickers": []},
+        )
+
+
+def test_validate_export_document_missing_version():
+    with pytest.raises(ValueError, match="unsupported_version"):
+        sticker_utils.validate_export_document(
+            {"format": "meshchatx-stickers", "stickers": []},
+        )
+
+
+def test_validate_export_document_not_dict():
+    with pytest.raises(ValueError, match="invalid_document"):
+        sticker_utils.validate_export_document([])
+
+
+def test_mime_for_image_type():
+    assert "image/" in sticker_utils.mime_for_image_type("png")
+    assert sticker_utils.mime_for_image_type("unknown") == "application/octet-stream"
+
+
+@settings(max_examples=200, deadline=None)
+@given(
+    raw=st.binary(min_size=0, max_size=sticker_utils.MAX_STICKER_BYTES + 1),
+    typ=st.one_of(
+        st.none(),
+        st.text(max_size=40),
+        st.sampled_from(
+            ["png", "jpeg", "jpg", "webp", "gif", "bmp", "svg", "image/png", ""]
+        ),
+    ),
+)
+def test_validate_sticker_payload_fuzz_never_raises_unexpected(raw, typ):
+    """Fuzz: validation either succeeds or raises ValueError with known reasons."""
+    try:
+        sticker_utils.validate_sticker_payload(raw, typ)
+    except ValueError:
+        pass
+
+
+@settings(max_examples=500, deadline=None)
+@given(raw=st.binary(min_size=0, max_size=4096))
+def test_detect_image_format_from_magic_fuzz_never_raises(raw):
+    out = sticker_utils.detect_image_format_from_magic(raw)
+    assert out is None or out in {"png", "jpeg", "gif", "webp", "bmp"}
+
+
+@settings(max_examples=100, deadline=None)
+@given(
+    doc=st.dictionaries(
+        keys=st.text(max_size=16),
+        values=st.recursive(
+            st.none()
+            | st.booleans()
+            | st.floats(allow_nan=False)
+            | st.text(max_size=40)
+            | st.binary(max_size=48),
+            lambda children: (
+                st.lists(children, max_size=3)
+                | st.dictionaries(st.text(max_size=6), children, max_size=3)
+            ),
+            max_leaves=12,
+        ),
+        max_size=8,
+    ),
+)
+def test_validate_export_document_fuzz_never_raises_unexpected(doc):
+    try:
+        sticker_utils.validate_export_document(doc)
+    except ValueError:
+        pass

tests/backend/test_stickers_dao.py

@@ -0,0 +1,154 @@
+"""Tests for UserStickersDAO and schema migration for user_stickers."""
+
+import base64
+
+import pytest
+
+from meshchatx.src.backend.database import Database
+from meshchatx.src.backend.database.schema import DatabaseSchema
+
+
+@pytest.fixture
+def db(tmp_path):
+    path = tmp_path / "t.db"
+    database = Database(str(path))
+    database.initialize()
+    assert DatabaseSchema.LATEST_VERSION >= 44
+    return database
+
+
+def _tiny_png():
+    return bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + b"\x00" * 32
+
+
+def test_insert_and_list(db):
+    identity = "ab" * 16
+    raw = _tiny_png()
+    row = db.stickers.insert(identity, "one", "png", raw, None)
+    assert row is not None
+    assert row["id"] >= 1
+    listed = db.stickers.list_for_identity(identity)
+    assert len(listed) == 1
+    assert listed[0]["image_size"] == len(raw)
+
+
+def test_duplicate_returns_none(db):
+    identity = "cd" * 16
+    raw = _tiny_png()
+    r1 = db.stickers.insert(identity, None, "png", raw, None)
+    assert r1 is not None
+    r2 = db.stickers.insert(identity, None, "png", raw, None)
+    assert r2 is None
+    assert db.stickers.count_for_identity(identity) == 1
+
+
+def test_delete_and_delete_all(db):
+    identity = "ef" * 16
+    db.stickers.insert(identity, "a", "png", _tiny_png(), None)
+    row = db.stickers.list_for_identity(identity)[0]
+    assert db.stickers.delete(row["id"], identity) is True
+    assert len(db.stickers.list_for_identity(identity)) == 0
+
+    db.stickers.insert(identity, "b", "png", _tiny_png(), None)
+    db.stickers.insert(
+        identity,
+        "c",
+        "png",
+        _tiny_png() + b"x",
+        None,
+    )
+    n = db.stickers.delete_all_for_identity(identity)
+    assert n == 2
+    assert db.stickers.count_for_identity(identity) == 0
+
+
+def test_wrong_identity_delete(db):
+    identity = "01" * 16
+    other = "02" * 16
+    row = db.stickers.insert(identity, None, "png", _tiny_png(), None)
+    assert db.stickers.delete(row["id"], other) is False
+
+
+def test_get_row_image_blob(db):
+    identity = "aa" * 16
+    raw = _tiny_png()
+    ins = db.stickers.insert(identity, "n", "png", raw, "msg1")
+    full = db.stickers.get_row(ins["id"], identity)
+    assert full["image_blob"] == raw
+    assert full["source_message_hash"] == "msg1"
+
+
+def test_update_name(db):
+    identity = "bb" * 16
+    ins = db.stickers.insert(identity, "old", "png", _tiny_png(), None)
+    assert db.stickers.update_name(ins["id"], identity, "new") is True
+    row = db.stickers.get_row(ins["id"], identity)
+    assert row["name"] == "new"
+
+
+def test_export_and_import_roundtrip(db):
+    identity = "cc" * 16
+    raw = _tiny_png()
+    db.stickers.insert(identity, "a", "png", raw, None)
+    payloads = db.stickers.export_payloads_for_identity(identity)
+    assert len(payloads) == 1
+    assert base64.b64decode(payloads[0]["image_bytes"]) == raw
+
+    other = "dd" * 16
+    items = [
+        {
+            "name": "x",
+            "image_type": "png",
+            "image_bytes_b64": payloads[0]["image_bytes"],
+            "source_message_hash": None,
+        },
+    ]
+    r = db.stickers.import_payloads(other, items, replace_duplicates=False)
+    assert r["imported"] == 1
+    assert db.stickers.count_for_identity(other) == 1
+    r2 = db.stickers.import_payloads(other, items, replace_duplicates=False)
+    assert r2["skipped_duplicates"] == 1
+
+
+def test_import_replace_duplicate(db):
+    identity = "ee" * 16
+    raw = _tiny_png()
+    db.stickers.insert(identity, "first", "png", raw, None)
+    items = [
+        {
+            "name": "second",
+            "image_type": "png",
+            "image_bytes_b64": base64.b64encode(raw).decode("ascii"),
+            "source_message_hash": None,
+        },
+    ]
+    r = db.stickers.import_payloads(identity, items, replace_duplicates=True)
+    assert r["imported"] == 1
+    rows = db.stickers.list_for_identity(identity)
+    assert len(rows) == 1
+    assert rows[0]["name"] == "second"
+
+
+def test_import_invalid_base64_skipped(db):
+    identity = "ff" * 16
+    items = [
+        {
+            "name": "x",
+            "image_type": "png",
+            "image_bytes_b64": "!!!not-base64!!!",
+            "source_message_hash": None,
+        },
+    ]
+    r = db.stickers.import_payloads(identity, items, replace_duplicates=False)
+    assert r["skipped_invalid"] >= 1
+
+
+def test_sticker_limit(db, monkeypatch):
+    from meshchatx.src.backend import sticker_utils
+
+    monkeypatch.setattr(sticker_utils, "MAX_STICKERS_PER_IDENTITY", 2)
+    identity = "11" * 16
+    db.stickers.insert(identity, None, "png", _tiny_png(), None)
+    db.stickers.insert(identity, None, "png", _tiny_png() + b"y", None)
+    with pytest.raises(ValueError, match="sticker_limit"):
+        db.stickers.insert(identity, None, "png", _tiny_png() + b"z", None)