Mercurial > hg
comparison tests/test-wireproto-serverreactor.py @ 37319:36d17f37db91
wireproto: convert human output frames to CBOR
This is easier than rolling our own encoding format.
As a bonus, some of our artificial limits around lengths of
things went away because we are no longer using fixed length
fields to hold sizes.
Differential Revision: https://phab.mercurial-scm.org/D3067
author | Gregory Szorc <gregory.szorc@gmail.com> |
---|---|
date | Fri, 30 Mar 2018 14:52:32 -0700 |
parents | 3d0e2cd86e05 |
children | e9dea82ea1f3 |
comparison
equal
deleted
inserted
replaced
37318:9954d0e2ad00 | 37319:36d17f37db91 |
---|---|
134 b"cbor:{b'name': b'command', b'args': {b'key1': b'key1value', " | 134 b"cbor:{b'name': b'command', b'args': {b'key1': b'key1value', " |
135 b"b'key2': b'key2value', b'key3': b'key3value'}}"), | 135 b"b'key2': b'key2value', b'key3': b'key3value'}}"), |
136 ffs(b'1 1 0 command-data eos %s' % data.getvalue()), | 136 ffs(b'1 1 0 command-data eos %s' % data.getvalue()), |
137 ]) | 137 ]) |
138 | 138 |
139 def testtextoutputexcessiveargs(self): | |
140 """At most 255 formatting arguments are allowed.""" | |
141 with self.assertRaisesRegexp(ValueError, | |
142 'cannot use more than 255 formatting'): | |
143 args = [b'x' for i in range(256)] | |
144 list(framing.createtextoutputframe(None, 1, | |
145 [(b'bleh', args, [])])) | |
146 | |
147 def testtextoutputexcessivelabels(self): | |
148 """At most 255 labels are allowed.""" | |
149 with self.assertRaisesRegexp(ValueError, | |
150 'cannot use more than 255 labels'): | |
151 labels = [b'l' for i in range(256)] | |
152 list(framing.createtextoutputframe(None, 1, | |
153 [(b'bleh', [], labels)])) | |
154 | |
155 def testtextoutputformattingstringtype(self): | 139 def testtextoutputformattingstringtype(self): |
156 """Formatting string must be bytes.""" | 140 """Formatting string must be bytes.""" |
157 with self.assertRaisesRegexp(ValueError, 'must use bytes formatting '): | 141 with self.assertRaisesRegexp(ValueError, 'must use bytes formatting '): |
158 list(framing.createtextoutputframe(None, 1, [ | 142 list(framing.createtextoutputframe(None, 1, [ |
159 (b'foo'.decode('ascii'), [], [])])) | 143 (b'foo'.decode('ascii'), [], [])])) |
166 def testtextoutputlabelbytes(self): | 150 def testtextoutputlabelbytes(self): |
167 with self.assertRaisesRegexp(ValueError, 'must use bytes for labels'): | 151 with self.assertRaisesRegexp(ValueError, 'must use bytes for labels'): |
168 list(framing.createtextoutputframe(None, 1, [ | 152 list(framing.createtextoutputframe(None, 1, [ |
169 (b'foo', [], [b'foo'.decode('ascii')])])) | 153 (b'foo', [], [b'foo'.decode('ascii')])])) |
170 | 154 |
171 def testtextoutputtoolongformatstring(self): | |
172 with self.assertRaisesRegexp(ValueError, | |
173 'formatting string cannot be longer than'): | |
174 list(framing.createtextoutputframe(None, 1, [ | |
175 (b'x' * 65536, [], [])])) | |
176 | |
177 def testtextoutputtoolongargumentstring(self): | |
178 with self.assertRaisesRegexp(ValueError, | |
179 'argument string cannot be longer than'): | |
180 list(framing.createtextoutputframe(None, 1, [ | |
181 (b'bleh', [b'x' * 65536], [])])) | |
182 | |
183 def testtextoutputtoolonglabelstring(self): | |
184 with self.assertRaisesRegexp(ValueError, | |
185 'label string cannot be longer than'): | |
186 list(framing.createtextoutputframe(None, 1, [ | |
187 (b'bleh', [], [b'x' * 65536])])) | |
188 | |
189 def testtextoutput1simpleatom(self): | 155 def testtextoutput1simpleatom(self): |
190 stream = framing.stream(1) | 156 stream = framing.stream(1) |
191 val = list(framing.createtextoutputframe(stream, 1, [ | 157 val = list(framing.createtextoutputframe(stream, 1, [ |
192 (b'foo', [], [])])) | 158 (b'foo', [], [])])) |
193 | 159 |
194 self.assertEqual(val, [ | 160 self.assertEqual(val, [ |
195 ffs(br'1 1 stream-begin text-output 0 \x03\x00\x00\x00foo'), | 161 ffs(b'1 1 stream-begin text-output 0 ' |
162 b"cbor:[{b'msg': b'foo'}]"), | |
196 ]) | 163 ]) |
197 | 164 |
198 def testtextoutput2simpleatoms(self): | 165 def testtextoutput2simpleatoms(self): |
199 stream = framing.stream(1) | 166 stream = framing.stream(1) |
200 val = list(framing.createtextoutputframe(stream, 1, [ | 167 val = list(framing.createtextoutputframe(stream, 1, [ |
201 (b'foo', [], []), | 168 (b'foo', [], []), |
202 (b'bar', [], []), | 169 (b'bar', [], []), |
203 ])) | 170 ])) |
204 | 171 |
205 self.assertEqual(val, [ | 172 self.assertEqual(val, [ |
206 ffs(br'1 1 stream-begin text-output 0 ' | 173 ffs(b'1 1 stream-begin text-output 0 ' |
207 br'\x03\x00\x00\x00foo\x03\x00\x00\x00bar'), | 174 b"cbor:[{b'msg': b'foo'}, {b'msg': b'bar'}]") |
208 ]) | 175 ]) |
209 | 176 |
210 def testtextoutput1arg(self): | 177 def testtextoutput1arg(self): |
211 stream = framing.stream(1) | 178 stream = framing.stream(1) |
212 val = list(framing.createtextoutputframe(stream, 1, [ | 179 val = list(framing.createtextoutputframe(stream, 1, [ |
213 (b'foo %s', [b'val1'], []), | 180 (b'foo %s', [b'val1'], []), |
214 ])) | 181 ])) |
215 | 182 |
216 self.assertEqual(val, [ | 183 self.assertEqual(val, [ |
217 ffs(br'1 1 stream-begin text-output 0 ' | 184 ffs(b'1 1 stream-begin text-output 0 ' |
218 br'\x06\x00\x00\x01\x04\x00foo %sval1'), | 185 b"cbor:[{b'msg': b'foo %s', b'args': [b'val1']}]") |
219 ]) | 186 ]) |
220 | 187 |
221 def testtextoutput2arg(self): | 188 def testtextoutput2arg(self): |
222 stream = framing.stream(1) | 189 stream = framing.stream(1) |
223 val = list(framing.createtextoutputframe(stream, 1, [ | 190 val = list(framing.createtextoutputframe(stream, 1, [ |
224 (b'foo %s %s', [b'val', b'value'], []), | 191 (b'foo %s %s', [b'val', b'value'], []), |
225 ])) | 192 ])) |
226 | 193 |
227 self.assertEqual(val, [ | 194 self.assertEqual(val, [ |
228 ffs(br'1 1 stream-begin text-output 0 ' | 195 ffs(b'1 1 stream-begin text-output 0 ' |
229 br'\x09\x00\x00\x02\x03\x00\x05\x00foo %s %svalvalue'), | 196 b"cbor:[{b'msg': b'foo %s %s', b'args': [b'val', b'value']}]") |
230 ]) | 197 ]) |
231 | 198 |
232 def testtextoutput1label(self): | 199 def testtextoutput1label(self): |
233 stream = framing.stream(1) | 200 stream = framing.stream(1) |
234 val = list(framing.createtextoutputframe(stream, 1, [ | 201 val = list(framing.createtextoutputframe(stream, 1, [ |
235 (b'foo', [], [b'label']), | 202 (b'foo', [], [b'label']), |
236 ])) | 203 ])) |
237 | 204 |
238 self.assertEqual(val, [ | 205 self.assertEqual(val, [ |
239 ffs(br'1 1 stream-begin text-output 0 ' | 206 ffs(b'1 1 stream-begin text-output 0 ' |
240 br'\x03\x00\x01\x00\x05foolabel'), | 207 b"cbor:[{b'msg': b'foo', b'labels': [b'label']}]") |
241 ]) | 208 ]) |
242 | 209 |
243 def testargandlabel(self): | 210 def testargandlabel(self): |
244 stream = framing.stream(1) | 211 stream = framing.stream(1) |
245 val = list(framing.createtextoutputframe(stream, 1, [ | 212 val = list(framing.createtextoutputframe(stream, 1, [ |
246 (b'foo %s', [b'arg'], [b'label']), | 213 (b'foo %s', [b'arg'], [b'label']), |
247 ])) | 214 ])) |
248 | 215 |
249 self.assertEqual(val, [ | 216 self.assertEqual(val, [ |
250 ffs(br'1 1 stream-begin text-output 0 ' | 217 ffs(b'1 1 stream-begin text-output 0 ' |
251 br'\x06\x00\x01\x01\x05\x03\x00foo %slabelarg'), | 218 b"cbor:[{b'msg': b'foo %s', b'args': [b'arg'], " |
219 b"b'labels': [b'label']}]") | |
252 ]) | 220 ]) |
253 | 221 |
254 class ServerReactorTests(unittest.TestCase): | 222 class ServerReactorTests(unittest.TestCase): |
255 def _sendsingleframe(self, reactor, f): | 223 def _sendsingleframe(self, reactor, f): |
256 results = list(sendframes(reactor, [f])) | 224 results = list(sendframes(reactor, [f])) |