This is an archive of the discontinued Mercurial Phabricator instance.

Differential D5522

tests: migrate test-wireproto-serverreactor.py to our internal CBOR
ClosedPublic

Authored by durin42 on Jan 7 2019, 6:49 PM.

Details

Reviewers
indygreg
pulkit
Group Reviewers
hg-reviewers
Commits
rHGa181a1c8af1d: tests: migrate test-wireproto-serverreactor.py to our internal CBOR
Summary

This leaves the only client of thirdparty.cbor as test-cbor.py, which appears
to be testing the behavior of cborutil against cbor. We should figure out some
appropriate test strings and drop thirdparty.cbor, but that's a mission for
another day.

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 13074

tests/test-wireproto-serverreactor.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
import unittest import unittest
from mercurial.thirdparty import (
cbor,
)
from mercurial import ( from mercurial import (
ui as uimod, ui as uimod,
util, util,
wireprotoframing as framing, wireprotoframing as framing,
) )
from mercurial.utils import ( from mercurial.utils import (
cborutil, cborutil,
) )
ffs = framing.makeframefromhumanstring ffs = framing.makeframefromhumanstring
OK = cbor.dumps({b'status': b'ok'}) OK = b''.join(cborutil.streamencode({b'status': b'ok'}))
def makereactor(deferoutput=False): def makereactor(deferoutput=False):
ui = uimod.ui() ui = uimod.ui()
return framing.serverreactor(ui, deferoutput=deferoutput) return framing.serverreactor(ui, deferoutput=deferoutput)
def sendframes(reactor, gen): def sendframes(reactor, gen):
"""Send a generator of frame bytearray to a reactor. """Send a generator of frame bytearray to a reactor.
self.assertaction(results[0], b'wantframe') self.assertaction(results[0], b'wantframe')
self.assertaction(results[1], b'error') self.assertaction(results[1], b'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
b'message': b'request with ID 1 already received', b'message': b'request with ID 1 already received',
}) })
def testinterleavedcommands(self): def testinterleavedcommands(self):
cbor1 = cbor.dumps({ cbor1 = b''.join(cborutil.streamencode({
b'name': b'command1', b'name': b'command1',
b'args': { b'args': {
b'foo': b'bar', b'foo': b'bar',
b'key1': b'val', b'key1': b'val',
} }
}, canonical=True) }))
cbor3 = cbor.dumps({ cbor3 = b''.join(cborutil.streamencode({
b'name': b'command3', b'name': b'command3',
b'args': { b'args': {
b'biz': b'baz', b'biz': b'baz',
b'key': b'val', b'key': b'val',
}, },
}, canonical=True) }))
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 1 stream-begin command-request new|more %s' % cbor1[0:6]), 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'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'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|more %s' % cbor3[10:13]),
ffs(b'3 1 0 command-request continuation %s' % cbor3[13:]), ffs(b'3 1 0 command-request continuation %s' % cbor3[13:]),
ffs(b'1 1 0 command-request continuation %s' % cbor1[9:]), ffs(b'1 1 0 command-request continuation %s' % cbor1[9:]),