This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2906

wireproto: start to associate frame generation with a stream
ClosedPublic

Authored by indygreg on Mar 20 2018, 5:41 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG3ed344546d9e: wireproto: start to associate frame generation with a stream
Summary

An upcoming commit will introduce "streams" into the frame-based wire
protocol. In preparation for this invasive change, we introduce a basic
"stream" class and have all operations that create frames also operate
alongside a stream instance.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Revision Contents
Changeset List

Diff 7548

mercurial/wireprotoframing.py

payload = fh.read(h.length) payload = fh.read(h.length)
if len(payload) != h.length: if len(payload) != h.length:
raise error.Abort(_('frame length error: expected %d; got %d') % raise error.Abort(_('frame length error: expected %d; got %d') %
(h.length, len(payload))) (h.length, len(payload)))
return frame(h.requestid, h.typeid, h.flags, payload) return frame(h.requestid, h.typeid, h.flags, payload)
def createcommandframes(requestid, cmd, args, datafh=None): def createcommandframes(stream, requestid, cmd, args, datafh=None):
"""Create frames necessary to transmit a request to run a command. """Create frames necessary to transmit a request to run a command.
This is a generator of bytearrays. Each item represents a frame This is a generator of bytearrays. Each item represents a frame
ready to be sent over the wire to a peer. ready to be sent over the wire to a peer.
""" """
flags = 0 flags = 0
if args: if args:
flags |= FLAG_COMMAND_NAME_HAVE_ARGS flags |= FLAG_COMMAND_NAME_HAVE_ARGS
if datafh: if datafh:
flags |= FLAG_COMMAND_NAME_HAVE_DATA flags |= FLAG_COMMAND_NAME_HAVE_DATA
if not flags: if not flags:
flags |= FLAG_COMMAND_NAME_EOS flags |= FLAG_COMMAND_NAME_EOS
yield makeframe(requestid=requestid, typeid=FRAME_TYPE_COMMAND_NAME, yield stream.makeframe(requestid=requestid, typeid=FRAME_TYPE_COMMAND_NAME,
flags=flags, payload=cmd) flags=flags, payload=cmd)
for i, k in enumerate(sorted(args)): for i, k in enumerate(sorted(args)):
v = args[k] v = args[k]
last = i == len(args) - 1 last = i == len(args) - 1
# TODO handle splitting of argument values across frames. # TODO handle splitting of argument values across frames.
payload = bytearray(ARGUMENT_FRAME_HEADER.size + len(k) + len(v)) payload = bytearray(ARGUMENT_FRAME_HEADER.size + len(k) + len(v))
offset = 0 offset = 0
ARGUMENT_FRAME_HEADER.pack_into(payload, offset, len(k), len(v)) ARGUMENT_FRAME_HEADER.pack_into(payload, offset, len(k), len(v))
offset += ARGUMENT_FRAME_HEADER.size offset += ARGUMENT_FRAME_HEADER.size
payload[offset:offset + len(k)] = k payload[offset:offset + len(k)] = k
offset += len(k) offset += len(k)
payload[offset:offset + len(v)] = v payload[offset:offset + len(v)] = v
flags = FLAG_COMMAND_ARGUMENT_EOA if last else 0 flags = FLAG_COMMAND_ARGUMENT_EOA if last else 0
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_ARGUMENT, typeid=FRAME_TYPE_COMMAND_ARGUMENT,
flags=flags, flags=flags,
payload=payload) payload=payload)
if datafh: if datafh:
while True: while True:
data = datafh.read(DEFAULT_MAX_FRAME_SIZE) data = datafh.read(DEFAULT_MAX_FRAME_SIZE)
done = False done = False
if len(data) == DEFAULT_MAX_FRAME_SIZE: if len(data) == DEFAULT_MAX_FRAME_SIZE:
flags = FLAG_COMMAND_DATA_CONTINUATION flags = FLAG_COMMAND_DATA_CONTINUATION
else: else:
flags = FLAG_COMMAND_DATA_EOS flags = FLAG_COMMAND_DATA_EOS
assert datafh.read(1) == b'' assert datafh.read(1) == b''
done = True done = True
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_DATA, typeid=FRAME_TYPE_COMMAND_DATA,
flags=flags, flags=flags,
payload=data) payload=data)
if done: if done:
break break
def createbytesresponseframesfrombytes(requestid, data, def createbytesresponseframesfrombytes(stream, requestid, data,
maxframesize=DEFAULT_MAX_FRAME_SIZE): maxframesize=DEFAULT_MAX_FRAME_SIZE):
"""Create a raw frame to send a bytes response from static bytes input. """Create a raw frame to send a bytes response from static bytes input.
Returns a generator of bytearrays. Returns a generator of bytearrays.
""" """
# Simple case of a single frame. # Simple case of a single frame.
if len(data) <= maxframesize: if len(data) <= maxframesize:
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_BYTES_RESPONSE, typeid=FRAME_TYPE_BYTES_RESPONSE,
flags=FLAG_BYTES_RESPONSE_EOS, flags=FLAG_BYTES_RESPONSE_EOS,
payload=data) payload=data)
return return
offset = 0 offset = 0
while True: while True:
chunk = data[offset:offset + maxframesize] chunk = data[offset:offset + maxframesize]
offset += len(chunk) offset += len(chunk)
done = offset == len(data) done = offset == len(data)
if done: if done:
flags = FLAG_BYTES_RESPONSE_EOS flags = FLAG_BYTES_RESPONSE_EOS
else: else:
flags = FLAG_BYTES_RESPONSE_CONTINUATION flags = FLAG_BYTES_RESPONSE_CONTINUATION
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_BYTES_RESPONSE, typeid=FRAME_TYPE_BYTES_RESPONSE,
flags=flags, flags=flags,
payload=chunk) payload=chunk)
if done: if done:
break break
def createerrorframe(requestid, msg, protocol=False, application=False): def createerrorframe(stream, requestid, msg, protocol=False, application=False):
# TODO properly handle frame size limits. # TODO properly handle frame size limits.
assert len(msg) <= DEFAULT_MAX_FRAME_SIZE assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
flags = 0 flags = 0
if protocol: if protocol:
flags |= FLAG_ERROR_RESPONSE_PROTOCOL flags |= FLAG_ERROR_RESPONSE_PROTOCOL
if application: if application:
flags |= FLAG_ERROR_RESPONSE_APPLICATION flags |= FLAG_ERROR_RESPONSE_APPLICATION
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_ERROR_RESPONSE, typeid=FRAME_TYPE_ERROR_RESPONSE,
flags=flags, flags=flags,
payload=msg) payload=msg)
def createtextoutputframe(requestid, atoms): def createtextoutputframe(stream, requestid, atoms):
"""Create a text output frame to render text to people. """Create a text output frame to render text to people.
``atoms`` is a 3-tuple of (formatting string, args, labels). ``atoms`` is a 3-tuple of (formatting string, args, labels).
The formatting string contains ``%s`` tokens to be replaced by the The formatting string contains ``%s`` tokens to be replaced by the
corresponding indexed entry in ``args``. ``labels`` is an iterable of corresponding indexed entry in ``args``. ``labels`` is an iterable of
formatters to be applied at rendering time. In terms of the ``ui`` formatters to be applied at rendering time. In terms of the ``ui``
class, each atom corresponds to a ``ui.write()``. class, each atom corresponds to a ``ui.write()``.
atom = b''.join(chunks) atom = b''.join(chunks)
atomchunks.append(atom) atomchunks.append(atom)
bytesleft -= len(atom) bytesleft -= len(atom)
if bytesleft < 0: if bytesleft < 0:
raise ValueError('cannot encode data in a single frame') raise ValueError('cannot encode data in a single frame')
yield makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_TEXT_OUTPUT, typeid=FRAME_TYPE_TEXT_OUTPUT,
flags=0, flags=0,
payload=b''.join(atomchunks)) payload=b''.join(atomchunks))
class stream(object):
"""Represents a logical unidirectional series of frames."""
def makeframe(self, requestid, typeid, flags, payload):
"""Create a frame to be sent out over this stream.
Only returns the frame instance. Does not actually send it.
"""
return makeframe(requestid, typeid, flags, payload)
class serverreactor(object): class serverreactor(object):
"""Holds state of a server handling frame-based protocol requests. """Holds state of a server handling frame-based protocol requests.
This class is the "brain" of the unified frame-based protocol server This class is the "brain" of the unified frame-based protocol server
component. While the protocol is stateless from the perspective of component. While the protocol is stateless from the perspective of
requests/commands, something needs to track which frames have been requests/commands, something needs to track which frames have been
received, what frames to expect, etc. This class is that thing. received, what frames to expect, etc. This class is that thing.
} }
meth = handlers.get(self._state) meth = handlers.get(self._state)
if not meth: if not meth:
raise error.ProgrammingError('unhandled state: %s' % self._state) raise error.ProgrammingError('unhandled state: %s' % self._state)
return meth(frame) return meth(frame)
def onbytesresponseready(self, requestid, data): def onbytesresponseready(self, stream, requestid, data):
"""Signal that a bytes response is ready to be sent to the client. """Signal that a bytes response is ready to be sent to the client.
The raw bytes response is passed as an argument. The raw bytes response is passed as an argument.
""" """
def sendframes(): def sendframes():
for frame in createbytesresponseframesfrombytes(requestid, data): for frame in createbytesresponseframesfrombytes(stream, requestid,
data):
yield frame yield frame
self._activecommands.remove(requestid) self._activecommands.remove(requestid)
result = sendframes() result = sendframes()
if self._deferoutput: if self._deferoutput:
self._bufferedframegens.append(result) self._bufferedframegens.append(result)
for gen in self._bufferedframegens: for gen in self._bufferedframegens:
for frame in gen: for frame in gen:
yield frame yield frame
return 'sendframes', { return 'sendframes', {
'framegen': makegen(), 'framegen': makegen(),
} }
def onapplicationerror(self, requestid, msg): def onapplicationerror(self, stream, requestid, msg):
return 'sendframes', { return 'sendframes', {
'framegen': createerrorframe(requestid, msg, application=True), 'framegen': createerrorframe(stream, requestid, msg,
application=True),
} }
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,
} }
def _makeruncommandresult(self, requestid): def _makeruncommandresult(self, requestid):

