From: Alexander Todorov <atodorov(a)redhat.com>
---
tests/lib/timer.py | 41 ++++++++++++
tests/pyanaconda_tests/iutil_test.py | 125 +++++++++++++++++++++++++++++------
2 files changed, 146 insertions(+), 20 deletions(-)
create mode 100644 tests/lib/timer.py
diff --git a/tests/lib/timer.py b/tests/lib/timer.py
new file mode 100644
index 0000000..f7f62c6
--- /dev/null
+++ b/tests/lib/timer.py
@@ -0,0 +1,41 @@
+#
+# timer.py: timer decorator for unittest functions
+#
+# Copyright (C) 2014 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published
+# by the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program 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 Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program. If not, see <
http://www.gnu.org/licenses/>.
+#
+# Author: David Shea <dshea(a)redhat.com>
+
+from contextlib import contextmanager
+import signal
+
+@contextmanager
+def timer(seconds):
+ """Return a timer context manager.
+
+ If the code within the context does not finish within the given number
+ of seconds, it will raise an AssertionError.
+ """
+ def _handle_sigalrm(signum, frame):
+ raise AssertionError("Test failed to complete within %d seconds" %
seconds)
+
+ old_handler = signal.signal(signal.SIGALRM, _handle_sigalrm)
+ try:
+ signal.alarm(seconds)
+ yield
+ finally:
+ # Put everything back
+ signal.alarm(0)
+ signal.signal(signal.SIGALRM, old_handler)
diff --git a/tests/pyanaconda_tests/iutil_test.py b/tests/pyanaconda_tests/iutil_test.py
index 22f5c11..ecb37db 100644
--- a/tests/pyanaconda_tests/iutil_test.py
+++ b/tests/pyanaconda_tests/iutil_test.py
@@ -21,11 +21,14 @@
from pyanaconda import iutil
import unittest
-import types
import os
+import tempfile
+import signal
import shutil
from test_constants import ANACONDA_TEST_DIR
+from timer import timer
+
class UpcaseFirstLetterTests(unittest.TestCase):
def setUp(self):
@@ -99,13 +102,99 @@ def exec_readlines_test(self):
# test some lines are returned
self.assertGreater(len(list(iutil.execReadlines("ls",
["--help"]))), 0)
- # check that it always returns a generator for both
+ # check that it always returns an iterator for both
# if there is some output and if there isn't any
- self.assertIsInstance(iutil.execReadlines("ls", ["--help"]),
- types.GeneratorType)
- self.assertIsInstance(iutil.execReadlines("true", []),
- types.GeneratorType)
-
+ self.assertTrue(hasattr(iutil.execReadlines("ls",
["--help"]), "__iter__"))
+ self.assertTrue(hasattr(iutil.execReadlines("true", []),
"__iter__"))
+
+ def exec_readlines_test_normal_output(self):
+ """Test the output of execReadlines."""
+
+ # Test regular-looking output
+ with tempfile.NamedTemporaryFile() as testscript:
+ testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three"
+exit 0
+""")
+ testscript.flush()
+
+ with timer(5):
+ rl_iterator = iutil.execReadlines("/bin/sh",
[testscript.name])
+ self.assertEqual(rl_iterator.next(), "one")
+ self.assertEqual(rl_iterator.next(), "two")
+ self.assertEqual(rl_iterator.next(), "three")
+ self.assertRaises(StopIteration, rl_iterator.next)
+
+ # Test output with no end of line
+ with tempfile.NamedTemporaryFile() as testscript:
+ testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo -n "three"
+exit 0
+""")
+ testscript.flush()
+
+ with timer(5):
+ rl_iterator = iutil.execReadlines("/bin/sh",
[testscript.name])
+ self.assertEqual(rl_iterator.next(), "one")
+ self.assertEqual(rl_iterator.next(), "two")
+ self.assertEqual(rl_iterator.next(), "three")
+ self.assertRaises(StopIteration, rl_iterator.next)
+
+ def exec_readlines_test_signals(self):
+ """Test execReadlines and signal receipt."""
+
+ # ignored signal
+ old_HUP_handler = signal.signal(signal.SIGHUP, signal.SIG_IGN)
+ try:
+ with tempfile.NamedTemporaryFile() as testscript:
+ testscript.write("""#!/bin/sh
+echo "one"
+kill -HUP $PPID
+echo "two"
+echo -n "three"
+exit 0
+""")
+ testscript.flush()
+
+ with timer(5):
+ rl_iterator = iutil.execReadlines("/bin/sh",
[testscript.name])
+ self.assertEqual(rl_iterator.next(), "one")
+ self.assertEqual(rl_iterator.next(), "two")
+ self.assertEqual(rl_iterator.next(), "three")
+ self.assertRaises(StopIteration, rl_iterator.next)
+ finally:
+ signal.signal(signal.SIGHUP, old_HUP_handler)
+
+ # caught signal
+ def _hup_handler(signum, frame):
+ pass
+ old_HUP_handler = signal.signal(signal.SIGHUP, _hup_handler)
+ try:
+ with tempfile.NamedTemporaryFile() as testscript:
+ testscript.write("""#!/bin/sh
+echo "one"
+kill -HUP $PPID
+echo "two"
+echo -n "three"
+exit 0
+""")
+ testscript.flush()
+
+ with timer(5):
+ rl_iterator = iutil.execReadlines("/bin/sh",
[testscript.name])
+ self.assertEqual(rl_iterator.next(), "one")
+ self.assertEqual(rl_iterator.next(), "two")
+ self.assertEqual(rl_iterator.next(), "three")
+ self.assertRaises(StopIteration, rl_iterator.next)
+ finally:
+ signal.signal(signal.SIGHUP, old_HUP_handler)
+
+
+class MiscTests(unittest.TestCase):
def get_dir_size_test(self):
"""Test the getDirSize."""
@@ -219,11 +308,16 @@ def vt_activate_test(self):
def raise_os_error(*args, **kwargs):
raise OSError
- # chvt does not exist on all platforms
- # and the function needs to correctly survie that
- iutil.vtActivate.func_globals['execWithRedirect'] = raise_os_error
+ _execWithRedirect = iutil.vtActivate.func_globals['execWithRedirect']
- self.assertEqual(iutil.vtActivate(2), False)
+ try:
+ # chvt does not exist on all platforms
+ # and the function needs to correctly survie that
+ iutil.vtActivate.func_globals['execWithRedirect'] = raise_os_error
+
+ self.assertEqual(iutil.vtActivate(2), False)
+ finally:
+ iutil.vtActivate.func_globals['execWithRedirect'] =
_execWithRedirect
def get_deep_attr_test(self):
"""Test getdeepattr."""
@@ -372,12 +466,3 @@ def parent_dir_test(self):
for d, r in dirs:
self.assertEquals(iutil.parent_dir(d), r)
-
- def parse_group_str_test(self):
- groups = [("acme", "acme", None), ("acme (2911)",
"acme", 2911),
- ("acme(2911)", "acme", 2911), ("acme
(2911)(569)", "acme", 2911),
- ("acme ()", "acme", None), ("",
"", None), ("()", "", None),
- ("(2911)", "", None)]
-
- for group_str, name, gid in groups:
- self.assertEquals(iutil.parse_group_str(group_str), (name, gid))
--
To view this commit on github, visit
https://github.com/rhinstaller/anaconda/commit/0a28e9717b97a990243c953dcb...