/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_foreign.py

• browse files at revision 7498.1.1
• compare with another revision
• download diff
• download tarball
• view history from revision 7498.1.1

• reverse the comparison (7498.1.1 to 6858.1.1)
• stop comparing with revision 6858.1.1

Show diffs side-by-side

added

removed

breezy/tests/test_foreign.py

38

38

 

39

39

from ..bzr import groupcompress_repo

40

40

 

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 

44

 

# make it "foreign". 

45

 

# 

 

43

# It is basically standard Bazaar with some minor modifications to

 

44

# make it "foreign".

 

45

#

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))

57

57

 

58

58

    def revision_id_bzr_to_foreign(self, bzr_revid):

59

 

        return tuple(bzr_revid[len("dummy-v1:"):].split("-")), self

 

59

        return tuple(bzr_revid[len(b"dummy-v1:"):].split(b"-")), self

60

60

 

61

61

    def revision_id_foreign_to_bzr(self, foreign_revid):

62

 

        return "dummy-v1:%s-%s-%s" % foreign_revid

 

62

        return b"dummy-v1:%s-%s-%s" % foreign_revid

63

63

 

64

64

 

65

65

class DummyForeignVcsMappingRegistry(foreign.VcsMappingRegistry):

66

66

 

67

67

    def revision_id_bzr_to_foreign(self, revid):

68

 

        if not revid.startswith("dummy-"):

 

68

        if not revid.startswith(b"dummy-"):

69

69

            raise errors.InvalidRevisionId(revid, None)

70

 

        mapping_version = revid[len("dummy-"):len("dummy-vx")]

 

70

        mapping_version = revid[len(b"dummy-"):len(b"dummy-vx")]

71

71

        mapping = self.get(mapping_version)

72

72

        return mapping.revision_id_bzr_to_foreign(revid)

73

73

 

80

80

 

81

81

    def __init__(self):

82

82

        self.mapping_registry = DummyForeignVcsMappingRegistry()

83

 

        self.mapping_registry.register("v1", DummyForeignVcsMapping(self),

 

83

        self.mapping_registry.register(b"v1", DummyForeignVcsMapping(self),

84

84

                                       "Version 1")

85

85

        self.abbreviation = "dummy"

86

86

 

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}

89

89

 

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,

109

 

            *args, **kwargs)

 

109

                                      *args, **kwargs)

110

110

 

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)

124

124

 

125

125

 

126

 

class DummyForeignCommitBuilder(vf_repository.VersionedFileRootCommitBuilder):

 

126

class DummyForeignCommitBuilder(vf_repository.VersionedFileCommitBuilder):

127

127

 

128

128

    def _generate_revision_if_needed(self, revid):

129

129

        mapping = DummyForeignVcsMapping(DummyForeignVcs())

130

130

        if self._lossy:

131

131

            self._new_revision_id = mapping.revision_id_foreign_to_bzr(

132

 

                (str(self._timestamp), str(self._timezone), "UNKNOWN"))

 

132

                (b'%d' % self._timestamp,

 

133

                 str(self._timezone).encode('ascii'), b"UNKNOWN"))

133

134

            self.random_revid = False

134

135

        elif revid is not None:

135

136

            self._new_revision_id = revid

140

141

 

141

142

 

142

143

class DummyForeignVcsRepository(groupcompress_repo.CHKInventoryRepository,

143

 

    foreign.ForeignRepository):

 

144

                                foreign.ForeignRepository):

144

145

    """Dummy foreign vcs repository."""

145

146

 

146

147

 

151

152

 

152

153

    @classmethod

153

154

    def get_format_string(cls):

154

 

        return "Dummy Foreign Vcs Repository"

 

155

        return b"Dummy Foreign Vcs Repository"

155

156

 

156

157

    def get_format_description(self):

157

158

        return "Dummy Foreign Vcs Repository"

159

160

 

160

161

def branch_history(graph, revid):

161

162

    ret = list(graph.iter_lefthand_ancestry(revid,

162

 

        (revision.NULL_REVISION,)))

 

163

                                            (revision.NULL_REVISION,)))

163

164

    ret.reverse()

164

165

    return ret

165

166

 

170

171

    def is_compatible(source, target):

171

172

        return isinstance(target, DummyForeignVcsBranch)

172

173

 

173

 

    def push(self, overwrite=False, stop_revision=None, lossy=False):

 

174

    def push(self, overwrite=False, stop_revision=None, lossy=False, tag_selector=None):

174

175

        if not lossy:

175

