/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_read_bundle.py

  • Committer: John Arbash Meinel
  • Date: 2006-07-18 18:57:54 UTC
  • mto: This revision was merged to the branch mainline in revision 1868.
  • Revision ID: john@arbash-meinel.com-20060718185754-4007745748e28db9
Commit timestamp restricted to 1ms precision.

The old code would restrict to 1s resolution if the timestamp was
supplied, while it preserved full resolution if the timestamp was
auto generated. Now both paths preserve only 1ms resolution.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_read_bundle.py
1
 
# Copyright (C) 2006 Canonical Ltd
 
1
# Copyright (C) 2006 by 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Test read_bundle works properly across various transports."""
18
18
 
23
23
from bzrlib.bundle.serializer import write_bundle
24
24
import bzrlib.bzrdir
25
25
import bzrlib.errors as errors
26
 
from bzrlib.symbol_versioning import deprecated_in
27
 
from bzrlib import tests
28
 
from bzrlib.tests.test_transport import TestTransportImplementation
29
 
from bzrlib.tests.per_transport import transport_test_permutations
 
26
from bzrlib.tests import TestCaseInTempDir
30
27
import bzrlib.transport
31
28
from bzrlib.transport.memory import MemoryTransport
32
29
import bzrlib.urlutils
33
30
 
34
31
 
35
 
def load_tests(standard_tests, module, loader):
36
 
    """Multiply tests for tranport implementations."""
37
 
    transport_tests, remaining_tests = tests.split_suite_by_condition(
38
 
        standard_tests, tests.condition_isinstance(TestReadBundleFromURL))
39
 
    return tests.multiply_tests(transport_tests, transport_test_permutations(),
40
 
        remaining_tests)
41
 
 
42
 
 
43
 
def create_bundle_file(test_case):
44
 
    test_case.build_tree(['tree/', 'tree/a', 'tree/subdir/'])
45
 
 
46
 
    format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
47
 
 
48
 
    bzrdir = format.initialize('tree')
49
 
    repo = bzrdir.create_repository()
50
 
    branch = repo.bzrdir.create_branch()
51
 
    wt = branch.bzrdir.create_workingtree()
52
 
 
53
 
    wt.add(['a', 'subdir/'])
54
 
    wt.commit('new project', rev_id='commit-1')
55
 
 
56
 
    out = cStringIO.StringIO()
57
 
    rev_ids = write_bundle(wt.branch.repository,
58
 
                           wt.get_parent_ids()[0], 'null:', out)
59
 
    out.seek(0)
60
 
    return out, wt
61
 
 
62
 
 
63
 
class TestDeprecations(tests.TestCaseInTempDir):
64
 
 
65
 
    def create_test_bundle(self):
66
 
        out, wt = create_bundle_file(self)
67
 
        f = open('test_bundle', 'wb')
68
 
        try:
69
 
            f.write(out.getvalue())
70
 
        finally:
71
 
            f.close()
72
 
        return wt
73
 
 
74
 
    def test_read_bundle_from_url_deprecated(self):
75
 
        wt = self.create_test_bundle()
76
 
        t = bzrlib.transport.get_transport(self.test_dir)
77
 
        url = t.abspath('test_bundle')
78
 
        self.callDeprecated([deprecated_in((1, 12, 0))
79
 
                             % 'bzrlib.bundle.read_bundle_from_url'],
80
 
                            bzrlib.bundle.read_bundle_from_url,
81
 
                            url)
82
 
 
83
 
 
84
 
class TestReadBundleFromURL(TestTransportImplementation):
 
32
# jam 20060615 Originally I thought I should use TestCaseWithTransport
 
33
#       but it turns out that doesn't play well with the transport adapter
 
34
class TestReadBundleFromURL(TestCaseInTempDir):
85
35
    """Test that read_bundle works properly across multiple transports"""
86
36
 
87
37
    def setUp(self):
88
38
        super(TestReadBundleFromURL, self).setUp()
89
 
        self.bundle_name = 'test_bundle'
90
 
        # read_mergeable_from_url will invoke get_transport which may *not*
91
 
        # respect self._transport (i.e. returns a transport that is different
92
 
        # from the one we want to test, so we must inject a correct transport
93
 
        # into possible_transports first).
94
 
        self.possible_transports = [self.get_transport(self.bundle_name)]
95
 
        self._captureVar('BZR_NO_SMART_VFS', None)
96
 
        wt = self.create_test_bundle()
 
39
        self._server = self.transport_server()
 
40
        self._server.setUp()
 
41
        self._transport = None
97
42
 
98
 
    def read_mergeable_from_url(self, url):
99
 
        return bzrlib.bundle.read_mergeable_from_url(
100
 
            url, possible_transports=self.possible_transports)
 
43
    def tearDown(self):
 
