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
|
import json
from tests_python.debugger_unittest import ReaderThread, IS_JYTHON
import pytest
import socket
from _pydev_bundle import pydev_localhost
from _pydevd_bundle.pydevd_comm import start_client
ABORT_CONNECTION = "ABORT_CONNECTION"
pytestmark = pytest.mark.skipif(IS_JYTHON, reason="Getting the actual port does not work when the port is == 0.")
class _DummySocket(object):
def __init__(self):
self._sock_for_reader_thread = None
self._sock_for_fixture_test = None
self._socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._localhost = host = pydev_localhost.get_localhost()
self._socket_server.bind((host, 0))
self._socket_server.listen(1)
def recv(self, *args, **kwargs):
if self._sock_for_reader_thread is None:
sock, _addr = self._socket_server.accept()
self._sock_for_reader_thread = sock
return self._sock_for_reader_thread.recv(*args, **kwargs)
def put(self, msg):
if not isinstance(msg, bytes):
msg = msg.encode("utf-8")
if self._sock_for_fixture_test is None:
self._sock_for_fixture_test = start_client(*self._socket_server.getsockname())
self._sock_for_fixture_test.sendall(msg)
def close(self):
self._socket_server.close()
if self._sock_for_fixture_test is not None:
self._sock_for_fixture_test.close()
if self._sock_for_reader_thread is not None:
self._sock_for_reader_thread.close()
@pytest.yield_fixture
def _dummy_socket():
sock = _DummySocket()
yield sock
sock.close()
def test_fixture_reader_thread1(_dummy_socket):
sock = _dummy_socket
reader_thread = ReaderThread(sock)
reader_thread.start()
json_part = json.dumps({"key": "val"})
json_part = json_part.replace(":", ":\n")
msg = json_part
msg = ("Content-Length: %s\r\n\r\n%s" % (len(msg), msg)).encode("utf-8")
# Check that receiving 2 messages at a time we're able to properly deal
# with each one.
sock.put(msg + msg)
assert reader_thread.get_next_message("check 1") == json_part
assert reader_thread.get_next_message("check 2") == json_part
def test_fixture_reader_thread2(_dummy_socket):
sock = _DummySocket()
reader_thread = ReaderThread(sock)
reader_thread.start()
json_part = json.dumps({"key": "val"})
json_part = json_part.replace(":", ":\n")
msg = json_part
http = "Content-Length: %s\r\n\r\n%s" % (len(msg), msg)
sock.put("msg1\nmsg2\nmsg3\n" + http + http)
assert reader_thread.get_next_message("check 1") == "msg1"
assert reader_thread.get_next_message("check 2") == "msg2"
assert reader_thread.get_next_message("check 3") == "msg3"
assert reader_thread.get_next_message("check 4") == json_part
assert reader_thread.get_next_message("check 5") == json_part
def test_fixture_reader_thread3(_dummy_socket):
sock = _DummySocket()
reader_thread = ReaderThread(sock)
reader_thread.start()
msg = "aaaaaaabbbbbbbccccccc"
http = "Content-Length: %s\r\n\r\n%s" % (len(msg), msg)
http *= 2
initial = http
for i in range(1, len(http)):
while http:
sock.put(http[:i])
http = http[i:]
assert reader_thread.get_next_message("check 1: %s" % i) == msg
assert reader_thread.get_next_message("check 2: %s" % i) == msg
http = initial
|