176

            raise errors.NoRoundtrippingSupport(self.source, self.target)

176

177

        result = branch.BranchPushResult()

182

183

            graph = self.source.repository.get_graph()

183

184

            # This just handles simple cases, but that's good enough for tests

184

185

            my_history = branch_history(self.target.repository.get_graph(),

185

 

                result.old_revid)

 

186

                                        result.old_revid)

186

187

            if stop_revision is None:

187

188

                stop_revision = self.source.last_revision()

188

189

            their_history = branch_history(graph, stop_revision)

193

194

            for revid in todo:

194

195

                rev = self.source.repository.get_revision(revid)

195

196

                tree = self.source.repository.revision_tree(revid)

196

 

                def get_file_with_stat(path, file_id=None):

197

 

                    return (tree.get_file(path, file_id), None)

 

197

                def get_file_with_stat(path):

 

198

                    return (tree.get_file(path), None)

198

199

                tree.get_file_with_stat = get_file_with_stat

199

200

                new_revid = self.target.mapping.revision_id_foreign_to_bzr(

200

 

                    (str(rev.timestamp), str(rev.timezone),

201

 

                        str(self.target.revno())))

202

 

                parent_revno, parent_revid= self.target.last_revision_info()

 

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()

203

204

                if parent_revid == revision.NULL_REVISION:

204

205

                    parent_revids = []

205

206

                else:

206

207

                    parent_revids = [parent_revid]

207

 

                builder = self.target.get_commit_builder(parent_revids,

208

 

                        self.target.get_config_stack(), rev.timestamp,

209

 

                        rev.timezone, rev.committer, rev.properties,

210

 

                        new_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,

 

211

                    new_revid)

211

212

                try:

212

213

                    parent_tree = self.target.repository.revision_tree(

213

214

                        parent_revid)

219

220

                    builder.abort()

220

221

                    raise

221

222

                revidmap[revid] = builder.commit(rev.message)

222

 

                self.target.set_last_revision_info(parent_revno+1,

223

 

                    revidmap[revid])

 

223

                self.target.set_last_revision_info(parent_revno + 1,

 

224

                                                   revidmap[revid])

224

225

                trace.mutter('lossily pushed revision %s -> %s',

225

 

                    revid, revidmap[revid])

 

226

                             revid, revidmap[revid])

226

227

        finally:

227

228

            self.source.unlock()

228

229

        result.new_revno, result.new_revid = self.target.last_revision_info()

234

235

 

235

236

    @classmethod

236

237

    def get_format_string(cls):

237

 

        return "Branch for Testing"

 

238

        return b"Branch for Testing"

238

239

 

239

240

    @property

240

241

    def _matchingcontroldir(self):

241

242

        return DummyForeignVcsDirFormat()

242

243

 

243

244

    def open(self, a_controldir, name=None, _found=False, ignore_fallbacks=False,

244

 

            found_repository=None):

 

245

             found_repository=None):

245

246

        if name is None:

246

247

            name = a_controldir._get_selected_branch()

247

248

        if not _found:

253

254

            if found_repository is None:

254

255

                found_repository = a_controldir.find_repository()

255

256

            return DummyForeignVcsBranch(_format=self,

256

 

                              _control_files=control_files,

257

 

                              a_controldir=a_controldir,

258

 

                              _repository=found_repository,

259

 

                              name=name)

 

257

                                         _control_files=control_files,

 

258

                                         a_controldir=a_controldir,

 

259

                                         _repository=found_repository,

 

260

                                         name=name)

260

261

        except errors.NoSuchFile:

261

262

            raise errors.NotBranchError(path=transport.base)

262

263

 

266

267

 

267

268

    @classmethod

268

269

    def get_format_string(cls):

269

 

        return "A Dummy VCS Dir"

 

270

        return b"A Dummy VCS Dir"

270

271

 

271

272

    @classmethod

272

273

    def get_format_description(cls):

288

289

        # Since we don't have a .bzr directory, inherit the

289

290

        # mode from the root directory

290

291

        temp_control = lockable_files.LockableFiles(transport,

291

 

                            '', lockable_files.TransportLock)

 

292

                                                    '', lockable_files.TransportLock)

292

293

        temp_control._transport.mkdir('.dummy',

293

294

                                      # FIXME: RBC 20060121 don't peek under

294

295

                                      # the covers

297

298

        bzrdir_transport = transport.clone('.dummy')

298

299

        # NB: no need to escape relative paths that are url safe.

299

300

        control_files = lockable_files.LockableFiles(bzrdir_transport,

300

 

            self._lock_file_name, self._lock_class)

 

301

                                                     self._lock_file_name, self._lock_class)

