/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_remote.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-01-12 18:39:25 UTC
  • mfrom: (5598.1.1 bzr)
  • Revision ID: pqm@pqm.ubuntu.com-20110112183925-rnn12p6xp91441r0
(vila) Fix PEP-8 violation in PythonVersionInfoBuilder's output (Soren
 Hansen)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_remote.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006-2011 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
30
30
    branch,
31
31
    bzrdir,
32
32
    config,
 
33
    controldir,
33
34
    errors,
34
35
    graph,
35
36
    inventory,
36
37
    inventory_delta,
37
 
    pack,
38
38
    remote,
39
39
    repository,
40
40
    tests,
 
41
    transport,
41
42
    treebuilder,
42
 
    urlutils,
43
43
    versionedfile,
44
44
    )
45
45
from bzrlib.branch import Branch
46
 
from bzrlib.bzrdir import BzrDir, BzrDirFormat
 
46
from bzrlib.bzrdir import (
 
47
    BzrDir,
 
48
    BzrDirFormat,
 
49
    RemoteBzrProber,
 
50
    )
47
51
from bzrlib.remote import (
48
52
    RemoteBranch,
49
53
    RemoteBranchFormat,
50
54
    RemoteBzrDir,
51
 
    RemoteBzrDirFormat,
52
55
    RemoteRepository,
53
56
    RemoteRepositoryFormat,
54
57
    )
58
61
from bzrlib.smart.client import _SmartClient
59
62
from bzrlib.smart.repository import SmartServerRepositoryGetParentMap
60
63
from bzrlib.tests import (
61
 
    condition_isinstance,
62
 
    split_suite_by_condition,
63
 
    multiply_tests,
64
64
    test_server,
65
65
    )
66
 
from bzrlib.transport import get_transport
 
66
from bzrlib.tests.scenarios import load_tests_apply_scenarios
67
67
from bzrlib.transport.memory import MemoryTransport
68
68
from bzrlib.transport.remote import (
69
69
    RemoteTransport,
70
70
    RemoteSSHTransport,
71
71
    RemoteTCPTransport,
72
 
)
73
 
 
74
 
def load_tests(standard_tests, module, loader):
75
 
    to_adapt, result = split_suite_by_condition(
76
 
        standard_tests, condition_isinstance(BasicRemoteObjectTests))
77
 
    smart_server_version_scenarios = [
 
72
    )
 
73
 
 
74
 
 
75
load_tests = load_tests_apply_scenarios
 
76
 
 
77
 
 
78
class BasicRemoteObjectTests(tests.TestCaseWithTransport):
 
79
 
 
80
    scenarios = [
78
81
        ('HPSS-v2',
79
 
         {'transport_server': test_server.SmartTCPServer_for_testing_v2_only}),
 
82
            {'transport_server': test_server.SmartTCPServer_for_testing_v2_only}),
80
83
        ('HPSS-v3',
81
 
         {'transport_server': test_server.SmartTCPServer_for_testing})]
82
 
    return multiply_tests(to_adapt, smart_server_version_scenarios, result)
83
 
 
84
 
 
85
 
class BasicRemoteObjectTests(tests.TestCaseWithTransport):
 
84
            {'transport_server': test_server.SmartTCPServer_for_testing})]
 
85
 
86
86
 
87
87
    def setUp(self):
88
88
        super(BasicRemoteObjectTests, self).setUp()
89
89
        self.transport = self.get_transport()
90
90
        # make a branch that can be opened over the smart transport
91
91
        self.local_wt = BzrDir.create_standalone_workingtree('.')
92
 
 
93
 
    def tearDown(self):
94
 
        self.transport.disconnect()
95
 
        tests.TestCaseWithTransport.tearDown(self)
 
92
        self.addCleanup(self.transport.disconnect)
96
93
 
97
94
    def test_create_remote_bzrdir(self):
98
95
        b = remote.RemoteBzrDir(self.transport, remote.RemoteBzrDirFormat())
122
119
    def test_find_correct_format(self):
123
120
        """Should open a RemoteBzrDir over a RemoteTransport"""
124
121
        fmt = BzrDirFormat.find_format(self.transport)
125
 
        self.assertTrue(RemoteBzrDirFormat
126
 
                        in BzrDirFormat._control_server_formats)
 
122
        self.assertTrue(bzrdir.RemoteBzrProber
 
123
                        in controldir.ControlDirFormat._server_probers)
127
124
        self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)
128
125
 
129
126
    def test_open_detected_smart_format(self):
359
356
        a given client_base and transport_base.
360
357
        """
361
358
        client_medium = medium.SmartClientMedium(client_base)
362
 
        transport = get_transport(transport_base)
363
 
        result = client_medium.remote_path_from_transport(transport)
 
359
        t = transport.get_transport(transport_base)
 
360
        result = client_medium.remote_path_from_transport(t)
364
361
        self.assertEqual(expected, result)
365
362
 
366
363
    def test_remote_path_from_transport(self):
377
374
        a given transport_base and relpath of that transport.  (Note that
378
375
        HttpTransportBase is a subclass of SmartClientMedium)
379
376
        """
380
 
        base_transport = get_transport(transport_base)
 
377
        base_transport = transport.get_transport(transport_base)
381
378
        client_medium = base_transport.get_smart_medium()
382
379
        cloned_transport = base_transport.clone(relpath)
383
380
        result = client_medium.remote_path_from_transport(cloned_transport)
