← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockable_files.py

  • Committer: Vincent Ladeuil
  • Date: 2010-07-07 15:03:14 UTC
  • mto: (5355.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5356.
  • Revision ID: v.ladeuil+lp@free.fr-20100707150314-7i5po3dwg8umiv8x
Fix remaining sphinx_conf references.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
1
 
# Copyright (C) 2005, 2006, 2008 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
21
21
    errors,
22
22
    lockdir,
23
23
    osutils,
 
24
    transport,
24
25
    )
25
26
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
26
27
from bzrlib.lockable_files import LockableFiles, TransportLock
37
38
                                 ReadOnlyTransaction,
38
39
                                 WriteTransaction,
39
40
                                 )
40
 
from bzrlib.transport import get_transport
41
41
 
42
42
 
43
43
# these tests are applied in each parameterized suite for LockableFiles
46
46
# so won't modernize them now. - mbp 20080430
47
47
class _TestLockableFiles_mixin(object):
48
48
 
49
 
    def test_read_write(self):
50
 
        self.assertRaises(NoSuchFile,
51
 
            self.applyDeprecated,
52
 
            deprecated_in((1, 5, 0)),
53
 
            self.lockable.get, 'foo')
54
 
        self.assertRaises(NoSuchFile,
55
 
            self.applyDeprecated,
56
 
            deprecated_in((1, 5, 0)),
57
 
            self.lockable.get_utf8, 'foo')
58
 
        self.lockable.lock_write()
59
 
        self.addCleanup(self.lockable.unlock)
60
 
        unicode_string = u'bar\u1234'
61
 
        self.assertEqual(4, len(unicode_string))
62
 
        byte_string = unicode_string.encode('utf-8')
63
 
        self.assertEqual(6, len(byte_string))
64
 
        self.assertRaises(UnicodeEncodeError,
65
 
            self.applyDeprecated,
66
 
            deprecated_in((1, 6, 0)),
67
 
            self.lockable.put, 'foo',
68
 
            StringIO(unicode_string))
69
 
        self.applyDeprecated(
70
 
            deprecated_in((1, 6, 0)),
71
 
            self.lockable.put,
72
 
            'foo', StringIO(byte_string))
73
 
        byte_stream = self.applyDeprecated(
74
 
            deprecated_in((1, 5, 0)),
75
 
            self.lockable.get,
76
 
            'foo')
77
 
        self.assertEqual(byte_string, byte_stream.read())
78
 
        unicode_stream = self.applyDeprecated(
79
 
            deprecated_in((1, 5, 0)),
80
 
            self.lockable.get_utf8,
81
 
            'foo')
82
 
        self.assertEqual(unicode_string,
83
 
            unicode_stream.read())
84
 
        self.assertRaises(BzrBadParameterNotString,
85
 
            self.applyDeprecated,
86
 
            deprecated_in((1, 6, 0)),
87
 
            self.lockable.put_utf8,
88
 
            'bar',
89
 
            StringIO(unicode_string))
90
 
        self.applyDeprecated(
91
 
            deprecated_in((1, 6, 0)),
92
 
            self.lockable.put_utf8,
93
 
            'bar',
94
 
            unicode_string)
95
 
        unicode_stream = self.applyDeprecated(
96
 
            deprecated_in((1, 5, 0)),
97
 
            self.lockable.get_utf8,
98
 
            'bar')
99
 
        self.assertEqual(unicode_string,
100
 
            unicode_stream.read())
101
 
        byte_stream = self.applyDeprecated(
102
 
            deprecated_in((1, 5, 0)),
103
 
            self.lockable.get,
104
 
            'bar')
105
 
        self.assertEqual(byte_string, byte_stream.read())
106
 
        self.applyDeprecated(
107
 
            deprecated_in((1, 6, 0)),
108
 
            self.lockable.put_bytes,
109
 
            'raw', 'raw\xffbytes')
110
 
        byte_stream = self.applyDeprecated(
111
 
            deprecated_in((1, 5, 0)),
112
 
            self.lockable.get,
113
 
            'raw')
114
 
        self.assertEqual('raw\xffbytes', byte_stream.read())
115
 
 
116
 
    def test_locks(self):
117
 
        self.lockable.lock_read()
118
 
        self.addCleanup(self.lockable.unlock)
119
 
        self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
120
 
                          StringIO('bar\u1234'))
121
 
 
122
49
    def test_transactions(self):
123
50
        self.assertIs(self.lockable.get_transaction().__class__,
124
51
                      PassThroughTransaction)
352
279
 
353
280
    def setUp(self):
354
281
        TestCaseInTempDir.setUp(self)
355
 
        transport = get_transport('.')
356
 
        transport.mkdir('.bzr')
357
 
        self.sub_transport = transport.clone('.bzr')
 
282
        t = transport.get_transport('.')
 
283
        t.mkdir('.bzr')
 
284
        self.sub_transport = t.clone('.bzr')
358
285
        self.lockable = self.get_lockable()
359
286
        self.lockable.create_lock()
360
287
 
361
 
    def tearDown(self):
362
 
        super(TestLockableFiles_TransportLock, self).tearDown()
 
288
    def stop_server(self):
 
289
        super(TestLockableFiles_TransportLock, self).stop_server()
363
290
        # free the subtransport so that we do not get a 5 second
364
291
        # timeout due to the SFTP connection cache.
365
292
        try:
377
304
 
378
305
    def setUp(self):
379
306
        TestCaseInTempDir.setUp(self)
380
 
        self.transport = get_transport('.')
 
307
        self.transport = transport.get_transport('.')
381
308
        self.lockable = self.get_lockable()
382
309
        # the lock creation here sets mode - test_permissions on branch
383
310
        # tests that implicitly, but it might be a good idea to factor
420
347
        # in test_remote and test_smart as usual.
421
348
        b = self.make_branch('foo')
422
349
        self.addCleanup(b.bzrdir.transport.disconnect)
423
 
        self.transport = get_transport('.')
 
350
        self.transport = transport.get_transport('.')
424
351
        self.lockable = self.get_lockable()
425
352
 
426
353
    def get_lockable(self):