← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_versionedfile.py

Add bzrlib.pyutils, which has get_named_object, a wrapper around __import__.

This is used to replace various ad hoc implementations of the same logic,
notably the version used in registry's _LazyObjectGetter which had a bug when
getting a module without also getting a member.  And of course, this new
function has unit tests, unlike the replaced code.

This also adds a KnownHooksRegistry subclass to provide a more natural home for
some other logic.

I'm not thrilled about the name of the new module or the new functions, but it's
hard to think of good names for such generic functionality.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_versionedfile.py
31
31
    knit as _mod_knit,
32
32
    osutils,
33
33
    progress,
 
34
    transport,
34
35
    ui,
35
36
    )
36
37
from bzrlib.errors import (
56
57
    )
57
58
from bzrlib.tests.http_utils import TestCaseWithWebserver
58
59
from bzrlib.trace import mutter
59
 
from bzrlib.transport import get_transport
60
60
from bzrlib.transport.memory import MemoryTransport
61
61
from bzrlib.tsort import topo_sort
62
62
from bzrlib.tuned_gzip import GzipFile
849
849
        self.assertEquals(('references_ghost', 'line_c\n'), origins[2])
850
850
 
851
851
    def test_readonly_mode(self):
852
 
        transport = get_transport(self.get_url('.'))
 
852
        t = transport.get_transport(self.get_url('.'))
853
853
        factory = self.get_factory()
854
 
        vf = factory('id', transport, 0777, create=True, access_mode='w')
855
 
        vf = factory('id', transport, access_mode='r')
 
854
        vf = factory('id', t, 0777, create=True, access_mode='w')
 
855
        vf = factory('id', t, access_mode='r')
856
856
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'base', [], [])
857
857
        self.assertRaises(errors.ReadOnlyError,
858
858
                          vf.add_lines_with_ghosts,
880
880
class TestWeave(TestCaseWithMemoryTransport, VersionedFileTestMixIn):
881
881
 
882
882
    def get_file(self, name='foo'):
883
 
        return WeaveFile(name, get_transport(self.get_url('.')), create=True,
884
 
            get_scope=self.get_transaction)
 
883
        return WeaveFile(name, transport.get_transport(self.get_url('.')),
 
884
                         create=True,
 
885
                         get_scope=self.get_transaction)
885
886
 
886
887
    def get_file_corrupted_text(self):
887
 
        w = WeaveFile('foo', get_transport(self.get_url('.')), create=True,
888
 
            get_scope=self.get_transaction)
 
888
        w = WeaveFile('foo', transport.get_transport(self.get_url('.')),
 
889
                      create=True,
 
890
                      get_scope=self.get_transaction)
889
891
        w.add_lines('v1', [], ['hello\n'])
890
892
        w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
891
893
 
919
921
        return w
920
922
 
921
923
    def reopen_file(self, name='foo', create=False):
922
 
        return WeaveFile(name, get_transport(self.get_url('.')), create=create,
923
 
            get_scope=self.get_transaction)
 
924
        return WeaveFile(name, transport.get_transport(self.get_url('.')),
 
925
                         create=create,
 
926
                         get_scope=self.get_transaction)
924
927
 
925
928
    def test_no_implicit_create(self):
926
929
        self.assertRaises(errors.NoSuchFile,
927
930
                          WeaveFile,
928
931
                          'foo',
929
 
                          get_transport(self.get_url('.')),
 
932
                          transport.get_transport(self.get_url('.')),
930
933
                          get_scope=self.get_transaction)
931
934
 
932
935
    def get_factory(self):
999
1002
        # we should be able to read from http with a versioned file.
1000
1003
        vf = self.get_file()
1001
1004
        # try an empty file access
1002
 
        readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
 
1005
        readonly_vf = self.get_factory()('foo', transport.get_transport(
 
1006
                self.get_readonly_url('.')))
1003
1007
        self.assertEqual([], readonly_vf.versions())
 
1008
 
 
1009
    def test_readonly_http_works_with_feeling(self):
 
1010
        # we should be able to read from http with a versioned file.
 
1011
        vf = self.get_file()
1004
1012
        # now with feeling.
1005
1013
        vf.add_lines('1', [], ['a\n'])
1006
1014
        vf.add_lines('2', ['1'], ['b\n', 'a\n'])
1007
 
        readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
 
1015
        readonly_vf = self.get_factory()('foo', transport.get_transport(
 
1016
                self.get_readonly_url('.')))
1008
1017
        self.assertEqual(['1', '2'], vf.versions())
 
1018
        self.assertEqual(['1', '2'], readonly_vf.versions())
1009
1019
        for version in readonly_vf.versions():
1010
1020
            readonly_vf.get_lines(version)
1011
1021
 
1013
1023
class TestWeaveHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
1014
1024
 
1015
1025
    def get_file(self):
1016
 
        return WeaveFile('foo', get_transport(self.get_url('.')), create=True,
1017
 
            get_scope=self.get_transaction)
 
1026
        return WeaveFile('foo', transport.get_transport(self.get_url('.')),
 
1027
                         create=True,
 
1028
                         get_scope=self.get_transaction)
1018
1029
 
1019
1030
    def get_factory(self):
1020
1031
        return WeaveFile
1264
1275
class TestWeaveMerge(TestCaseWithMemoryTransport, MergeCasesMixin):
1265
1276
 
1266
1277
    def get_file(self, name='foo'):
1267
 
        return WeaveFile(name, get_transport(self.get_url('.')), create=True)
 
1278
        return WeaveFile(name, transport.get_transport(self.get_url('.')),
 
1279
                         create=True)
1268
1280
 
1269
1281
    def log_contents(self, w):
1270
1282
        self.log('weave is:')