Revision c74cda62 test/ganeti.utils_unittest.py
b/test/ganeti.utils_unittest.py | ||
---|---|---|
237 | 237 |
testutils.GanetiTestCase.setUp(self) |
238 | 238 |
self.magic = time.ctime() + " ganeti test" |
239 | 239 |
self.fname = self._CreateTempFile() |
240 |
self.fifo_tmpdir = tempfile.mkdtemp() |
|
241 |
self.fifo_file = os.path.join(self.fifo_tmpdir, "ganeti_test_fifo") |
|
242 |
os.mkfifo(self.fifo_file) |
|
243 |
|
|
244 |
def tearDown(self): |
|
245 |
shutil.rmtree(self.fifo_tmpdir) |
|
240 | 246 |
|
241 | 247 |
def testOk(self): |
242 | 248 |
"""Test successful exit code""" |
... | ... | |
284 | 290 |
self.assertEqual(result.signal, 15) |
285 | 291 |
self.assertEqual(result.output, "") |
286 | 292 |
|
293 |
def testTimeoutClean(self): |
|
294 |
cmd = "trap 'exit 0' TERM; read < %s" % self.fifo_file |
|
295 |
result = RunCmd(["/bin/sh", "-c", cmd], timeout=0.2) |
|
296 |
self.assertEqual(result.exit_code, 0) |
|
297 |
|
|
298 |
def testTimeoutCleanInteractive(self): |
|
299 |
cmd = "trap 'exit 0' TERM; read" |
|
300 |
result = RunCmd(["/bin/sh", "-c", cmd], interactive=True, timeout=0.2) |
|
301 |
self.assertEqual(result.exit_code, 0) |
|
302 |
|
|
303 |
def testTimeoutNonClean(self): |
|
304 |
for exit_code in (1, 10, 17, 29): |
|
305 |
cmd = "trap 'exit %i' TERM; read" % exit_code |
|
306 |
result = RunCmd(["/bin/sh", "-c", cmd], interactive=True, timeout=0.2) |
|
307 |
self.assert_(result.failed) |
|
308 |
self.assertEqual(result.exit_code, exit_code) |
|
309 |
|
|
310 |
def testTimeoutKill(self): |
|
311 |
cmd = "trap '' TERM; read < %s" % self.fifo_file |
|
312 |
timeout = 0.2 |
|
313 |
strcmd = utils.ShellQuoteArgs(["/bin/sh", "-c", cmd]) |
|
314 |
out, err, status, ta = utils._RunCmdPipe(strcmd, {}, True, "/", False, |
|
315 |
timeout, _linger_timeout=0.2) |
|
316 |
self.assert_(status < 0) |
|
317 |
self.assertEqual(-status, signal.SIGKILL) |
|
318 |
|
|
319 |
def testTimeoutOutputAfterTerm(self): |
|
320 |
cmd = "trap 'echo sigtermed; exit 1' TERM; read < %s" % self.fifo_file |
|
321 |
result = RunCmd(["/bin/sh", "-c", cmd], timeout=0.2) |
|
322 |
self.assert_(result.failed) |
|
323 |
self.assertEqual(result.stdout, "sigtermed\n") |
|
324 |
|
|
287 | 325 |
def testListRun(self): |
288 | 326 |
"""Test list runs""" |
289 | 327 |
result = RunCmd(["true"]) |
Also available in: Unified diff