Mercurial > public > mercurial-scm > hg

comparison tests/test-commandserver.py @ 22566:480b7fefbb08

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
test-commandserver: split helper functions to new hgclient module This prepares for porting test-commandserver.py to .t test. Though command-server test needs many Python codes, .t test will be more readable than .py test thanks to inlined output.
author Yuya Nishihara <yuya@tcha.org>
date Sun, 28 Sep 2014 13:31:16 +0900
parents 68f2f8bfe9ae
children
comparison
equal deleted inserted replaced
22565:8d45a42b0c0f 22566:480b7fefbb08
1 import sys, os, struct, subprocess, cStringIO, re, shutil 1 import sys, os, cStringIO, re, shutil
2 2
3 def connect(path=None): 3 sys.path.insert(0, os.path.join(os.environ['TESTDIR'], '..', 'contrib'))
4 cmdline = ['hg', 'serve', '--cmdserver', 'pipe'] 4 from hgclient import readchannel, sep, runcommand, check
5 if path:
6 cmdline += ['-R', path]
7
8 server = subprocess.Popen(cmdline, stdin=subprocess.PIPE,
9 stdout=subprocess.PIPE)
10
11 return server
12
13 def writeblock(server, data):
14 server.stdin.write(struct.pack('>I', len(data)))
15 server.stdin.write(data)
16 server.stdin.flush()
17
18 def readchannel(server):
19 data = server.stdout.read(5)
20 if not data:
21 raise EOFError
22 channel, length = struct.unpack('>cI', data)
23 if channel in 'IL':
24 return channel, length
25 else:
26 return channel, server.stdout.read(length)
27
28 def sep(text):
29 return text.replace('\\', '/')
30
31 def runcommand(server, args, output=sys.stdout, error=sys.stderr, input=None,
32 outfilter=lambda x: x):
33 print ' runcommand', ' '.join(args)
34 sys.stdout.flush()
35 server.stdin.write('runcommand\n')
36 writeblock(server, '\0'.join(args))
37
38 if not input:
39 input = cStringIO.StringIO()
40
41 while True:
42 ch, data = readchannel(server)
43 if ch == 'o':
44 output.write(outfilter(data))
45 output.flush()
46 elif ch == 'e':
47 error.write(data)
48 error.flush()
49 elif ch == 'I':
50 writeblock(server, input.read(data))
51 elif ch == 'L':
52 writeblock(server, input.readline(data))
53 elif ch == 'r':
54 ret, = struct.unpack('>i', data)
55 if ret != 0:
56 print ' [%d]' % ret
57 return ret
58 else:
59 print "unexpected channel %c: %r" % (ch, data)
60 if ch.isupper():
61 return
62
63 def check(func, repopath=None):
64 print
65 print 'testing %s:' % func.__name__
66 print
67 sys.stdout.flush()
68 server = connect(repopath)
69 try:
70 return func(server)
71 finally:
72 server.stdin.close()
73 server.wait()
74 5
75 def unknowncommand(server): 6 def unknowncommand(server):
76 server.stdin.write('unknowncommand\n') 7 server.stdin.write('unknowncommand\n')
77 8
78 def hellomessage(server): 9 def hellomessage(server):