263
265
repo = wt.branch.repository
264
266
repo.lock_write()
265
267
repo.start_write_group()
266
repo.sign_revision('rev1', gpg.LoopbackGPGStrategy(None))
269
repo.sign_revision('rev1', gpg.LoopbackGPGStrategy(None))
270
except errors.UnsupportedOperation:
271
self.assertFalse(repo._format.supports_revision_signatures)
272
raise TestNotApplicable("repository format does not support signatures")
267
273
repo.commit_write_group()
349
355
self.addCleanup(source.unlock)
350
356
target.fetch(source, revision_id='B-id')
353
class TestSource(TestCaseWithRepository):
354
"""Tests for/about the results of Repository._get_source."""
356
def test_no_absent_records_in_stream_with_ghosts(self):
357
# XXX: Arguably should be in per_interrepository but
358
# doesn't actually gain coverage there; need a specific set of
359
# permutations to cover it.
360
# bug lp:376255 was reported about this.
361
builder = self.make_branch_builder('repo')
362
builder.start_series()
363
builder.build_snapshot('tip', ['ghost'],
364
[('add', ('', 'ROOT_ID', 'directory', ''))],
365
allow_leftmost_as_ghost=True)
366
builder.finish_series()
367
b = builder.get_branch()
369
self.addCleanup(b.unlock)
371
source = repo._get_source(repo._format)
372
search = graph.PendingAncestryResult(['tip'], repo)
373
stream = source.get_stream(search)
374
for substream_type, substream in stream:
375
for record in substream:
376
self.assertNotEqual('absent', record.storage_kind,
377
"Absent record for %s" % (((substream_type,) + record.key),))