Mercurial > hg
view tests/test-wireproto.py @ 33115:fa9a90d5ad89
tests: save the original PATH and PYTHONPATH variables
When running the tests, define ORIG_PATH and ORIG_PYTHONPATH environment
variables that contain the original contents of PATH and PYTHONPATH, before
they were modified by run-tests.py
This will make it possible for tests to refer to the original contents of these
variables if necessary. In particular, this is necessary for invoking the
correct version of hg for examining the local repository (the mercurial
repository itself, not the temporary test repositories). Various tests examine
the local repository to check the file lists and contents of commit messages.
author | Adam Simpkins <simpkins@fb.com> |
---|---|
date | Tue, 27 Jun 2017 17:24:31 -0700 |
parents | 86db5cb55d46 |
children | b47fe9733d76 |
line wrap: on
line source
from __future__ import absolute_import, print_function from mercurial import ( util, wireproto, ) stringio = util.stringio class proto(object): def __init__(self, args): self.args = args def getargs(self, spec): args = self.args args.setdefault('*', {}) names = spec.split() return [args[n] for n in names] class clientpeer(wireproto.wirepeer): def __init__(self, serverrepo): self.serverrepo = serverrepo def _capabilities(self): return ['batch'] def _call(self, cmd, **args): return wireproto.dispatch(self.serverrepo, proto(args), cmd) def _callstream(self, cmd, **args): return stringio(self._call(cmd, **args)) @wireproto.batchable def greet(self, name): f = wireproto.future() yield {'name': mangle(name)}, f yield unmangle(f.value) class serverrepo(object): def greet(self, name): return "Hello, " + name def filtered(self, name): return self def mangle(s): return ''.join(chr(ord(c) + 1) for c in s) def unmangle(s): return ''.join(chr(ord(c) - 1) for c in s) def greet(repo, proto, name): return mangle(repo.greet(unmangle(name))) wireproto.commands['greet'] = (greet, 'name',) srv = serverrepo() clt = clientpeer(srv) print(clt.greet("Foobar")) b = clt.batch() fs = [b.greet(s) for s in ["Fo, =;:<o", "Bar"]] b.submit() print([f.value for f in fs])