-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_protocol.py
More file actions
53 lines (34 loc) · 1.57 KB
/
Copy pathtest_protocol.py
File metadata and controls
53 lines (34 loc) · 1.57 KB
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
import pytest
from pyritone.protocol import decode_line, decode_message, encode_line, encode_message, new_request
def test_new_request_shape():
request = new_request("ping", {})
assert request["type"] == "request"
assert request["method"] == "ping"
assert request["params"] == {}
assert isinstance(request["id"], str)
def test_new_request_honors_explicit_request_id_and_default_params():
request = new_request("ping", request_id="req-123")
assert request["id"] == "req-123"
assert request["params"] == {}
def test_encode_line_adds_newline_framing():
payload = {"type": "response", "id": "abc", "ok": True, "result": {"pong": True}}
encoded = encode_line(payload)
assert isinstance(encoded, bytes)
assert encoded.endswith(b"\n")
def test_encode_decode_message_roundtrip():
payload = {"type": "event", "event": "task.progress", "data": {"task_id": "x"}}
encoded = encode_message(payload)
decoded = decode_message(encoded)
assert decoded == payload
def test_encode_decode_roundtrip():
payload = {"type": "response", "id": "abc", "ok": True, "result": {"pong": True}}
encoded = encode_line(payload)
decoded = decode_line(encoded)
assert decoded == payload
def test_decode_line_accepts_text_input():
payload = decode_line('{"type":"event","event":"task.completed","data":{"task_id":"abc"}}')
assert payload["type"] == "event"
assert payload["data"]["task_id"] == "abc"
def test_decode_line_rejects_non_object_payloads():
with pytest.raises(ValueError, match="JSON object"):
decode_line("[]")