← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_foreign.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-10-13 06:08:53 UTC
  • mfrom: (4737.1.1 merge-2.0-into-devel)
  • Revision ID: pqm@pqm.ubuntu.com-20091013060853-erk2aaj80fnkrv25
(andrew) Merge lp:bzr/2.0 into lp:bzr, including fixes for #322807,
        #389413, #402623 and documentation improvements.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_foreign.py
1
 
# Copyright (C) 2008-2011 Canonical Ltd
 
1
# Copyright (C) 2008 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
from bzrlib import (
22
22
    branch,
23
23
    bzrdir,
24
 
    controldir,
25
24
    errors,
26
25
    foreign,
27
26
    lockable_files,
28
27
    lockdir,
29
 
    repository,
30
28
    revision,
31
29
    tests,
32
30
    trace,
33
 
    vf_repository,
34
31
    )
35
32
 
36
 
from bzrlib.repofmt import groupcompress_repo
37
 
 
38
33
# This is the dummy foreign revision control system, used 
39
34
# mainly here in the testsuite to test the foreign VCS infrastructure.
40
35
# It is basically standard Bazaar with some minor modifications to 
79
74
        self.mapping_registry = DummyForeignVcsMappingRegistry()
80
75
        self.mapping_registry.register("v1", DummyForeignVcsMapping(self),
81
76
                                       "Version 1")
82
 
        self.abbreviation = "dummy"
83
77
 
84
78
    def show_foreign_revid(self, foreign_revid):
85
79
        return { "dummy ding": "%s/%s\\%s" % foreign_revid }
86
80
 
87
 
    def serialize_foreign_revid(self, foreign_revid):
88
 
        return "%s|%s|%s" % foreign_revid
89
 
 
90
81
 
91
82
class DummyForeignVcsBranch(branch.BzrBranch6,foreign.ForeignBranch):
92
83
    """A Dummy VCS Branch."""
93
84
 
94
 
    @property
95
 
    def user_transport(self):
96
 
        return self.bzrdir.user_transport
97
 
 
98
85
    def __init__(self, _format, _control_files, a_bzrdir, *args, **kwargs):
99
86
        self._format = _format
100
87
        self._base = a_bzrdir.transport.base
101
88
        self._ignore_fallbacks = False
102
 
        self.bzrdir = a_bzrdir
103
 
        foreign.ForeignBranch.__init__(self,
 
89
        foreign.ForeignBranch.__init__(self, 
104
90
            DummyForeignVcsMapping(DummyForeignVcs()))
105
 
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir,
 
91
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir, 
106
92
            *args, **kwargs)
107
93
 
108
 
    def _get_checkout_format(self, lightweight=False):
109
 
        """Return the most suitable metadir for a checkout of this branch.
110
 
        Weaves are used if this branch's repository uses weaves.
111
 
        """
112
 
        return self.bzrdir.checkout_metadir()
113
 
 
114
 
    def import_last_revision_info_and_tags(self, source, revno, revid,
115
 
                                           lossy=False):
116
 
        interbranch = InterToDummyVcsBranch(source, self)
117
 
        result = interbranch.push(stop_revision=revid, lossy=True)
118
 
        if lossy:
119
 
            revid = result.revidmap[revid]
120
 
        return (revno, revid)
121
 
 
122
 
 
123
 
class DummyForeignCommitBuilder(vf_repository.VersionedFileRootCommitBuilder):
124
 
 
125
 
    def _generate_revision_if_needed(self):
126
 
        mapping = DummyForeignVcsMapping(DummyForeignVcs())
127
 
        if self._lossy:
128
 
            self._new_revision_id = mapping.revision_id_foreign_to_bzr(
129
 
                (str(self._timestamp), str(self._timezone), "UNKNOWN"))
130
 
            self.random_revid = False
131
 
        elif self._new_revision_id is not None:
132
 
            self.random_revid = False
133
 
        else:
134
 
            self._new_revision_id = self._gen_revision_id()
135
 
            self.random_revid = True
136
 
 
137
 
 
138
 
class DummyForeignVcsRepository(groupcompress_repo.CHKInventoryRepository,
139
 
    foreign.ForeignRepository):
140
 
    """Dummy foreign vcs repository."""
