1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
#!@PYTHON@
# This file is part of GNUnet.
# (C) 2010 Christian Grothoff (and other contributing authors)
#
# GNUnet is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; either version 2, or (at your
# option) any later version.
#
# GNUnet is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNUnet; see the file COPYING. If not, write to the
# Free Software Foundation, Inc., 59 Temple Place - Suite 330,
# Boston, MA 02111-1307, USA.
#
# Testcase for gnunet-peerinfo
#import pexpect
from __future__ import print_function
import os
#import signal
import re
import subprocess
import sys
import shutil
import time
class pexpect (object):
def __init__ (self):
super (pexpect, self).__init__ ()
def spawn (self, stdin, arglist, *pargs, **kwargs):
self.proc = subprocess.Popen (arglist, *pargs, **kwargs)
if self.proc is None:
print ("Failed to spawn a process {0}".format (arglist))
sys.exit (1)
if stdin is not None:
self.stdo, self.stde = self.proc.communicate (stdin)
else:
self.stdo, self.stde = self.proc.communicate ()
return self.proc
def expect (self, s, r, flags=0):
stream = self.stdo if s == 'stdout' else self.stde
if isinstance (r, str):
if r == "EOF":
if len (stream) == 0:
return True
else:
print ("Failed to match {0} with `{1}'. {0} is `{2}'".format (s, r, stream))
sys.exit (2)
raise ValueError ("Argument `r' should be an instance of re.RegexObject or a special string, but is `{0}'".format (r))
m = r.match (stream, flags)
if not m:
print ("Failed to match {0} with `{1}'. {0} is `{2}'".format (s, r.pattern, stream))
sys.exit (2)
stream = stream[m.end ():]
if s == 'stdout':
self.stdo = stream
else:
self.stde = stream
return m
def read (self, s, size=-1):
stream = self.stdo if s == 'stdout' else self.stde
result = ""
if size < 0:
result = stream
new_stream = ""
else:
result = stream[0:size]
new_stream = stream[size:]
if s == 'stdout':
self.stdo = new_stream
else:
self.stde = new_stream
return result
if os.name == 'posix':
peerinfo = 'gnunet-peerinfo'
gnunetarm = 'gnunet-arm'
elif os.name == 'nt':
peerinfo = 'gnunet-peerinfo.exe'
gnunetarm = 'gnunet-arm.exe'
pinfo = pexpect ()
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', '-L', 'ERROR'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile (r'Error in communication with PEERINFO service\r?\n'))
pinfo.expect ("stdout", "EOF")
if os.name == "nt":
shutil.rmtree (os.path.join (os.getenv ("TEMP"), "tmp", "gnunet-test-peerinfo"), True)
else:
shutil.rmtree ("/tmp/gnunet-test-peerinfo", True)
arm = subprocess.Popen ([gnunetarm, '-sq', '-c', 'test_gnunet_peerinfo_data.conf'])
arm.communicate ()
try:
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', '-s'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile (r'I am peer `.*\'.\r?\n'))
pinfo.expect ("stdout", "EOF")
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', '-qs'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile (r'.......................................................................................................\r?\n'))
pinfo.expect ("stdout", "EOF")
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', 'invalid'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile (r'Invalid command line argument `invalid\'\r?\n'))
pinfo.expect ("stdout", "EOF")
arm = subprocess.Popen ([gnunetarm, '-q', '-i', 'transport', '-c', 'test_gnunet_peerinfo_data.conf'])
arm.communicate ()
time.sleep (1)
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile ("Peer `.*'\r?\n"))
m = pinfo.expect ("stdout", re.compile ("\s.*:24357\r?\n"))
while len (m.group (0)) > 0:
m = pinfo.expect ("stdout", re.compile ("(\s.*:24357\r?\n|\r?\n|)"))
pinfo.expect ("stdout", "EOF")
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', '-n'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pinfo.expect ("stdout", re.compile ("Peer `.*'\r?\n"))
m = pinfo.expect ("stdout", re.compile ("\s.*:24357\r?\n"))
while len (m.group (0)) > 0:
m = pinfo.expect ("stdout", re.compile ("(\s.*:24357\r?\n|\r?\n|)"))
pinfo.expect ("stdout", "EOF")
pinfo.spawn (None, [peerinfo, '-c', 'test_gnunet_peerinfo_data.conf', '-qs'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pid = pinfo.read ("stdout")
pid.strip ()
finally:
arm = subprocess.Popen ([gnunetarm, '-eq', '-c', 'test_gnunet_peerinfo_data.conf'])
arm.communicate ()
if os.name == "nt":
shutil.rmtree (os.path.join (os.getenv ("TEMP"), "tmp", "gnunet-test-peerinfo"), True)
else:
shutil.rmtree ("/tmp/gnunet-test-peerinfo", True)
|
