[PATCH 1/1] backport iutil_test.py updates from master

atodorov installerbot-noreply at redhat.com
Fri Feb 27 10:45:21 UTC 2015


From: Alexander Todorov <atodorov at 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 at redhat.com>
+
+from contextlib import contextmanager
+import signal
+
+ at 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/0a28e9717b97a990243c953dcbe1f559ec4aea6c


More information about the anaconda-patches mailing list