File: test_websocket_writer.py

package info (click to toggle)
python-aiohttp 3.8.4-1%2Bdeb12u1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm
  • size: 14,492 kB
  • sloc: python: 41,859; ansic: 25,006; makefile: 369; javascript: 31
file content (106 lines) | stat: -rw-r--r-- 3,356 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import random
from unittest import mock

import pytest

from aiohttp.http import WebSocketWriter
from aiohttp.test_utils import make_mocked_coro


@pytest.fixture
def protocol():
    ret = mock.Mock()
    ret._drain_helper = make_mocked_coro()
    return ret


@pytest.fixture
def transport():
    ret = mock.Mock()
    ret.is_closing.return_value = False
    return ret


@pytest.fixture
def writer(protocol, transport):
    return WebSocketWriter(protocol, transport, use_mask=False)


async def test_pong(writer) -> None:
    await writer.pong()
    writer.transport.write.assert_called_with(b"\x8a\x00")


async def test_ping(writer) -> None:
    await writer.ping()
    writer.transport.write.assert_called_with(b"\x89\x00")


async def test_send_text(writer) -> None:
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\x81\x04text")


async def test_send_binary(writer) -> None:
    await writer.send("binary", True)
    writer.transport.write.assert_called_with(b"\x82\x06binary")


async def test_send_binary_long(writer) -> None:
    await writer.send(b"b" * 127, True)
    assert writer.transport.write.call_args[0][0].startswith(b"\x82~\x00\x7fb")


async def test_send_binary_very_long(writer) -> None:
    await writer.send(b"b" * 65537, True)
    assert (
        writer.transport.write.call_args_list[0][0][0]
        == b"\x82\x7f\x00\x00\x00\x00\x00\x01\x00\x01"
    )
    assert writer.transport.write.call_args_list[1][0][0] == b"b" * 65537


async def test_close(writer) -> None:
    await writer.close(1001, "msg")
    writer.transport.write.assert_called_with(b"\x88\x05\x03\xe9msg")

    await writer.close(1001, b"msg")
    writer.transport.write.assert_called_with(b"\x88\x05\x03\xe9msg")

    # Test that Service Restart close code is also supported
    await writer.close(1012, b"msg")
    writer.transport.write.assert_called_with(b"\x88\x05\x03\xf4msg")


async def test_send_text_masked(protocol, transport) -> None:
    writer = WebSocketWriter(
        protocol, transport, use_mask=True, random=random.Random(123)
    )
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\x81\x84\rg\xb3fy\x02\xcb\x12")


async def test_send_compress_text(protocol, transport) -> None:
    writer = WebSocketWriter(protocol, transport, compress=15)
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00")
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\xc1\x05*\x01b\x00\x00")


async def test_send_compress_text_notakeover(protocol, transport) -> None:
    writer = WebSocketWriter(protocol, transport, compress=15, notakeover=True)
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00")
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00")


async def test_send_compress_text_per_message(protocol, transport) -> None:
    writer = WebSocketWriter(protocol, transport)
    await writer.send(b"text", compress=15)
    writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00")
    await writer.send(b"text")
    writer.transport.write.assert_called_with(b"\x81\x04text")
    await writer.send(b"text", compress=15)
    writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00")