Coverage for tests/test_lexer.py: 100%

1"""Test BEEP lexing.""" 

2 

3from io import BytesIO 

4from unittest import TestCase 

5 

6from galaxy_graph import Connection, Node 

7 

8from galaxy_beep_frame.converter import Lexer, State, TokenType 

9from galaxy_beep_frame.exception import ( 

10 BadEndError, 

11 BadHeaderEndError, 

12 InvalidByteError, 

13 LexerError, 

14 NoCurrentTokenError, 

15 NodeNoneError, 

16 NotFoundError, 

17) 

18 

19 

20class TestLexer(TestCase): 

21 """Test BEEP conversion from bytes.""" 

22 

23 def test_lexer_bad_context(self) -> None: 

24 """Test creating token in a bad context.""" 

25 

26 lexer = Lexer() 

27 string = b"MSG 0 1" 

28 lexer.feed(string) 

29 

30 self.assertIs(lexer.state, State.MESSAGE) 

31 

32 # fake a " " reading, because current code is correct 

33 lexer.state = State.START 

34 # header is initialized but we are in the START state 

35 with self.assertRaises(LexerError): 

36 lexer._new_token( 

37 TokenType.CONTINUATION_INDICATOR, 

38 "has continuation indicator", 

39 ) 

40 

41 def test_lexer_bad_current_token(self) -> None: 

42 """ 

43 Test _ensure_current_token in bad condition. 

44 

45 As _ensure_current_token() is an internal method only use for 

46 static type analysis, this code is kinda useless, except it make 

47 a 100% code coverage possible. 

48 """ 

49 

50 lexer = Lexer() 

51 with self.assertRaises(NoCurrentTokenError): 

52 lexer._ensure_current_token() 

53 

54 def test_lexer_bad_data(self) -> None: 

55 """ 

56 Test _ensure_data in bad condition. 

57 

58 Same as test_lexer_bad_current_token. 

59 """ 

60 

61 lexer = Lexer() 

62 with self.assertRaises(NodeNoneError): 

63 lexer._ensure_data(Node()) 

64 

65 def test_bad_header_end(self) -> None: 

66 """ 

67 Test _header_end in bad condition. 

68 

69 As _header_end is only used in internal code: same than 

70 test_lexer_bad_current_token. 

71 """ 

72 

73 lexer = Lexer() 

74 with self.assertRaises(NodeNoneError): 

75 lexer._header_end(ord("\n")) 

76 

77 def test_bad_end(self) -> None: 

78 """ 

79 Test _end in bad condition. 

80 

81 Same as test_bad_header_end. 

82 """ 

83 

84 lexer = Lexer() 

85 lexer.end = b"END\r" 

86 with self.assertRaises(NodeNoneError): 

87 lexer._end(ord("\n")) 

88 

89 def test_bad_connection_to_humans(self) -> None: 

90 """ 

91 Test connections_to_humans when connections are not setup. 

92 

93 Again, this is intended for debugging purpose, so no real use 

94 here. 

95 """ 

96 

97 connection = Connection() 

98 lexer = Lexer() 

99 with self.assertRaises(LexerError): 

100 lexer.connections_to_human({connection}) 

101 

102 def test_lexer_empty(self) -> None: 

103 """Test lexing specific frame.""" 

104 

105 lexer = Lexer() 

106 with self.assertRaises(InvalidByteError): 

107 lexer.feed(b" ") 

108 

109 def test_lexer_bad_message_type(self) -> None: 

110 """Test lexing when no message type or bad value.""" 

111 

112 no_message_type = b"0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

113 not_message_type = b" 0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

114 empty_message_type = b" 0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

115 bad_message_type = b"- 0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

116 bad_message_type_2 = b"M- 0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

117 

118 with self.assertRaises(InvalidByteError): 

119 lexer = Lexer() 

120 lexer.feed(no_message_type) 

121 with self.assertRaises(InvalidByteError): 

122 lexer = Lexer() 

123 lexer.feed(not_message_type) 

124 with self.assertRaises(InvalidByteError): 

125 lexer = Lexer() 

126 lexer.feed(empty_message_type) 

127 with self.assertRaises(InvalidByteError): 

128 lexer = Lexer() 

129 lexer.feed(bad_message_type) 

130 with self.assertRaises(InvalidByteError): 

131 lexer = Lexer() 

132 lexer.feed(bad_message_type_2) 

133 

134 def test_lexer_bad_channel_number(self) -> None: 

135 """Test lexing when no channel number or bad value.""" 

136 

137 lexer = Lexer() 

138 

139 no_channel_number = b"MSG 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

140 not_channel_number = b"MSG 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

