ia64/xen-unstable

view tools/xm-test/lib/XmTestLib/Test.py @ 12768:de8abd5ce652

Revert the change to destroy all VMs when skipping or failing tests -- this
needs more thought, because people shouldn't need to set up their Xen-API server
to run the non-Xen-API bits of the test, and at the moment, you get
authentication failures on every SKIP.

Signed-off-by: Ewan Mellor <ewan@xensource.com>
author Ewan Mellor <ewan@xensource.com>
date Sun Dec 03 13:30:23 2006 +0000 (2006-12-03)
parents b37e66b08bf5
children
line source
1 #!/usr/bin/python
3 """
4 Copyright (C) International Business Machines Corp., 2005
5 Author: Dan Smith <danms@us.ibm.com>
7 This program is free software; you can redistribute it and/or modify
8 it under the terms of the GNU General Public License as published by
9 the Free Software Foundation; under version 2 of the License.
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, write to the Free Software
18 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
20 """
22 ##
23 ## These are utility functions for test cases
24 ##
26 import sys
27 import commands
28 import os
29 import pwd
30 import time
31 import pty
32 import select
33 import signal
34 import re
35 import glob
37 TEST_PASS = 0
38 TEST_FAIL = 255
39 TEST_SKIP = 77
41 # We currently advise waiting this many seconds for the ramdisk to
42 # boot inside a domU
43 TEST_DOMU_BOOT_DELAY = 20
45 if os.environ.get("TEST_VERBOSE"):
46 verbose = True
47 else:
48 verbose = False
50 class TimeoutError(Exception):
51 def __init__(self, msg, outputSoFar):
52 self.msg = msg
53 self.output = outputSoFar
55 def __str__(self):
56 return str(self.msg)
58 def runWithTimeout(cmd, timeout):
59 args = cmd.split()
61 pid, fd = pty.fork();
63 startTime = time.time()
65 if pid == 0:
66 os.execvp(args[0], args)
68 output = ""
70 while time.time() - startTime < timeout:
71 i, o, e = select.select([fd], [], [], timeout)
73 if fd in i:
74 try:
75 str = os.read(fd, 1)
76 output += str
77 except OSError, e:
78 exitPid, status = os.waitpid(pid, os.WNOHANG)
80 if exitPid == pid:
81 if verbose:
82 print "Child exited with %i" % status
83 return status, output
85 if verbose:
86 print "Command timed out: killing pid %i" % pid
88 os.kill(pid, signal.SIGINT)
89 raise TimeoutError("Command execution time exceeded %i seconds" % timeout,
90 outputSoFar=output)
92 def traceCommand(command, timeout=None, logOutput=True):
93 if verbose:
94 print "[dom0] Running `%s'" % command
96 if timeout:
97 status, output = runWithTimeout(command, timeout)
98 else:
99 status, output = commands.getstatusoutput(command)
101 if logOutput and verbose:
102 print output
104 return status, output
106 def getTestName():
107 script = sys.argv[0]
108 fname = os.path.basename(script)
109 match = re.match("([^\.]+)\.[a-z]+", fname)
110 if match:
111 tname = match.group(1)
112 else:
113 tname = "UNKNOWN"
115 return tname
117 def becomeNonRoot():
118 """Become a non-root user, or FAIL if this is not possible. This call
119 succeeds if we are already running as a non-root user.
120 """
122 if os.geteuid() == 0:
123 # Try and become "nobody". This user is commonly in place, but this
124 # could be extended to consider any number of users to be acceptable,
125 # if there are systems where "nobody" is not present.
126 allusers = pwd.getpwall()
127 for u in allusers:
128 if u[0] == "nobody":
129 os.setreuid(u[2], u[2])
130 break
131 if os.geteuid() == 0:
132 FAIL("Could not become a non-root user")
134 def FAIL(format, *args):
135 print "\nREASON:", (format % args)
136 sys.exit(TEST_FAIL)
138 def SKIP(format, *args):
139 print "\nREASON:", (format % args)
140 sys.exit(TEST_SKIP)
142 def saveLog(logText, filename=None):
143 if not filename:
144 filename = "log";
145 logfile = open(filename, 'w');
146 date = commands.getoutput("date");
147 logfile.write("-- BEGIN XmTest Log @" + date + "\n");
148 logfile.write(logText);
149 logfile.write("\n-- END XmTest Log\n");
150 logfile.close();
152 def waitForBoot():
153 if verbose:
154 print "[dom0] Waiting %i seconds for domU boot..." % TEST_DOMU_BOOT_DELAY
155 time.sleep(TEST_DOMU_BOOT_DELAY)
157 def timeStamp():
158 name = getTestName()
160 t = time.asctime(time.localtime())
162 print "*** Test %s started at %s %s" % (name, t,
163 time.tzname[time.daylight])
165 #
166 # Try to start a domain and attach a console to it to see if
167 # the console system is working
168 #
169 def isConsoleDead():
171 from XmTestLib import XmTestDomain, DomainError, XmConsole, ConsoleError
173 domain = XmTestDomain()
175 try:
176 console = domain.start()
177 console.runCmd("ls")
178 except DomainError, e:
179 return True
180 except ConsoleError, e:
181 domain.destroy()
182 return True
184 domain.destroy()
186 return False
188 #
189 # We currently can only load as many concurrent HVM domains as loop
190 # devices, need to find how many devices the system has.
191 def getMaxHVMDomains():
192 nodes = glob.glob("/dev/loop*")
193 maxd = len(nodes)
195 return maxd
198 if __name__ == "__main__":
200 timeStamp()
202 FAIL("foo")