[master 2/4] Add some tests for execReadlines

David Shea dshea at redhat.com
Fri Aug 8 19:32:48 UTC 2014


---
 tests/lib/timer.py                   |  41 +++++++++
 tests/pyanaconda_tests/iutil_test.py | 162 +++++++++++++++++++++++++++++++++++
 2 files changed, 203 insertions(+)
 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 a782c36..ecee3a6 100644
--- a/tests/pyanaconda_tests/iutil_test.py
+++ b/tests/pyanaconda_tests/iutil_test.py
@@ -23,9 +23,13 @@ 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):
@@ -106,6 +110,164 @@ class RunProgramTests(unittest.TestCase):
         self.assertIsInstance(iutil.execReadlines("true", []),
                               types.GeneratorType)
 
+    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_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(StopIteration, rl_generator.next)
+
+        # Test output with no end of line
+        with tempfile.NamedTemporaryFile() as testscript:
+            testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three" | tr -d '\\n'
+exit 0
+""")
+            testscript.flush()
+
+            with timer(5):
+                rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(StopIteration, rl_generator.next)
+
+    def exec_readlines_test_exits(self):
+        """Test execReadlines in different child exit situations."""
+
+        # These tests raise OSError once output has been consumed
+
+        # Test a normal, non-0 exit
+        with tempfile.NamedTemporaryFile() as testscript:
+            testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three"
+exit 1
+""")
+            testscript.flush()
+
+            with timer(5):
+                rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(OSError, rl_generator.next)
+
+        # Test exit on signal
+        with tempfile.NamedTemporaryFile() as testscript:
+            testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three"
+kill -TERM $$
+""")
+            testscript.flush()
+
+            with timer(5):
+                rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(OSError, rl_generator.next)
+
+        # Repeat the above two tests, but exit before a final newline
+        with tempfile.NamedTemporaryFile() as testscript:
+            testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three" | tr -d '\\n'
+exit 1
+""")
+            testscript.flush()
+
+            with timer(5):
+                rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(OSError, rl_generator.next)
+
+        with tempfile.NamedTemporaryFile() as testscript:
+            testscript.write("""#!/bin/sh
+echo "one"
+echo "two"
+echo "three" | tr -d '\\n'
+kill -TERM $$
+""")
+            testscript.flush()
+
+            with timer(5):
+                rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                self.assertEqual(rl_generator.next(), "one")
+                self.assertEqual(rl_generator.next(), "two")
+                self.assertEqual(rl_generator.next(), "three")
+                self.assertRaises(OSError, rl_generator.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 "three" | tr -d '\\n'
+exit 0
+""")
+                testscript.flush()
+
+                with timer(5):
+                    rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                    self.assertEqual(rl_generator.next(), "one")
+                    self.assertEqual(rl_generator.next(), "two")
+                    self.assertEqual(rl_generator.next(), "three")
+                    self.assertRaises(StopIteration, rl_generator.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 "three" | tr -d '\\n'
+exit 0
+""")
+                testscript.flush()
+
+                with timer(5):
+                    rl_generator = iutil.execReadlines("/bin/sh", [testscript.name])
+                    self.assertEqual(rl_generator.next(), "one")
+                    self.assertEqual(rl_generator.next(), "two")
+                    self.assertEqual(rl_generator.next(), "three")
+                    self.assertRaises(StopIteration, rl_generator.next)
+        finally:
+            signal.signal(signal.SIGHUP, old_HUP_handler)
+
 class MiscTests(unittest.TestCase):
     def get_dir_size_test(self):
         """Test the getDirSize."""
-- 
1.9.3



More information about the anaconda-patches mailing list