141 empty_channel_number = b"MSG 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

142 bad_channel_number = b"MSG - 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

143 

144 with self.assertRaises(InvalidByteError): 

145 lexer = Lexer() 

146 lexer.feed(no_channel_number) 

147 with self.assertRaises(NotFoundError): 

148 lexer = Lexer() 

149 lexer.feed(not_channel_number) 

150 with self.assertRaises(NotFoundError): 

151 lexer = Lexer() 

152 lexer.feed(empty_channel_number) 

153 with self.assertRaises(InvalidByteError): 

154 lexer = Lexer() 

155 lexer.feed(bad_channel_number) 

156 

157 def test_lexer_bad_message_number(self) -> None: 

158 """Test lexing when no message number or bad value.""" 

159 

160 no_message_number = b"MSG 0 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

161 not_message_number = b"MSG 0 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

162 empty_message_number = b"MSG 0 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

163 bad_message_number = b"MSG 0 - . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

164 

165 with self.assertRaises(InvalidByteError): 

166 lexer = Lexer() 

167 lexer.feed(no_message_number) 

168 with self.assertRaises(NotFoundError): 

169 lexer = Lexer() 

170 lexer.feed(not_message_number) 

171 with self.assertRaises(NotFoundError): 

172 lexer = Lexer() 

173 lexer.feed(empty_message_number) 

174 with self.assertRaises(InvalidByteError): 

175 lexer = Lexer() 

176 lexer.feed(bad_message_number) 

177 

178 def test_lexer_bad_continuation_indicator(self) -> None: 

179 """Test lexing when no continuation integer or bad value.""" 

180 

181 no_continuation_indicator = b"MSG 0 1 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

182 not_continuation_indicator = b"MSG 0 1 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

183 empty_continuation_indicator = b"MSG 0 0 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

184 bad_continuation_indicator = b"MSG 0 1 - 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

185 long_continuation_indicator = b"MSG 0 1 ** 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

186 

187 with self.assertRaises(InvalidByteError): 

188 lexer = Lexer() 

189 lexer.feed(no_continuation_indicator) 

190 with self.assertRaises(InvalidByteError): 

191 lexer = Lexer() 

192 lexer.feed(not_continuation_indicator) 

193 with self.assertRaises(InvalidByteError): 

194 lexer = Lexer() 

195 lexer.feed(empty_continuation_indicator) 

196 with self.assertRaises(InvalidByteError): 

197 lexer = Lexer() 

198 lexer.feed(bad_continuation_indicator) 

199 with self.assertRaises(InvalidByteError): 

200 lexer = Lexer() 

201 lexer.feed(long_continuation_indicator) 

202 

203 def test_lexer_bad_sequence_number(self) -> None: 

204 """Test lexing when no sequence number or bad value.""" 

205 

206 no_sequence_number = b"MSG 0 1 . 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

207 not_sequence_number = b"MSG 0 1 . 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

208 empty_sequence_number = b"MSG 0 1 . 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

209 bad_sequence_number = b"MSG 0 1 . - 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

210 

211 with self.assertRaises(InvalidByteError): 

212 lexer = Lexer() 

213 lexer.feed(no_sequence_number) 

214 with self.assertRaises(NotFoundError): 

215 lexer = Lexer() 

216 lexer.feed(not_sequence_number) 

217 with self.assertRaises(NotFoundError): 

218 lexer = Lexer() 

219 lexer.feed(empty_sequence_number) 

220 with self.assertRaises(InvalidByteError): 

221 lexer = Lexer() 

222 lexer.feed(bad_sequence_number) 

223 

224 def test_lexer_bad_size(self) -> None: 

225 """Test lexing when no size or bad value.""" 

226 

227 no_size = b"MSG 0 1 . 52\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

228 not_size = b"MSG 0 1 . 52 \r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

229 empty_size = b"MSG 0 1 . 52 \r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

230 bad_size = b"MSG 0 1 . 52 -\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

231 bad_start_header = b"MSG 0 1 . 52 120\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

232 bad_header = b"MSG 0 1 . 52 120\rContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

233 bad_end_header = b"MSG 0 1 . 52 120\r-Content-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

234 

235 with self.assertRaises(InvalidByteError): 

236 lexer = Lexer() 

237 lexer.feed(no_size) 

238 with self.assertRaises(BadEndError): 

239 lexer = Lexer() 

240 lexer.feed(not_size) 

241 with self.assertRaises(NotFoundError): 

242 lexer = Lexer() 

243 lexer.feed(empty_size) 

244 with self.assertRaises(InvalidByteError): 

245 lexer = Lexer() 

246 lexer.feed(bad_size) 

