test-wireproto-serverreactor.py
490 lines
| 19.0 KiB
| text/x-python
|
PythonLexer
/ tests / test-wireproto-serverreactor.py
|
r37070 | from __future__ import absolute_import, print_function | ||
import unittest | ||||
|
r37308 | from mercurial.thirdparty import ( | ||
cbor, | ||||
) | ||||
|
r37070 | from mercurial import ( | ||
util, | ||||
wireprotoframing as framing, | ||||
) | ||||
ffs = framing.makeframefromhumanstring | ||||
|
r37743 | OK = cbor.dumps({b'status': b'ok'}) | ||
|
r37074 | def makereactor(deferoutput=False): | ||
return framing.serverreactor(deferoutput=deferoutput) | ||||
|
r37070 | |||
def sendframes(reactor, gen): | ||||
"""Send a generator of frame bytearray to a reactor. | ||||
Emits a generator of results from ``onframerecv()`` calls. | ||||
""" | ||||
for frame in gen: | ||||
|
r37079 | header = framing.parseheader(frame) | ||
|
r37070 | payload = frame[framing.FRAME_HEADER_SIZE:] | ||
|
r37079 | assert len(payload) == header.length | ||
|
r37070 | |||
|
r37079 | yield reactor.onframerecv(framing.frame(header.requestid, | ||
|
r37304 | header.streamid, | ||
header.streamflags, | ||||
|
r37079 | header.typeid, | ||
header.flags, | ||||
payload)) | ||||
|
r37070 | |||
|
r37303 | def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None): | ||
|
r37070 | """Generate frames to run a command and send them to a reactor.""" | ||
|
r37075 | return sendframes(reactor, | ||
|
r37303 | framing.createcommandframes(stream, rid, cmd, args, | ||
datafh)) | ||||
|
r37070 | |||
|
r37078 | |||
|
r37070 | class ServerReactorTests(unittest.TestCase): | ||
|
r37302 | def _sendsingleframe(self, reactor, f): | ||
results = list(sendframes(reactor, [f])) | ||||
|
r37070 | self.assertEqual(len(results), 1) | ||
return results[0] | ||||
def assertaction(self, res, expected): | ||||
self.assertIsInstance(res, tuple) | ||||
self.assertEqual(len(res), 2) | ||||
self.assertIsInstance(res[1], dict) | ||||
self.assertEqual(res[0], expected) | ||||
|
r37073 | def assertframesequal(self, frames, framestrings): | ||
expected = [ffs(s) for s in framestrings] | ||||
self.assertEqual(list(frames), expected) | ||||
|
r37070 | def test1framecommand(self): | ||
"""Receiving a command in a single frame yields request to run it.""" | ||||
reactor = makereactor() | ||||
|
r37304 | stream = framing.stream(1) | ||
|
r37303 | results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {})) | ||
|
r37070 | self.assertEqual(len(results), 1) | ||
|
r37700 | self.assertaction(results[0], b'runcommand') | ||
|
r37070 | self.assertEqual(results[0][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'mycommand', | ||||
b'args': {}, | ||||
b'data': None, | ||||
|
r37070 | }) | ||
|
r37074 | result = reactor.oninputeof() | ||
|
r37700 | self.assertaction(result, b'noop') | ||
|
r37074 | |||
|
r37070 | def test1argument(self): | ||
reactor = makereactor() | ||||
|
r37304 | stream = framing.stream(1) | ||
|
r37303 | results = list(sendcommandframes(reactor, stream, 41, b'mycommand', | ||
|
r37070 | {b'foo': b'bar'})) | ||
|
r37308 | self.assertEqual(len(results), 1) | ||
|
r37700 | self.assertaction(results[0], b'runcommand') | ||
|
r37308 | self.assertEqual(results[0][1], { | ||
|
r37700 | b'requestid': 41, | ||
b'command': b'mycommand', | ||||
b'args': {b'foo': b'bar'}, | ||||
b'data': None, | ||||
|
r37070 | }) | ||
def testmultiarguments(self): | ||||
reactor = makereactor() | ||||
|
r37304 | stream = framing.stream(1) | ||
|
r37303 | results = list(sendcommandframes(reactor, stream, 1, b'mycommand', | ||
|
r37070 | {b'foo': b'bar', b'biz': b'baz'})) | ||
|
r37308 | self.assertEqual(len(results), 1) | ||
|
r37700 | self.assertaction(results[0], b'runcommand') | ||
|
r37308 | self.assertEqual(results[0][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'mycommand', | ||||
b'args': {b'foo': b'bar', b'biz': b'baz'}, | ||||
b'data': None, | ||||
|
r37070 | }) | ||
def testsimplecommanddata(self): | ||||
reactor = makereactor() | ||||
|
r37304 | stream = framing.stream(1) | ||
|
r37303 | results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}, | ||
|
r37070 | util.bytesio(b'data!'))) | ||
self.assertEqual(len(results), 2) | ||||
|
r37700 | self.assertaction(results[0], b'wantframe') | ||
self.assertaction(results[1], b'runcommand') | ||||
|
r37070 | self.assertEqual(results[1][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'mycommand', | ||||
b'args': {}, | ||||
b'data': b'data!', | ||||
|
r37070 | }) | ||
def testmultipledataframes(self): | ||||
frames = [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new|have-data ' | ||
b"cbor:{b'name': b'mycommand'}"), | ||||
|
r37304 | ffs(b'1 1 0 command-data continuation data1'), | ||
ffs(b'1 1 0 command-data continuation data2'), | ||||
ffs(b'1 1 0 command-data eos data3'), | ||||
|
r37070 | ] | ||
reactor = makereactor() | ||||
results = list(sendframes(reactor, frames)) | ||||
self.assertEqual(len(results), 4) | ||||
for i in range(3): | ||||
|
r37700 | self.assertaction(results[i], b'wantframe') | ||
self.assertaction(results[3], b'runcommand') | ||||
|
r37070 | self.assertEqual(results[3][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'mycommand', | ||||
b'args': {}, | ||||
b'data': b'data1data2data3', | ||||
|
r37070 | }) | ||
def testargumentanddata(self): | ||||
frames = [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new|have-data ' | ||
b"cbor:{b'name': b'command', b'args': {b'key': b'val'," | ||||
b"b'foo': b'bar'}}"), | ||||
|
r37304 | ffs(b'1 1 0 command-data continuation value1'), | ||
ffs(b'1 1 0 command-data eos value2'), | ||||
|
r37070 | ] | ||
reactor = makereactor() | ||||
results = list(sendframes(reactor, frames)) | ||||
|
r37700 | self.assertaction(results[-1], b'runcommand') | ||
|
r37070 | self.assertEqual(results[-1][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'command', | ||||
b'args': { | ||||
|
r37070 | b'key': b'val', | ||
b'foo': b'bar', | ||||
}, | ||||
|
r37700 | b'data': b'value1value2', | ||
|
r37070 | }) | ||
|
r37308 | def testnewandcontinuation(self): | ||
result = self._sendsingleframe(makereactor(), | ||||
ffs(b'1 1 stream-begin command-request new|continuation ')) | ||||
|
r37700 | self.assertaction(result, b'error') | ||
|
r37070 | self.assertEqual(result[1], { | ||
|
r37700 | b'message': b'received command request frame with both new and ' | ||
b'continuation flags set', | ||||
|
r37070 | }) | ||
|
r37308 | def testneithernewnorcontinuation(self): | ||
result = self._sendsingleframe(makereactor(), | ||||
ffs(b'1 1 stream-begin command-request 0 ')) | ||||
|
r37700 | self.assertaction(result, b'error') | ||
|
r37308 | self.assertEqual(result[1], { | ||
|
r37700 | b'message': b'received command request frame with neither new nor ' | ||
b'continuation flags set', | ||||
|
r37076 | }) | ||
|
r37070 | def testunexpectedcommanddata(self): | ||
|
r37308 | """Command data frame when not running a command is an error.""" | ||
result = self._sendsingleframe(makereactor(), | ||||
ffs(b'1 1 stream-begin command-data 0 ignored')) | ||||
|
r37700 | self.assertaction(result, b'error') | ||
|
r37070 | self.assertEqual(result[1], { | ||
|
r37741 | b'message': b'expected command request frame; got 2', | ||
|
r37070 | }) | ||
|
r37076 | def testunexpectedcommanddatareceiving(self): | ||
"""Same as above except the command is receiving.""" | ||||
results = list(sendframes(makereactor(), [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new|more ' | ||
b"cbor:{b'name': b'ignored'}"), | ||||
|
r37304 | ffs(b'1 1 0 command-data eos ignored'), | ||
|
r37076 | ])) | ||
|
r37700 | self.assertaction(results[0], b'wantframe') | ||
self.assertaction(results[1], b'error') | ||||
|
r37076 | self.assertEqual(results[1][1], { | ||
|
r37700 | b'message': b'received command data frame for request that is not ' | ||
b'expecting data: 1', | ||||
|
r37076 | }) | ||
|
r37302 | def testconflictingrequestidallowed(self): | ||
|
r37076 | """Multiple fully serviced commands with same request ID is allowed.""" | ||
|
r37302 | reactor = makereactor() | ||
results = [] | ||||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37302 | results.append(self._sendsingleframe( | ||
|
r37308 | reactor, ffs(b'1 1 stream-begin command-request new ' | ||
b"cbor:{b'name': b'command'}"))) | ||||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response1') | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
list(result[1][b'framegen']) | ||||
|
r37302 | results.append(self._sendsingleframe( | ||
|
r37308 | reactor, ffs(b'1 1 stream-begin command-request new ' | ||
b"cbor:{b'name': b'command'}"))) | ||||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response2') | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
list(result[1][b'framegen']) | ||||
|
r37302 | results.append(self._sendsingleframe( | ||
|
r37308 | reactor, ffs(b'1 1 stream-begin command-request new ' | ||
b"cbor:{b'name': b'command'}"))) | ||||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response3') | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
list(result[1][b'framegen']) | ||||
|
r37302 | |||
|
r37076 | for i in range(3): | ||
|
r37700 | self.assertaction(results[i], b'runcommand') | ||
|
r37076 | self.assertEqual(results[i][1], { | ||
|
r37700 | b'requestid': 1, | ||
b'command': b'command', | ||||
b'args': {}, | ||||
b'data': None, | ||||
|
r37076 | }) | ||
def testconflictingrequestid(self): | ||||
"""Request ID for new command matching in-flight command is illegal.""" | ||||
results = list(sendframes(makereactor(), [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new|more ' | ||
b"cbor:{b'name': b'command'}"), | ||||
ffs(b'1 1 0 command-request new ' | ||||
b"cbor:{b'name': b'command1'}"), | ||||
|
r37076 | ])) | ||
|
r37700 | self.assertaction(results[0], b'wantframe') | ||
self.assertaction(results[1], b'error') | ||||
|
r37076 | self.assertEqual(results[1][1], { | ||
|
r37700 | b'message': b'request with ID 1 already received', | ||
|
r37076 | }) | ||
def testinterleavedcommands(self): | ||||
|
r37308 | cbor1 = cbor.dumps({ | ||
b'name': b'command1', | ||||
b'args': { | ||||
b'foo': b'bar', | ||||
b'key1': b'val', | ||||
} | ||||
}, canonical=True) | ||||
cbor3 = cbor.dumps({ | ||||
b'name': b'command3', | ||||
b'args': { | ||||
b'biz': b'baz', | ||||
b'key': b'val', | ||||
}, | ||||
}, canonical=True) | ||||
|
r37076 | results = list(sendframes(makereactor(), [ | ||
|
r37308 | ffs(b'1 1 stream-begin command-request new|more %s' % cbor1[0:6]), | ||
ffs(b'3 1 0 command-request new|more %s' % cbor3[0:10]), | ||||
ffs(b'1 1 0 command-request continuation|more %s' % cbor1[6:9]), | ||||
ffs(b'3 1 0 command-request continuation|more %s' % cbor3[10:13]), | ||||
ffs(b'3 1 0 command-request continuation %s' % cbor3[13:]), | ||||
ffs(b'1 1 0 command-request continuation %s' % cbor1[9:]), | ||||
|
r37076 | ])) | ||
self.assertEqual([t[0] for t in results], [ | ||||
|
r37700 | b'wantframe', | ||
b'wantframe', | ||||
b'wantframe', | ||||
b'wantframe', | ||||
b'runcommand', | ||||
b'runcommand', | ||||
|
r37076 | ]) | ||
self.assertEqual(results[4][1], { | ||||
|
r37700 | b'requestid': 3, | ||
b'command': b'command3', | ||||
b'args': {b'biz': b'baz', b'key': b'val'}, | ||||
b'data': None, | ||||
|
r37076 | }) | ||
self.assertEqual(results[5][1], { | ||||
|
r37700 | b'requestid': 1, | ||
b'command': b'command1', | ||||
b'args': {b'foo': b'bar', b'key1': b'val'}, | ||||
b'data': None, | ||||
|
r37076 | }) | ||
|
r37070 | def testmissingcommanddataframe(self): | ||
|
r37076 | # The reactor doesn't currently handle partially received commands. | ||
# So this test is failing to do anything with request 1. | ||||
|
r37070 | frames = [ | ||
|
r37308 | ffs(b'1 1 stream-begin command-request new|have-data ' | ||
b"cbor:{b'name': b'command1'}"), | ||||
ffs(b'3 1 0 command-request new ' | ||||
b"cbor:{b'name': b'command2'}"), | ||||
|
r37070 | ] | ||
results = list(sendframes(makereactor(), frames)) | ||||
self.assertEqual(len(results), 2) | ||||
|
r37700 | self.assertaction(results[0], b'wantframe') | ||
self.assertaction(results[1], b'runcommand') | ||||
|
r37070 | |||
def testmissingcommanddataframeflags(self): | ||||
frames = [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new|have-data ' | ||
b"cbor:{b'name': b'command1'}"), | ||||
|
r37304 | ffs(b'1 1 0 command-data 0 data'), | ||
|
r37070 | ] | ||
results = list(sendframes(makereactor(), frames)) | ||||
self.assertEqual(len(results), 2) | ||||
|
r37700 | self.assertaction(results[0], b'wantframe') | ||
self.assertaction(results[1], b'error') | ||||
|
r37070 | self.assertEqual(results[1][1], { | ||
|
r37700 | b'message': b'command data frame without flags', | ||
|
r37070 | }) | ||
|
r37076 | def testframefornonreceivingrequest(self): | ||
"""Receiving a frame for a command that is not receiving is illegal.""" | ||||
results = list(sendframes(makereactor(), [ | ||||
|
r37308 | ffs(b'1 1 stream-begin command-request new ' | ||
b"cbor:{b'name': b'command1'}"), | ||||
ffs(b'3 1 0 command-request new|have-data ' | ||||
b"cbor:{b'name': b'command3'}"), | ||||
ffs(b'5 1 0 command-data eos ignored'), | ||||
|
r37076 | ])) | ||
|
r37700 | self.assertaction(results[2], b'error') | ||
|
r37076 | self.assertEqual(results[2][1], { | ||
|
r37700 | b'message': b'received frame for request that is not receiving: 5', | ||
|
r37076 | }) | ||
|
r37073 | def testsimpleresponse(self): | ||
"""Bytes response to command sends result frames.""" | ||||
reactor = makereactor() | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) | ||
|
r37073 | |||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response') | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37743 | b'1 2 stream-begin command-response eos %sresponse' % OK, | ||
|
r37073 | ]) | ||
def testmultiframeresponse(self): | ||||
"""Bytes response spanning multiple frames is handled.""" | ||||
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE | ||||
second = b'y' * 100 | ||||
reactor = makereactor() | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) | ||
|
r37073 | |||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, first + second) | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37743 | b'1 2 stream-begin command-response continuation %s' % OK, | ||
b'1 2 0 command-response continuation %s' % first, | ||||
|
r37742 | b'1 2 0 command-response eos %s' % second, | ||
|
r37073 | ]) | ||
|
r37744 | def testservererror(self): | ||
|
r37073 | reactor = makereactor() | ||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) | ||
|
r37073 | |||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37744 | result = reactor.onservererror(outstream, 1, b'some message') | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37744 | b"1 2 stream-begin error-response 0 " | ||
b"cbor:{b'type': b'server', " | ||||
b"b'message': [{b'msg': b'some message'}]}", | ||||
|
r37073 | ]) | ||
|
r37074 | def test1commanddeferresponse(self): | ||
"""Responses when in deferred output mode are delayed until EOF.""" | ||||
reactor = makereactor(deferoutput=True) | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | results = list(sendcommandframes(reactor, instream, 1, b'mycommand', | ||
{})) | ||||
|
r37074 | self.assertEqual(len(results), 1) | ||
|
r37700 | self.assertaction(results[0], b'runcommand') | ||
|
r37074 | |||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response') | ||
|
r37700 | self.assertaction(result, b'noop') | ||
|
r37074 | result = reactor.oninputeof() | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37743 | b'1 2 stream-begin command-response eos %sresponse' % OK, | ||
|
r37074 | ]) | ||
def testmultiplecommanddeferresponse(self): | ||||
reactor = makereactor(deferoutput=True) | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
list(sendcommandframes(reactor, instream, 3, b'command2', {})) | ||||
|
r37074 | |||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | result = reactor.oncommandresponseready(outstream, 1, b'response1') | ||
|
r37700 | self.assertaction(result, b'noop') | ||
|
r37742 | result = reactor.oncommandresponseready(outstream, 3, b'response2') | ||
|
r37700 | self.assertaction(result, b'noop') | ||
|
r37074 | result = reactor.oninputeof() | ||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37743 | b'1 2 stream-begin command-response eos %sresponse1' % OK, | ||
b'3 2 0 command-response eos %sresponse2' % OK, | ||||
|
r37075 | ]) | ||
def testrequestidtracking(self): | ||||
reactor = makereactor(deferoutput=True) | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
list(sendcommandframes(reactor, instream, 3, b'command2', {})) | ||||
list(sendcommandframes(reactor, instream, 5, b'command3', {})) | ||||
|
r37075 | |||
# Register results for commands out of order. | ||||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | reactor.oncommandresponseready(outstream, 3, b'response3') | ||
reactor.oncommandresponseready(outstream, 1, b'response1') | ||||
reactor.oncommandresponseready(outstream, 5, b'response5') | ||||
|
r37075 | |||
result = reactor.oninputeof() | ||||
|
r37700 | self.assertaction(result, b'sendframes') | ||
self.assertframesequal(result[1][b'framegen'], [ | ||||
|
r37743 | b'3 2 stream-begin command-response eos %sresponse3' % OK, | ||
b'1 2 0 command-response eos %sresponse1' % OK, | ||||
b'5 2 0 command-response eos %sresponse5' % OK, | ||||
|
r37074 | ]) | ||
|
r37081 | def testduplicaterequestonactivecommand(self): | ||
"""Receiving a request ID that matches a request that isn't finished.""" | ||||
reactor = makereactor() | ||||
|
r37304 | stream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, stream, 1, b'command1', {})) | ||
results = list(sendcommandframes(reactor, stream, 1, b'command1', {})) | ||||
|
r37081 | |||
|
r37700 | self.assertaction(results[0], b'error') | ||
|
r37081 | self.assertEqual(results[0][1], { | ||
|
r37700 | b'message': b'request with ID 1 is already active', | ||
|
r37081 | }) | ||
def testduplicaterequestonactivecommandnosend(self): | ||||
"""Same as above but we've registered a response but haven't sent it.""" | ||||
reactor = makereactor() | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | reactor.oncommandresponseready(outstream, 1, b'response') | ||
|
r37081 | |||
# We've registered the response but haven't sent it. From the | ||||
# perspective of the reactor, the command is still active. | ||||
|
r37303 | results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
|
r37700 | self.assertaction(results[0], b'error') | ||
|
r37081 | self.assertEqual(results[0][1], { | ||
|
r37700 | b'message': b'request with ID 1 is already active', | ||
|
r37081 | }) | ||
def testduplicaterequestaftersend(self): | ||||
"""We can use a duplicate request ID after we've sent the response.""" | ||||
reactor = makereactor() | ||||
|
r37304 | instream = framing.stream(1) | ||
|
r37303 | list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
|
r37305 | outstream = reactor.makeoutputstream() | ||
|
r37742 | res = reactor.oncommandresponseready(outstream, 1, b'response') | ||
|
r37700 | list(res[1][b'framegen']) | ||
|
r37081 | |||
|
r37303 | results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) | ||
|
r37700 | self.assertaction(results[0], b'runcommand') | ||
|
r37081 | |||
|
r37070 | if __name__ == '__main__': | ||
import silenttestrunner | ||||
silenttestrunner.main(__name__) | ||||