~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_remote.py

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

• reverse the comparison (6071 to 5971.1.76)
• stop comparing with revision 5971.1.76

Show diffs side-by-side

added

removed

bzrlib/tests/test_remote.py

1183

1183

        client.add_expected_call(

1184

1184

            'Branch.last_revision_info', ('quack/',),

1185

1185

            'success', ('ok', '1', 'rev-tip'))

 

1186

        client.add_expected_call(

 

1187

            'Branch.get_config_file', ('quack/',),

 

1188

            'success', ('ok',), '')

 

1189

        transport.mkdir('quack')

 

1190

        transport = transport.clone('quack')

 

1191

        branch = self.make_remote_branch(transport, client)

 

1192

        result = branch.heads_to_fetch()

 

1193

        self.assertFinished(client)

 

1194

        self.assertEqual((set(['rev-tip']), set()), result)

 

1195

 

 

1196

    def test_uses_last_revision_info_and_tags_when_set(self):

 

1197

        transport = MemoryTransport()

 

1198

        client = FakeClient(transport.base)

 

1199

        client.add_expected_call(

 

1200

            'Branch.get_stacked_on_url', ('quack/',),

 

1201

            'error', ('NotStacked',))

 

1202

        client.add_expected_call(

 

1203

            'Branch.last_revision_info', ('quack/',),

 

1204

            'success', ('ok', '1', 'rev-tip'))

 

1205

        client.add_expected_call(

 

1206

            'Branch.get_config_file', ('quack/',),

 

1207

            'success', ('ok',), 'branch.fetch_tags = True')

1186

1208

        # XXX: this will break if the default format's serialization of tags

1187

1209

        # changes, or if the RPC for fetching tags changes from get_tags_bytes.

1188

1210

        client.add_expected_call(

1213

1235

        self.assertFinished(client)

1214

1236

        self.assertEqual((set(['tip']), set(['tagged-1', 'tagged-2'])), result)

1215

1237

 

1216

 

    def test_backwards_compatible(self):

 

1238

    def make_branch_with_tags(self):

1217

1239

        self.setup_smart_server_with_call_log()

1218

1240

        # Make a branch with a single revision.

1219

1241

        builder = self.make_branch_builder('foo')

1225

1247

        # Add two tags to that branch

1226

1248

        branch.tags.set_tag('tag-1', 'rev-1')

1227

1249

        branch.tags.set_tag('tag-2', 'rev-2')

 

1250

        return branch

 

1251

 

 

1252

    def test_backwards_compatible(self):

 

1253

        branch = self.make_branch_with_tags()

 

1254

        c = branch.get_config()

 

1255

        c.set_user_option('branch.fetch_tags', 'True')

1228

1256

        self.addCleanup(branch.lock_read().unlock)

1229

1257

        # Disable the heads_to_fetch verb

1230

1258

        verb = 'Branch.heads_to_fetch'

1233

1261

        result = branch.heads_to_fetch()

1234

1262

        self.assertEqual((set(['tip']), set(['rev-1', 'rev-2'])), result)

1235

1263

        self.assertEqual(

1236

 

            ['Branch.last_revision_info', 'Branch.get_tags_bytes'],

 

1264

            ['Branch.last_revision_info', 'Branch.get_config_file',

 

1265

             'Branch.get_tags_bytes'],

 

1266

            [call.call.method for call in self.hpss_calls])

 

1267

 

 

1268

    def test_backwards_compatible_no_tags(self):

 

1269

        branch = self.make_branch_with_tags()

 

1270

        c = branch.get_config()

 

1271

        c.set_user_option('branch.fetch_tags', 'False')

 

1272

        self.addCleanup(branch.lock_read().unlock)

 

1273

        # Disable the heads_to_fetch verb

 

1274

        verb = 'Branch.heads_to_fetch'

 

1275

        self.disable_verb(verb)

 

1276

        self.reset_smart_call_log()

 

1277

        result = branch.heads_to_fetch()

 

1278

        self.assertEqual((set(['tip']), set()), result)

 

1279

        self.assertEqual(

 

1280

            ['Branch.last_revision_info', 'Branch.get_config_file'],

1237

1281

            [call.call.method for call in self.hpss_calls])

1238

1282

 

1239

1283

 

3359

3403

        remote_branch_url = self.smart_server.get_url() + 'remote'

3360

3404

        remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()

3361

3405

        self.hpss_calls = []

3362

 

        local.repository.fetch(remote_branch.repository,

3363

 

                fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))

 

