Coverage for tests/test_frame_generation.py: 100%

1"""Test for BEEP Frame generation.""" 

2 

3from unittest import TestCase 

4 

5from galaxy_beep_frame.frame import ( 

6 ContinuationIndicator, 

7 Frame, 

8 HeaderType, 

9) 

10from galaxy_beep_frame.header_type import PossibleHeaderType 

11 

12 

13class TestFrameGeneration(TestCase): 

14 """Test BEEP Frame generation.""" 

15 

16 def test_frame_generation_unknown(self) -> None: 

17 """Mostly to get 100% coverage.""" 

18 

19 header_type = HeaderType(state=0) # type: ignore[ty:bad-argument-type] 

20 result = header_type.build(0) 

21 self.assertEqual(result, NotImplemented) 

22 

23 def test_frame_generation_message(self) -> None: 

24 """Test MSG Frame generation.""" 

25 

26 content = ( 

27 "Content-Type: application/beep+xml\r\n" 

28 "\r\n" 

29 "<start number='1'>\r\n" 

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

31 "</start>\r\n" 

32 ) 

33 frame = Frame( 

34 payload=content.encode("utf-8"), 

35 channel_number=0, 

36 message_number=1, 

37 sequence_number=52, 

38 continuation_indicator=ContinuationIndicator.COMPLETE, 

39 header_type=PossibleHeaderType.MSG, 

40 answer_number=None, 

41 ) 

42 expected = ( 

43 "MSG 0 1 . 52 120\r\n" 

44 "Content-Type: application/beep+xml\r\n" 

45 "\r\n" 

46 "<start number='1'>\r\n" 

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

48 "</start>\r\n" 

49 "END\r\n" 

50 ) 

51 

52 self.assertEqual(frame.build().decode("utf-8"), expected) 

53 

54 def test_frame_generation_answer(self) -> None: 

55 """Test ANS Frame generation.""" 

56 

57 content = ( 

58 "Content-Type: application/beep+xml\r\n" 

59 "\r\n" 

60 "<start number='1'>\r\n" 

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

62 "</start>\r\n" 

63 ) 

64 frame = Frame( 

65 payload=content.encode("utf-8"), 

66 channel_number=0, 

67 message_number=1, 

68 sequence_number=52, 

69 continuation_indicator=ContinuationIndicator.COMPLETE, 

70 header_type=PossibleHeaderType.ANS, 

71 answer_number=12, 

72 ) 

73 expected = ( 

74 "ANS 0 1 . 52 120 12\r\n" 

75 "Content-Type: application/beep+xml\r\n" 

76 "\r\n" 

77 "<start number='1'>\r\n" 

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

79 "</start>\r\n" 

80 "END\r\n" 

81 ) 

82 

83 self.assertEqual(frame.build().decode("utf-8"), expected) 

84 

85 def test_frame_generation_error(self) -> None: 

86 """Test ERR Frame generation.""" 

87 

88 content = ( 

89 "Content-Type: application/beep+xml\r\n" 

90 "\r\n" 

91 "<start number='1'>\r\n" 

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

93 "</start>\r\n" 

94 ) 

95 frame = Frame( 

96 payload=content.encode("utf-8"), 

97 channel_number=13, 

98 message_number=1, 

99 sequence_number=53, 

100 continuation_indicator=ContinuationIndicator.COMPLETE, 

101 header_type=PossibleHeaderType.ERR, 

102 answer_number=None, 

103 ) 

104 expected = ( 

105 "ERR 13 1 . 53 120\r\n" 

106 "Content-Type: application/beep+xml\r\n" 

107 "\r\n" 

108 "<start number='1'>\r\n" 

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

110 "</start>\r\n" 

111 "END\r\n" 

112 ) 

113 

114 self.assertEqual(frame.build().decode("utf-8"), expected) 

115 

116 def test_frame_generation_reply(self) -> None: 

117 """Test RPY Frame generation.""" 

118 

119 content = ( 

120 "Content-Type: application/beep+xml\r\n" 

121 "\r\n" 

122 "<start number='1'>\r\n" 

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

124 "</start>\r\n" 

125 ) 

126 frame = Frame( 

127 payload=content.encode("utf-8"), 

128 channel_number=13, 

129 message_number=4, 

130 sequence_number=53, 

131 continuation_indicator=ContinuationIndicator.INTERMEDIATE, 

132 header_type=PossibleHeaderType.RPY, 

133 answer_number=None, 

134 ) 

135 expected = ( 

136 "RPY 13 4 * 53 120\r\n" 

137 "Content-Type: application/beep+xml\r\n" 

138 "\r\n" 

139 "<start number='1'>\r\n" 

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

141 "</start>\r\n" 

142 "END\r\n" 

143 ) 

144 

145 self.assertEqual(frame.build().decode("utf-8"), expected) 

146 

147 def test_frame_generation_null(self) -> None: 

148 """Test NUL Frame generation.""" 

149 

150 content = ( 

151 "Content-Type: application/beep+xml\r\n" 

152 "\r\n" 

153 "<start number='1'>\r\n" 

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

155 "</start>\r\n" 

156 ) 

157 frame = Frame( 

158 payload=content.encode("utf-8"), 

159 channel_number=13, 

160 message_number=1, 

161 sequence_number=53, 

162 continuation_indicator=ContinuationIndicator.COMPLETE, 

163 header_type=PossibleHeaderType.NUL, 

164 answer_number=None, 

165 ) 

166 expected = ( 

167 "NUL 13 1 . 53 120\r\n" 

168 "Content-Type: application/beep+xml\r\n" 

169 "\r\n" 

170 "<start number='1'>\r\n" 

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

172 "</start>\r\n" 

173 "END\r\n" 

174 ) 

175 

176 self.assertEqual(frame.build().decode("utf-8"), expected)