/ - Diff - snf-ganeti - Greek Research and Technology Network's projects

Revision a9d68e40

       if backup and not dry_run and os.path.isfile(file_name):
         CreateBackup(file_name)
       dir_name, base_name = os.path.split(file_name)
       fd, new_name = tempfile.mkstemp('.new', base_name, dir_name)
       # Whether temporary file needs to be removed (e.g. if any error occurs)
       do_remove = True
       # here we need to make sure we remove the temp file, if any error
       # leaves it in place
       # Function result
       result = None
       (dir_name, base_name) = os.path.split(file_name)
       (fd, new_name) = tempfile.mkstemp(suffix=".new", prefix=base_name,
                                         dir=dir_name)
       try:
         if uid != -1 or gid != -1:
           os.chown(new_name, uid, gid)
         if mode:
           os.chmod(new_name, mode)
         if callable(prewrite):
           prewrite(fd)
         if data is not None:
           os.write(fd, data)
         else:
           fn(fd)
         if callable(postwrite):
           postwrite(fd)
         os.fsync(fd)
         if atime is not None and mtime is not None:
           os.utime(new_name, (atime, mtime))
         try:
           if uid != -1 or gid != -1:
             os.chown(new_name, uid, gid)
           if mode:
             os.chmod(new_name, mode)
           if callable(prewrite):
             prewrite(fd)
           if data is not None:
             os.write(fd, data)
           else:
             fn(fd)
           if callable(postwrite):
             postwrite(fd)
           os.fsync(fd)
           if atime is not None and mtime is not None:
             os.utime(new_name, (atime, mtime))
         finally:
           # Close file unless the file descriptor should be returned
           if close:
             os.close(fd)
           else:
             result = fd
         # Rename file to destination name
         if not dry_run:
           os.rename(new_name, file_name)
           # Successful, no need to remove anymore
           do_remove = False
       finally:
         if close:
           os.close(fd)
           result = None
         else:
           result = fd
         if do_remove:
           RemoveFile(new_name)

     class TestWriteFile(unittest.TestCase):
       def setUp(self):
         self.tmpdir = None
         self.tfile = tempfile.NamedTemporaryFile()
         self.did_pre = False
         self.did_post = False
         self.did_write = False
       def tearDown(self):
         if self.tmpdir:
           shutil.rmtree(self.tmpdir)
       def markPre(self, fd):
         self.did_pre = True
-...
         self.assertRaises(errors.ProgrammerError, utils.WriteFile,
                           self.tfile.name, data="test", atime=0)
       def testCalls(self):
         utils.WriteFile(self.tfile.name, fn=self.markWrite,
                         prewrite=self.markPre, postwrite=self.markPost)
       def testPreWrite(self):
         utils.WriteFile(self.tfile.name, data="", prewrite=self.markPre)
         self.assertTrue(self.did_pre)
         self.assertFalse(self.did_post)
         self.assertFalse(self.did_write)
       def testPostWrite(self):
         utils.WriteFile(self.tfile.name, data="", postwrite=self.markPost)
         self.assertFalse(self.did_pre)
         self.assertTrue(self.did_post)
         self.assertFalse(self.did_write)
       def testWriteFunction(self):
         utils.WriteFile(self.tfile.name, fn=self.markWrite)
         self.assertFalse(self.did_pre)
         self.assertFalse(self.did_post)
         self.assertTrue(self.did_write)
       def testDryRun(self):
-...
         finally:
           os.close(fd)
       def testNoLeftovers(self):
         self.tmpdir = tempfile.mkdtemp()
         self.assertEqual(utils.WriteFile(utils.PathJoin(self.tmpdir, "test"),
                                          data="abc"),
                          None)
         self.assertEqual(os.listdir(self.tmpdir), ["test"])
       def testFailRename(self):
         self.tmpdir = tempfile.mkdtemp()
         target = utils.PathJoin(self.tmpdir, "target")
         os.mkdir(target)
         self.assertRaises(OSError, utils.WriteFile, target, data="abc")
         self.assertTrue(os.path.isdir(target))
         self.assertEqual(os.listdir(self.tmpdir), ["target"])
         self.assertFalse(os.listdir(target))
       def testFailRenameDryRun(self):
         self.tmpdir = tempfile.mkdtemp()
         target = utils.PathJoin(self.tmpdir, "target")
         os.mkdir(target)
         self.assertEqual(utils.WriteFile(target, data="abc", dry_run=True), None)
         self.assertTrue(os.path.isdir(target))
         self.assertEqual(os.listdir(self.tmpdir), ["target"])
         self.assertFalse(os.listdir(target))
       def testBackup(self):
         self.tmpdir = tempfile.mkdtemp()
         testfile = utils.PathJoin(self.tmpdir, "test")
         self.assertEqual(utils.WriteFile(testfile, data="foo", backup=True), None)
         self.assertEqual(utils.ReadFile(testfile), "foo")
         self.assertEqual(os.listdir(self.tmpdir), ["test"])
         # Write again
         assert os.path.isfile(testfile)
         self.assertEqual(utils.WriteFile(testfile, data="bar", backup=True), None)
         self.assertEqual(utils.ReadFile(testfile), "bar")
         self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)
         self.assertTrue("test" in os.listdir(self.tmpdir))
         self.assertEqual(len(os.listdir(self.tmpdir)), 2)
         # Write again as dry-run
         assert os.path.isfile(testfile)
         self.assertEqual(utils.WriteFile(testfile, data="000", backup=True,
                                          dry_run=True),
                          None)
         self.assertEqual(utils.ReadFile(testfile), "bar")
         self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)
         self.assertTrue("test" in os.listdir(self.tmpdir))
         self.assertEqual(len(os.listdir(self.tmpdir)), 2)
     class TestFileID(testutils.GanetiTestCase):
       def testEquality(self):

Also available in: Unified diff

Synnefo » snf-ganeti

Revision a9d68e40