← 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: John Arbash Meinel
  • Date: 2010-08-24 19:21:32 UTC
  • mto: This revision was merged to the branch mainline in revision 5390.
  • Revision ID: john@arbash-meinel.com-20100824192132-2ktt5adkbk5bk1ct
Handle test_source and extensions. Also define an 'extern' protocol, to allow
the test suite to recognize that returning an object of that type is a Python object.

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, 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
93
        self.bzrdir = a_bzrdir
103
 
        foreign.ForeignBranch.__init__(self,
 
94
        foreign.ForeignBranch.__init__(self, 
104
95
            DummyForeignVcsMapping(DummyForeignVcs()))
105
 
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir,
 
96
        branch.BzrBranch6.__init__(self, _format, _control_files, a_bzrdir, 
106
97
            *args, **kwargs)
107
98
 
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):
 
99
 
 
100
class InterToDummyVcsBranch(branch.GenericInterBranch,
 
101
                            foreign.InterToForeignBranch):
164
102
 
165
103
    @staticmethod
166
104
    def is_compatible(source, target):
167
105
        return isinstance(target, DummyForeignVcsBranch)
168
106
 
169
 
    def push(self, overwrite=False, stop_revision=None, lossy=False):
170
 
        if not lossy:
171
 
            raise errors.NoRoundtrippingSupport(self.source, self.target)
 
107
    def push(self, overwrite=False, stop_revision=None):
 
108
        raise errors.NoRoundtrippingSupport(self.source, self.target)
 
109
 
 
110
    def lossy_push(self, stop_revision=None):
172
111
        result = branch.BranchPushResult()
173
112
        result.source_branch = self.source
174
113
        result.target_branch = self.target
175
114
        result.old_revno, result.old_revid = self.target.last_revision_info()
176
115
        self.source.lock_read()
177
116
        try:
178
 
            graph = self.source.repository.get_graph()
179
117
            # 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)
 
118
            my_history = self.target.revision_history()
 
119
            their_history = self.source.revision_history()
185
120
            if their_history[:min(len(my_history), len(their_history))] != my_history:
186
121
                raise errors.DivergedBranches(self.target, self.source)
187
122
            todo = their_history[len(my_history):]
193
128
                    return (tree.get_file(file_id), None)
194
129
                tree.get_file_with_stat = get_file_with_stat
195
130
                new_revid = self.target.mapping.revision_id_foreign_to_bzr(
196
 
                    (str(rev.timestamp), str(rev.timezone),
 
131
                    (str(rev.timestamp), str(rev.timezone), 
197
132
                        str(self.target.revno())))
198
133
                parent_revno, parent_revid= self.target.last_revision_info()
199
134
                if parent_revid == revision.NULL_REVISION:
201
136
                else:
202
137
                    parent_revids = [parent_revid]
203
138
                builder = self.target.get_commit_builder(parent_revids, 
204
 
                        self.target.get_config_stack(), rev.timestamp,
 
139
                        self.target.get_config(), rev.timestamp,
205
140
                        rev.timezone, rev.committer, rev.properties,
206
141
                        new_revid)
207
142
                try:
208
 
                    parent_tree = self.target.repository.revision_tree(
209
 
                        parent_revid)
210
 
                    for path, ie in tree.iter_entries_by_dir():
 
143
                    for path, ie in tree.inventory.iter_entries():
211
144
                        new_ie = ie.copy()
212
145
                        new_ie.revision = None
213
146
                        builder.record_entry_contents(new_ie, 
214
 
                            [parent_tree.root_inventory],
 
147
                            [self.target.repository.revision_tree(parent_revid).inventory],
215
148
                            path, tree, 
216
149
                            (ie.kind, ie.text_size, ie.executable, ie.text_sha1))
217
150
                    builder.finish_inventory()
232
165
 
233
166
class DummyForeignVcsBranchFormat(branch.BzrBranchFormat6):
234
167
 
235
 
    @classmethod
236
 
    def get_format_string(cls):
 
168
    def get_format_string(self):
237
169
        return "Branch for Testing"
238
170
 
239
 
    @property
240
 
    def _matchingbzrdir(self):
241
 
        return DummyForeignVcsDirFormat()
 
171
    def __init__(self):
 
172
        super(DummyForeignVcsBranchFormat, self).__init__()
 
173
        self._matchingbzrdir = DummyForeignVcsDirFormat()
242
174
 
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()
 
175
    def open(self, a_bzrdir, name=None, _found=False):
247
176
        if not _found:
248
177
            raise NotImplementedError
249
178
        try:
250
179
            transport = a_bzrdir.get_branch_transport(None, name=name)
251
180
            control_files = lockable_files.LockableFiles(transport, 'lock',
252
181
                                                         lockdir.LockDir)
253
 
            if found_repository is None:
254
 
                found_repository = a_bzrdir.find_repository()
255
182
            return DummyForeignVcsBranch(_format=self,
256
183
                              _control_files=control_files,
257
184
                              a_bzrdir=a_bzrdir,
258
 
                              _repository=found_repository,
259
 
                              name=name)
 
185
                              _repository=a_bzrdir.find_repository())
