view tests/test-wireproto.py @ 35473:02f54a1ec9eb

lfs: add note messages indicating what store holds the lfs blob

The following corruption related patches were written prior to adding the user
level cache, and it took awhile to track down why the tests changed.  (It
generally made things more resilient.)  But I think this will be useful to the
end user as well.  I didn't make it --debug level, because there can be a ton of
info coming out of clone/push/pull --debug.  The pointers are sorted for test
stability.

I opted for ui.note() instead of checking ui.verbose and then using ui.write()
for convenience, but I see most of this extension does the latter.  I have no
idea what the preferred form is.

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

    @property
    def ui(self):
        return self.serverrepo.ui

    def url(self):
        return 'test'

    def local(self):
        return None

    def peer(self):
        return self

    def canpush(self):
        return True

    def close(self):
        pass

    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.iterbatch()
map(b.greet, ('Fo, =;:<o', 'Bar'))
b.submit()
print([r for r in b.results()])