~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart.py

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

• reverse the comparison (5200.3.7 to 6170)
• stop comparing with revision 6170

Show diffs side-by-side

added

removed

bzrlib/tests/test_smart.py

1

 

# Copyright (C) 2006-2011 Canonical Ltd

 

1

# Copyright (C) 2006-2010 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

25

25

"""

26

26

 

27

27

import bz2

 

28

from cStringIO import StringIO

 

29

import tarfile

28

30

 

29

31

from bzrlib import (

 

32

    bencode,

30

33

    branch as _mod_branch,

31

34

    bzrdir,

32

35

    errors,

 

36

    pack,

33

37

    tests,

34

38

    transport,

35

39

    urlutils,

85

89

            backing_transport = tests.TestCaseWithTransport.get_transport(self)

86

90

            self._chroot_server = chroot.ChrootServer(backing_transport)

87

91

            self.start_server(self._chroot_server)

88

 

        t = transport.get_transport_from_url(self._chroot_server.get_url())

 

92

        t = transport.get_transport(self._chroot_server.get_url())

89

93

        if relpath is not None:

90

94

            t = t.clone(relpath)

91

95

        return t

99

103

        # the default or a parameterized class, but rather use the

100

104

        # TestCaseWithTransport infrastructure to set up a smart server and

101

105

        # transport.

102

 

        self.overrideAttr(self, "transport_server", self.make_transport_server)

 

106

        self.transport_server = self.make_transport_server

103

107

 

104

108

    def make_transport_server(self):

105

109

        return test_server.SmartTCPServer_for_testing('-' + self.id())

798

802

        branch.unlock()

799

803

 

800

804

 

801

 

class TestSmartServerBranchRequestSetConfigOptionDict(TestLockedBranch):

802

 

 

803

 

    def setUp(self):

804

 

        TestLockedBranch.setUp(self)

805

 

        # A dict with non-ascii keys and values to exercise unicode

806

 

        # roundtripping.

807

 

        self.encoded_value_dict = (

808

 

            'd5:ascii1:a11:unicode \xe2\x8c\x9a3:\xe2\x80\xbde')

809

 

        self.value_dict = {

810

 

            'ascii': 'a', u'unicode \N{WATCH}': u'\N{INTERROBANG}'}

811

 

 

812

 

    def test_value_name(self):

813

 

        branch = self.make_branch('.')

814

 

        request = smart_branch.SmartServerBranchRequestSetConfigOptionDict(

815

 

            branch.bzrdir.root_transport)

816

 

        branch_token, repo_token = self.get_lock_tokens(branch)

817

 

        config = branch._get_config()

818

 

        result = request.execute('', branch_token, repo_token,

819

 

            self.encoded_value_dict, 'foo', '')

820

 

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)

821

 

        self.assertEqual(self.value_dict, config.get_option('foo'))

822

 

        # Cleanup

823

 

        branch.unlock()

824

 

 

825

 

    def test_value_name_section(self):

826

 

        branch = self.make_branch('.')

827

 

        request = smart_branch.SmartServerBranchRequestSetConfigOptionDict(

828

 

            branch.bzrdir.root_transport)

829

 

        branch_token, repo_token = self.get_lock_tokens(branch)

830

 

        config = branch._get_config()

831

 

        result = request.execute('', branch_token, repo_token,

832

 

            self.encoded_value_dict, 'foo', 'gam')

833

 

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)

834

 

        self.assertEqual(self.value_dict, config.get_option('foo', 'gam'))

835

 

        # Cleanup

836

 

        branch.unlock()

837

 

 

838

 

 

839

805

class TestSmartServerBranchRequestSetTagsBytes(TestLockedBranch):

840

806

    # Only called when the branch format and tags match [yay factory

841

807

    # methods] so only need to test straight forward cases.

932

898

        # its repository.

933

899

        self.make_tree_with_two_commits()

934

900

        rev_id_utf8 = u'\xc8'.encode('utf-8')

935

 

        self.tree.branch.set_last_revision_info(0, 'null:')

 

901

        self.tree.branch.set_revision_history([])

936

902

        self.assertEqual(

937

903

            (0, 'null:'), self.tree.branch.last_revision_info())

938

904

        # We can update the branch to a revision that is present in the

1470

1436

            request.execute('stacked', 1, (3, r3)))

1471

1437

 

1472

1438

 

1473

 

class GetStreamTestBase(tests.TestCaseWithMemoryTransport):

 

1439

class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):

1474

1440

 

1475

1441

    def make_two_commit_repo(self):

1476

1442

        tree = self.make_branch_and_memory_tree('.')

1482

1448

        repo = tree.branch.repository

1483

1449

        return repo, r1, r2

1484

1450

 

1485

 

 

1486

 

class TestSmartServerRepositoryGetStream(GetStreamTestBase):

1487

 

 

1488

1451

    def test_ancestry_of(self):

1489

1452

        """The search argument may be a 'ancestry-of' some heads'."""

