test-wireproto.py
124 lines
| 2.6 KiB
| text/x-python
|
PythonLexer
/ tests / test-wireproto.py
|
r28675 | from __future__ import absolute_import, print_function | ||
|
r27301 | |||
|
r37942 | import sys | ||
|
r28860 | from mercurial import ( | ||
|
r36091 | error, | ||
|
r36573 | pycompat, | ||
|
r36962 | ui as uimod, | ||
|
r28861 | util, | ||
|
r36091 | wireprototypes, | ||
|
r37632 | wireprotov1peer, | ||
|
r37803 | wireprotov1server, | ||
|
r28860 | ) | ||
|
r43346 | from mercurial.utils import stringutil | ||
|
r28861 | stringio = util.stringio | ||
|
r14765 | |||
|
r43346 | |||
|
r14765 | class proto(object): | ||
def __init__(self, args): | ||||
self.args = args | ||||
|
r37311 | self.name = 'dummyproto' | ||
|
r14765 | def getargs(self, spec): | ||
args = self.args | ||||
|
r36573 | args.setdefault(b'*', {}) | ||
|
r14765 | names = spec.split() | ||
return [args[n] for n in names] | ||||
|
r36819 | def checkperm(self, perm): | ||
pass | ||||
|
r43346 | |||
|
r37311 | wireprototypes.TRANSPORTS['dummyproto'] = { | ||
'transport': 'dummy', | ||||
'version': 1, | ||||
} | ||||
|
r43346 | |||
|
r37632 | class clientpeer(wireprotov1peer.wirepeer): | ||
|
r36962 | def __init__(self, serverrepo, ui): | ||
|
r14765 | self.serverrepo = serverrepo | ||
|
r37337 | self.ui = ui | ||
|
r33805 | |||
def url(self): | ||||
|
r36573 | return b'test' | ||
|
r33805 | |||
def local(self): | ||||
return None | ||||
def peer(self): | ||||
return self | ||||
def canpush(self): | ||||
return True | ||||
def close(self): | ||||
pass | ||||
def capabilities(self): | ||||
|
r36573 | return [b'batch'] | ||
|
r25912 | |||
|
r14765 | def _call(self, cmd, **args): | ||
|
r36573 | args = pycompat.byteskwargs(args) | ||
|
r37803 | res = wireprotov1server.dispatch(self.serverrepo, proto(args), cmd) | ||
|
r36091 | if isinstance(res, wireprototypes.bytesresponse): | ||
return res.data | ||||
elif isinstance(res, bytes): | ||||
return res | ||||
else: | ||||
raise error.Abort('dummy client does not support response type') | ||||
|
r14765 | |||
|
r28438 | def _callstream(self, cmd, **args): | ||
|
r28861 | return stringio(self._call(cmd, **args)) | ||
|
r28438 | |||
|
r37632 | @wireprotov1peer.batchable | ||
|
r14765 | def greet(self, name): | ||
|
r48687 | return {b'name': mangle(name)}, unmangle | ||
|
r14765 | |||
|
r43346 | |||
|
r14765 | class serverrepo(object): | ||
|
r42006 | def __init__(self, ui): | ||
self.ui = ui | ||||
|
r14765 | def greet(self, name): | ||
|
r36573 | return b"Hello, " + name | ||
|
r14765 | |||
|
r18278 | def filtered(self, name): | ||
return self | ||||
|
r43346 | |||
|
r14765 | def mangle(s): | ||
|
r36573 | return b''.join(pycompat.bytechr(ord(c) + 1) for c in pycompat.bytestr(s)) | ||
|
r43346 | |||
|
r14765 | def unmangle(s): | ||
|
r36573 | return b''.join(pycompat.bytechr(ord(c) - 1) for c in pycompat.bytestr(s)) | ||
|
r14765 | |||
|
r43346 | |||
|
r14765 | def greet(repo, proto, name): | ||
return mangle(repo.greet(unmangle(name))) | ||||
|
r43346 | |||
|
r37803 | wireprotov1server.commands[b'greet'] = (greet, b'name') | ||
|
r14765 | |||
|
r42006 | srv = serverrepo(uimod.ui()) | ||
|
r36962 | clt = clientpeer(srv, uimod.ui()) | ||
|
r14765 | |||
|
r43346 | |||
|
r37942 | def printb(data, end=b'\n'): | ||
out = getattr(sys.stdout, 'buffer', sys.stdout) | ||||
out.write(data + end) | ||||
out.flush() | ||||
|
r43346 | |||
|
r37942 | printb(clt.greet(b"Foobar")) | ||
|
r37651 | |||
with clt.commandexecutor() as e: | ||||
fgreet1 = e.callcommand(b'greet', {b'name': b'Fo, =;:<o'}) | ||||
fgreet2 = e.callcommand(b'greet', {b'name': b'Bar'}) | ||||
|
r43346 | printb( | ||
stringutil.pprint([f.result() for f in (fgreet1, fgreet2)], bprefix=True) | ||||
) | ||||