view tests/test-wireproto.py @ 29499:9c5325c79683

sslutil: issue warning when unable to load certificates on OS X Previously, failure to load system certificates on OS X would lead to a certificate verify failure and that's it. We now print a warning message with a URL that will contain information on how to configure certificates on OS X. As the inline comment states, there is room to improve here. I think we could try harder to detect Homebrew and MacPorts installed certificate files, for example. It's worth noting that Homebrew's openssl package uses `security find-certificate -a -p` during package installation to export the system keychain root CAs to etc/openssl/cert.pem. This is something we could consider adding to setup.py. We could also encourage packagers to do this. For now, I'd just like to get this warning (which matches Windows behavior) landed. We should have time to improve things before release.
author Gregory Szorc <gregory.szorc@gmail.com>
date Wed, 06 Jul 2016 20:46:05 -0700
parents 86db5cb55d46
children b47fe9733d76
line wrap: on
line source

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