Mercurial > hg
view tests/test-wireproto.py @ 37283:d4e62df1c73d
debugcommands: drop offset and length from debugindex by default
These fields are an implementation detail of revlog storage. As
such, they are not part of the generic storage "index" interface
and shouldn't be displayed by default.
Because we don't have another way to display these fields, we've
retained support for printing these fields via --verbose.
Yes, I know we should probably be doing all this formatting using
modern formatting/templater APIs. I didn't feel like scope
bloating this patch.
Differential Revision: https://phab.mercurial-scm.org/D3028
author | Gregory Szorc <gregory.szorc@gmail.com> |
---|---|
date | Mon, 02 Apr 2018 16:47:53 -0700 |
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()])