1
# Copyright (C) 2005 by Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
from bzrlib.tests import TestCaseInTempDir
18
from bzrlib.transport import get_transport
19
from bzrlib.transactions import PassThroughTransaction, ReadOnlyTransaction
20
from bzrlib.lockable_files import LockableFiles
21
from bzrlib.errors import NoSuchFile, ReadOnlyError
22
from StringIO import StringIO
24
class TestLockableFiles(TestCaseInTempDir):
26
super(self.__class__, self).setUp()
27
transport = get_transport('.')
28
transport.mkdir('.bzr')
29
transport.put('.bzr/my-lock', StringIO(''))
30
self.lockable = LockableFiles(transport.clone('.bzr'), 'my-lock')
32
def test_read_write(self):
33
self.assertRaises(NoSuchFile, self.lockable.controlfile, 'foo')
34
self.lockable.lock_write()
36
unicode_string = u'bar\u1234'
37
self.assertEqual(4, len(unicode_string))
38
byte_string = unicode_string.encode('utf-8')
39
self.assertEqual(6, len(byte_string))
40
self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo',
41
StringIO(unicode_string))
42
self.lockable.put('foo', StringIO(byte_string))
43
self.assertEqual(byte_string,
44
self.lockable.controlfile('foo', 'rb').read())
45
self.assertEqual(unicode_string,
46
self.lockable.controlfile('foo', 'r').read())
47
self.lockable.put_utf8('bar', StringIO(unicode_string))
48
self.assertEqual(unicode_string,
49
self.lockable.controlfile('bar', 'r').read())
50
self.assertEqual(byte_string,
51
self.lockable.controlfile('bar', 'rb').read())
53
self.lockable.unlock()
56
self.lockable.lock_read()
57
self.lockable.unlock()
58
self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
59
StringIO('bar\u1234'))
61
def test_transactions(self):
62
self.assertIs(self.lockable.get_transaction().__class__,
63
PassThroughTransaction)
64
self.lockable.lock_read()
66
self.assertIs(self.lockable.get_transaction().__class__,
69
self.lockable.unlock()
70
self.assertIs(self.lockable.get_transaction().__class__,
71
PassThroughTransaction)
72
self.lockable.lock_write()
73
self.assertIs(self.lockable.get_transaction().__class__,
74
PassThroughTransaction)
75
self.lockable.unlock()