141
 
 
142
 
 
143
 
class DummyForeignVcsRepositoryFormat(groupcompress_repo.RepositoryFormat2a):
144
 
 
145
 
    repository_class = DummyForeignVcsRepository
146
 
    _commit_builder_class = DummyForeignCommitBuilder
147
 
 
148
 
    @classmethod
149
 
    def get_format_string(cls):
150
 
        return "Dummy Foreign Vcs Repository"
151
 
 
152
 
    def get_format_description(self):
153
 
        return "Dummy Foreign Vcs Repository"
154
 
 
155
 
 
156
 
def branch_history(graph, revid):
157
 
    ret = list(graph.iter_lefthand_ancestry(revid,
158
 
        (revision.NULL_REVISION,)))
159
 
    ret.reverse()
160
 
    return ret
161
 
 
162
 
 
163
 
class InterToDummyVcsBranch(branch.GenericInterBranch):
 
94
 
 
95
class InterToDummyVcsBranch(branch.GenericInterBranch,
 
96
                            foreign.InterToForeignBranch):
164
97
 
165
98
    @staticmethod
166
99
    def is_compatible(source, target):
167
100
        return isinstance(target, DummyForeignVcsBranch)
168
101
 
169
 
    def push(self, overwrite=False, stop_revision=None, lossy=False):
170
 
        if not lossy:
171
 
            raise errors.NoRoundtrippingSupport(self.source, self.target)
 
102
    def lossy_push(self, stop_revision=None):
172
103
        result = branch.BranchPushResult()
173
104
        result.source_branch = self.source
174
105
        result.target_branch = self.target
175
106
        result.old_revno, result.old_revid = self.target.last_revision_info()
176
107
        self.source.lock_read()
177
108
        try:
178
 
            graph = self.source.repository.get_graph()
179
109
            # This just handles simple cases, but that's good enough for tests
180
 
            my_history = branch_history(self.target.repository.get_graph(),
181
 
                result.old_revid)
182
 
            if stop_revision is None:
183
 
                stop_revision = self.source.last_revision()
184
 
            their_history = branch_history(graph, stop_revision)
 
110
            my_history = self.target.revision_history()
 
111
            their_history = self.source.revision_history()
185
112
            if their_history[:min(len(my_history), len(their_history))] != my_history:
186
113
                raise errors.DivergedBranches(self.target, self.source)
187
114
            todo = their_history[len(my_history):]
193
120
                    return (tree.get_file(file_id), None)
194
121
                tree.get_file_with_stat = get_file_with_stat
195
122
                new_revid = self.target.mapping.revision_id_foreign_to_bzr(
196
 
                    (str(rev.timestamp), str(rev.timezone),
 
123
                    (str(rev.timestamp), str(rev.timezone), 
197
124
                        str(self.target.revno())))
198
125
                parent_revno, parent_revid= self.target.last_revision_info()
199
126
                if parent_revid == revision.NULL_REVISION:
201
128
                else:
202
129
                    parent_revids = [parent_revid]
203
130
                builder = self.target.get_commit_builder(parent_revids, 
204
 
                        self.target.get_config_stack(), rev.timestamp,
 
131
                        self.target.get_config(), rev.timestamp,
205
132
                        rev.timezone, rev.committer, rev.properties,
206
133
                        new_revid)
207
134
                try:
208
 
                    parent_tree = self.target.repository.revision_tree(
209
 
                        parent_revid)
210
 
                    for path, ie in tree.iter_entries_by_dir():
 
135
                    for path, ie in tree.inventory.iter_entries():
211
136
                        new_ie = ie.copy()
212
137
                        new_ie.revision = None
213
138
                        builder.record_entry_contents(new_ie, 
214
 
                            [parent_tree.root_inventory],
 
139
                            [self.target.repository.revision_tree(parent_revid).inventory],
215
140
                            path, tree, 
216
141
                            (ie.kind, ie.text_size, ie.executable, ie.text_sha1))
217
142
                    builder.finish_inventory()
232
157
 
233
158
class DummyForeignVcsBranchFormat(branch.BzrBranchFormat6):
234
159
 
235
 
    @classmethod
236
 
    def get_format_string(cls):
 
160
    def get_format_string(self):
237
161
        return "Branch for Testing"
238
162
 
239
 
    @property
240
 
    def _matchingbzrdir(self):
241
 
        return DummyForeignVcsDirFormat()
 
163
    def __init__(self):
 
164
        super(DummyForeignVcsBranchFormat, self).__init__()
 
165
        self._matchingbzrdir = DummyForeignVcsDirFormat()
242
166
 
243
 
    def open(self, a_bzrdir, name=None, _found=False, ignore_fallbacks=False,
244
 
            found_repository=None):
245
 
        if name is None:
246
 
            name = a_bzrdir._get_selected_branch()
 
167
    def open(self, a_bzrdir, _found=False):
247
168
        if not _found:
248
169
            raise NotImplementedError
249
170
        try:
250
 
            transport = a_bzrdir.get_branch_transport(None, name=name)
 
171
            transport = a_bzrdir.get_branch_transport(None)
251
172
            control_files = lockable_files.LockableFiles(transport, 'lock',
252
173
                                                         lockdir.LockDir)
253
 
            if found_repository is None:
254
 
                found_repository = a_bzrdir.find_repository()
255
174
            return DummyForeignVcsBranch(_format=self,
256
175
                              _control_files=control_files,
257
176
                              a_bzrdir=a_bzrdir,
258
 
                              _repository=found_repository,
259
 
                              name=name)
 
177
                              _repository=a_bzrdir.find_repository())
