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

show: construct changeset templater during dispatch...

show: construct changeset templater during dispatch Previously, we constructed a formatter from a specific template topic. Then from show() we reached into the internals of the formatter to resolve a template string to be used to construct a changeset templater. A downside to this approach was it limited us to having the entire template defined in a single entry in the map file. You couldn't reference other entries in the map file and this would lead to long templates and redundancy in the map file. This commit teaches @showview how to instantiate a changeset templater so we can construct a templater with full access to the map file. To prove it works, we've split "showwork" into components.

timeless - - Load All Authors

File last commit:

r28861:86db5cb5 default


                r33046:11f76825

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