tests/test-commandserver.t

28835:68a946e83188

  ... 
  ...     # server exits with 1 if the pipe closed while reading the command
  ...     print 'server exit code =', server.wait()
  server exit code = 1

  >>> import cStringIO
  >>> from hgclient import readchannel, runcommand, check
  >>> @check
  ... def serverinput(server):
  ...     readchannel(server)
  ... 
  ...     patch = """

  ... +++ b/a	Thu Jan 01 00:00:00 1970 +0000
  ... @@ -0,0 +1,1 @@
  ... +1
  ... """
  ... 
  ...     runcommand(server, ['import', '-'], input=cStringIO.StringIO(patch))
  ...     runcommand(server, ['log'])
  *** runcommand import -
  applying patch from stdin
  *** runcommand log
  changeset:   0:eff892de26ec

  > def hook(**args):
  >     print 'hook talking'
  >     print 'now try to read something: %r' % sys.stdin.read()
  > EOF

  >>> import cStringIO
  >>> from hgclient import readchannel, runcommand, check
  >>> @check
  ... def hookoutput(server):
  ...     readchannel(server)
  ...     runcommand(server, ['--config',
  ...                         'hooks.pre-identify=python:hook.hook',
  ...                         'id'],
  ...                input=cStringIO.StringIO('some input'))
  *** runcommand --config hooks.pre-identify=python:hook.hook id
  hook talking
  now try to read something: 'some input'
  eff892de26ec tip

  $ cat <<EOF >> .hg/hgrc
  > [extensions]
  > dbgui = dbgui.py
  > EOF

  >>> import cStringIO
  >>> from hgclient import readchannel, runcommand, check
  >>> @check
  ... def getpass(server):
  ...     readchannel(server)
  ...     runcommand(server, ['debuggetpass', '--config',
  ...                         'ui.interactive=True'],
  ...                input=cStringIO.StringIO('1234\n'))
  ...     runcommand(server, ['debugprompt', '--config',
  ...                         'ui.interactive=True'],
  ...                input=cStringIO.StringIO('5678\n'))
  ...     runcommand(server, ['debugreadstdin'])
  ...     runcommand(server, ['debugwritestdout'])
  *** runcommand debuggetpass --config ui.interactive=True
  password: 1234
  *** runcommand debugprompt --config ui.interactive=True

  *** runcommand debugwritestdout


run commandserver in commandserver, which is silly but should work:

  >>> import cStringIO
  >>> from hgclient import readchannel, runcommand, check
  >>> @check
  ... def nested(server):
  ...     print '%c, %r' % readchannel(server)
  ...     class nestedserver(object):
  ...         stdin = cStringIO.StringIO('getencoding\n')
  ...         stdout = cStringIO.StringIO()
  ...     runcommand(server, ['serve', '--cmdserver', 'pipe'],
  ...                output=nestedserver.stdout, input=nestedserver.stdin)
  ...     nestedserver.stdout.seek(0)
  ...     print '%c, %r' % readchannel(nestedserver)  # hello
  ...     print '%c, %r' % readchannel(nestedserver)  # getencoding

  $ cd repo
  $ hg update -q

#if unix-socket unix-permissions

  >>> import cStringIO
  >>> from hgclient import unixserver, readchannel, runcommand, check
  >>> server = unixserver('.hg/server.sock', '.hg/server.log')
  >>> def hellomessage(conn):
  ...     ch, data = readchannel(conn)
  ...     print '%c, %r' % (ch, data)
  ...     runcommand(conn, ['id'])

  ... +++ b/a
  ... @@ -1,1 +1,2 @@
  ...  1
  ... +2
  ... """
  ...     runcommand(conn, ['import', '-'], input=cStringIO.StringIO(patch))
  ...     runcommand(conn, ['log', '-rtip', '-q'])
  >>> check(serverinput, server.connect)
  *** runcommand import -
  applying patch from stdin
  *** runcommand log -rtip -q

28836:3f45488d70df

  ... 
  ...     # server exits with 1 if the pipe closed while reading the command
  ...     print 'server exit code =', server.wait()
  server exit code = 1

  >>> from hgclient import readchannel, runcommand, check, stringio
  >>> @check
  ... def serverinput(server):
  ...     readchannel(server)
  ... 
  ...     patch = """

  ... +++ b/a	Thu Jan 01 00:00:00 1970 +0000
  ... @@ -0,0 +1,1 @@
  ... +1
  ... """
  ... 
  ...     runcommand(server, ['import', '-'], input=stringio(patch))
  ...     runcommand(server, ['log'])
  *** runcommand import -
  applying patch from stdin
  *** runcommand log
  changeset:   0:eff892de26ec

  > def hook(**args):
  >     print 'hook talking'
  >     print 'now try to read something: %r' % sys.stdin.read()
  > EOF

  >>> from hgclient import readchannel, runcommand, check, stringio
  >>> @check
  ... def hookoutput(server):
  ...     readchannel(server)
  ...     runcommand(server, ['--config',
  ...                         'hooks.pre-identify=python:hook.hook',
  ...                         'id'],
  ...                input=stringio('some input'))
  *** runcommand --config hooks.pre-identify=python:hook.hook id
  hook talking
  now try to read something: 'some input'
  eff892de26ec tip

  $ cat <<EOF >> .hg/hgrc
  > [extensions]
  > dbgui = dbgui.py
  > EOF

  >>> from hgclient import readchannel, runcommand, check, stringio
  >>> @check
  ... def getpass(server):
  ...     readchannel(server)
  ...     runcommand(server, ['debuggetpass', '--config',
  ...                         'ui.interactive=True'],
  ...                input=stringio('1234\n'))
  ...     runcommand(server, ['debugprompt', '--config',
  ...                         'ui.interactive=True'],
  ...                input=stringio('5678\n'))
  ...     runcommand(server, ['debugreadstdin'])
  ...     runcommand(server, ['debugwritestdout'])
  *** runcommand debuggetpass --config ui.interactive=True
  password: 1234
  *** runcommand debugprompt --config ui.interactive=True

  *** runcommand debugwritestdout


run commandserver in commandserver, which is silly but should work:

  >>> from hgclient import readchannel, runcommand, check, stringio
  >>> @check
  ... def nested(server):
  ...     print '%c, %r' % readchannel(server)
  ...     class nestedserver(object):
  ...         stdin = stringio('getencoding\n')
  ...         stdout = stringio()
  ...     runcommand(server, ['serve', '--cmdserver', 'pipe'],
  ...                output=nestedserver.stdout, input=nestedserver.stdin)
  ...     nestedserver.stdout.seek(0)
  ...     print '%c, %r' % readchannel(nestedserver)  # hello
  ...     print '%c, %r' % readchannel(nestedserver)  # getencoding

  $ cd repo
  $ hg update -q

#if unix-socket unix-permissions

  >>> from hgclient import unixserver, readchannel, runcommand, check, stringio
  >>> server = unixserver('.hg/server.sock', '.hg/server.log')
  >>> def hellomessage(conn):
  ...     ch, data = readchannel(conn)
  ...     print '%c, %r' % (ch, data)
  ...     runcommand(conn, ['id'])

  ... +++ b/a
  ... @@ -1,1 +1,2 @@
  ...  1
  ... +2
  ... """
  ...     runcommand(conn, ['import', '-'], input=stringio(patch))
  ...     runcommand(conn, ['log', '-rtip', '-q'])
  >>> check(serverinput, server.connect)
  *** runcommand import -
  applying patch from stdin
  *** runcommand log -rtip -q