← 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: Vincent Ladeuil
  • Date: 2010-03-10 09:33:04 UTC
  • mto: (5082.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5083.
  • Revision ID: v.ladeuil+lp@free.fr-20100310093304-4245t4tazd4sxoav
Cleanup test from overly cautious checks.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_foreign.py
1
 
# Copyright (C) 2008-2012, 2016 Canonical Ltd
 
1
# Copyright (C) 2008, 2009, 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
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 
91
86
class DummyForeignVcsBranch(branch.BzrBranch6,foreign.ForeignBranch):
92
87
    """A Dummy VCS Branch."""
93
88
 
94
 
    @property
95
 
    def user_transport(self):
96
 
        return self.bzrdir.user_transport
97
 
 
98
89
    def __init__(self, _format, _control_files, a_bzrdir, *args, **kwargs):
99
90
        self._format = _format
100
91
        self._base = a_bzrdir.transport.base
101
92
        self._ignore_fallbacks = False
102
 
        self.bzrdir = a_bzrdir
103
 
        foreign.ForeignBranch.__init__(self,
 
93
        foreign.ForeignBranch.__init__(self, 
104
94
            DummyForeignVcsMapping(DummyForeignVcs()))
105
 
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir,
 
95
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir, 
106
96
            *args, **kwargs)
107
97
 
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):
 
98
 
 
99
class InterToDummyVcsBranch(branch.GenericInterBranch,
 
100
                            foreign.InterToForeignBranch):
164
101
 
165
102
    @staticmethod
166
103
    def is_compatible(source, target):
167
104
        return isinstance(target, DummyForeignVcsBranch)
168
105
 
169
 
    def push(self, overwrite=False, stop_revision=None, lossy=False):
170
 
        if not lossy:
171
 
            raise errors.NoRoundtrippingSupport(self.source, self.target)
 
106
    def push(self, overwrite=False, stop_revision=None):
 
107
        raise errors.NoRoundtrippingSupport(self.source, self.target)
 
108
 
 
109
    def lossy_push(self, stop_revision=None):
172
110
        result = branch.BranchPushResult()
173
111
        result.source_branch = self.source
174
112
        result.target_branch = self.target
175
113
        result.old_revno, result.old_revid = self.target.last_revision_info()
176
114
        self.source.lock_read()
177
115
        try:
178
 
            graph = self.source.repository.get_graph()
179
116
            # 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)
 
117
            my_history = self.target.revision_history()
 
118
            their_history = self.source.revision_history()
185
119
            if their_history[:min(len(my_history), len(their_history))] != my_history:
186
120
                raise errors.DivergedBranches(self.target, self.source)
187
121
            todo = their_history[len(my_history):]
193
127
                    return (tree.get_file(file_id), None)
194
128
                tree.get_file_with_stat = get_file_with_stat
