39
39
from ..bzr import groupcompress_repo
41
# This is the dummy foreign revision control system, used
41
# This is the dummy foreign revision control system, used
42
42
# mainly here in the testsuite to test the foreign VCS infrastructure.
43
# It is basically standard Bazaar with some minor modifications to
43
# It is basically standard Bazaar with some minor modifications to
46
46
# It has the following differences to "regular" Bazaar:
47
47
# - The control directory is named ".dummy", not ".bzr".
48
48
# - The revision ids are tuples, not strings.
56
56
return isinstance(self, type(other))
58
58
def revision_id_bzr_to_foreign(self, bzr_revid):
59
return tuple(bzr_revid[len(b"dummy-v1:"):].split(b"-")), self
59
return tuple(bzr_revid[len("dummy-v1:"):].split("-")), self
61
61
def revision_id_foreign_to_bzr(self, foreign_revid):
62
return b"dummy-v1:%s-%s-%s" % foreign_revid
62
return "dummy-v1:%s-%s-%s" % foreign_revid
65
65
class DummyForeignVcsMappingRegistry(foreign.VcsMappingRegistry):
67
67
def revision_id_bzr_to_foreign(self, revid):
68
if not revid.startswith(b"dummy-"):
68
if not revid.startswith("dummy-"):
69
69
raise errors.InvalidRevisionId(revid, None)
70
mapping_version = revid[len(b"dummy-"):len(b"dummy-vx")]
70
mapping_version = revid[len("dummy-"):len("dummy-vx")]
71
71
mapping = self.get(mapping_version)
72
72
return mapping.revision_id_bzr_to_foreign(revid)
81
81
def __init__(self):
82
82
self.mapping_registry = DummyForeignVcsMappingRegistry()
83
self.mapping_registry.register(b"v1", DummyForeignVcsMapping(self),
83
self.mapping_registry.register("v1", DummyForeignVcsMapping(self),
85
85
self.abbreviation = "dummy"
87
87
def show_foreign_revid(self, foreign_revid):
88
return {"dummy ding": "%s/%s\\%s" % foreign_revid}
88
return { "dummy ding": "%s/%s\\%s" % foreign_revid }
90
90
def serialize_foreign_revid(self, foreign_revid):
91
91
return "%s|%s|%s" % foreign_revid
104
104
self._ignore_fallbacks = False
105
105
self.controldir = a_controldir
106
106
foreign.ForeignBranch.__init__(self,
107
DummyForeignVcsMapping(DummyForeignVcs()))
107
DummyForeignVcsMapping(DummyForeignVcs()))
108
108
bzrbranch.BzrBranch6.__init__(self, _format, _control_files, a_controldir,
111
111
def _get_checkout_format(self, lightweight=False):
112
112
"""Return the most suitable metadir for a checkout of this branch.
123
123
return (revno, revid)
126
class DummyForeignCommitBuilder(vf_repository.VersionedFileCommitBuilder):
126
class DummyForeignCommitBuilder(vf_repository.VersionedFileRootCommitBuilder):
128
128
def _generate_revision_if_needed(self, revid):
129
129
mapping = DummyForeignVcsMapping(DummyForeignVcs())
131
131
self._new_revision_id = mapping.revision_id_foreign_to_bzr(
132
(b'%d' % self._timestamp,
133
str(self._timezone).encode('ascii'), b"UNKNOWN"))
132
(str(self._timestamp), str(self._timezone), "UNKNOWN"))
134
133
self.random_revid = False
135
134
elif revid is not None:
136
135
self._new_revision_id = revid
143
142
class DummyForeignVcsRepository(groupcompress_repo.CHKInventoryRepository,
144
foreign.ForeignRepository):
143
foreign.ForeignRepository):
145
144
"""Dummy foreign vcs repository."""
161
160
def branch_history(graph, revid):
162
161
ret = list(graph.iter_lefthand_ancestry(revid,
163
(revision.NULL_REVISION,)))
162
(revision.NULL_REVISION,)))
171
170
def is_compatible(source, target):
172
171
return isinstance(target, DummyForeignVcsBranch)
174
def push(self, overwrite=False, stop_revision=None, lossy=False, tag_selector=None):
173
def push(self, overwrite=False, stop_revision=None, lossy=False):
176
175
raise errors.NoRoundtrippingSupport(self.source, self.target)
177
176
result = branch.BranchPushResult()
183
182
graph = self.source.repository.get_graph()
184
183
# This just handles simple cases, but that's good enough for tests
185
184
my_history = branch_history(self.target.repository.get_graph(),
187
186
if stop_revision is None:
188
187
stop_revision = self.source.last_revision()
189
188
their_history = branch_history(graph, stop_revision)
194
193
for revid in todo:
195
194
rev = self.source.repository.get_revision(revid)
196
195
tree = self.source.repository.revision_tree(revid)
197
def get_file_with_stat(path):
198
return (tree.get_file(path), None)
196
def get_file_with_stat(path, file_id=None):
197
return (tree.get_file(path, file_id), None)
199
198
tree.get_file_with_stat = get_file_with_stat
200
199
new_revid = self.target.mapping.revision_id_foreign_to_bzr(
201
(b'%d' % rev.timestamp, str(rev.timezone).encode('ascii'),
202
str(self.target.revno()).encode('ascii')))
203
parent_revno, parent_revid = self.target.last_revision_info()
200
(str(rev.timestamp), str(rev.timezone),
201
str(self.target.revno())))
202
parent_revno, parent_revid= self.target.last_revision_info()
204
203
if parent_revid == revision.NULL_REVISION:
205
204
parent_revids = []
207
206
parent_revids = [parent_revid]
208
builder = self.target.get_commit_builder(
209
parent_revids, self.target.get_config_stack(), rev.timestamp,
210
rev.timezone, rev.committer, rev.properties,
207
builder = self.target.get_commit_builder(parent_revids,
208
self.target.get_config_stack(), rev.timestamp,
209
rev.timezone, rev.committer, rev.properties,
213
212
parent_tree = self.target.repository.revision_tree(
222
221
revidmap[revid] = builder.commit(rev.message)
223
self.target.set_last_revision_info(parent_revno + 1,
222
self.target.set_last_revision_info(parent_revno+1,
225
224
trace.mutter('lossily pushed revision %s -> %s',
226
revid, revidmap[revid])
225
revid, revidmap[revid])
228
227
self.source.unlock()
229
228
result.new_revno, result.new_revid = self.target.last_revision_info()
242
241
return DummyForeignVcsDirFormat()
244
243
def open(self, a_controldir, name=None, _found=False, ignore_fallbacks=False,
245
found_repository=None):
244
found_repository=None):
247
246
name = a_controldir._get_selected_branch()
254
253
if found_repository is None:
255
254
found_repository = a_controldir.find_repository()
256
255
return DummyForeignVcsBranch(_format=self,
257
_control_files=control_files,
258
a_controldir=a_controldir,
259
_repository=found_repository,
256
_control_files=control_files,
257
a_controldir=a_controldir,
258
_repository=found_repository,
261
260
except errors.NoSuchFile:
262
261
raise errors.NotBranchError(path=transport.base)
289
288
# Since we don't have a .bzr directory, inherit the
290
289
# mode from the root directory
291
290
temp_control = lockable_files.LockableFiles(transport,
292
'', lockable_files.TransportLock)
291
'', lockable_files.TransportLock)
293
292
temp_control._transport.mkdir('.dummy',
294
293
# FIXME: RBC 20060121 don't peek under
298
297
bzrdir_transport = transport.clone('.dummy')
299
298
# NB: no need to escape relative paths that are url safe.
300
299
control_files = lockable_files.LockableFiles(bzrdir_transport,
301
self._lock_file_name, self._lock_class)
300
self._lock_file_name, self._lock_class)
302
301
control_files.create_lock()
303
302
return self.open(transport, _found=True)
314
313
self.root_transport = _transport
315
314
self._mode_check_done = False
316
315
self._control_files = lockable_files.LockableFiles(self.transport,
317
"lock", lockable_files.TransportLock)
316
"lock", lockable_files.TransportLock)
319
318
def create_workingtree(self):
320
319
# dirstate requires a ".bzr" entry to exist
321
self.root_transport.put_bytes(".bzr", b"foo")
320
self.root_transport.put_bytes(".bzr", "foo")
322
321
return super(DummyForeignVcsDir, self).create_workingtree()
324
323
def open_branch(self, name=None, unsupported=False, ignore_fallbacks=True,
386
385
def test_parse_revision_id_no_dash(self):
387
386
reg = foreign.ForeignVcsRegistry()
388
387
self.assertRaises(errors.InvalidRevisionId,
389
reg.parse_revision_id, b"invalid")
388
reg.parse_revision_id, "invalid")
391
390
def test_parse_revision_id_unknown_mapping(self):
392
391
reg = foreign.ForeignVcsRegistry()
393
392
self.assertRaises(errors.InvalidRevisionId,
394
reg.parse_revision_id, b"unknown-foreignrevid")
393
reg.parse_revision_id, "unknown-foreignrevid")
396
395
def test_parse_revision_id(self):
397
396
reg = foreign.ForeignVcsRegistry()
398
397
vcs = DummyForeignVcs()
399
398
reg.register("dummy", vcs, "Dummy VCS")
400
399
self.assertEqual((
401
(b"some", b"foreign", b"revid"), DummyForeignVcsMapping(vcs)),
402
reg.parse_revision_id(b"dummy-v1:some-foreign-revid"))
400
("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),
401
reg.parse_revision_id("dummy-v1:some-foreign-revid"))
405
404
class ForeignRevisionTests(tests.TestCase):
408
407
def test_create(self):
409
408
mapp = DummyForeignVcsMapping(DummyForeignVcs())
410
rev = foreign.ForeignRevision((b"a", b"foreign", b"revid"),
411
mapp, b"roundtripped-revid")
412
self.assertEqual(b"", rev.inventory_sha1)
413
self.assertEqual((b"a", b"foreign", b"revid"), rev.foreign_revid)
409
rev = foreign.ForeignRevision(("a", "foreign", "revid"),
410
mapp, "roundtripped-revid")
411
self.assertEqual("", rev.inventory_sha1)
412
self.assertEqual(("a", "foreign", "revid"), rev.foreign_revid)
414
413
self.assertEqual(mapp, rev.mapping)
416
class WorkingTreeFileUpdateTests(tests.TestCaseWithTransport):
417
"""Tests for update_workingtree_fileids()."""
419
def test_update_workingtree(self):
420
wt = self.make_branch_and_tree('br1')
421
self.build_tree_contents([('br1/bla', b'original contents\n')])
422
wt.add('bla', 'bla-a')
424
root_id = wt.get_root_id()
425
target = wt.controldir.sprout('br2').open_workingtree()
426
target.unversion(['bla'])
427
target.add('bla', 'bla-b')
428
target.commit('bla-b')
429
target_basis = target.basis_tree()
430
target_basis.lock_read()
431
self.addCleanup(target_basis.unlock)
432
foreign.update_workingtree_fileids(wt, target_basis)
435
self.assertEqual({'', "bla"}, set(wt.all_versioned_paths()))
417
440
class DummyForeignVcsTests(tests.TestCaseWithTransport):
418
441
"""Very basic test for DummyForeignVcs."""
425
448
"""Test we can create dummies."""
426
449
self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
427
450
dir = controldir.ControlDir.open("d")
428
self.assertEqual(b"A Dummy VCS Dir", dir._format.get_format_string())
451
self.assertEqual("A Dummy VCS Dir", dir._format.get_format_string())
429
452
dir.open_repository()
430
453
dir.open_branch()
431
454
dir.open_workingtree()
435
458
self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())
436
459
dir = controldir.ControlDir.open("d")
437
460
newdir = dir.sprout("e")
438
self.assertNotEqual(b"A Dummy VCS Dir",
461
self.assertNotEqual("A Dummy VCS Dir",
439
462
newdir._format.get_format_string())
441
464
def test_push_not_supported(self):
added
removed
breezy/tests/test_foreign.py