mercurial/wireprotoserver.py

res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(_('command in frame must match command in URL')) res.setbodybytes(_('command in frame must match command in URL'))
return True return True
rsp = wireproto.dispatch(repo, proto, command['command']) rsp = wireproto.dispatch(repo, proto, command['command'])
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
stream = wireprotoframing.stream()
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
action, meta = reactor.onbytesresponseready(command['requestid'], action, meta = reactor.onbytesresponseready(stream,
command['requestid'],
rsp.data) rsp.data)
else: else:
action, meta = reactor.onapplicationerror( action, meta = reactor.onapplicationerror(
_('unhandled response type from wire proto command')) _('unhandled response type from wire proto command'))
if action == 'sendframes': if action == 'sendframes':
res.setbodygen(meta['framegen']) res.setbodygen(meta['framegen'])
return True return True

tests/test-wireproto-serverreactor.py

payload = frame[framing.FRAME_HEADER_SIZE:] payload = frame[framing.FRAME_HEADER_SIZE:]
assert len(payload) == header.length assert len(payload) == header.length
yield reactor.onframerecv(framing.frame(header.requestid, yield reactor.onframerecv(framing.frame(header.requestid,
header.typeid, header.typeid,
header.flags, header.flags,
payload)) payload))
def sendcommandframes(reactor, rid, cmd, args, datafh=None): def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None):
"""Generate frames to run a command and send them to a reactor.""" """Generate frames to run a command and send them to a reactor."""
return sendframes(reactor, return sendframes(reactor,
framing.createcommandframes(rid, cmd, args, datafh)) framing.createcommandframes(stream, rid, cmd, args,
datafh))
class FrameTests(unittest.TestCase): class FrameTests(unittest.TestCase):
def testdataexactframesize(self): def testdataexactframesize(self):
data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE) data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE)
frames = list(framing.createcommandframes(1, b'command', {}, data)) stream = framing.stream()
frames = list(framing.createcommandframes(stream, 1, b'command',
{}, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-data command'), ffs(b'1 command-name have-data command'),
ffs(b'1 command-data continuation %s' % data.getvalue()), ffs(b'1 command-data continuation %s' % data.getvalue()),
ffs(b'1 command-data eos ') ffs(b'1 command-data eos ')
]) ])
def testdatamultipleframes(self): def testdatamultipleframes(self):
data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1)) data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1))
frames = list(framing.createcommandframes(1, b'command', {}, data))
stream = framing.stream()
frames = list(framing.createcommandframes(stream, 1, b'command', {},
data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-data command'), ffs(b'1 command-name have-data command'),
ffs(b'1 command-data continuation %s' % ( ffs(b'1 command-data continuation %s' % (
b'x' * framing.DEFAULT_MAX_FRAME_SIZE)), b'x' * framing.DEFAULT_MAX_FRAME_SIZE)),
ffs(b'1 command-data eos x'), ffs(b'1 command-data eos x'),
]) ])
def testargsanddata(self): def testargsanddata(self):
data = util.bytesio(b'x' * 100) data = util.bytesio(b'x' * 100)
frames = list(framing.createcommandframes(1, b'command', { stream = framing.stream()
frames = list(framing.createcommandframes(stream, 1, b'command', {
b'key1': b'key1value', b'key1': b'key1value',
b'key2': b'key2value', b'key2': b'key2value',
b'key3': b'key3value', b'key3': b'key3value',
}, data)) }, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-args|have-data command'), ffs(b'1 command-name have-args|have-data command'),
ffs(br'1 command-argument 0 \x04\x00\x09\x00key1key1value'), ffs(br'1 command-argument 0 \x04\x00\x09\x00key1key1value'),
ffs(br'1 command-argument 0 \x04\x00\x09\x00key2key2value'), ffs(br'1 command-argument 0 \x04\x00\x09\x00key2key2value'),
ffs(br'1 command-argument eoa \x04\x00\x09\x00key3key3value'), ffs(br'1 command-argument eoa \x04\x00\x09\x00key3key3value'),
ffs(b'1 command-data eos %s' % data.getvalue()), ffs(b'1 command-data eos %s' % data.getvalue()),
]) ])
def testtextoutputexcessiveargs(self): def testtextoutputexcessiveargs(self):
"""At most 255 formatting arguments are allowed.""" """At most 255 formatting arguments are allowed."""
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'cannot use more than 255 formatting'): 'cannot use more than 255 formatting'):
args = [b'x' for i in range(256)] args = [b'x' for i in range(256)]
list(framing.createtextoutputframe(1, [(b'bleh', args, [])])) list(framing.createtextoutputframe(None, 1,
[(b'bleh', args, [])]))
def testtextoutputexcessivelabels(self): def testtextoutputexcessivelabels(self):
"""At most 255 labels are allowed.""" """At most 255 labels are allowed."""
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'cannot use more than 255 labels'): 'cannot use more than 255 labels'):
labels = [b'l' for i in range(256)] labels = [b'l' for i in range(256)]
list(framing.createtextoutputframe(1, [(b'bleh', [], labels)])) list(framing.createtextoutputframe(None, 1,
[(b'bleh', [], labels)]))
def testtextoutputformattingstringtype(self): def testtextoutputformattingstringtype(self):
"""Formatting string must be bytes.""" """Formatting string must be bytes."""
with self.assertRaisesRegexp(ValueError, 'must use bytes formatting '): with self.assertRaisesRegexp(ValueError, 'must use bytes formatting '):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'foo'.decode('ascii'), [], [])])) (b'foo'.decode('ascii'), [], [])]))
def testtextoutputargumentbytes(self): def testtextoutputargumentbytes(self):
with self.assertRaisesRegexp(ValueError, 'must use bytes for argument'): with self.assertRaisesRegexp(ValueError, 'must use bytes for argument'):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'foo', [b'foo'.decode('ascii')], [])])) (b'foo', [b'foo'.decode('ascii')], [])]))
def testtextoutputlabelbytes(self): def testtextoutputlabelbytes(self):
with self.assertRaisesRegexp(ValueError, 'must use bytes for labels'): with self.assertRaisesRegexp(ValueError, 'must use bytes for labels'):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'foo', [], [b'foo'.decode('ascii')])])) (b'foo', [], [b'foo'.decode('ascii')])]))
def testtextoutputtoolongformatstring(self): def testtextoutputtoolongformatstring(self):
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'formatting string cannot be longer than'): 'formatting string cannot be longer than'):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'x' * 65536, [], [])])) (b'x' * 65536, [], [])]))
def testtextoutputtoolongargumentstring(self): def testtextoutputtoolongargumentstring(self):
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'argument string cannot be longer than'): 'argument string cannot be longer than'):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'bleh', [b'x' * 65536], [])])) (b'bleh', [b'x' * 65536], [])]))
def testtextoutputtoolonglabelstring(self): def testtextoutputtoolonglabelstring(self):
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'label string cannot be longer than'): 'label string cannot be longer than'):
list(framing.createtextoutputframe(1, [ list(framing.createtextoutputframe(None, 1, [
(b'bleh', [], [b'x' * 65536])])) (b'bleh', [], [b'x' * 65536])]))
def testtextoutput1simpleatom(self): def testtextoutput1simpleatom(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], [])])) (b'foo', [], [])]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x00\x00foo'), ffs(br'1 text-output 0 \x03\x00\x00\x00foo'),
]) ])
def testtextoutput2simpleatoms(self): def testtextoutput2simpleatoms(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], []), (b'foo', [], []),
(b'bar', [], []), (b'bar', [], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x00\x00foo\x03\x00\x00\x00bar'), ffs(br'1 text-output 0 \x03\x00\x00\x00foo\x03\x00\x00\x00bar'),
]) ])
def testtextoutput1arg(self): def testtextoutput1arg(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s', [b'val1'], []), (b'foo %s', [b'val1'], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x06\x00\x00\x01\x04\x00foo %sval1'), ffs(br'1 text-output 0 \x06\x00\x00\x01\x04\x00foo %sval1'),
]) ])
def testtextoutput2arg(self): def testtextoutput2arg(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s %s', [b'val', b'value'], []), (b'foo %s %s', [b'val', b'value'], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x09\x00\x00\x02\x03\x00\x05\x00' ffs(br'1 text-output 0 \x09\x00\x00\x02\x03\x00\x05\x00'
br'foo %s %svalvalue'), br'foo %s %svalvalue'),
]) ])
def testtextoutput1label(self): def testtextoutput1label(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], [b'label']), (b'foo', [], [b'label']),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x01\x00\x05foolabel'), ffs(br'1 text-output 0 \x03\x00\x01\x00\x05foolabel'),
]) ])
def testargandlabel(self): def testargandlabel(self):
val = list(framing.createtextoutputframe(1, [ stream = framing.stream()
val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s', [b'arg'], [b'label']), (b'foo %s', [b'arg'], [b'label']),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x06\x00\x01\x01\x05\x03\x00foo %slabelarg'), ffs(br'1 text-output 0 \x06\x00\x01\x01\x05\x03\x00foo %slabelarg'),
]) ])
class ServerReactorTests(unittest.TestCase): class ServerReactorTests(unittest.TestCase):
def assertframesequal(self, frames, framestrings): def assertframesequal(self, frames, framestrings):
expected = [ffs(s) for s in framestrings] expected = [ffs(s) for s in framestrings]
self.assertEqual(list(frames), expected) self.assertEqual(list(frames), expected)
def test1framecommand(self): def test1framecommand(self):
"""Receiving a command in a single frame yields request to run it.""" """Receiving a command in a single frame yields request to run it."""
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, 1, b'mycommand', {})) stream = framing.stream()
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': None, 'data': None,
}) })
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
def test1argument(self): def test1argument(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, 41, b'mycommand', stream = framing.stream()
results = list(sendcommandframes(reactor, stream, 41, b'mycommand',
{b'foo': b'bar'})) {b'foo': b'bar'}))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 41, 'requestid': 41,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar'}, 'args': {b'foo': b'bar'},
'data': None, 'data': None,
}) })
def testmultiarguments(self): def testmultiarguments(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, 1, b'mycommand', stream = framing.stream()
results = list(sendcommandframes(reactor, stream, 1, b'mycommand',
{b'foo': b'bar', b'biz': b'baz'})) {b'foo': b'bar', b'biz': b'baz'}))
self.assertEqual(len(results), 3) self.assertEqual(len(results), 3)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'wantframe') self.assertaction(results[1], 'wantframe')
self.assertaction(results[2], 'runcommand') self.assertaction(results[2], 'runcommand')
self.assertEqual(results[2][1], { self.assertEqual(results[2][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar', b'biz': b'baz'}, 'args': {b'foo': b'bar', b'biz': b'baz'},
'data': None, 'data': None,
}) })
def testsimplecommanddata(self): def testsimplecommanddata(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, 1, b'mycommand', {}, stream = framing.stream()
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {},
util.bytesio(b'data!'))) util.bytesio(b'data!')))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'missing frame flags on command frame', 'message': b'missing frame flags on command frame',
}) })
def testconflictingrequestidallowed(self): def testconflictingrequestidallowed(self):
"""Multiple fully serviced commands with same request ID is allowed.""" """Multiple fully serviced commands with same request ID is allowed."""
reactor = makereactor() reactor = makereactor()
results = [] results = []
outstream = framing.stream()
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 command-name eos command')))
result = reactor.onbytesresponseready(1, b'response1') result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 command-name eos command')))
result = reactor.onbytesresponseready(1, b'response2') result = reactor.onbytesresponseready(outstream, 1, b'response2')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 command-name eos command')))
result = reactor.onbytesresponseready(1, b'response3') result = reactor.onbytesresponseready(outstream, 1, b'response3')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
for i in range(3): for i in range(3):
self.assertaction(results[i], 'runcommand') self.assertaction(results[i], 'runcommand')
self.assertEqual(results[i][1], { self.assertEqual(results[i][1], {
'requestid': 1, 'requestid': 1,
'command': b'command', 'command': b'command',
self.assertaction(results[2], 'error') self.assertaction(results[2], 'error')
self.assertEqual(results[2][1], { self.assertEqual(results[2][1], {
'message': b'received frame for request that is not receiving: 5', 'message': b'received frame for request that is not receiving: 5',
}) })
def testsimpleresponse(self): def testsimpleresponse(self):
"""Bytes response to command sends result frames.""" """Bytes response to command sends result frames."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'mycommand', {})) instream = framing.stream()
list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
result = reactor.onbytesresponseready(1, b'response') outstream = framing.stream()
result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response', b'1 bytes-response eos response',
]) ])
def testmultiframeresponse(self): def testmultiframeresponse(self):
"""Bytes response spanning multiple frames is handled.""" """Bytes response spanning multiple frames is handled."""
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE
second = b'y' * 100 second = b'y' * 100
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'mycommand', {})) instream = framing.stream()
list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
result = reactor.onbytesresponseready(1, first + second) outstream = framing.stream()
result = reactor.onbytesresponseready(outstream, 1, first + second)
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response continuation %s' % first, b'1 bytes-response continuation %s' % first,
b'1 bytes-response eos %s' % second, b'1 bytes-response eos %s' % second,
]) ])
def testapplicationerror(self): def testapplicationerror(self):
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'mycommand', {})) instream = framing.stream()
list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
result = reactor.onapplicationerror(1, b'some message') outstream = framing.stream()
result = reactor.onapplicationerror(outstream, 1, b'some message')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 error-response application some message', b'1 error-response application some message',
]) ])
def test1commanddeferresponse(self): def test1commanddeferresponse(self):
"""Responses when in deferred output mode are delayed until EOF.""" """Responses when in deferred output mode are delayed until EOF."""
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
results = list(sendcommandframes(reactor, 1, b'mycommand', {})) instream = framing.stream()
results = list(sendcommandframes(reactor, instream, 1, b'mycommand',
{}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
result = reactor.onbytesresponseready(1, b'response') outstream = framing.stream()
result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response', b'1 bytes-response eos response',
]) ])
def testmultiplecommanddeferresponse(self): def testmultiplecommanddeferresponse(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
list(sendcommandframes(reactor, 1, b'command1', {})) instream = framing.stream()
list(sendcommandframes(reactor, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, instream, 3, b'command2', {}))
result = reactor.onbytesresponseready(1, b'response1') outstream = framing.stream()
result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.onbytesresponseready(3, b'response2') result = reactor.onbytesresponseready(outstream, 3, b'response2')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response1', b'1 bytes-response eos response1',
b'3 bytes-response eos response2' b'3 bytes-response eos response2'
]) ])
def testrequestidtracking(self): def testrequestidtracking(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
list(sendcommandframes(reactor, 1, b'command1', {})) instream = framing.stream()
list(sendcommandframes(reactor, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, 5, b'command3', {})) list(sendcommandframes(reactor, instream, 3, b'command2', {}))
list(sendcommandframes(reactor, instream, 5, b'command3', {}))
# Register results for commands out of order. # Register results for commands out of order.
reactor.onbytesresponseready(3, b'response3') outstream = framing.stream()
reactor.onbytesresponseready(1, b'response1') reactor.onbytesresponseready(outstream, 3, b'response3')
reactor.onbytesresponseready(5, b'response5') reactor.onbytesresponseready(outstream, 1, b'response1')
reactor.onbytesresponseready(outstream, 5, b'response5')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'3 bytes-response eos response3', b'3 bytes-response eos response3',
b'1 bytes-response eos response1', b'1 bytes-response eos response1',
b'5 bytes-response eos response5', b'5 bytes-response eos response5',
]) ])
def testduplicaterequestonactivecommand(self): def testduplicaterequestonactivecommand(self):
"""Receiving a request ID that matches a request that isn't finished.""" """Receiving a request ID that matches a request that isn't finished."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'command1', {})) stream = framing.stream()
results = list(sendcommandframes(reactor, 1, b'command1', {})) list(sendcommandframes(reactor, stream, 1, b'command1', {}))
results = list(sendcommandframes(reactor, stream, 1, b'command1', {}))
self.assertaction(results[0], 'error') self.assertaction(results[0], 'error')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'message': b'request with ID 1 is already active', 'message': b'request with ID 1 is already active',
}) })
def testduplicaterequestonactivecommandnosend(self): def testduplicaterequestonactivecommandnosend(self):
"""Same as above but we've registered a response but haven't sent it.""" """Same as above but we've registered a response but haven't sent it."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'command1', {})) instream = framing.stream()
reactor.onbytesresponseready(1, b'response') list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream()
reactor.onbytesresponseready(outstream, 1, b'response')
# We've registered the response but haven't sent it. From the # We've registered the response but haven't sent it. From the
# perspective of the reactor, the command is still active. # perspective of the reactor, the command is still active.
results = list(sendcommandframes(reactor, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'error') self.assertaction(results[0], 'error')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'message': b'request with ID 1 is already active', 'message': b'request with ID 1 is already active',
}) })
def testduplicaterequestargumentframe(self): def testduplicaterequestargumentframe(self):
"""Variant on above except we sent an argument frame instead of name.""" """Variant on above except we sent an argument frame instead of name."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'command', {})) stream = framing.stream()
list(sendcommandframes(reactor, stream, 1, b'command', {}))
results = list(sendframes(reactor, [ results = list(sendframes(reactor, [
ffs(b'3 command-name have-args command'), ffs(b'3 command-name have-args command'),
ffs(b'1 command-argument 0 ignored'), ffs(b'1 command-argument 0 ignored'),
])) ]))
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': 'received frame for request that is still active: 1', 'message': 'received frame for request that is still active: 1',
}) })
def testduplicaterequestaftersend(self): def testduplicaterequestaftersend(self):
"""We can use a duplicate request ID after we've sent the response.""" """We can use a duplicate request ID after we've sent the response."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'command1', {})) instream = framing.stream()
res = reactor.onbytesresponseready(1, b'response') list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream()
res = reactor.onbytesresponseready(outstream, 1, b'response')
list(res[1]['framegen']) list(res[1]['framegen'])
results = list(sendcommandframes(reactor, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)