dulwich/tests/test_file.py

   1 # test_file.py -- Test for git files
   2 # Copyright (C) 2010 Google, Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; version 2
   7 # of the License or (at your option) a later version of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  17 # MA  02110-1301, USA.
  18
  19 import errno
  20 import os
  21 import shutil
  22 import sys
  23 import tempfile
  24
  25 from dulwich.file import GitFile, fancy_rename
  26 from dulwich.tests import (
  27     TestCase,
  28     TestSkipped,
  29     )
  30
  31
  32 class FancyRenameTests(TestCase):
  33
  34     def setUp(self):
  35         super(FancyRenameTests, self).setUp()
  36         self._tempdir = tempfile.mkdtemp()
  37         self.foo = self.path('foo')
  38         self.bar = self.path('bar')
  39         self.create(self.foo, 'foo contents')
  40
  41     def tearDown(self):
  42         shutil.rmtree(self._tempdir)
  43         super(FancyRenameTests, self).tearDown()
  44
  45     def path(self, filename):
  46         return os.path.join(self._tempdir, filename)
  47
  48     def create(self, path, contents):
  49         f = open(path, 'wb')
  50         f.write(contents)
  51         f.close()
  52
  53     def test_no_dest_exists(self):
  54         self.assertFalse(os.path.exists(self.bar))
  55         fancy_rename(self.foo, self.bar)
  56         self.assertFalse(os.path.exists(self.foo))
  57
  58         new_f = open(self.bar, 'rb')
  59         self.assertEquals('foo contents', new_f.read())
  60         new_f.close()
  61
  62     def test_dest_exists(self):
  63         self.create(self.bar, 'bar contents')
  64         fancy_rename(self.foo, self.bar)
  65         self.assertFalse(os.path.exists(self.foo))
  66
  67         new_f = open(self.bar, 'rb')
  68         self.assertEquals('foo contents', new_f.read())
  69         new_f.close()
  70
  71     def test_dest_opened(self):
  72         if sys.platform != "win32":
  73             raise TestSkipped("platform allows overwriting open files")
  74         self.create(self.bar, 'bar contents')
  75         dest_f = open(self.bar, 'rb')
  76         self.assertRaises(OSError, fancy_rename, self.foo, self.bar)
  77         dest_f.close()
  78         self.assertTrue(os.path.exists(self.path('foo')))
  79
  80         new_f = open(self.foo, 'rb')
  81         self.assertEquals('foo contents', new_f.read())
  82         new_f.close()
  83
  84         new_f = open(self.bar, 'rb')
  85         self.assertEquals('bar contents', new_f.read())
  86         new_f.close()
  87
  88
  89 class GitFileTests(TestCase):
  90
  91     def setUp(self):
  92         super(GitFileTests, self).setUp()
  93         self._tempdir = tempfile.mkdtemp()
  94         f = open(self.path('foo'), 'wb')
  95         f.write('foo contents')
  96         f.close()
  97
  98     def tearDown(self):
  99         shutil.rmtree(self._tempdir)
 100         super(GitFileTests, self).tearDown()
 101
 102     def path(self, filename):
 103         return os.path.join(self._tempdir, filename)
 104
 105     def test_invalid(self):
 106         foo = self.path('foo')
 107         self.assertRaises(IOError, GitFile, foo, mode='r')
 108         self.assertRaises(IOError, GitFile, foo, mode='ab')
 109         self.assertRaises(IOError, GitFile, foo, mode='r+b')
 110         self.assertRaises(IOError, GitFile, foo, mode='w+b')
 111         self.assertRaises(IOError, GitFile, foo, mode='a+bU')
 112
 113     def test_readonly(self):
 114         f = GitFile(self.path('foo'), 'rb')
 115         self.assertTrue(isinstance(f, file))
 116         self.assertEquals('foo contents', f.read())
 117         self.assertEquals('', f.read())
 118         f.seek(4)
 119         self.assertEquals('contents', f.read())
 120         f.close()
 121
 122     def test_default_mode(self):
 123         f = GitFile(self.path('foo'))
 124         self.assertEquals('foo contents', f.read())
 125         f.close()
 126
 127     def test_write(self):
 128         foo = self.path('foo')
 129         foo_lock = '%s.lock' % foo
 130
 131         orig_f = open(foo, 'rb')
 132         self.assertEquals(orig_f.read(), 'foo contents')
 133         orig_f.close()
 134
 135         self.assertFalse(os.path.exists(foo_lock))
 136         f = GitFile(foo, 'wb')
 137         self.assertFalse(f.closed)
 138         self.assertRaises(AttributeError, getattr, f, 'not_a_file_property')
 139
 140         self.assertTrue(os.path.exists(foo_lock))
 141         f.write('new stuff')
 142         f.seek(4)
 143         f.write('contents')
 144         f.close()
 145         self.assertFalse(os.path.exists(foo_lock))
 146
 147         new_f = open(foo, 'rb')
 148         self.assertEquals('new contents', new_f.read())
 149         new_f.close()
 150
 151     def test_open_twice(self):
 152         foo = self.path('foo')
 153         f1 = GitFile(foo, 'wb')
 154         f1.write('new')
 155         try:
 156             f2 = GitFile(foo, 'wb')
 157             self.fail()
 158         except OSError, e:
 159             self.assertEquals(errno.EEXIST, e.errno)
 160         f1.write(' contents')
 161         f1.close()
 162
 163         # Ensure trying to open twice doesn't affect original.
 164         f = open(foo, 'rb')
 165         self.assertEquals('new contents', f.read())
 166         f.close()
 167
 168     def test_abort(self):
 169         foo = self.path('foo')
 170         foo_lock = '%s.lock' % foo
 171
 172         orig_f = open(foo, 'rb')
 173         self.assertEquals(orig_f.read(), 'foo contents')
 174         orig_f.close()
 175
 176         f = GitFile(foo, 'wb')
 177         f.write('new contents')
 178         f.abort()
 179         self.assertTrue(f.closed)
 180         self.assertFalse(os.path.exists(foo_lock))
 181
 182         new_orig_f = open(foo, 'rb')
 183         self.assertEquals(new_orig_f.read(), 'foo contents')
 184         new_orig_f.close()
 185
 186     def test_abort_close(self):
 187         foo = self.path('foo')
 188         f = GitFile(foo, 'wb')
 189         f.abort()
 190         try:
 191             f.close()
 192         except (IOError, OSError):
 193             self.fail()
 194
 195         f = GitFile(foo, 'wb')
 196         f.close()
 197         try:
 198             f.abort()
 199         except (IOError, OSError):
 200             self.fail()
 201
 202     def test_abort_close_removed(self):
 203         foo = self.path('foo')
 204         f = GitFile(foo, 'wb')
 205
 206         f._file.close()
 207         os.remove(foo+".lock")
 208
 209         f.abort()
 210         self.assertTrue(f._closed)