195
129
                new_revid = self.target.mapping.revision_id_foreign_to_bzr(
196
 
                    (str(rev.timestamp), str(rev.timezone),
 
130
                    (str(rev.timestamp), str(rev.timezone), 
197
131
                        str(self.target.revno())))
198
132
                parent_revno, parent_revid= self.target.last_revision_info()
199
133
                if parent_revid == revision.NULL_REVISION:
201
135
                else:
202
136
                    parent_revids = [parent_revid]
203
137
                builder = self.target.get_commit_builder(parent_revids, 
204
 
                        self.target.get_config_stack(), rev.timestamp,
 
138
                        self.target.get_config(), rev.timestamp,
205
139
                        rev.timezone, rev.committer, rev.properties,
206
140
                        new_revid)
207
141
                try:
208
 
                    parent_tree = self.target.repository.revision_tree(
209
 
                        parent_revid)
210
 
                    for path, ie in tree.iter_entries_by_dir():
 
142
                    for path, ie in tree.inventory.iter_entries():
211
143
                        new_ie = ie.copy()
212
144
                        new_ie.revision = None
213
145
                        builder.record_entry_contents(new_ie, 
214
 
                            [parent_tree.root_inventory],
 
146
                            [self.target.repository.revision_tree(parent_revid).inventory],
215
147
                            path, tree, 
216
148
                            (ie.kind, ie.text_size, ie.executable, ie.text_sha1))
217
149
                    builder.finish_inventory()
232
164
 
233
165
class DummyForeignVcsBranchFormat(branch.BzrBranchFormat6):
234
166
 
235
 
    @classmethod
236
 
    def get_format_string(cls):
 
167
    def get_format_string(self):
237
168
        return "Branch for Testing"
238
169
 
239
 
    @property
240
 
    def _matchingbzrdir(self):
241
 
        return DummyForeignVcsDirFormat()
 
170
    def __init__(self):
 
171
        super(DummyForeignVcsBranchFormat, self).__init__()
 
172
        self._matchingbzrdir = DummyForeignVcsDirFormat()
242
173
 
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()
 
174
    def open(self, a_bzrdir, _found=False):
247
175
        if not _found:
248
176
            raise NotImplementedError
249
177
        try:
250
 
            transport = a_bzrdir.get_branch_transport(None, name=name)
 
178
            transport = a_bzrdir.get_branch_transport(None)
251
179
            control_files = lockable_files.LockableFiles(transport, 'lock',
252
180
                                                         lockdir.LockDir)
253
 
            if found_repository is None:
254
 
                found_repository = a_bzrdir.find_repository()
255
181
            return DummyForeignVcsBranch(_format=self,
256
182
                              _control_files=control_files,
257
183
                              a_bzrdir=a_bzrdir,
258
 
                              _repository=found_repository,
259
 
                              name=name)
 
184
                              _repository=a_bzrdir.find_repository())
260
185
        except errors.NoSuchFile:
261
186
            raise errors.NotBranchError(path=transport.base)
262
187
 
279
204
    def get_branch_format(self):
280
205
        return DummyForeignVcsBranchFormat()
281
206
 
282
 
    @property
283
 
    def repository_format(self):
284
 
        return DummyForeignVcsRepositoryFormat()
 
207
    @classmethod
 
208
    def probe_transport(klass, transport):
 
209
        """Return the .bzrdir style format present in a directory."""
 
210
        if not transport.has('.dummy'):
 
211
            raise errors.NotBranchError(path=transport.base)
 
212
        return klass()
285
213
 
286
214
    def initialize_on_transport(self, transport):
287
215
        """Initialize a new bzrdir in the base directory of a Transport."""
315
243
        self._control_files = lockable_files.LockableFiles(self.transport,
316
244
            "lock", lockable_files.TransportLock)
317
245
 
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 != "":
 
246
    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=True):
 
247
        if name is not None:
328
248
            raise errors.NoColocatedBranchSupport(self)
329
249
        return self._format.get_branch_format().open(self, _found=True)
330
250
 
331
251
    def cloning_metadir(self, stacked=False):
