Mercurial > hg
view tests/test-wireproto.py @ 37132:a54113fcc8c9
lfs: move the 'supportedoutgoingversions' handling to changegroup.py
This handling already exists here for the narrow extension. We still need to
either figure out how to enable changegroup v3 without the extension, or figure
out how to let the server detect that the client doesn't have it loaded, and
emit a user friendly error[1]. I can't tell if D1944 is the appropriate vehicle
for the latter.
[1] https://www.mercurial-scm.org/pipermail/mercurial-devel/2018-January/109550.html
author | Matt Harbison <matt_harbison@yahoo.com> |
---|---|
date | Mon, 26 Mar 2018 23:02:50 -0400 |
parents | 65615c29e74f |
children | 45b39c69fae0 |
line wrap: on
line source
from __future__ import absolute_import, print_function from mercurial import ( error, pycompat, ui as uimod, util, wireproto, wireprototypes, ) stringio = util.stringio class proto(object): def __init__(self, args): self.args = args def getargs(self, spec): args = self.args args.setdefault(b'*', {}) names = spec.split() return [args[n] for n in names] def checkperm(self, perm): pass class clientpeer(wireproto.wirepeer): def __init__(self, serverrepo, ui): self.serverrepo = serverrepo self._ui = ui @property def ui(self): return self._ui def url(self): return b'test' def local(self): return None def peer(self): return self def canpush(self): return True def close(self): pass def capabilities(self): return [b'batch'] def _call(self, cmd, **args): args = pycompat.byteskwargs(args) res = wireproto.dispatch(self.serverrepo, proto(args), cmd) 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') def _callstream(self, cmd, **args): return stringio(self._call(cmd, **args)) @wireproto.batchable def greet(self, name): f = wireproto.future() yield {b'name': mangle(name)}, f yield unmangle(f.value) class serverrepo(object): def greet(self, name): return b"Hello, " + name def filtered(self, name): return self def mangle(s): return b''.join(pycompat.bytechr(ord(c) + 1) for c in pycompat.bytestr(s)) def unmangle(s): return b''.join(pycompat.bytechr(ord(c) - 1) for c in pycompat.bytestr(s)) def greet(repo, proto, name): return mangle(repo.greet(unmangle(name))) wireproto.commands[b'greet'] = (greet, b'name',) srv = serverrepo() clt = clientpeer(srv, uimod.ui()) print(clt.greet(b"Foobar")) b = clt.iterbatch() list(map(b.greet, (b'Fo, =;:<o', b'Bar'))) b.submit() print([r for r in b.results()])