upstream/mercurial-mirror Files · tests/test-wireproto.py

sslutil: try to find CA certficates in well-known locations...

sslutil: try to find CA certficates in well-known locations Many Linux distros and other Nixen have CA certificates in well-defined locations. Rather than potentially fail to load any CA certificates at all (which will always result in a certificate verification failure), we scan for paths to known CA certificate files and load one if seen. Because a proper Mercurial install will have the path to the CA certificate file defined at install time, we print a warning that the install isn't proper and provide a URL with instructions to correct things. We only perform path-based fallback on Pythons that don't know how to call into OpenSSL to load the default verify locations. This is because we trust that Python/OpenSSL is properly configured and knows better than Mercurial. So this new code effectively only runs on Python <2.7.9 (technically Pythons without the modern ssl module).

timeless - - Load All Authors

File last commit:

r28861:86db5cb5 default


                r29500:4b16a5bd

default

Download file

             test-wireproto.py
        
                    61 lines
            
             | 1.4 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / tests / test-wireproto.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

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

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

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