view tests/test-wireproto.py @ 28180:2836a43c7722

run-tests: allow run-tests.py to run tests outside current directory When reloading tests, run-tests.py was assuming that it could look up the test by the basename, which only works if you are running tests which are in the current directory. This patch changes that lookup to use the full path. This is all that was needed, and does not appear to cause any problems for any of the existing testing work flows based on running the suggested commands at the top of run-tests.py. Motivation: In order to test Mercurial with Hypothesis (according to https://www.mercurial-scm.org/wiki/HypothesisPlan) it is useful to be able to generate temporary test files and execute them. Generating temporary files in the tests/ directory leads to a lot of suboptimal clutter.
author David R. MacIver <david@drmaciver.com>
date Thu, 18 Feb 2016 08:52:15 +0000
parents 5defcb7d6e5f
children 48fd02dac1d4
line wrap: on
line source

from __future__ import absolute_import

from mercurial import wireproto

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)

    @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]