332
252
        """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()
 
253
        return bzrdir.format_registry.make_bzrdir("default")
337
254
 
338
255
    def sprout(self, url, revision_id=None, force_new_repo=False,
339
256
               recurse='down', possible_transports=None,
340
257
               accelerator_tree=None, hardlink=False, stacked=False,
341
258
               source_branch=None):
342
 
        # dirstate doesn't cope with accelerator_trees well
 
259
        # dirstate doesn't cope with accelerator_trees well 
343
260
        # that have a different control dir
344
 
        return super(DummyForeignVcsDir, self).sprout(
345
 
            url=url,
346
 
            revision_id=revision_id, force_new_repo=force_new_repo,
347
 
            recurse=recurse, possible_transports=possible_transports,
348
 
            hardlink=hardlink, stacked=stacked, source_branch=source_branch)
 
261
        return super(DummyForeignVcsDir, self).sprout(url=url, 
 
262
                revision_id=revision_id, force_new_repo=force_new_repo, 
 
263
                recurse=recurse, possible_transports=possible_transports, 
 
264
                hardlink=hardlink, stacked=stacked, source_branch=source_branch)
349
265
 
350
266
 
351
267
def register_dummy_foreign_for_test(testcase):
352
 
    controldir.ControlDirFormat.register_prober(DummyForeignProber)
353
 
    testcase.addCleanup(controldir.ControlDirFormat.unregister_prober,
354
 
                        DummyForeignProber)
355
 
    repository.format_registry.register(DummyForeignVcsRepositoryFormat())
356
 
    testcase.addCleanup(repository.format_registry.remove,
357
 
                        DummyForeignVcsRepositoryFormat())
358
 
    branch.format_registry.register(DummyForeignVcsBranchFormat())
359
 
    testcase.addCleanup(branch.format_registry.remove,
360
 
                        DummyForeignVcsBranchFormat())
 
268
    bzrdir.BzrDirFormat.register_control_format(DummyForeignVcsDirFormat)
 
269
    testcase.addCleanup(bzrdir.BzrDirFormat.unregister_control_format,
 
270
                        DummyForeignVcsDirFormat)
361
271
    # We need to register the optimiser to make the dummy appears really
362
272
    # different from a regular bzr repository.
363
273
    branch.InterBranch.register_optimiser(InterToDummyVcsBranch)
365
275
                        InterToDummyVcsBranch)
366
276
 
367
277
 
368
 
class DummyForeignProber(controldir.Prober):
369
 
 
370
 
    @classmethod
371
 
    def probe_transport(klass, transport):
372
 
        """Return the .bzrdir style format present in a directory."""
373
 
        if not transport.has('.dummy'):
374
 
            raise errors.NotBranchError(path=transport.base)
375
 
        return DummyForeignVcsDirFormat()
376
 
 
377
 
    @classmethod
378
 
    def known_formats(cls):
379
 
        return set([DummyForeignVcsDirFormat()])
380
 
 
381
 
 
382
278
class ForeignVcsRegistryTests(tests.TestCase):
383
279
    """Tests for the ForeignVcsRegistry class."""
384
280
 
396
292
        reg = foreign.ForeignVcsRegistry()
397
293
        vcs = DummyForeignVcs()
398
294
        reg.register("dummy", vcs, "Dummy VCS")
399
 
        self.assertEqual((
400
 
            ("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
401
 
            reg.parse_revision_id("dummy-v1:some-foreign-revid"))
 
295
        self.assertEquals((("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
 
296
                          reg.parse_revision_id("dummy-v1:some-foreign-revid"))
402
297
 
403
298
 
404
299
class ForeignRevisionTests(tests.TestCase):
408
303
        mapp = DummyForeignVcsMapping(DummyForeignVcs())
409
304
        rev = foreign.ForeignRevision(("a", "foreign", "revid"),
410
305
                                      mapp, "roundtripped-revid")
411
 
        self.assertEqual("", rev.inventory_sha1)
412
 
        self.assertEqual(("a", "foreign", "revid"), rev.foreign_revid)
413
 
        self.assertEqual(mapp, rev.mapping)
 
306
        self.assertEquals("", rev.inventory_sha1)
 
307
        self.assertEquals(("a", "foreign", "revid"), rev.foreign_revid)
 
308
        self.assertEquals(mapp, rev.mapping)
414
309
 
415
310
 
416
311
class WorkingTreeFileUpdateTests(tests.TestCaseWithTransport):
432
327
        foreign.update_workingtree_fileids(wt, target_basis)
433
328
        wt.lock_read()
434
329
        try:
435
 
            self.assertEqual(set([root_id, "bla-b"]), set(wt.all_file_ids()))
 
330
            self.assertEquals(set([root_id, "bla-b"]), set(wt.inventory))
436
331
        finally:
437
332
            wt.unlock()
438
333
 
447
342
    def test_create(self):
448
343
        """Test we can create dummies."""
449
344
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
450
 
        dir = controldir.ControlDir.open("d")
451
 
        self.assertEqual("A Dummy VCS Dir", dir._format.get_format_string())
 
345
        dir = bzrdir.BzrDir.open("d")
 
346
        self.assertEquals("A Dummy VCS Dir", dir._format.get_format_string())
452
347
        dir.open_repository()
453
348
        dir.open_branch()
454
349
        dir.open_workingtree()
456
351
    def test_sprout(self):
457
352
        """Test we can clone dummies and that the format is not preserved."""
458
353
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
459
 
        dir = controldir.ControlDir.open("d")
 
354
        dir = bzrdir.BzrDir.open("d")
460
355
        newdir = dir.sprout("e")
461
 
        self.assertNotEqual("A Dummy VCS Dir",
462
 
                            newdir._format.get_format_string())
 
356
        self.assertNotEquals("A Dummy VCS Dir",
 
357
                             newdir._format.get_format_string())
463
358
 
464
359
    def test_push_not_supported(self):
465
360
        source_tree = self.make_branch_and_tree("source")
466
 
        target_tree = self.make_branch_and_tree(
467
 
            "target", format=DummyForeignVcsDirFormat())
468
 
        self.assertRaises(errors.NoRoundtrippingSupport,
469
 
                          source_tree.branch.push, target_tree.branch)
 
361
        target_tree = self.make_branch_and_tree("target", 
 
362
            format=DummyForeignVcsDirFormat())
 
363
        self.assertRaises(errors.NoRoundtrippingSupport, 
 
364
            source_tree.branch.push, target_tree.branch)
470
365
 
471
366
    def test_lossy_push_empty(self):
472
367
        source_tree = self.make_branch_and_tree("source")
473
 
        target_tree = self.make_branch_and_tree(
474
 
            "target", format=DummyForeignVcsDirFormat())
475
 
        pushresult = source_tree.branch.push(target_tree.branch, lossy=True)
476
 
        self.assertEqual(revision.NULL_REVISION, pushresult.old_revid)
477
 
        self.assertEqual(revision.NULL_REVISION, pushresult.new_revid)
478
 
        self.assertEqual({}, pushresult.revidmap)
 
368
        target_tree = self.make_branch_and_tree("target", 
 
369
            format=DummyForeignVcsDirFormat())
 
370
        pushresult = source_tree.branch.lossy_push(target_tree.branch)
 
371
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)
 
372
        self.assertEquals(revision.NULL_REVISION, pushresult.new_revid)
 
373
        self.assertEquals({}, pushresult.revidmap)
479
374
 
480
375
    def test_lossy_push_simple(self):
481
376
        source_tree = self.make_branch_and_tree("source")
482
377
        self.build_tree(['source/a', 'source/b'])
483
378
        source_tree.add(['a', 'b'])
484
379
        revid1 = source_tree.commit("msg")
485
 
        target_tree = self.make_branch_and_tree(
486
 
            "target", format=DummyForeignVcsDirFormat())
 
380
        target_tree = self.make_branch_and_tree("target", 
 
381
            format=DummyForeignVcsDirFormat())
487
382
        target_tree.branch.lock_write()
488
383
        try:
489
 
            pushresult = source_tree.branch.push(
490
 
                target_tree.branch, lossy=True)
 
384
            pushresult = source_tree.branch.lossy_push(target_tree.branch)
491
385
        finally:
492
386
            target_tree.branch.unlock()
493
 
        self.assertEqual(revision.NULL_REVISION, pushresult.old_revid)
494
 
        self.assertEqual({revid1: target_tree.branch.last_revision()},
495
 
                         pushresult.revidmap)
496
 
        self.assertEqual(pushresult.revidmap[revid1], pushresult.new_revid)
 
387
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)
 
388
        self.assertEquals({revid1:target_tree.branch.last_revision()}, 
 
389
                           pushresult.revidmap)
 
390
        self.assertEquals(pushresult.revidmap[revid1], pushresult.new_revid)