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 import unittest
  25
  26 from dulwich.file import GitFile, fancy_rename
  27 from dulwich.tests import TestSkipped
  28
  29
  30 class FancyRenameTests(unittest.TestCase):
  31
  32     def setUp(self):
  33         self._tempdir = tempfile.mkdtemp()
  34         self.foo = self.path('foo')
  35         self.bar = self.path('bar')
  36         self.create(self.foo, 'foo contents')
  37
  38     def tearDown(self):
  39         shutil.rmtree(self._tempdir)
  40
  41     def path(self, filename):
  42         return os.path.join(self._tempdir, filename)
  43
  44     def create(self, path, contents):
  45         f = open(path, 'wb')
  46         f.write(contents)
  47         f.close()
  48
  49     def test_no_dest_exists(self):
  50         self.assertFalse(os.path.exists(self.bar))
  51         fancy_rename(self.foo, self.bar)
  52         self.assertFalse(os.path.exists(self.foo))
  53
  54         new_f = open(self.bar, 'rb')
  55         self.assertEquals('foo contents', new_f.read())
  56         new_f.close()
  57
  58     def test_dest_exists(self):
  59         self.create(self.bar, 'bar contents')
  60         fancy_rename(self.foo, self.bar)
  61         self.assertFalse(os.path.exists(self.foo))
  62
  63         new_f = open(self.bar, 'rb')
  64         self.assertEquals('foo contents', new_f.read())
  65         new_f.close()
  66
  67     def test_dest_opened(self):
  68         if sys.platform != "win32":
  69             raise TestSkipped("platform allows overwriting open files")
  70         self.create(self.bar, 'bar contents')
  71         dest_f = open(self.bar, 'rb')
  72         self.assertRaises(OSError, fancy_rename, self.foo, self.bar)
  73         dest_f.close()
  74         self.assertTrue(os.path.exists(self.path('foo')))
  75
  76         new_f = open(self.foo, 'rb')
  77         self.assertEquals('foo contents', new_f.read())
  78         new_f.close()
  79
  80         new_f = open(self.bar, 'rb')
  81         self.assertEquals('bar contents', new_f.read())
  82         new_f.close()
  83
  84
  85 class GitFileTests(unittest.TestCase):
  86
  87     def setUp(self):
  88         self._tempdir = tempfile.mkdtemp()
  89         f = open(self.path('foo'), 'wb')
  90         f.write('foo contents')
  91         f.close()
  92
  93     def tearDown(self):
  94         shutil.rmtree(self._tempdir)
  95
  96     def path(self, filename):
  97         return os.path.join(self._tempdir, filename)
  98
  99     def test_invalid(self):
 100         foo = self.path('foo')
 101         self.assertRaises(IOError, GitFile, foo, mode='r')
 102         self.assertRaises(IOError, GitFile, foo, mode='ab')
 103         self.assertRaises(IOError, GitFile, foo, mode='r+b')
 104         self.assertRaises(IOError, GitFile, foo, mode='w+b')
 105         self.assertRaises(IOError, GitFile, foo, mode='a+bU')
 106
 107     def test_readonly(self):
 108         f = GitFile(self.path('foo'), 'rb')
 109         self.assertTrue(isinstance(f, file))
 110         self.assertEquals('foo contents', f.read())
 111         self.assertEquals('', f.read())
 112         f.seek(4)
 113         self.assertEquals('contents', f.read())
 114         f.close()
 115
 116     def test_write(self):
 117         foo = self.path('foo')
 118         foo_lock = '%s.lock' % foo
 119
 120         orig_f = open(foo, 'rb')
 121         self.assertEquals(orig_f.read(), 'foo contents')
 122         orig_f.close()
 123
 124         self.assertFalse(os.path.exists(foo_lock))
 125         f = GitFile(foo, 'wb')
 126         self.assertFalse(f.closed)
 127         self.assertRaises(AttributeError, getattr, f, 'not_a_file_property')
 128
 129         self.assertTrue(os.path.exists(foo_lock))
 130         f.write('new stuff')
 131         f.seek(4)
 132         f.write('contents')
 133         f.close()
 134         self.assertFalse(os.path.exists(foo_lock))
 135
 136         new_f = open(foo, 'rb')
 137         self.assertEquals('new contents', new_f.read())
 138         new_f.close()
 139
 140     def test_open_twice(self):
 141         foo = self.path('foo')
 142         f1 = GitFile(foo, 'wb')
 143         f1.write('new')
 144         try:
 145             f2 = GitFile(foo, 'wb')
 146             self.fail()
 147         except OSError, e:
 148             self.assertEquals(errno.EEXIST, e.errno)
 149         f1.write(' contents')
 150         f1.close()
 151
 152         # Ensure trying to open twice doesn't affect original.
 153         f = open(foo, 'rb')
 154         self.assertEquals('new contents', f.read())
 155         f.close()
 156
 157     def test_abort(self):
 158         foo = self.path('foo')
 159         foo_lock = '%s.lock' % foo
 160
 161         orig_f = open(foo, 'rb')
 162         self.assertEquals(orig_f.read(), 'foo contents')
 163         orig_f.close()
 164
 165         f = GitFile(foo, 'wb')
 166         f.write('new contents')
 167         f.abort()
 168         self.assertTrue(f.closed)
 169         self.assertFalse(os.path.exists(foo_lock))
 170
 171         new_orig_f = open(foo, 'rb')
 172         self.assertEquals(new_orig_f.read(), 'foo contents')
 173         new_orig_f.close()
 174
 175     def test_abort_close(self):
 176         foo = self.path('foo')
 177         f = GitFile(foo, 'wb')
 178         f.abort()
 179         try:
 180             f.close()
 181         except (IOError, OSError):
 182             self.fail()
 183
 184         f = GitFile(foo, 'wb')
 185         f.close()
 186         try:
 187             f.abort()
 188         except (IOError, OSError):
 189             self.fail()
 190
 191     def test_abort_close_removed(self):
 192         foo = self.path('foo')
 193         f = GitFile(foo, 'wb')
 194         os.remove(foo+".lock")
 195         f.abort()
 196         self.assertTrue(f._closed)