609
606
        # _get_tree_branch is a form of open_branch, but it should only ask for
610
607
        # branch opening, not any other network requests.
611
608
        calls = []
612
 
        def open_branch():
 
609
        def open_branch(name=None):
613
610
            calls.append("Called")
614
611
            return "a-branch"
615
612
        transport = MemoryTransport()
686
683
        old.
687
684
        """
688
685
        self.assertRaises(errors.NotBranchError,
689
 
            RemoteBzrDirFormat.probe_transport, OldServerTransport())
 
686
            RemoteBzrProber.probe_transport, OldServerTransport())
690
687
 
691
688
 
692
689
class TestBzrDirCreateBranch(TestRemote):
1618
1615
    def test_get_multi_line_branch_conf(self):
1619
1616
        # Make sure that multiple-line branch.conf files are supported
1620
1617
        #
1621
 
        # https://bugs.edge.launchpad.net/bzr/+bug/354075
 
1618
        # https://bugs.launchpad.net/bzr/+bug/354075
1622
1619
        client = FakeClient()
1623
1620
        client.add_expected_call(
1624
1621
            'Branch.get_stacked_on_url', ('memory:///',),
1652
1649
        branch.unlock()
1653
1650
        self.assertFinished(client)
1654
1651
 
 
1652
    def test_set_option_with_dict(self):
 
1653
        client = FakeClient()
 
1654
        client.add_expected_call(
 
1655
            'Branch.get_stacked_on_url', ('memory:///',),
 
1656
            'error', ('NotStacked',),)
 
1657
        client.add_expected_call(
 
1658
            'Branch.lock_write', ('memory:///', '', ''),
 
1659
            'success', ('ok', 'branch token', 'repo token'))
 
1660
        encoded_dict_value = 'd5:ascii1:a11:unicode \xe2\x8c\x9a3:\xe2\x80\xbde'
 
1661
        client.add_expected_call(
 
1662
            'Branch.set_config_option_dict', ('memory:///', 'branch token',
 
1663
            'repo token', encoded_dict_value, 'foo', ''),
 
1664
            'success', ())
 
1665
        client.add_expected_call(
 
1666
            'Branch.unlock', ('memory:///', 'branch token', 'repo token'),
 
1667
            'success', ('ok',))
 
1668
        transport = MemoryTransport()
 
1669
        branch = self.make_remote_branch(transport, client)
 
1670
        branch.lock_write()
 
1671
        config = branch._get_config()
 
1672
        config.set_option(
 
1673
            {'ascii': 'a', u'unicode \N{WATCH}': u'\N{INTERROBANG}'},
 
1674
            'foo')
 
1675
        branch.unlock()
 
1676
        self.assertFinished(client)
 
1677
 
1655
1678
    def test_backwards_compat_set_option(self):
1656
1679
        self.setup_smart_server_with_call_log()
1657
1680
        branch = self.make_branch('.')
1664
1687
        self.assertLength(10, self.hpss_calls)
1665
1688
        self.assertEqual('value', branch._get_config().get_option('name'))
1666
1689
 
 
1690
    def test_backwards_compat_set_option_with_dict(self):
 
1691
        self.setup_smart_server_with_call_log()
 
1692
        branch = self.make_branch('.')
 
1693
        verb = 'Branch.set_config_option_dict'
 
1694
        self.disable_verb(verb)
 
1695
        branch.lock_write()
 
1696
        self.addCleanup(branch.unlock)
 
1697
        self.reset_smart_call_log()
 
1698
        config = branch._get_config()
 
1699
        value_dict = {'ascii': 'a', u'unicode \N{WATCH}': u'\N{INTERROBANG}'}
 
1700
        config.set_option(value_dict, 'name')
 
1701
        self.assertLength(10, self.hpss_calls)
 
1702
        self.assertEqual(value_dict, branch._get_config().get_option('name'))
 
1703
 
1667
1704
 
1668
1705
class TestBranchLockWrite(RemoteBranchTestCase):
1669
1706
 
1828
1865
class TestRepositoryFormat(TestRemoteRepository):
1829
1866
 
1830
1867
    def test_fast_delta(self):
1831
 
        true_name = groupcompress_repo.RepositoryFormatCHK1().network_name()
 
1868
        true_name = groupcompress_repo.RepositoryFormat2a().network_name()
1832
1869
        true_format = RemoteRepositoryFormat()
1833
1870
        true_format._network_name = true_name
1834
1871
        self.assertEqual(True, true_format.fast_deltas)
2257
2294
        self.setup_smart_server_with_call_log()
2258
2295
        tree = self.make_branch_and_memory_tree('.')
2259
2296
        tree.lock_write()
 
2297
        tree.add('')
2260
2298
        rev1 = tree.commit('First')
2261
2299
        rev2 = tree.commit('Second')
2262
2300
        tree.unlock()
2301
2339
        transport_path = 'quack'
2302
2340
        repo, client = self.setup_fake_client_and_repository(transport_path)
2303
2341
        client.add_success_response('ok', 'a token')
2304
 
        result = repo.lock_write()
 
2342
        token = repo.lock_write().repository_token
2305
2343
        self.assertEqual(
2306
2344
            [('call', 'Repository.lock_write', ('quack/', ''))],
2307
2345
            client._calls)
2308
 
        self.assertEqual('a token', result)
 
2346
        self.assertEqual('a token', token)
2309
2347
 
2310
2348
    def test_lock_write_already_locked(self):
2311
2349
        transport_path = 'quack'