python-jsonrpc-server/test/test_streams.py at 96038995ed8f3d8672b655c79ba042b54d1c1a10 · palantir/python-jsonrpc-server · GitHub

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
# Copyright 2018 Palantir Technologies, Inc.
# pylint: disable=redefined-outer-name
from io import BytesIO
import mock
import pytest

from jsonrpc.streams import JsonRpcStreamReader, JsonRpcStreamWriter


@pytest.fixture()
def rfile():
    return BytesIO()


@pytest.fixture()
def wfile():
    return BytesIO()


@pytest.fixture()
def reader(rfile):
    return JsonRpcStreamReader(rfile)


@pytest.fixture()
def writer(wfile):
    return JsonRpcStreamWriter(wfile, sort_keys=True)


@pytest.mark.parametrize("data", [
    (
        b'Content-Length: 49\r\n'
        b'Content-Type: application/vscode-jsonrpc; charset=utf8\r\n'
        b'\r\n'
        b'{"id": "hello", "method": "method", "params": {}}'
    ),
    (
        b'Content-Type: application/vscode-jsonrpc; charset=utf8\r\n'
        b'Content-Length: 49\r\n'
        b'\r\n'
        b'{"id": "hello", "method": "method", "params": {}}'
    ),
], ids=["Content-Length first", "Content-Length middle"])
def test_reader(rfile, reader, data):
    rfile.write(data)
    rfile.seek(0)

    consumer = mock.Mock()
    reader.listen(consumer)

    consumer.assert_called_once_with({
        'id': 'hello',
        'method': 'method',
        'params': {}
    })


@pytest.mark.parametrize("data", [
    (
        b'hello'
    ),
    (
        b'Content-Type: application/vscode-jsonrpc; charset=utf8\r\n'
        b'Content-Length: NOT_AN_INT\r\n'
        b'\r\n'
    ),
    (
        b'Content-Length: 8\r\n'
        b'Content-Type: application/vscode-jsonrpc; charset=utf8\r\n'
        b'\r\n'
        b'{hello}}'
    ),
], ids=["hello", "Invalid Content-Length", "Bad json"])
def test_reader_bad_message(rfile, reader, data):
    rfile.write(data)
    rfile.seek(0)

    # Ensure the listener doesn't throw
    consumer = mock.Mock()
    reader.listen(consumer)
    consumer.assert_not_called()


def test_writer(wfile, writer):
    writer.write({
        'id': 'hello',
        'method': 'method',
        'params': {}
    })
    assert wfile.getvalue() == (
        b'Content-Length: 49\r\n'
        b'Content-Type: application/vscode-jsonrpc; charset=utf8\r\n'
        b'\r\n'
        b'{"id": "hello", "method": "method", "params": {}}'
    )


def test_writer_bad_message(wfile, writer):
    # A datetime isn't serializable, ensure the write method doesn't throw
    import datetime
    writer.write(datetime.datetime.now())
    assert wfile.getvalue() == b''