view tests/test-wireproto.py @ 27874:15c6eb0a51bd

context: use a the nofsauditor when matching file in history (issue4749) Before this change, asking for file from history (eg: 'hg cat -r 42 foo/bar') could fail because of the current content of the working copy (eg: current "foo" being a symlink). As the working copy state have no influence on the content of the history, we can safely skip these checks. The working copy context class have a different 'match' implementation. That implementation still use the repo.auditor will still catch symlink traversal. I've audited all stuff calling "match" and they all go through a ctx in a sensible way. The most unclear case was diff which still seemed okay. You raised my paranoid level today and I double checked through tests. They behave properly. The odds of someone using the wrong (matching with a changectx for operation that will eventually touch the file system) is non-zero because you are never sure of what people will do. But I dunno if we can fight against that. So I would not commit to "never" for "at this level" and "in the future" if someone write especially bad code. However, as a last defense, the vfs itself is running path auditor in all cases outside of .hg/. So I think anything passing the 'matcher' for buggy reason would growl at the vfs layer.
author Pierre-Yves David <pierre-yves.david@fb.com>
date Thu, 03 Dec 2015 13:23:46 -0800
parents cbbdd085c991
children 5defcb7d6e5f
line wrap: on
line source

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]