260
178
        except errors.NoSuchFile:
261
179
            raise errors.NotBranchError(path=transport.base)
262
180
 
279
197
    def get_branch_format(self):
280
198
        return DummyForeignVcsBranchFormat()
281
199
 
282
 
    @property
283
 
    def repository_format(self):
284
 
        return DummyForeignVcsRepositoryFormat()
 
200
    @classmethod
 
201
    def probe_transport(klass, transport):
 
202
        """Return the .bzrdir style format present in a directory."""
 
203
        if not transport.has('.dummy'):
 
204
            raise errors.NotBranchError(path=transport.base)
 
205
        return klass()
285
206
 
286
207
    def initialize_on_transport(self, transport):
287
208
        """Initialize a new bzrdir in the base directory of a Transport."""
315
236
        self._control_files = lockable_files.LockableFiles(self.transport,
316
237
            "lock", lockable_files.TransportLock)
317
238
 
318
 
    def create_workingtree(self):
319
 
        # dirstate requires a ".bzr" entry to exist
320
 
        self.root_transport.put_bytes(".bzr", "foo")
321
 
        return super(DummyForeignVcsDir, self).create_workingtree()
322
 
 
323
 
    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=True,
324
 
            possible_transports=None):
325
 
        if name is None:
326
 
            name = self._get_selected_branch()
327
 
        if name != "":
328
 
            raise errors.NoColocatedBranchSupport(self)
 
239
    def open_branch(self, ignore_fallbacks=True):
329
240
        return self._format.get_branch_format().open(self, _found=True)
330
241
 
331
242
    def cloning_metadir(self, stacked=False):
332
243
        """Produce a metadir suitable for cloning with."""
333
 
        return controldir.format_registry.make_bzrdir("default")
334
 
 
335
 
    def checkout_metadir(self):
336
 
        return self.cloning_metadir()
 
244
        return bzrdir.format_registry.make_bzrdir("default")
337
245
 
338
246
    def sprout(self, url, revision_id=None, force_new_repo=False,
339
247
               recurse='down', possible_transports=None,
348
256
 
349
257
 
350
258
def register_dummy_foreign_for_test(testcase):
351
 
    controldir.ControlDirFormat.register_prober(DummyForeignProber)
352
 
    testcase.addCleanup(controldir.ControlDirFormat.unregister_prober,
353
 
        DummyForeignProber)
354
 
    repository.format_registry.register(DummyForeignVcsRepositoryFormat())
355
 
    testcase.addCleanup(repository.format_registry.remove,
356
 
            DummyForeignVcsRepositoryFormat())
357
 
    branch.format_registry.register(DummyForeignVcsBranchFormat())
358
 
    testcase.addCleanup(branch.format_registry.remove,
359
 
            DummyForeignVcsBranchFormat())
 
259
    bzrdir.BzrDirFormat.register_control_format(DummyForeignVcsDirFormat)
 
260
    testcase.addCleanup(bzrdir.BzrDirFormat.unregister_control_format,
 
261
                        DummyForeignVcsDirFormat)
360
262
    # We need to register the optimiser to make the dummy appears really
361
263
    # different from a regular bzr repository.
362
264
    branch.InterBranch.register_optimiser(InterToDummyVcsBranch)
364
266
                        InterToDummyVcsBranch)
