# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """Envelope parsing across both message shapes.""" from __future__ import annotations import json import pytest from signal_bot.models import parse_envelope from tests.conftest import WATCHED_GROUP, envelope def test_parses_group_data_message() -> None: message = parse_envelope(envelope(text="!cat")) assert message is not None assert message.text == "!cat" assert message.group_id == WATCHED_GROUP assert message.source_uuid == "uuid-alice" assert message.source_name == "Alice" assert not message.is_direct_message def test_parses_own_account_sync_message() -> None: message = parse_envelope(envelope(text="!dog", sync=True)) assert message is not None assert message.text == "!dog" assert message.group_id == WATCHED_GROUP @pytest.mark.parametrize("shape", [{"sync": False}, {"sync": True}]) def test_detects_sticker_in_both_shapes(shape: dict[str, bool]) -> None: """The original bot only ever looked under dataMessage, so its own stickers were never filtered out of the tally.""" message = parse_envelope(envelope(extra="sticker", **shape)) assert message is not None assert message.has_sticker @pytest.mark.parametrize("shape", [{"sync": False}, {"sync": True}]) def test_detects_reaction_in_both_shapes(shape: dict[str, bool]) -> None: message = parse_envelope(envelope(extra="reaction", **shape)) assert message is not None assert message.is_reaction def test_direct_message_has_no_group() -> None: message = parse_envelope(envelope(group_id=None)) assert message is not None assert message.group_id is None assert message.is_direct_message def test_missing_text_is_none() -> None: message = parse_envelope(envelope(text=None)) assert message is not None assert message.text is None def test_non_string_text_is_none() -> None: raw = json.dumps({"envelope": {"dataMessage": {"message": 42}}}) message = parse_envelope(raw) assert message is not None assert message.text is None @pytest.mark.parametrize( "raw", [ "not json at all", json.dumps(["a", "list"]), json.dumps({"envelope": "not a dict"}), json.dumps({"account": "+1"}), json.dumps({"envelope": {"receiptMessage": {"when": 1}}}), json.dumps({"envelope": {"syncMessage": "not a dict"}}), json.dumps({"envelope": {"syncMessage": {"readMessages": []}}}), ], ) def test_unparseable_frames_return_none(raw: str) -> None: assert parse_envelope(raw) is None def test_accepts_bytes() -> None: message = parse_envelope(envelope(text="hi").encode()) assert message is not None assert message.text == "hi" def test_invalid_utf8_bytes_return_none() -> None: assert parse_envelope(b"\xff\xfe not utf8") is None @pytest.mark.parametrize( "group_info", ["not a dict", {"type": "DELIVER"}, {"groupId": ""}, {"groupId": 12}], ) def test_malformed_group_info_reads_as_direct_message(group_info: object) -> None: raw = json.dumps({"envelope": {"dataMessage": {"message": "hi", "groupInfo": group_info}}}) message = parse_envelope(raw) assert message is not None assert message.is_direct_message # The exact shape of a live frame captured on 2026-08-30: a text reply to # somebody else's photo. The photo is nested inside `quote`, not at the top # level, and counting it would credit the replier with a photo they did not # post -- the same wrong-nesting-level mistake that once made the bot ignore # its own stickers. REPLY_TO_A_PHOTO = json.dumps( { "envelope": { "sourceUuid": "u1", "sourceName": "Mithra", "sourceNumber": None, "timestamp": 1756576440000, "dataMessage": { "timestamp": 1756576440000, "message": "ha", "viewOnce": False, "expiresInSeconds": 0, "isExpirationUpdate": False, "groupInfo": { "groupId": WATCHED_GROUP, "groupName": "x", "revision": 1, "type": "DELIVER", }, "quote": { "id": 1756576000000, "author": "x", "authorNumber": "+15555550123", "authorUuid": "u2", "text": "", "attachments": [ { "contentType": "image/jpeg", "filename": "photo.jpg", "thumbnail": { "contentType": "image/jpeg", "caption": None, "id": "t", "size": 1, "width": 1, "height": 1, "filename": "thumb.jpg", "uploadTimestamp": 1756576000000, }, } ], }, }, } } ) def test_a_quoted_photo_is_not_the_replier_s_photo() -> None: """`quote` carries its own attachments array. Only the top level counts.""" message = parse_envelope(REPLY_TO_A_PHOTO) assert message is not None assert message.attachment_types == () assert message.is_reply is True assert message.text == "ha"