4 |
4 |
5 from mercurial import ( |
5 from mercurial import ( |
6 error, |
6 error, |
7 wireprotoframing as framing, |
7 wireprotoframing as framing, |
8 ) |
8 ) |
|
9 |
|
10 ffs = framing.makeframefromhumanstring |
|
11 |
|
12 def sendframe(reactor, frame): |
|
13 """Send a frame bytearray to a reactor.""" |
|
14 header = framing.parseheader(frame) |
|
15 payload = frame[framing.FRAME_HEADER_SIZE:] |
|
16 assert len(payload) == header.length |
|
17 |
|
18 return reactor.onframerecv(framing.frame(header.requestid, |
|
19 header.streamid, |
|
20 header.streamflags, |
|
21 header.typeid, |
|
22 header.flags, |
|
23 payload)) |
9 |
24 |
10 class SingleSendTests(unittest.TestCase): |
25 class SingleSendTests(unittest.TestCase): |
11 """A reactor that can only send once rejects subsequent sends.""" |
26 """A reactor that can only send once rejects subsequent sends.""" |
12 def testbasic(self): |
27 def testbasic(self): |
13 reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) |
28 reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) |
59 for frame in meta['framegen']: |
74 for frame in meta['framegen']: |
60 self.assertEqual(request.state, 'sending') |
75 self.assertEqual(request.state, 'sending') |
61 |
76 |
62 self.assertEqual(request.state, 'sent') |
77 self.assertEqual(request.state, 'sent') |
63 |
78 |
|
79 class BadFrameRecvTests(unittest.TestCase): |
|
80 def testoddstream(self): |
|
81 reactor = framing.clientreactor() |
|
82 |
|
83 action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo')) |
|
84 self.assertEqual(action, 'error') |
|
85 self.assertEqual(meta['message'], |
|
86 'received frame with odd numbered stream ID: 1') |
|
87 |
|
88 def testunknownstream(self): |
|
89 reactor = framing.clientreactor() |
|
90 |
|
91 action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo')) |
|
92 self.assertEqual(action, 'error') |
|
93 self.assertEqual(meta['message'], |
|
94 'received frame on unknown stream without beginning ' |
|
95 'of stream flag set') |
|
96 |
|
97 def testunhandledframetype(self): |
|
98 reactor = framing.clientreactor(buffersends=False) |
|
99 |
|
100 request, action, meta = reactor.callcommand(b'foo', {}) |
|
101 for frame in meta['framegen']: |
|
102 pass |
|
103 |
|
104 with self.assertRaisesRegexp(error.ProgrammingError, |
|
105 'unhandled frame type'): |
|
106 sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo')) |
|
107 |
64 if __name__ == '__main__': |
108 if __name__ == '__main__': |
65 import silenttestrunner |
109 import silenttestrunner |
66 silenttestrunner.main(__name__) |
110 silenttestrunner.main(__name__) |