1490

1453

        backing = self.get_transport()

1511

1474

        stream_bytes = ''.join(response.body_stream)

1512

1475

        self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')

1513

1476

 

1514

 

    def test_search_everything(self):

1515

 

        """A search of 'everything' returns a stream."""

1516

 

        backing = self.get_transport()

1517

 

        request = smart_repo.SmartServerRepositoryGetStream_1_19(backing)

1518

 

        repo, r1, r2 = self.make_two_commit_repo()

1519

 

        serialised_fetch_spec = 'everything'

1520

 

        request.execute('', repo._format.network_name())

1521

 

        response = request.do_body(serialised_fetch_spec)

1522

 

        self.assertEqual(('ok',), response.args)

1523

 

        stream_bytes = ''.join(response.body_stream)

1524

 

        self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')

1525

 

 

1526

1477

 

1527

1478

class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):

1528

1479

 

1922

1873

            smart_repo.SmartServerRepositoryGetRevisionGraph)

1923

1874

        self.assertHandlerEqual('Repository.get_stream',

1924

1875

            smart_repo.SmartServerRepositoryGetStream)

1925

 

        self.assertHandlerEqual('Repository.get_stream_1.19',

1926

 

            smart_repo.SmartServerRepositoryGetStream_1_19)

1927

1876

        self.assertHandlerEqual('Repository.has_revision',

1928

1877

            smart_repo.SmartServerRequestHasRevision)

1929

1878

        self.assertHandlerEqual('Repository.insert_stream',

1940

1889

            smart_repo.SmartServerRepositoryUnlock)

1941

1890

        self.assertHandlerEqual('Transport.is_readonly',

1942

1891

            smart_req.SmartServerIsReadonly)

1943

 

 

1944

 

 

1945

 

class SmartTCPServerHookTests(tests.TestCaseWithMemoryTransport):

1946

 

    """Tests for SmartTCPServer hooks."""

1947

 

 

1948

 

    def setUp(self):

1949

 

        super(SmartTCPServerHookTests, self).setUp()

1950

 

        self.server = server.SmartTCPServer(self.get_transport())

1951

 

 

1952

 

    def test_run_server_started_hooks(self):

1953

 

        """Test the server started hooks get fired properly."""

1954

 

        started_calls = []

1955

 

        server.SmartTCPServer.hooks.install_named_hook('server_started',

1956

 

            lambda backing_urls, url: started_calls.append((backing_urls, url)),

1957

 

            None)

1958

 

        started_ex_calls = []

1959

 

        server.SmartTCPServer.hooks.install_named_hook('server_started_ex',

1960

 

            lambda backing_urls, url: started_ex_calls.append((backing_urls, url)),

1961

 

            None)

1962

 

        self.server._sockname = ('example.com', 42)

1963

 

        self.server.run_server_started_hooks()

1964

 

        self.assertEquals(started_calls,

1965

 

            [([self.get_transport().base], 'bzr://example.com:42/')])

1966

 

        self.assertEquals(started_ex_calls,

1967

 

            [([self.get_transport().base], self.server)])

1968

 

 

1969

 

    def test_run_server_started_hooks_ipv6(self):

1970

 

        """Test that socknames can contain 4-tuples."""

1971

 

        self.server._sockname = ('::', 42, 0, 0)

1972

 

        started_calls = []

1973

 

        server.SmartTCPServer.hooks.install_named_hook('server_started',

1974

 

            lambda backing_urls, url: started_calls.append((backing_urls, url)),

1975

 

            None)

1976

 

        self.server.run_server_started_hooks()

1977

 

        self.assertEquals(started_calls,

1978

 

                [([self.get_transport().base], 'bzr://:::42/')])

1979

 

 

1980

 

    def test_run_server_stopped_hooks(self):

1981

 

        """Test the server stopped hooks."""

1982

 

        self.server._sockname = ('example.com', 42)

1983

 

        stopped_calls = []

1984

 

        server.SmartTCPServer.hooks.install_named_hook('server_stopped',

1985

 

            lambda backing_urls, url: stopped_calls.append((backing_urls, url)),

1986

 

            None)

1987

 

        self.server.run_server_stopped_hooks()

1988

 

        self.assertEquals(stopped_calls,

1989

 

            [([self.get_transport().base], 'bzr://example.com:42/')])

• Older »

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