301

302

        control_files.create_lock()

302

303

        return self.open(transport, _found=True)

303

304

 

313

314

        self.root_transport = _transport

314

315

        self._mode_check_done = False

315

316

        self._control_files = lockable_files.LockableFiles(self.transport,

316

 

            "lock", lockable_files.TransportLock)

 

317

                                                           "lock", lockable_files.TransportLock)

317

318

 

318

319

    def create_workingtree(self):

319

320

        # dirstate requires a ".bzr" entry to exist

320

 

        self.root_transport.put_bytes(".bzr", "foo")

 

321

        self.root_transport.put_bytes(".bzr", b"foo")

321

322

        return super(DummyForeignVcsDir, self).create_workingtree()

322

323

 

323

324

    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=True,

376

377

 

377

378

    @classmethod

378

379

    def known_formats(cls):

379

 

        return {DummyForeignVcsDirFormat()}

 

380

        return [DummyForeignVcsDirFormat()]

380

381

 

381

382

 

382

383

class ForeignVcsRegistryTests(tests.TestCase):

385

386

    def test_parse_revision_id_no_dash(self):

386

387

        reg = foreign.ForeignVcsRegistry()

387

388

        self.assertRaises(errors.InvalidRevisionId,

388

 

                          reg.parse_revision_id, "invalid")

 

389

                          reg.parse_revision_id, b"invalid")

389

390

 

390

391

    def test_parse_revision_id_unknown_mapping(self):

391

392

        reg = foreign.ForeignVcsRegistry()

392

393

        self.assertRaises(errors.InvalidRevisionId,

393

 

                          reg.parse_revision_id, "unknown-foreignrevid")

 

394

                          reg.parse_revision_id, b"unknown-foreignrevid")

394

395

 

395

396

    def test_parse_revision_id(self):

396

397

        reg = foreign.ForeignVcsRegistry()

397

398

        vcs = DummyForeignVcs()

398

399

        reg.register("dummy", vcs, "Dummy VCS")

399

400

        self.assertEqual((

400

 

            ("some", "foreign", "revid"), DummyForeignVcsMapping(vcs)),

401

 

            reg.parse_revision_id("dummy-v1:some-foreign-revid"))

 

401

            (b"some", b"foreign", b"revid"), DummyForeignVcsMapping(vcs)),

 

402

            reg.parse_revision_id(b"dummy-v1:some-foreign-revid"))

402

403

 

403

404

 

404

405

class ForeignRevisionTests(tests.TestCase):

406

407

 

407

408

    def test_create(self):

408

409

        mapp = DummyForeignVcsMapping(DummyForeignVcs())

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)

 

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)

413

414

        self.assertEqual(mapp, rev.mapping)

414

415

 

415

416

 

416

 

class WorkingTreeFileUpdateTests(tests.TestCaseWithTransport):

417

 

    """Tests for update_workingtree_fileids()."""

418

 

 

419

 

    def test_update_workingtree(self):

420

 

        wt = self.make_branch_and_tree('br1')

421

 

        self.build_tree_contents([('br1/bla', 'original contents\n')])

422

 

        wt.add('bla', 'bla-a')

423

 

        wt.commit('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)

433

 

        wt.lock_read()

434

 

        try:

435

 

            self.assertEqual({'', "bla"}, set(wt.all_versioned_paths()))

436

 

        finally:

437

 

            wt.unlock()

438

 

 

439

 

 

440

417

class DummyForeignVcsTests(tests.TestCaseWithTransport):

441

418

    """Very basic test for DummyForeignVcs."""

442

419

 

448

425

        """Test we can create dummies."""

449

426

        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())

450

427

        dir = controldir.ControlDir.open("d")

451

 

        self.assertEqual("A Dummy VCS Dir", dir._format.get_format_string())

 

428

        self.assertEqual(b"A Dummy VCS Dir", dir._format.get_format_string())

452

429

        dir.open_repository()

453

430

        dir.open_branch()

454

431

        dir.open_workingtree()

458

435

        self.make_branch_and_tree("d", format=DummyForeignVcsDirFormat())

459

436

        dir = controldir.ControlDir.open("d")

460

437

        newdir = dir.sprout("e")

461

 

        self.assertNotEqual("A Dummy VCS Dir",

 

438

        self.assertNotEqual(b"A Dummy VCS Dir",

462

439

                            newdir._format.get_format_string())

463

440

 

464

441

    def test_push_not_supported(self):

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0