247 with self.assertRaises(InvalidByteError): 

248 lexer = Lexer() 

249 lexer.feed(bad_start_header) 

250 with self.assertRaises(BadHeaderEndError): 

251 lexer = Lexer() 

252 lexer.feed(bad_header) 

253 with self.assertRaises(BadHeaderEndError): 

254 lexer = Lexer() 

255 lexer.feed(bad_end_header) 

256 

257 def test_lexer_bad_answer_number(self) -> None: 

258 """Test lexing when no answer number or bad value.""" 

259 

260 not_answer_number = b"ANS 0 1 . 52 120 \r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

261 empty_answer_number = b"ANS 0 1 . 52 120 \r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

262 bad_answer_number = b"ANS 0 1 . 52 120 1-\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

263 

264 with self.assertRaises(NotFoundError): 

265 lexer = Lexer() 

266 lexer.feed(not_answer_number) 

267 with self.assertRaises(InvalidByteError): 

268 lexer = Lexer() 

269 lexer.feed(empty_answer_number) 

270 with self.assertRaises(InvalidByteError): 

271 lexer = Lexer() 

272 lexer.feed(bad_answer_number) 

273 

274 def test_lexer_bad_payload(self) -> None: 

275 """Test lexing when no payload or bad value.""" 

276 

277 empty_payload = b"MSG 0 1 . 52 1\r\nEND\r\n" # should be 0 

278 no_end = b"MSG 0 1 . 52 0\r\n-\r\n" 

279 bad_end = b"MSG 0 1 . 52 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nENT\r\n" 

280 

281 with self.assertRaises(BadEndError): 

282 lexer = Lexer() 

283 lexer.feed(empty_payload) 

284 with self.assertRaises(BadEndError): 

285 lexer = Lexer() 

286 lexer.feed(no_end) 

287 with self.assertRaises(BadEndError): 

288 lexer = Lexer() 

289 lexer.feed(bad_end) 

290 

291 def test_lexer_good(self) -> None: 

292 """Test lexing good format.""" 

293 

294 lexer = Lexer() 

295 

296 contents = [ 

297 b"MSG 0 1 . 42 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n", 

298 b"MSG 0 1 . 42 0\r\nEND\r\n", 

299 ] 

300 

301 for content in contents: 

302 lexer.feed(content) 

303 

304 def test_lexer_iterator(self) -> None: 

305 """Test lexing via the lex and lex_iterator function.""" 

306 

307 initial_str = b"" 

308 initial_str += b"RPY 0 0 . 0 110\r\n" 

309 initial_str += b"Content-Type: application/beep+xml\r\n" 

310 initial_str += b"\r\n" 

311 initial_str += b"<greeting>\r\n" 

312 initial_str += ( 

313 b" <profile uri='http://iana.org/beep/TLS' />\r\n" 

314 ) 

315 initial_str += b"</greeting>\r\n" 

316 initial_str += b"END\r\n" 

317 initial_str += b"RPY 0 0 . 0 52\r\n" 

318 initial_str += b"Content-Type: application/beep+xml\r\n" 

319 initial_str += b"\r\n" 

320 initial_str += b"<greeting />\r\n" 

321 initial_str += b"END\r\n" 

322 initial_str += b"MSG 0 1 . 52 120\r\n" 

323 initial_str += b"Content-Type: application/beep+xml\r\n" 

324 initial_str += b"\r\n" 

325 initial_str += b"<start number='1'>\r\n" 

326 initial_str += ( 

327 b" <profile uri='http://iana.org/beep/SASL/OTP' />\r\n" 

328 ) 

329 initial_str += b"</start>\r\n" 

330 initial_str += b"END\r\n" 

331 initial_str += b"RPY 0 1 . 307 0\r\n" 

332 initial_str += b"END\r\n" 

333 

334 lexer = Lexer() 

335 frames = lexer.lex(BytesIO(initial_str)) 

336 

337 self.assertEqual(len(frames), 4) 

338 

339 lexer = Lexer() 

340 i = 0 

341 for _ in lexer.lex_iterator(BytesIO(initial_str)): 

342 i += 1 

343 self.assertEqual(i, 4) 

344 

345 def test_lexer_human_output(self) -> None: 

346 """Human respresentation. For coverage, but not tested.""" 

347 

348 content = b"MSG 0 1 . 42 120\r\nContent-Type: application/beep+xml\r\n\r\n<start number='1'>\r\n <profile uri='http://iana.org/beep/SASL/OTP' />\r\n</start>\r\nEND\r\n" 

349 

350 lexer = Lexer() 

351 frames = lexer.feed(content) 

352 

353 lexer.node_to_human(frames[0])