Mercurial > hg
view tests/test-wireproto.py @ 36827:12492794bf8c
py3: silence f.write() in test-annotate.t
author | Yuya Nishihara <yuya@tcha.org> |
---|---|
date | Sat, 10 Mar 2018 16:55:54 +0900 |
parents | 66de4555cefd |
children | 65615c29e74f |
line wrap: on
line source
from __future__ import absolute_import, print_function from mercurial import ( error, pycompat, 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): self.serverrepo = serverrepo @property def ui(self): return self.serverrepo.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) 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()])