X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/17b97ab3d27b766e64b8868019b7914c97d08fe1..43c16a8a1adfd543751fcaf60ad4c8e04cf83688:/test/ganeti.utils_unittest.py?ds=sidebyside diff --git a/test/ganeti.utils_unittest.py b/test/ganeti.utils_unittest.py index ea136f3..80e737c 100755 --- a/test/ganeti.utils_unittest.py +++ b/test/ganeti.utils_unittest.py @@ -48,519 +48,6 @@ from ganeti.utils import RunCmd, \ RunParts -class TestIsProcessAlive(unittest.TestCase): - """Testing case for IsProcessAlive""" - - def testExists(self): - mypid = os.getpid() - self.assert_(utils.IsProcessAlive(mypid), "can't find myself running") - - def testNotExisting(self): - pid_non_existing = os.fork() - if pid_non_existing == 0: - os._exit(0) - elif pid_non_existing < 0: - raise SystemError("can't fork") - os.waitpid(pid_non_existing, 0) - self.assertFalse(utils.IsProcessAlive(pid_non_existing), - "nonexisting process detected") - - -class TestGetProcStatusPath(unittest.TestCase): - def test(self): - self.assert_("/1234/" in utils._GetProcStatusPath(1234)) - self.assertNotEqual(utils._GetProcStatusPath(1), - utils._GetProcStatusPath(2)) - - -class TestIsProcessHandlingSignal(unittest.TestCase): - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - - def tearDown(self): - shutil.rmtree(self.tmpdir) - - def testParseSigsetT(self): - self.assertEqual(len(utils._ParseSigsetT("0")), 0) - self.assertEqual(utils._ParseSigsetT("1"), set([1])) - self.assertEqual(utils._ParseSigsetT("1000a"), set([2, 4, 17])) - self.assertEqual(utils._ParseSigsetT("810002"), set([2, 17, 24, ])) - self.assertEqual(utils._ParseSigsetT("0000000180000202"), - set([2, 10, 32, 33])) - self.assertEqual(utils._ParseSigsetT("0000000180000002"), - set([2, 32, 33])) - self.assertEqual(utils._ParseSigsetT("0000000188000002"), - set([2, 28, 32, 33])) - self.assertEqual(utils._ParseSigsetT("000000004b813efb"), - set([1, 2, 4, 5, 6, 7, 8, 10, 11, 12, 13, 14, 17, - 24, 25, 26, 28, 31])) - self.assertEqual(utils._ParseSigsetT("ffffff"), set(range(1, 25))) - - def testGetProcStatusField(self): - for field in ["SigCgt", "Name", "FDSize"]: - for value in ["", "0", "cat", " 1234 KB"]: - pstatus = "\n".join([ - "VmPeak: 999 kB", - "%s: %s" % (field, value), - "TracerPid: 0", - ]) - result = utils._GetProcStatusField(pstatus, field) - self.assertEqual(result, value.strip()) - - def test(self): - sp = utils.PathJoin(self.tmpdir, "status") - - utils.WriteFile(sp, data="\n".join([ - "Name: bash", - "State: S (sleeping)", - "SleepAVG: 98%", - "Pid: 22250", - "PPid: 10858", - "TracerPid: 0", - "SigBlk: 0000000000010000", - "SigIgn: 0000000000384004", - "SigCgt: 000000004b813efb", - "CapEff: 0000000000000000", - ])) - - self.assert_(utils.IsProcessHandlingSignal(1234, 10, status_path=sp)) - - def testNoSigCgt(self): - sp = utils.PathJoin(self.tmpdir, "status") - - utils.WriteFile(sp, data="\n".join([ - "Name: bash", - ])) - - self.assertRaises(RuntimeError, utils.IsProcessHandlingSignal, - 1234, 10, status_path=sp) - - def testNoSuchFile(self): - sp = utils.PathJoin(self.tmpdir, "notexist") - - self.assertFalse(utils.IsProcessHandlingSignal(1234, 10, status_path=sp)) - - @staticmethod - def _TestRealProcess(): - signal.signal(signal.SIGUSR1, signal.SIG_DFL) - if utils.IsProcessHandlingSignal(os.getpid(), signal.SIGUSR1): - raise Exception("SIGUSR1 is handled when it should not be") - - signal.signal(signal.SIGUSR1, lambda signum, frame: None) - if not utils.IsProcessHandlingSignal(os.getpid(), signal.SIGUSR1): - raise Exception("SIGUSR1 is not handled when it should be") - - signal.signal(signal.SIGUSR1, signal.SIG_IGN) - if utils.IsProcessHandlingSignal(os.getpid(), signal.SIGUSR1): - raise Exception("SIGUSR1 is not handled when it should be") - - signal.signal(signal.SIGUSR1, signal.SIG_DFL) - if utils.IsProcessHandlingSignal(os.getpid(), signal.SIGUSR1): - raise Exception("SIGUSR1 is handled when it should not be") - - return True - - def testRealProcess(self): - self.assert_(utils.RunInSeparateProcess(self._TestRealProcess)) - - -class TestRunCmd(testutils.GanetiTestCase): - """Testing case for the RunCmd function""" - - def setUp(self): - testutils.GanetiTestCase.setUp(self) - self.magic = time.ctime() + " ganeti test" - self.fname = self._CreateTempFile() - self.fifo_tmpdir = tempfile.mkdtemp() - self.fifo_file = os.path.join(self.fifo_tmpdir, "ganeti_test_fifo") - os.mkfifo(self.fifo_file) - - def tearDown(self): - shutil.rmtree(self.fifo_tmpdir) - testutils.GanetiTestCase.tearDown(self) - - def testOk(self): - """Test successful exit code""" - result = RunCmd("/bin/sh -c 'exit 0'") - self.assertEqual(result.exit_code, 0) - self.assertEqual(result.output, "") - - def testFail(self): - """Test fail exit code""" - result = RunCmd("/bin/sh -c 'exit 1'") - self.assertEqual(result.exit_code, 1) - self.assertEqual(result.output, "") - - def testStdout(self): - """Test standard output""" - cmd = 'echo -n "%s"' % self.magic - result = RunCmd("/bin/sh -c '%s'" % cmd) - self.assertEqual(result.stdout, self.magic) - result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname) - self.assertEqual(result.output, "") - self.assertFileContent(self.fname, self.magic) - - def testStderr(self): - """Test standard error""" - cmd = 'echo -n "%s"' % self.magic - result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd) - self.assertEqual(result.stderr, self.magic) - result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd, output=self.fname) - self.assertEqual(result.output, "") - self.assertFileContent(self.fname, self.magic) - - def testCombined(self): - """Test combined output""" - cmd = 'echo -n "A%s"; echo -n "B%s" 1>&2' % (self.magic, self.magic) - expected = "A" + self.magic + "B" + self.magic - result = RunCmd("/bin/sh -c '%s'" % cmd) - self.assertEqual(result.output, expected) - result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname) - self.assertEqual(result.output, "") - self.assertFileContent(self.fname, expected) - - def testSignal(self): - """Test signal""" - result = RunCmd(["python", "-c", "import os; os.kill(os.getpid(), 15)"]) - self.assertEqual(result.signal, 15) - self.assertEqual(result.output, "") - - def testTimeoutClean(self): - cmd = "trap 'exit 0' TERM; read < %s" % self.fifo_file - result = RunCmd(["/bin/sh", "-c", cmd], timeout=0.2) - self.assertEqual(result.exit_code, 0) - - def testTimeoutKill(self): - cmd = ["/bin/sh", "-c", "trap '' TERM; read < %s" % self.fifo_file] - timeout = 0.2 - out, err, status, ta = utils._RunCmdPipe(cmd, {}, False, "/", False, - timeout, _linger_timeout=0.2) - self.assert_(status < 0) - self.assertEqual(-status, signal.SIGKILL) - - def testTimeoutOutputAfterTerm(self): - cmd = "trap 'echo sigtermed; exit 1' TERM; read < %s" % self.fifo_file - result = RunCmd(["/bin/sh", "-c", cmd], timeout=0.2) - self.assert_(result.failed) - self.assertEqual(result.stdout, "sigtermed\n") - - def testListRun(self): - """Test list runs""" - result = RunCmd(["true"]) - self.assertEqual(result.signal, None) - self.assertEqual(result.exit_code, 0) - result = RunCmd(["/bin/sh", "-c", "exit 1"]) - self.assertEqual(result.signal, None) - self.assertEqual(result.exit_code, 1) - result = RunCmd(["echo", "-n", self.magic]) - self.assertEqual(result.signal, None) - self.assertEqual(result.exit_code, 0) - self.assertEqual(result.stdout, self.magic) - - def testFileEmptyOutput(self): - """Test file output""" - result = RunCmd(["true"], output=self.fname) - self.assertEqual(result.signal, None) - self.assertEqual(result.exit_code, 0) - self.assertFileContent(self.fname, "") - - def testLang(self): - """Test locale environment""" - old_env = os.environ.copy() - try: - os.environ["LANG"] = "en_US.UTF-8" - os.environ["LC_ALL"] = "en_US.UTF-8" - result = RunCmd(["locale"]) - for line in result.output.splitlines(): - key, value = line.split("=", 1) - # Ignore these variables, they're overridden by LC_ALL - if key == "LANG" or key == "LANGUAGE": - continue - self.failIf(value and value != "C" and value != '"C"', - "Variable %s is set to the invalid value '%s'" % (key, value)) - finally: - os.environ = old_env - - def testDefaultCwd(self): - """Test default working directory""" - self.failUnlessEqual(RunCmd(["pwd"]).stdout.strip(), "/") - - def testCwd(self): - """Test default working directory""" - self.failUnlessEqual(RunCmd(["pwd"], cwd="/").stdout.strip(), "/") - self.failUnlessEqual(RunCmd(["pwd"], cwd="/tmp").stdout.strip(), "/tmp") - cwd = os.getcwd() - self.failUnlessEqual(RunCmd(["pwd"], cwd=cwd).stdout.strip(), cwd) - - def testResetEnv(self): - """Test environment reset functionality""" - self.failUnlessEqual(RunCmd(["env"], reset_env=True).stdout.strip(), "") - self.failUnlessEqual(RunCmd(["env"], reset_env=True, - env={"FOO": "bar",}).stdout.strip(), "FOO=bar") - - def testNoFork(self): - """Test that nofork raise an error""" - self.assertFalse(utils._no_fork) - utils.DisableFork() - try: - self.assertTrue(utils._no_fork) - self.assertRaises(errors.ProgrammerError, RunCmd, ["true"]) - finally: - utils._no_fork = False - - def testWrongParams(self): - """Test wrong parameters""" - self.assertRaises(errors.ProgrammerError, RunCmd, ["true"], - output="/dev/null", interactive=True) - - -class TestRunParts(testutils.GanetiTestCase): - """Testing case for the RunParts function""" - - def setUp(self): - self.rundir = tempfile.mkdtemp(prefix="ganeti-test", suffix=".tmp") - - def tearDown(self): - shutil.rmtree(self.rundir) - - def testEmpty(self): - """Test on an empty dir""" - self.failUnlessEqual(RunParts(self.rundir, reset_env=True), []) - - def testSkipWrongName(self): - """Test that wrong files are skipped""" - fname = os.path.join(self.rundir, "00test.dot") - utils.WriteFile(fname, data="") - os.chmod(fname, stat.S_IREAD | stat.S_IEXEC) - relname = os.path.basename(fname) - self.failUnlessEqual(RunParts(self.rundir, reset_env=True), - [(relname, constants.RUNPARTS_SKIP, None)]) - - def testSkipNonExec(self): - """Test that non executable files are skipped""" - fname = os.path.join(self.rundir, "00test") - utils.WriteFile(fname, data="") - relname = os.path.basename(fname) - self.failUnlessEqual(RunParts(self.rundir, reset_env=True), - [(relname, constants.RUNPARTS_SKIP, None)]) - - def testError(self): - """Test error on a broken executable""" - fname = os.path.join(self.rundir, "00test") - utils.WriteFile(fname, data="") - os.chmod(fname, stat.S_IREAD | stat.S_IEXEC) - (relname, status, error) = RunParts(self.rundir, reset_env=True)[0] - self.failUnlessEqual(relname, os.path.basename(fname)) - self.failUnlessEqual(status, constants.RUNPARTS_ERR) - self.failUnless(error) - - def testSorted(self): - """Test executions are sorted""" - files = [] - files.append(os.path.join(self.rundir, "64test")) - files.append(os.path.join(self.rundir, "00test")) - files.append(os.path.join(self.rundir, "42test")) - - for fname in files: - utils.WriteFile(fname, data="") - - results = RunParts(self.rundir, reset_env=True) - - for fname in sorted(files): - self.failUnlessEqual(os.path.basename(fname), results.pop(0)[0]) - - def testOk(self): - """Test correct execution""" - fname = os.path.join(self.rundir, "00test") - utils.WriteFile(fname, data="#!/bin/sh\n\necho -n ciao") - os.chmod(fname, stat.S_IREAD | stat.S_IEXEC) - (relname, status, runresult) = RunParts(self.rundir, reset_env=True)[0] - self.failUnlessEqual(relname, os.path.basename(fname)) - self.failUnlessEqual(status, constants.RUNPARTS_RUN) - self.failUnlessEqual(runresult.stdout, "ciao") - - def testRunFail(self): - """Test correct execution, with run failure""" - fname = os.path.join(self.rundir, "00test") - utils.WriteFile(fname, data="#!/bin/sh\n\nexit 1") - os.chmod(fname, stat.S_IREAD | stat.S_IEXEC) - (relname, status, runresult) = RunParts(self.rundir, reset_env=True)[0] - self.failUnlessEqual(relname, os.path.basename(fname)) - self.failUnlessEqual(status, constants.RUNPARTS_RUN) - self.failUnlessEqual(runresult.exit_code, 1) - self.failUnless(runresult.failed) - - def testRunMix(self): - files = [] - files.append(os.path.join(self.rundir, "00test")) - files.append(os.path.join(self.rundir, "42test")) - files.append(os.path.join(self.rundir, "64test")) - files.append(os.path.join(self.rundir, "99test")) - - files.sort() - - # 1st has errors in execution - utils.WriteFile(files[0], data="#!/bin/sh\n\nexit 1") - os.chmod(files[0], stat.S_IREAD | stat.S_IEXEC) - - # 2nd is skipped - utils.WriteFile(files[1], data="") - - # 3rd cannot execute properly - utils.WriteFile(files[2], data="") - os.chmod(files[2], stat.S_IREAD | stat.S_IEXEC) - - # 4th execs - utils.WriteFile(files[3], data="#!/bin/sh\n\necho -n ciao") - os.chmod(files[3], stat.S_IREAD | stat.S_IEXEC) - - results = RunParts(self.rundir, reset_env=True) - - (relname, status, runresult) = results[0] - self.failUnlessEqual(relname, os.path.basename(files[0])) - self.failUnlessEqual(status, constants.RUNPARTS_RUN) - self.failUnlessEqual(runresult.exit_code, 1) - self.failUnless(runresult.failed) - - (relname, status, runresult) = results[1] - self.failUnlessEqual(relname, os.path.basename(files[1])) - self.failUnlessEqual(status, constants.RUNPARTS_SKIP) - self.failUnlessEqual(runresult, None) - - (relname, status, runresult) = results[2] - self.failUnlessEqual(relname, os.path.basename(files[2])) - self.failUnlessEqual(status, constants.RUNPARTS_ERR) - self.failUnless(runresult) - - (relname, status, runresult) = results[3] - self.failUnlessEqual(relname, os.path.basename(files[3])) - self.failUnlessEqual(status, constants.RUNPARTS_RUN) - self.failUnlessEqual(runresult.output, "ciao") - self.failUnlessEqual(runresult.exit_code, 0) - self.failUnless(not runresult.failed) - - def testMissingDirectory(self): - nosuchdir = utils.PathJoin(self.rundir, "no/such/directory") - self.assertEqual(RunParts(nosuchdir), []) - - -class TestStartDaemon(testutils.GanetiTestCase): - def setUp(self): - self.tmpdir = tempfile.mkdtemp(prefix="ganeti-test") - self.tmpfile = os.path.join(self.tmpdir, "test") - - def tearDown(self): - shutil.rmtree(self.tmpdir) - - def testShell(self): - utils.StartDaemon("echo Hello World > %s" % self.tmpfile) - self._wait(self.tmpfile, 60.0, "Hello World") - - def testShellOutput(self): - utils.StartDaemon("echo Hello World", output=self.tmpfile) - self._wait(self.tmpfile, 60.0, "Hello World") - - def testNoShellNoOutput(self): - utils.StartDaemon(["pwd"]) - - def testNoShellNoOutputTouch(self): - testfile = os.path.join(self.tmpdir, "check") - self.failIf(os.path.exists(testfile)) - utils.StartDaemon(["touch", testfile]) - self._wait(testfile, 60.0, "") - - def testNoShellOutput(self): - utils.StartDaemon(["pwd"], output=self.tmpfile) - self._wait(self.tmpfile, 60.0, "/") - - def testNoShellOutputCwd(self): - utils.StartDaemon(["pwd"], output=self.tmpfile, cwd=os.getcwd()) - self._wait(self.tmpfile, 60.0, os.getcwd()) - - def testShellEnv(self): - utils.StartDaemon("echo \"$GNT_TEST_VAR\"", output=self.tmpfile, - env={ "GNT_TEST_VAR": "Hello World", }) - self._wait(self.tmpfile, 60.0, "Hello World") - - def testNoShellEnv(self): - utils.StartDaemon(["printenv", "GNT_TEST_VAR"], output=self.tmpfile, - env={ "GNT_TEST_VAR": "Hello World", }) - self._wait(self.tmpfile, 60.0, "Hello World") - - def testOutputFd(self): - fd = os.open(self.tmpfile, os.O_WRONLY | os.O_CREAT) - try: - utils.StartDaemon(["pwd"], output_fd=fd, cwd=os.getcwd()) - finally: - os.close(fd) - self._wait(self.tmpfile, 60.0, os.getcwd()) - - def testPid(self): - pid = utils.StartDaemon("echo $$ > %s" % self.tmpfile) - self._wait(self.tmpfile, 60.0, str(pid)) - - def testPidFile(self): - pidfile = os.path.join(self.tmpdir, "pid") - checkfile = os.path.join(self.tmpdir, "abort") - - pid = utils.StartDaemon("while sleep 5; do :; done", pidfile=pidfile, - output=self.tmpfile) - try: - fd = os.open(pidfile, os.O_RDONLY) - try: - # Check file is locked - self.assertRaises(errors.LockError, utils.LockFile, fd) - - pidtext = os.read(fd, 100) - finally: - os.close(fd) - - self.assertEqual(int(pidtext.strip()), pid) - - self.assert_(utils.IsProcessAlive(pid)) - finally: - # No matter what happens, kill daemon - utils.KillProcess(pid, timeout=5.0, waitpid=False) - self.failIf(utils.IsProcessAlive(pid)) - - self.assertEqual(utils.ReadFile(self.tmpfile), "") - - def _wait(self, path, timeout, expected): - # Due to the asynchronous nature of daemon processes, polling is necessary. - # A timeout makes sure the test doesn't hang forever. - def _CheckFile(): - if not (os.path.isfile(path) and - utils.ReadFile(path).strip() == expected): - raise utils.RetryAgain() - - try: - utils.Retry(_CheckFile, (0.01, 1.5, 1.0), timeout) - except utils.RetryTimeout: - self.fail("Apparently the daemon didn't run in %s seconds and/or" - " didn't write the correct output" % timeout) - - def testError(self): - self.assertRaises(errors.OpExecError, utils.StartDaemon, - ["./does-NOT-EXIST/here/0123456789"]) - self.assertRaises(errors.OpExecError, utils.StartDaemon, - ["./does-NOT-EXIST/here/0123456789"], - output=os.path.join(self.tmpdir, "DIR/NOT/EXIST")) - self.assertRaises(errors.OpExecError, utils.StartDaemon, - ["./does-NOT-EXIST/here/0123456789"], - cwd=os.path.join(self.tmpdir, "DIR/NOT/EXIST")) - self.assertRaises(errors.OpExecError, utils.StartDaemon, - ["./does-NOT-EXIST/here/0123456789"], - output=os.path.join(self.tmpdir, "DIR/NOT/EXIST")) - - fd = os.open(self.tmpfile, os.O_WRONLY | os.O_CREAT) - try: - self.assertRaises(errors.ProgrammerError, utils.StartDaemon, - ["./does-NOT-EXIST/here/0123456789"], - output=self.tmpfile, output_fd=fd) - finally: - os.close(fd) - - class TestParseCpuMask(unittest.TestCase): """Test case for the ParseCpuMask function.""" @@ -574,6 +61,22 @@ class TestParseCpuMask(unittest.TestCase): self.assertRaises(errors.ParseError, utils.ParseCpuMask, data) +class TestParseMultiCpuMask(unittest.TestCase): + """Test case for the ParseMultiCpuMask function.""" + + def testWellFormed(self): + self.assertEqual(utils.ParseMultiCpuMask(""), []) + self.assertEqual(utils.ParseMultiCpuMask("1"), [[1]]) + self.assertEqual(utils.ParseMultiCpuMask("0-2,4,5-5"), [[0, 1, 2, 4, 5]]) + self.assertEqual(utils.ParseMultiCpuMask("all"), [[-1]]) + self.assertEqual(utils.ParseMultiCpuMask("0-2:all:4,6-8"), + [[0, 1, 2], [-1], [4, 6, 7, 8]]) + + def testInvalidInput(self): + for data in ["garbage", "0,", "0-1-2", "2-1", "1-a", "all-all"]: + self.assertRaises(errors.ParseError, utils.ParseCpuMask, data) + + class TestGetMounts(unittest.TestCase): """Test case for GetMounts().""" @@ -594,12 +97,6 @@ class TestGetMounts(unittest.TestCase): ("none", "/proc", "proc", "rw,nosuid,nodev,noexec,relatime"), ]) -class TestNewUUID(unittest.TestCase): - """Test case for NewUUID""" - - def runTest(self): - self.failUnless(utils.UUID_RE.match(utils.NewUUID())) - class TestFirstFree(unittest.TestCase): """Test case for the FirstFree function""" @@ -714,44 +211,6 @@ class TestForceDictType(unittest.TestCase): {"b": "hello"}, {"b": "no-such-type"}) -class RunInSeparateProcess(unittest.TestCase): - def test(self): - for exp in [True, False]: - def _child(): - return exp - - self.assertEqual(exp, utils.RunInSeparateProcess(_child)) - - def testArgs(self): - for arg in [0, 1, 999, "Hello World", (1, 2, 3)]: - def _child(carg1, carg2): - return carg1 == "Foo" and carg2 == arg - - self.assert_(utils.RunInSeparateProcess(_child, "Foo", arg)) - - def testPid(self): - parent_pid = os.getpid() - - def _check(): - return os.getpid() == parent_pid - - self.failIf(utils.RunInSeparateProcess(_check)) - - def testSignal(self): - def _kill(): - os.kill(os.getpid(), signal.SIGTERM) - - self.assertRaises(errors.GenericError, - utils.RunInSeparateProcess, _kill) - - def testException(self): - def _exc(): - raise errors.GenericError("This is a test") - - self.assertRaises(errors.GenericError, - utils.RunInSeparateProcess, _exc) - - class TestValidateServiceName(unittest.TestCase): def testValid(self): testnames = [ @@ -845,42 +304,6 @@ class TestFindMatch(unittest.TestCase): self.assert_(utils.FindMatch(data, "Hello World") is None) -class TimeMock: - def __init__(self, values): - self.values = values - - def __call__(self): - return self.values.pop(0) - - -class TestRunningTimeout(unittest.TestCase): - def setUp(self): - self.time_fn = TimeMock([0.0, 0.3, 4.6, 6.5]) - - def testRemainingFloat(self): - timeout = utils.RunningTimeout(5.0, True, _time_fn=self.time_fn) - self.assertAlmostEqual(timeout.Remaining(), 4.7) - self.assertAlmostEqual(timeout.Remaining(), 0.4) - self.assertAlmostEqual(timeout.Remaining(), -1.5) - - def testRemaining(self): - self.time_fn = TimeMock([0, 2, 4, 5, 6]) - timeout = utils.RunningTimeout(5, True, _time_fn=self.time_fn) - self.assertEqual(timeout.Remaining(), 3) - self.assertEqual(timeout.Remaining(), 1) - self.assertEqual(timeout.Remaining(), 0) - self.assertEqual(timeout.Remaining(), -1) - - def testRemainingNonNegative(self): - timeout = utils.RunningTimeout(5.0, False, _time_fn=self.time_fn) - self.assertAlmostEqual(timeout.Remaining(), 4.7) - self.assertAlmostEqual(timeout.Remaining(), 0.4) - self.assertEqual(timeout.Remaining(), 0.0) - - def testNegativeTimeout(self): - self.assertRaises(ValueError, utils.RunningTimeout, -1.0, True) - - class TestTryConvert(unittest.TestCase): def test(self): for src, fn, result in [ @@ -892,21 +315,5 @@ class TestTryConvert(unittest.TestCase): self.assertEqual(utils.TryConvert(fn, src), result) -class TestIsValidShellParam(unittest.TestCase): - def test(self): - for val, result in [ - ("abc", True), - ("ab;cd", False), - ]: - self.assertEqual(utils.IsValidShellParam(val), result) - - -class TestBuildShellCmd(unittest.TestCase): - def test(self): - self.assertRaises(errors.ProgrammerError, utils.BuildShellCmd, - "ls %s", "ab;cd") - self.assertEqual(utils.BuildShellCmd("ls %s", "ab"), "ls ab") - - if __name__ == '__main__': testutils.GanetiTestProgram()