3406

        local.repository.fetch(

 

3407

            remote_branch.repository,

 

3408

            fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))

3364

3409

        self.assertEqual(['Repository.get_stream_1.19'], self.hpss_calls)

3365

3410

 

3366

3411

    def override_verb(self, verb_name, verb):

3381

3426

            """A version of the Repository.get_stream_1.19 verb patched to

3382

3427

            reject 'everything' searches the way 2.3 and earlier do.

3383

3428

            """

3384

 

            def recreate_search(self, repository, search_bytes, discard_excess=False):

 

3429

            def recreate_search(self, repository, search_bytes,

 

3430

                                discard_excess=False):

3385

3431

                verb_log.append(search_bytes.split('\n', 1)[0])

3386

3432

                if search_bytes == 'everything':

3387

 

                    return (None, request.FailedSmartServerResponse(('BadSearch',)))

 

3433

                    return (None,

 

3434

                            request.FailedSmartServerResponse(('BadSearch',)))

3388

3435

                return super(OldGetStreamVerb,

3389

3436

                        self).recreate_search(repository, search_bytes,

3390

3437

                            discard_excess=discard_excess)

3395

3442

        remote_branch_url = self.smart_server.get_url() + 'remote'

3396

3443

        remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()

3397

3444

        self.hpss_calls = []

3398

 

        local.repository.fetch(remote_branch.repository,

3399

 

                fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))

 

3445

        local.repository.fetch(

 

3446

            remote_branch.repository,

 

3447

            fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))

3400

3448

        # make sure the overridden verb was used

3401

3449

        self.assertLength(1, verb_log)

3402

3450

        # more than one HPSS call is needed, but because it's a VFS callback

3403

3451

        # its hard to predict exactly how many.

3404

3452

        self.assertTrue(len(self.hpss_calls) > 1)

3405

3453

 

 

3454

 

 

3455

class TestUpdateBoundBranchWithModifiedBoundLocation(

 

3456

    tests.TestCaseWithTransport):

 

3457

    """Ensure correct handling of bound_location modifications.

 

3458

 

 

3459

    This is tested against a smart server as http://pad.lv/786980 was about a

 

3460

    ReadOnlyError (write attempt during a read-only transaction) which can only

 

3461

    happen in this context.

 

3462

    """

 

3463

 

 

3464

    def setUp(self):

 

3465

        super(TestUpdateBoundBranchWithModifiedBoundLocation, self).setUp()

 

3466

        self.transport_server = test_server.SmartTCPServer_for_testing

 

3467

 

 

3468

    def make_master_and_checkout(self, master_name, checkout_name):

 

3469

        # Create the master branch and its associated checkout

 

3470

        self.master = self.make_branch_and_tree(master_name)

 

3471

        self.checkout = self.master.branch.create_checkout(checkout_name)

 

3472

        # Modify the master branch so there is something to update

 

3473

        self.master.commit('add stuff')

 

3474

        self.last_revid = self.master.commit('even more stuff')

 

3475

        self.bound_location = self.checkout.branch.get_bound_location()

 

3476

 

 

3477

    def assertUpdateSucceeds(self, new_location):

 

3478

        self.checkout.branch.set_bound_location(new_location)

 

3479

        self.checkout.update()

 

3480

        self.assertEquals(self.last_revid, self.checkout.last_revision())

 

3481

 

 

3482

    def test_without_final_slash(self):

 

3483

        self.make_master_and_checkout('master', 'checkout')

 

3484

        # For unclear reasons some users have a bound_location without a final

 

3485

        # '/', simulate that by forcing such a value

 

3486

        self.assertEndsWith(self.bound_location, '/')

 

3487

        self.assertUpdateSucceeds(self.bound_location.rstrip('/'))

 

3488

 

 

3489

    def test_plus_sign(self):

 

3490

        self.make_master_and_checkout('+master', 'checkout')

 

3491

        self.assertUpdateSucceeds(self.bound_location.replace('%2B', '+', 1))

 

3492

 

 

3493

    def test_tilda(self):

 

3494

        # Embed ~ in the middle of the path just to avoid any $HOME

 

3495

        # interpretation

 

3496

        self.make_master_and_checkout('mas~ter', 'checkout')

 

3497

        self.assertUpdateSucceeds(self.bound_location.replace('%2E', '~', 1))

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1