32 def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None): |
32 def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None): |
33 """Generate frames to run a command and send them to a reactor.""" |
33 """Generate frames to run a command and send them to a reactor.""" |
34 return sendframes(reactor, |
34 return sendframes(reactor, |
35 framing.createcommandframes(stream, rid, cmd, args, |
35 framing.createcommandframes(stream, rid, cmd, args, |
36 datafh)) |
36 datafh)) |
|
37 |
|
38 class FrameHumanStringTests(unittest.TestCase): |
|
39 def testbasic(self): |
|
40 self.assertEqual(ffs(b'1 1 0 1 0 '), |
|
41 b'\x00\x00\x00\x01\x00\x01\x00\x10') |
|
42 |
|
43 self.assertEqual(ffs(b'2 4 0 1 0 '), |
|
44 b'\x00\x00\x00\x02\x00\x04\x00\x10') |
|
45 |
|
46 self.assertEqual(ffs(b'2 4 0 1 0 foo'), |
|
47 b'\x03\x00\x00\x02\x00\x04\x00\x10foo') |
|
48 |
|
49 def testcborint(self): |
|
50 self.assertEqual(ffs(b'1 1 0 1 0 cbor:15'), |
|
51 b'\x01\x00\x00\x01\x00\x01\x00\x10\x0f') |
|
52 |
|
53 self.assertEqual(ffs(b'1 1 0 1 0 cbor:42'), |
|
54 b'\x02\x00\x00\x01\x00\x01\x00\x10\x18*') |
|
55 |
|
56 self.assertEqual(ffs(b'1 1 0 1 0 cbor:1048576'), |
|
57 b'\x05\x00\x00\x01\x00\x01\x00\x10\x1a' |
|
58 b'\x00\x10\x00\x00') |
|
59 |
|
60 self.assertEqual(ffs(b'1 1 0 1 0 cbor:0'), |
|
61 b'\x01\x00\x00\x01\x00\x01\x00\x10\x00') |
|
62 |
|
63 self.assertEqual(ffs(b'1 1 0 1 0 cbor:-1'), |
|
64 b'\x01\x00\x00\x01\x00\x01\x00\x10 ') |
|
65 |
|
66 self.assertEqual(ffs(b'1 1 0 1 0 cbor:-342542'), |
|
67 b'\x05\x00\x00\x01\x00\x01\x00\x10:\x00\x05:\r') |
|
68 |
|
69 def testcborstrings(self): |
|
70 # String literals should be unicode. |
|
71 self.assertEqual(ffs(b"1 1 0 1 0 cbor:'foo'"), |
|
72 b'\x04\x00\x00\x01\x00\x01\x00\x10cfoo') |
|
73 |
|
74 self.assertEqual(ffs(b"1 1 0 1 0 cbor:b'foo'"), |
|
75 b'\x04\x00\x00\x01\x00\x01\x00\x10Cfoo') |
|
76 |
|
77 self.assertEqual(ffs(b"1 1 0 1 0 cbor:u'foo'"), |
|
78 b'\x04\x00\x00\x01\x00\x01\x00\x10cfoo') |
|
79 |
|
80 def testcborlists(self): |
|
81 self.assertEqual(ffs(b"1 1 0 1 0 cbor:[None, True, False, 42, b'foo']"), |
|
82 b'\n\x00\x00\x01\x00\x01\x00\x10\x85\xf6\xf5\xf4' |
|
83 b'\x18*Cfoo') |
|
84 |
|
85 def testcbordicts(self): |
|
86 self.assertEqual(ffs(b"1 1 0 1 0 " |
|
87 b"cbor:{b'foo': b'val1', b'bar': b'val2'}"), |
|
88 b'\x13\x00\x00\x01\x00\x01\x00\x10\xa2' |
|
89 b'CbarDval2CfooDval1') |
37 |
90 |
38 class FrameTests(unittest.TestCase): |
91 class FrameTests(unittest.TestCase): |
39 def testdataexactframesize(self): |
92 def testdataexactframesize(self): |
40 data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE) |
93 data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE) |
41 |
94 |