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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
#
# gensio - A library for abstracting stream I/O
# Copyright (C) 2018 Corey Minyard <minyard@acm.org>
#
# SPDX-License-Identifier: GPL-2.0-only
#
from utils import *
import gensio
class SigRspHandler:
def __init__(self, o, sigval):
self.sigval = sigval
self.waiter = gensio.waiter(o)
return
def control_done(self, io, err, value):
if (err):
raise Exception("Error getting signature: %s" % err)
value = value.decode(encoding='utf-8')
if (value != self.sigval):
raise Exception("Signature value was '%s', expected '%s'" %
(value, self.sigval))
self.waiter.wake();
return
def signature(self, sio, err, value):
if (err):
raise Exception("Error getting signature: %s" % err)
value = value.decode(encoding='utf-8')
if (value != self.sigval):
raise Exception("Signature value was '%s', expected '%s'" %
(value, self.sigval))
self.waiter.wake();
return
def wait_timeout(self, timeout):
return self.waiter.wait_timeout(1, timeout)
class CtrlRspHandler:
def __init__(self, o, val):
self.val = val
self.waiter = gensio.waiter(o)
return
def control_done(self, io, err, value):
if (err):
raise Exception("Error getting signature: %s" % err)
value = value.decode(encoding='utf-8')
if (value != str(self.val)):
raise Exception("Value was '%s', expected '%s'" %
(value, self.val))
self.waiter.wake();
return
def wait_timeout(self, timeout):
return self.waiter.wait_timeout(1, timeout)
import sys
def do_telnet_test(io1, io2):
# Modemstate must be the first test
io1.handler.set_expected_modemstate(0)
io1.read_cb_enable(True);
io2.control(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_CONTROL_SER_SEND_MODEMSTATE, "0")
if (io1.handler.wait_timeout(2000) == 0):
raise Exception("%s: %s: Timed out waiting for telnet modemstate 1" %
("test open", io1.handler.name))
do_test(io1, io2)
io1.read_cb_enable(True);
io2.read_cb_enable(True);
io2.handler.set_expected_modemstate_mask(3)
h = SigRspHandler(o, "3")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_SET_MODEMSTATE_MASK, "3",
h, -1)
if (io2.handler.wait_timeout(2000) == 0):
raise Exception("%s: %s: Timed out waiting for telnet modemstate 2" %
("test open", io1.handler.name))
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client telnet modemstate rsp")
io2.handler.set_expected_linestate_mask(7)
h = SigRspHandler(o, "7")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_SET_LINESTATE_MASK, "7",
h, -1)
if (io2.handler.wait_timeout(2000) == 0):
raise Exception("%s: %s: Timed out waiting for telnet linestate 2" %
("test open", io1.handler.name))
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client telnet linestate rsp")
io2.handler.set_expected_win_size(12, 83)
io1.control(0, False, gensio.GENSIO_CONTROL_WIN_SIZE, "12:83");
if (io2.handler.wait_timeout(2000) == 0):
raise Exception("%s: Timed out waiting for telnet win size" %
io1.handler.name)
h = SigRspHandler(o, "testsig")
io2.handler.set_expected_sig_server_cb("testsig")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_SIGNATURE, "testsig", h, -1)
if (h.wait_timeout(1000) == 0):
raise Exception("Timeout waiting for signature")
h = CtrlRspHandler(o, 2000)
io2.handler.set_expected_server_cb("baud", 1000, 2000)
io1.acontrol(0, gensio.GENSIO_CONTROL_SET, gensio.GENSIO_ACONTROL_SER_BAUD,
"1000", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server baud set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client baud response")
h = CtrlRspHandler(o, 6)
io2.handler.set_expected_server_cb("datasize", 5, 6)
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_DATASIZE,
"5", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server datasize set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client datasize response")
h = CtrlRspHandler(o, "space")
io2.handler.set_expected_server_cb("parity", gensio.GENSIO_SER_PARITY_NONE,
"space")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_PARITY,
"none", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server parity set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client parity response")
h = CtrlRspHandler(o, 1)
io2.handler.set_expected_server_cb("stopbits", 2, 1)
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_STOPBITS,
"2", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server stopbits set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client stopbits response")
h = CtrlRspHandler(o, "xonxoff")
io2.handler.set_expected_server_cb("flowcontrol",
gensio.GENSIO_SER_FLOWCONTROL_NONE,
"xonxoff")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_FLOWCONTROL,
"none", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server flowcontrol set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client flowcontrol response")
h = CtrlRspHandler(o, "dsr")
io2.handler.set_expected_server_cb("iflowcontrol",
gensio.GENSIO_SER_FLOWCONTROL_DCD,
"dsr")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_IFLOWCONTROL,
"dcd", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server flowcontrol set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client flowcontrol response")
h = CtrlRspHandler(o, "off")
io2.handler.set_expected_server_cb("sbreak",
gensio.GENSIO_SER_ON,
"off")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_SBREAK,
"on", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server sbreak set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client sbreak response")
h = CtrlRspHandler(o, "on")
io2.handler.set_expected_server_cb("dtr",
gensio.GENSIO_SER_OFF,
"on")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_DTR,
"off", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server dtr set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client dtr response")
h = CtrlRspHandler(o, "on")
io2.handler.set_expected_server_cb("rts",
gensio.GENSIO_SER_OFF,
"on")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_RTS,
"off", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server rts set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client rts response")
h = CtrlRspHandler(o, "both")
io2.handler.set_expected_server_cb("flush",
gensio.GENSIO_SER_FLUSH_BOTH,
"both")
io1.acontrol(0, gensio.GENSIO_CONTROL_SET,
gensio.GENSIO_ACONTROL_SER_FLUSH,
"both", h, -1)
if io2.handler.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for server flush set")
if h.wait_timeout(1000) == 0:
raise Exception("Timeout waiting for client flush response")
io1.read_cb_enable(False)
io2.read_cb_enable(False)
return
print("Test accept telnet")
TestAccept(o, "telnet(rfc2217,winsize),tcp,localhost,",
"telnet(rfc2217=true,winsize),tcp,localhost,0", do_telnet_test)
del o
test_shutdown()
|
