tests/test-wireproto.py

test-rebase: add a brute force test

Rebase is becoming more complex and it looks like a good idea to try some
brute force enumeration to cover cases that are hard to find manually.

Using brute force to generate repos in different shapes and enumerating the
rebase source and destination would generate too many cases that takes too
long to compute. This patch limits the "brute force" to only the "rebase
source" part. Repo and destination are still manual.

The added test cases are crafted manually to reveal some behaviors that are
not covered by other tests:

  - "revlog index out of range" crash
  - after rebase, p1 == p2, p2 != null
  - "nothing to merge" abort

In the future we might want to add more tests here. For now I'm more
interested in revealing interesting behaviors in a minified way. I tried
some more complex cases but didn't find other interesting behaviors.

Differential Revision: https://phab.mercurial-scm.org/D262

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