Coverage for tests/test_lexer_async.py: 100%

1"""Test BEEP lexing with async.""" 

2 

3from asyncio import StreamReader, create_task, sleep 

4from unittest import IsolatedAsyncioTestCase 

5 

6from galaxy_beep_frame.converter import Lexer 

7 

8 

9class TestLexerAsync(IsolatedAsyncioTestCase): 

10 """Test async lexing.""" 

11 

12 async def test_lexer_async(self) -> None: 

13 """Test lexing with async.""" 

14 

15 initial_str = b"" 

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

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

18 initial_str += b"\r\n" 

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

20 initial_str += ( 

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

22 ) 

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

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

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

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

27 initial_str += b"\r\n" 

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

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

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

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

32 initial_str += b"\r\n" 

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

34 initial_str += ( 

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

36 ) 

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

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

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

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

41 

42 reader = StreamReader() 

43 

44 async def feed() -> None: 

45 for chunk in ( 

46 initial_str[:40], 

47 initial_str[40:100], 

48 initial_str[100:], 

49 ): 

50 reader.feed_data(chunk) 

51 await sleep(0) 

52 reader.feed_eof() 

53 

54 feed_task = create_task(feed()) 

55 

56 lexer = Lexer() 

57 frames = await lexer.lex_stream(reader) 

58 

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

60 

61 await feed_task 

62 

63 reader = StreamReader() 

64 

65 feed_task = create_task(feed()) 

66 

67 lexer = Lexer() 

68 i = 0 

69 async for _ in lexer.lex_stream_iterator(reader): 

70 i += 1 

71 self.assertEqual(i, 4) 

72 

73 await feed_task