23
23
from bzrlib.bundle.serializer import write_bundle
24
24
import bzrlib.bzrdir
25
25
import bzrlib.errors as errors
26
from bzrlib.tests import TestCaseInTempDir
26
from bzrlib import tests
27
27
from bzrlib.tests.test_transport import TestTransportImplementation
28
from bzrlib.tests.test_transport_implementations import TransportTestProviderAdapter
28
29
import bzrlib.transport
29
30
from bzrlib.transport.memory import MemoryTransport
30
31
import bzrlib.urlutils
34
def load_tests(standard_tests, module, loader):
35
"""Multiply tests for tranport implementations."""
36
result = loader.suiteClass()
37
adapter = TransportTestProviderAdapter()
38
for test in tests.iter_suite_tests(standard_tests):
39
result.addTests(adapter.adapt(test))
43
def create_bundle_file(test_case):
44
test_case.build_tree(['tree/', 'tree/a', 'tree/subdir/'])
46
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
48
bzrdir = format.initialize('tree')
49
repo = bzrdir.create_repository()
50
branch = repo.bzrdir.create_branch()
51
wt = branch.bzrdir.create_workingtree()
53
wt.add(['a', 'subdir/'])
54
wt.commit('new project', rev_id='commit-1')
56
out = cStringIO.StringIO()
57
rev_ids = write_bundle(wt.branch.repository,
58
wt.get_parent_ids()[0], 'null:', out)
33
63
class TestReadBundleFromURL(TestTransportImplementation):
34
64
"""Test that read_bundle works properly across multiple transports"""
37
67
return bzrlib.urlutils.join(self._server.get_url(), relpath)
39
69
def create_test_bundle(self):
40
# Can't use self.get_transport() because that asserts that
41
# it is not readonly, so just skip tests where the server is readonly
42
self._transport = self.get_transport()
43
#if isinstance(self._transport, MemoryTransport):
45
self.build_tree(['tree/', 'tree/a', 'tree/subdir/'])
47
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
50
bzrdir = format.initialize('tree')
51
repo = bzrdir.create_repository()
52
branch = repo.bzrdir.create_branch()
53
wt = branch.bzrdir.create_workingtree()
55
wt.add(['a', 'subdir/'])
56
wt.commit('new project', rev_id='commit-1')
58
out = cStringIO.StringIO()
59
rev_ids = write_bundle(wt.branch.repository,
60
wt.last_revision(), None, out)
62
if self._transport.is_readonly():
70
out, wt = create_bundle_file(self)
71
if self.get_transport().is_readonly():
63
72
f = open('test_bundle', 'wb')
64
73
f.write(out.getvalue())
67
self._transport.put('test_bundle', out)
76
self.get_transport().put_file('test_bundle', out)
68
77
self.log('Put to: %s', self.get_url('test_bundle'))
71
80
def test_read_bundle_from_url(self):
81
self._captureVar('BZR_NO_SMART_VFS', None)
72
82
wt = self.create_test_bundle()
76
85
info = bzrlib.bundle.read_bundle_from_url(
77
self.get_url('test_bundle'))
78
bundle_tree = info.revision_tree(wt.branch.repository, info.target)
79
self.assertEqual('commit-1', bundle_tree.revision_id)
86
unicode(self.get_url('test_bundle')))
87
revision = info.real_revisions[-1]
88
self.assertEqual('commit-1', revision.revision_id)
81
90
def test_read_fail(self):
82
91
# Trying to read from a directory, or non-bundle file
83
92
# should fail with NotABundle
93
self._captureVar('BZR_NO_SMART_VFS', None)
84
94
wt = self.create_test_bundle()
91
101
self.assertRaises(errors.NotABundle,
92
102
bzrlib.bundle.read_bundle_from_url,
93
103
self.get_url('tree/a'))
105
def test_read_mergeable_populates_possible_transports(self):
106
self._captureVar('BZR_NO_SMART_VFS', None)
107
wt = self.create_test_bundle()
110
possible_transports = []
111
url = unicode(self.get_url('test_bundle'))
112
info = bzrlib.bundle.read_mergeable_from_url(url,
113
possible_transports=possible_transports)
114
self.assertEqual(1, len(possible_transports))