4 # Copyright (C) 2010, 2012 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Script for testing utils.ResetTempfileModule"""
32 from ganeti import utils
37 # This constant is usually at a much higher value. Setting it lower for test
42 class TestResetTempfileModule(unittest.TestCase):
44 self.tmpdir = tempfile.mkdtemp()
47 shutil.rmtree(self.tmpdir)
49 def testNoReset(self):
50 if ((sys.hexversion >= 0x020703F0 and sys.hexversion < 0x03000000) or
51 sys.hexversion >= 0x030203F0):
52 # We can't test the no_reset case on Python 2.7+
55 if (hasattr(tempfile._RandomNameSequence, "rng") and
56 type(tempfile._RandomNameSequence.rng) == property):
63 def _Test(self, reset):
64 self.failIf(tempfile.TMP_MAX > 10)
66 # Initialize tempfile module
67 (fd, _) = tempfile.mkstemp(dir=self.tmpdir, prefix="init.", suffix="")
70 (notify_read, notify_write) = os.pipe()
78 utils.ResetTempfileModule()
80 os.close(notify_write)
82 # Wait for parent to close pipe
83 os.read(notify_read, 1)
86 # This is a short-lived process, not caring about closing file
88 (_, path) = tempfile.mkstemp(dir=self.tmpdir,
89 prefix="test.", suffix="")
90 except EnvironmentError, err:
91 if err.errno == errno.EEXIST:
92 # Couldnt' create temporary file (e.g. because we run out of
97 logging.debug("Child created %s", path)
101 logging.exception("Unhandled error")
106 os.close(notify_read)
108 # Create parent's temporary files
109 for _ in range(tempfile.TMP_MAX):
110 (fd, path) = tempfile.mkstemp(dir=self.tmpdir,
111 prefix="test.", suffix="")
113 logging.debug("Parent created %s", path)
115 # Notify child by closing pipe
116 os.close(notify_write)
118 (_, status) = os.waitpid(pid, 0)
120 self.failIf(os.WIFSIGNALED(status))
123 # If the tempfile module was reset, it should not fail to create
129 self.assertEqual(os.WEXITSTATUS(status), expected)
132 if __name__ == "__main__":
133 testutils.GanetiTestProgram()