tests/test-wireproto.py
author Jun Wu <quark@fb.com>
Fri, 24 Mar 2017 14:59:19 -0700
changeset 31622 2243ba216f66
parent 28861 86db5cb55d46
child 33767 b47fe9733d76
permissions -rw-r--r--
statfs: change Linux feature detection Previously we check three things: "statfs" function, "linux/magic.h" and "sys/vfs.h" headers. But we didn't check "struct statfs" or the "f_type" field. That means if a system has "statfs" but "struct statfs" is not defined in the two header files we check, or defined without the "f_type" field, the compilation will fail. This patch combines the checks (2 headers + 1 function + 1 field) together and sets "HAVE_LINUX_STATFS". It makes setup.py faster (less checks), and more reliable (immutable to the issue above).

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