← 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: 2010-08-24 23:20:14 UTC
  • mfrom: (5365.5.29 2.3-btree-chk-leaf)
  • Revision ID: pqm@pqm.ubuntu.com-20100824232014-nu9owzel2zym2jk2
(jam) Use a custom C type for CHK index entries, saves memory

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