260
186
        except errors.NoSuchFile:
261
187
            raise errors.NotBranchError(path=transport.base)
262
188
 
279
205
    def get_branch_format(self):
280
206
        return DummyForeignVcsBranchFormat()
281
207
 
282
 
    @property
283
 
    def repository_format(self):
284
 
        return DummyForeignVcsRepositoryFormat()
 
208
    @classmethod
 
209
    def probe_transport(klass, transport):
 
210
        """Return the .bzrdir style format present in a directory."""
 
211
        if not transport.has('.dummy'):
 
212
            raise errors.NotBranchError(path=transport.base)
 
213
        return klass()
285
214
 
286
215
    def initialize_on_transport(self, transport):
287
216
        """Initialize a new bzrdir in the base directory of a Transport."""
315
244
        self._control_files = lockable_files.LockableFiles(self.transport,
316
245
            "lock", lockable_files.TransportLock)
317
246
 
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 != "":
 
247
    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=True):
 
248
        if name is not None:
328
249
            raise errors.NoColocatedBranchSupport(self)
329
250
        return self._format.get_branch_format().open(self, _found=True)
330
251
 
331
252
    def cloning_metadir(self, stacked=False):
332
253
        """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()
 
254
        return bzrdir.format_registry.make_bzrdir("default")
337
255
 
338
256
    def sprout(self, url, revision_id=None, force_new_repo=False,
339
257
               recurse='down', possible_transports=None,
348
266
 
349
267
 
350
268
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())
 
269
    bzrdir.BzrDirFormat.register_control_format(DummyForeignVcsDirFormat)
 
270
    testcase.addCleanup(bzrdir.BzrDirFormat.unregister_control_format,
 
271
                        DummyForeignVcsDirFormat)
360
272
    # We need to register the optimiser to make the dummy appears really
361
273
    # different from a regular bzr repository.
362
274
    branch.InterBranch.register_optimiser(InterToDummyVcsBranch)
364
276
                        InterToDummyVcsBranch)
365
277
 
366
278
 
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
279
class ForeignVcsRegistryTests(tests.TestCase):
382
280
    """Tests for the ForeignVcsRegistry class."""
383
281
 
395
293
        reg = foreign.ForeignVcsRegistry()
396
294
        vcs = DummyForeignVcs()
397
295
        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"))
 
296
        self.assertEquals((("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
 
297
                          reg.parse_revision_id("dummy-v1:some-foreign-revid"))
401
298
 
402
299
 
403
300
class ForeignRevisionTests(tests.TestCase):
431
328
        foreign.update_workingtree_fileids(wt, target_basis)
432
329
        wt.lock_read()
433
330
        try:
434
 
            self.assertEquals(set([root_id, "bla-b"]), set(wt.all_file_ids()))
 
331
            self.assertEquals(set([root_id, "bla-b"]), set(wt.inventory))
435
332
        finally:
436
333
            wt.unlock()
437
334
 
446
343
    def test_create(self):
447
344
        """Test we can create dummies."""
448
345
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
449
 
        dir = controldir.ControlDir.open("d")
 
346
        dir = bzrdir.BzrDir.open("d")
450
347
        self.assertEquals("A Dummy VCS Dir", dir._format.get_format_string())
451
348
        dir.open_repository()
452
349
        dir.open_branch()
455
352
    def test_sprout(self):
456
353
        """Test we can clone dummies and that the format is not preserved."""
457
354
        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
458
 
        dir = controldir.ControlDir.open("d")
 
355
        dir = bzrdir.BzrDir.open("d")
459
356
        newdir = dir.sprout("e")
460
357
        self.assertNotEquals("A Dummy VCS Dir",
461
358
                             newdir._format.get_format_string())
471
368
        source_tree = self.make_branch_and_tree("source")
472
369
        target_tree = self.make_branch_and_tree("target", 
473
370
            format=DummyForeignVcsDirFormat())
474
 
        pushresult = source_tree.branch.push(target_tree.branch, lossy=True)
 
371
        pushresult = source_tree.branch.lossy_push(target_tree.branch)
475
372
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)
476
373
        self.assertEquals(revision.NULL_REVISION, pushresult.new_revid)
477
374
        self.assertEquals({}, pushresult.revidmap)
485
382
            format=DummyForeignVcsDirFormat())
486
383
        target_tree.branch.lock_write()
487
384
        try:
488
 
            pushresult = source_tree.branch.push(
489
 
                target_tree.branch, lossy=True)
 
385
            pushresult = source_tree.branch.lossy_push(target_tree.branch)
490
386
        finally:
491
387
            target_tree.branch.unlock()
492
388
        self.assertEquals(revision.NULL_REVISION, pushresult.old_revid)