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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
from pathlib import Path
import h2
import pytest
from nio.client import HttpClient, RequestInfo, TransportType
from nio.exceptions import LocalProtocolError
from nio.http import Http2Response
from nio.responses import LoginResponse, SyncResponse
class TestClass:
example_response_headers = [(":status", "200"), ("server", "fake-serv/0.1.0")]
def login_response(self, stream_id, frame_factory):
f = frame_factory.build_headers_frame(
headers=self.example_response_headers, stream_id=stream_id
)
login_body = Path("tests/data/login_response.json").read_bytes()
data = frame_factory.build_data_frame(
data=login_body, stream_id=stream_id, flags=["END_STREAM"]
)
return f.serialize() + data.serialize()
def sync_response(self, stream_id, frame_factory):
f = frame_factory.build_headers_frame(
headers=self.example_response_headers, stream_id=stream_id
)
body = Path("tests/data/sync.json").read_bytes()
data = frame_factory.build_data_frame(
data=body, stream_id=stream_id, flags=["END_STREAM"]
)
return f.serialize() + data.serialize()
def test_client_lag(self, frame_factory):
client = HttpClient("localhost", "example")
client.connect(TransportType.HTTP2)
response = Http2Response()
response.send_time = 0
response.receive_time = 30
response.timeout = 25 * 1000
response2 = Http2Response()
response2.send_time = 0
response2.receive_time = 31
response2.timeout = 25 * 1000
client.connection._responses[response.uuid] = response
client.connection._responses[response2.uuid] = response2
typed_response = RequestInfo("sync", 25 * 1000)
client.requests_made[response.uuid] = typed_response
assert client.lag == 6
def test_client_local_error(self, frame_factory):
client = HttpClient("localhost", "example")
with pytest.raises(LocalProtocolError):
_uuid, _request = client.login("wordpass")
client.connect(TransportType.HTTP2)
_uuid, _request = client.login("wordpass")
with pytest.raises(LocalProtocolError):
_uuid, _request = client.sync()
client.receive(self.login_response(1, frame_factory))
client.next_response()
_uuid, _request = client.sync()
def test_client_receive(self, frame_factory):
client = HttpClient("localhost", "example")
client.connect(TransportType.HTTP2)
uuid, request = client.login("wordpass")
conf = h2.config.H2Configuration(client_side=True)
server = h2.connection.H2Connection(conf)
server.max_inbound_frame_size = 64 * 1024
server.initiate_connection()
server.receive_data(frame_factory.preamble())
server.receive_data(request)
# assert events[0].headers == []
client.receive(self.login_response(1, frame_factory))
response = client.next_response()
assert isinstance(response, LoginResponse)
assert response.uuid == uuid
uuid, request = client.sync()
server.receive_data(request)
client.receive(self.sync_response(3, frame_factory))
response = client.next_response()
assert isinstance(response, SyncResponse)
assert response.uuid == uuid
_sync_uuid, request = client.sync()
server.receive_data(request)
content = {"body": "test", "msgtype": "m.text"}
_send_uuid, _send_request = client.room_send(
"!test:localhost", "m.room.message", content
)
def test_frame_splitting(self, frame_factory):
client = HttpClient("localhost", "example")
data = client.connect(TransportType.HTTP2)
client.connection._connection.outbound_flow_control_window = 5
_uuid, request = client.login("wordpass")
assert client.connection._data_to_send
to_send = data + request
while to_send:
f = frame_factory.build_window_update_frame(
stream_id=0,
increment=5,
)
client.receive(f.serialize())
to_send = client.data_to_send()
assert not client.connection._data_to_send
|