44
        self._transport = None
 
45
        self._server.tearDown()
 
46
        super(TestReadBundleFromURL, self).tearDown()
 
47
        
 
48
    def get_transport(self):
 
49
        """Return a connected transport to the local directory."""
 
50
        base_url = self._server.get_url()
 
51
        t = bzrlib.transport.get_transport(base_url)
 
52
        if not isinstance(t, self.transport_class):
 
53
            # we want to make sure to construct one particular class, even if
 
54
            # there are several available implementations of this transport;
 
55
            # therefore construct it by hand rather than through the regular
 
56
            # get_transport method
 
57
            t = self.transport_class(base_url)
 
58
        return t
101
59
 
102
60
    def get_url(self, relpath=''):
103
61
        return bzrlib.urlutils.join(self._server.get_url(), relpath)
104
62
 
105
63
    def create_test_bundle(self):
106
 
        out, wt = create_bundle_file(self)
107
 
        if self.get_transport().is_readonly():
108
 
            self.build_tree_contents([(self.bundle_name, out.getvalue())])
 
64
        # Can't use self.get_transport() because that asserts that 
 
65
        # it is not readonly, so just skip tests where the server is readonly
 
66
        self._transport = self.get_transport()
 
67
        #if isinstance(self._transport, MemoryTransport):
 
68
        #    return None
 
69
        self.build_tree(['tree/', 'tree/a', 'tree/subdir/'])
 
70
 
 
71
        format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
 
72
 
 
73
 
 
74
        bzrdir = format.initialize('tree')
 
75
        repo = bzrdir.create_repository()
 
76
        branch = repo.bzrdir.create_branch()
 
77
        wt = branch.bzrdir.create_workingtree()
 
78
 
 
79
        wt.add(['a', 'subdir/'])
 
80
        wt.commit('new project', rev_id='commit-1')
 
81
 
 
82
        out = cStringIO.StringIO()
 
83
        rev_ids = write_bundle(wt.branch.repository,
 
84
                               wt.last_revision(), None, out)
 
85
        out.seek(0)
 
86
        if self._transport.is_readonly():
 
87
            f = open('test_bundle', 'wb')
 
88
            f.write(out.getvalue())
 
89
            f.close()
109
90
        else:
110
 
            self.get_transport().put_file(self.bundle_name, out)
111
 
            self.log('Put to: %s', self.get_url(self.bundle_name))
 
91
            self._transport.put('test_bundle', out)
 
92
            self.log('Put to: %s', self.get_url('test_bundle'))
112
93
        return wt
113
94
 
114
 
    def test_read_mergeable_from_url(self):
115
 
        info = self.read_mergeable_from_url(
116
 
            unicode(self.get_url(self.bundle_name)))
117
 
        revision = info.real_revisions[-1]
118
 
        self.assertEqual('commit-1', revision.revision_id)
 
95
    def test_read_bundle_from_url(self):
 
96
        wt = self.create_test_bundle()
 
97
        if wt is None:
 
98
            return
 
99
 
 
100
        info = bzrlib.bundle.read_bundle_from_url(
 
101
                    self.get_url('test_bundle'))
 
102
        bundle_tree = info.revision_tree(wt.branch.repository, info.target)
 
103
        self.assertEqual('commit-1', bundle_tree.revision_id)
119
104
 
120
105
    def test_read_fail(self):
121
106
        # Trying to read from a directory, or non-bundle file
122
107
        # should fail with NotABundle
123
 
        self.assertRaises(errors.NotABundle,
124
 
                          self.read_mergeable_from_url, self.get_url('tree'))
125
 
        self.assertRaises(errors.NotABundle,
126
 
                          self.read_mergeable_from_url, self.get_url('tree/a'))
 
108
        wt = self.create_test_bundle()
 
109
        if wt is None:
 
110
            return
127
111
 
128
 
    def test_read_mergeable_respects_possible_transports(self):
129
 
        if not isinstance(self.get_transport(self.bundle_name),
130
 
                          bzrlib.transport.ConnectedTransport):
131
 
            # There is no point testing transport reuse for not connected
132
 
            # transports (the test will fail even).
133
 
            raise tests.TestSkipped(
134
 
                'Need a ConnectedTransport to test transport reuse')
135
 
        url = unicode(self.get_url(self.bundle_name))
136
 
        info = self.read_mergeable_from_url(url)
137
 
        self.assertEqual(1, len(self.possible_transports))
 
112
        self.assertRaises(errors.NotABundle, 
 
113
            bzrlib.bundle.read_bundle_from_url, 
 
114
            self.get_url('tree'))
 
115
        self.assertRaises(errors.NotABundle, 
 
116
            bzrlib.bundle.read_bundle_from_url, 
 
117
            self.get_url('tree/a'))