miguelgrinberg.python-socketio/tests/common/test_packet.py

import pytest

from socketio import packet


class TestPacket:
    def test_encode_default_packet(self):
        pkt = packet.Packet()
        assert pkt.packet_type == packet.EVENT
        assert pkt.data is None
        assert pkt.namespace is None
        assert pkt.id is None
        assert pkt.attachment_count == 0
        assert pkt.encode() == "2"

    def test_decode_default_packet(self):
        pkt = packet.Packet(encoded_packet="2")
        assert pkt.encode(), "2"

    def test_encode_text_event_packet(self):
        pkt = packet.Packet(packet_type=packet.EVENT, data=["foo"])
        assert pkt.packet_type == packet.EVENT
        assert pkt.data == ["foo"]
        assert pkt.encode() == '2["foo"]'

    def test_decode_text_event_packet(self):
        pkt = packet.Packet(encoded_packet='2["foo"]')
        assert pkt.packet_type == packet.EVENT
        assert pkt.data == ["foo"]
        assert pkt.encode() == '2["foo"]'

    def test_decode_empty_event_packet(self):
        pkt = packet.Packet(encoded_packet="1")
        assert pkt.packet_type == packet.DISCONNECT
        # same thing, but with a numeric payload
        pkt = packet.Packet(encoded_packet=1)
        assert pkt.packet_type == packet.DISCONNECT

    def test_encode_binary_event_packet(self):
        pkt = packet.Packet(packet_type=packet.EVENT, data=b"1234")
        assert pkt.packet_type == packet.BINARY_EVENT
        assert pkt.data == b"1234"
        a = ['51-{"_placeholder":true,"num":0}', b"1234"]
        b = ['51-{"num":0,"_placeholder":true}', b"1234"]
        encoded_packet = pkt.encode()
        assert encoded_packet == a or encoded_packet == b

    def test_decode_binary_event_packet(self):
        pkt = packet.Packet(encoded_packet='51-{"_placeholder":true,"num":0}')
        assert pkt.add_attachment(b"1234")
        assert pkt.packet_type == packet.BINARY_EVENT
        assert pkt.data == b"1234"

    def test_encode_text_ack_packet(self):
        pkt = packet.Packet(packet_type=packet.ACK, data=["foo"])
        assert pkt.packet_type == packet.ACK
        assert pkt.data == ["foo"]
        assert pkt.encode() == '3["foo"]'

    def test_decode_text_ack_packet(self):
        pkt = packet.Packet(encoded_packet='3["foo"]')
        assert pkt.packet_type == packet.ACK
        assert pkt.data == ["foo"]
        assert pkt.encode() == '3["foo"]'

    def test_encode_binary_ack_packet(self):
        pkt = packet.Packet(packet_type=packet.ACK, data=b"1234")
        assert pkt.packet_type == packet.BINARY_ACK
        assert pkt.data == b"1234"
        a = ['61-{"_placeholder":true,"num":0}', b"1234"]
        b = ['61-{"num":0,"_placeholder":true}', b"1234"]
        encoded_packet = pkt.encode()
        assert encoded_packet == a or encoded_packet == b

    def test_decode_binary_ack_packet(self):
        pkt = packet.Packet(encoded_packet='61-{"_placeholder":true,"num":0}')
        assert pkt.add_attachment(b"1234")
        assert pkt.packet_type == packet.BINARY_ACK
        assert pkt.data == b"1234"

    def test_invalid_binary_packet(self):
        with pytest.raises(ValueError):
            packet.Packet(packet_type=packet.CONNECT_ERROR, data=b"123")

    def test_encode_namespace(self):
        pkt = packet.Packet(
            packet_type=packet.EVENT,
            data=["foo"],
            namespace="/bar",
        )
        assert pkt.namespace == "/bar"
        assert pkt.encode() == '2/bar,["foo"]'

    def test_decode_namespace(self):
        pkt = packet.Packet(encoded_packet='2/bar,["foo"]')
        assert pkt.namespace == "/bar"
        assert pkt.encode() == '2/bar,["foo"]'

    def test_decode_namespace_with_query_string(self):
        # some Socket.IO clients mistakenly attach the query string to the
        # namespace
        pkt = packet.Packet(encoded_packet='2/bar?a=b,["foo"]')
        assert pkt.namespace == "/bar"
        assert pkt.encode() == '2/bar,["foo"]'

    def test_encode_namespace_no_data(self):
        pkt = packet.Packet(packet_type=packet.EVENT, namespace="/bar")
        assert pkt.encode() == "2/bar,"

    def test_decode_namespace_no_data(self):
        pkt = packet.Packet(encoded_packet="2/bar,")
        assert pkt.namespace == "/bar"
        assert pkt.data is None
        assert pkt.encode() == "2/bar,"

    def test_encode_namespace_with_hyphens(self):
        pkt = packet.Packet(
            packet_type=packet.EVENT,
            data=["foo"],
            namespace="/b-a-r",
        )
        assert pkt.namespace == "/b-a-r"
        assert pkt.encode() == '2/b-a-r,["foo"]'

    def test_decode_namespace_with_hyphens(self):
        pkt = packet.Packet(encoded_packet='2/b-a-r,["foo"]')
        assert pkt.namespace == "/b-a-r"
        assert pkt.encode() == '2/b-a-r,["foo"]'

    def test_encode_event_with_hyphens(self):
        pkt = packet.Packet(packet_type=packet.EVENT, data=["f-o-o"])
        assert pkt.namespace is None
        assert pkt.encode() == '2["f-o-o"]'

    def test_decode_event_with_hyphens(self):
        pkt = packet.Packet(encoded_packet='2["f-o-o"]')
        assert pkt.namespace is None
        assert pkt.encode() == '2["f-o-o"]'

    def test_encode_id(self):
        pkt = packet.Packet(packet_type=packet.EVENT, data=["foo"], id=123)
        assert pkt.id == 123
        assert pkt.encode() == '2123["foo"]'

    def test_decode_id(self):
        pkt = packet.Packet(encoded_packet='2123["foo"]')
        assert pkt.id == 123
        assert pkt.encode() == '2123["foo"]'

    def test_decode_id_long(self):
        pkt = packet.Packet(encoded_packet="2" + "1" * 100 + '["foo"]')
        assert pkt.id == int("1" * 100)
        assert pkt.data == ["foo"]

    def test_decode_id_too_long(self):
        with pytest.raises(ValueError):
            packet.Packet(encoded_packet="2" + "1" * 101)
        with pytest.raises(ValueError):
            packet.Packet(encoded_packet="2" + "1" * 101 + '["foo"]')

    def test_encode_id_no_data(self):
        pkt = packet.Packet(packet_type=packet.EVENT, id=123)
        assert pkt.id == 123
        assert pkt.data is None
        assert pkt.encode() == "2123"

    def test_decode_id_no_data(self):
        pkt = packet.Packet(encoded_packet="2123")
        assert pkt.id == 123
        assert pkt.data is None
        assert pkt.encode() == "2123"

    def test_encode_namespace_and_id(self):
        pkt = packet.Packet(
            packet_type=packet.EVENT,
            data=["foo"],
            namespace="/bar",
            id=123,
        )
        assert pkt.namespace == "/bar"
        assert pkt.id == 123
        assert pkt.encode() == '2/bar,123["foo"]'

    def test_decode_namespace_and_id(self):
        pkt = packet.Packet(encoded_packet='2/bar,123["foo"]')
        assert pkt.namespace == "/bar"
        assert pkt.id == 123
        assert pkt.encode() == '2/bar,123["foo"]'

    def test_encode_many_binary(self):
        pkt = packet.Packet(
            packet_type=packet.EVENT,
            data={"a": "123", "b": b"456", "c": [b"789", 123]},
        )
        assert pkt.packet_type == packet.BINARY_EVENT
        ep = pkt.encode()
        assert len(ep) == 3
        assert b"456" in ep
        assert b"789" in ep

    def test_encode_many_binary_ack(self):
        pkt = packet.Packet(
            packet_type=packet.ACK,
            data={"a": "123", "b": b"456", "c": [b"789", 123]},
        )
        assert pkt.packet_type == packet.BINARY_ACK
        ep = pkt.encode()
        assert len(ep) == 3
        assert b"456" in ep
        assert b"789" in ep

    def test_decode_many_binary(self):
        pkt = packet.Packet(
            encoded_packet=(
                '52-{"a":"123","b":{"_placeholder":true,"num":0},'
                '"c":[{"_placeholder":true,"num":1},123]}'
            )
        )
        assert not pkt.add_attachment(b"456")
        assert pkt.add_attachment(b"789")
        assert pkt.packet_type == packet.BINARY_EVENT
        assert pkt.data["a"] == "123"
        assert pkt.data["b"] == b"456"
        assert pkt.data["c"] == [b"789", 123]

    def test_decode_many_binary_ack(self):
        pkt = packet.Packet(
            encoded_packet=(
                '62-{"a":"123","b":{"_placeholder":true,"num":0},'
                '"c":[{"_placeholder":true,"num":1},123]}'
            )
        )
        assert not pkt.add_attachment(b"456")
        assert pkt.add_attachment(b"789")
        assert pkt.packet_type == packet.BINARY_ACK
        assert pkt.data["a"] == "123"
        assert pkt.data["b"] == b"456"
        assert pkt.data["c"] == [b"789", 123]

    def test_decode_too_many_binary_packets(self):
        pkt = packet.Packet(
            encoded_packet=(
                '62-{"a":"123","b":{"_placeholder":true,"num":0},'
                '"c":[{"_placeholder":true,"num":1},123]}'
            )
        )
        assert not pkt.add_attachment(b"456")
        assert pkt.add_attachment(b"789")
        with pytest.raises(ValueError):
            pkt.add_attachment(b"123")

    def test_decode_attachment_count_too_long(self):
        with pytest.raises(ValueError):
            packet.Packet(encoded_packet="6" + ("1" * 11) + '-{"a":"123"}')

    def test_decode_dash_in_payload(self):
        pkt = packet.Packet(encoded_packet='6{"a":"0123456789-"}')
        assert pkt.data["a"] == "0123456789-"
        assert pkt.attachment_count == 0

    def test_data_is_binary_list(self):
        pkt = packet.Packet()
        assert not pkt._data_is_binary(["foo"])
        assert not pkt._data_is_binary([])
        assert pkt._data_is_binary([b"foo"])
        assert pkt._data_is_binary(["foo", b"bar"])

    def test_data_is_binary_dict(self):
        pkt = packet.Packet()
        assert not pkt._data_is_binary({"a": "foo"})
        assert not pkt._data_is_binary({})
        assert pkt._data_is_binary({"a": b"foo"})
        assert pkt._data_is_binary({"a": "foo", "b": b"bar"})