365
267
 
366
268
 
367
 
class DummyForeignProber(controldir.Prober):
368
 
 
369
 
    @classmethod
370
 
    def probe_transport(klass, transport):
371
 
        """Return the .bzrdir style format present in a directory."""
372
 
        if not transport.has('.dummy'):
373
 
            raise errors.NotBranchError(path=transport.base)
374
 
        return DummyForeignVcsDirFormat()
375
 
 
376
 
    @classmethod
377
 
    def known_formats(cls):
378
 
        return set([DummyForeignVcsDirFormat()])
379
 
 
380
 
 
381
269
class ForeignVcsRegistryTests(tests.TestCase):
382
270
    """Tests for the ForeignVcsRegistry class."""
383
271
 
395
283
        reg = foreign.ForeignVcsRegistry()
396
284
        vcs = DummyForeignVcs()
397
285
        reg.register("dummy", vcs, "Dummy VCS")
398
 
        self.assertEquals((
399
 
            ("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
400
 
            reg.parse_revision_id("dummy-v1:some-foreign-revid"))
 
286
        self.assertEquals((("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
 
287
                          reg.parse_revision_id("dummy-v1:some-foreign-revid"))
401
288
 
402
289
 
403
290
class ForeignRevisionTests(tests.TestCase):
431
318
        foreign.update_workingtree_fileids(wt, target_basis)
432
319
        wt.lock_read()
433
320
        try:
434
 
            self.assertEquals(set([root_id, "bla-b"]), set(wt.all_file_ids()))
 
321
            self.assertEquals(set([root_id, "bla-b"]), set(wt.inventory))
435
322
        finally:
436
323
            wt.unlock()
437
324
 
446
333
    def test_create(self):
447
334
        """Test we can create dummies."""
448
335
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
449
 
        dir = controldir.ControlDir.open("d")
 
336
        dir = bzrdir.BzrDir.open("d")
450
337
        self.assertEquals("A Dummy VCS Dir", dir._format.get_format_string())
451
338
        dir.open_repository()
452
339
        dir.open_branch()
455
342
    def test_sprout(self):
456
343
        """Test we can clone dummies and that the format is not preserved."""
457
344
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
458
 
        dir = controldir.ControlDir.open("d")
 
345
        dir = bzrdir.BzrDir.open("d")
459
346
        newdir = dir.sprout("e")
460
347
        self.assertNotEquals("A Dummy VCS Dir",
461
348
                             newdir._format.get_format_string())
462
349
 
463
 
    def test_push_not_supported(self):
464
 
        source_tree = self.make_branch_and_tree("source")
465
 
        target_tree = self.make_branch_and_tree("target", 
466
 
            format=DummyForeignVcsDirFormat())
467
 
        self.assertRaises(errors.NoRoundtrippingSupport, 
468
 
            source_tree.branch.push, target_tree.branch)
469
 
 
470
350
    def test_lossy_push_empty(self):
471
351
        source_tree = self.make_branch_and_tree("source")
472
352
        target_tree = self.make_branch_and_tree("target", 
473
353
            format=DummyForeignVcsDirFormat())
474
 
        pushresult = source_tree.branch.push(target_tree.branch, lossy=True)
 
354
        pushresult = source_tree.branch.lossy_push(target_tree.branch)
475
355
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)
476
356
        self.assertEquals(revision.NULL_REVISION, pushresult.new_revid)
477
357
        self.assertEquals({}, pushresult.revidmap)
485
365
            format=DummyForeignVcsDirFormat())
486
366
        target_tree.branch.lock_write()
487
367
        try:
488
 
            pushresult = source_tree.branch.push(
489
 
                target_tree.branch, lossy=True)
 
368
            pushresult = source_tree.branch.lossy_push(target_tree.branch)
490
369
        finally:
491
370
            target_tree.branch.unlock()
492
371
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)