14
14
from bzrlib.bzrdir import (
19
from bzrlib.controldir import ControlDirFormat
20
from bzrlib.errors import NotBranchError
21
18
from bzrlib.repofmt.knitpack_repo import RepositoryFormatKnitPack1
22
19
from bzrlib.tests import TestCaseWithTransport
23
20
from bzrlib.transport import chroot
58
55
self._reference_values[references[i]] = references[i + 1]
59
56
self.follow_reference_calls = []
61
def followReference(self, url):
58
def followReference(self, url, open_dir=None):
62
59
self.follow_reference_calls.append(url)
63
60
return self._reference_values[url]
71
68
def testCheckInitialURL(self):
72
69
# checkSource rejects all URLs that are not allowed.
73
70
opener = self.makeBranchOpener(None, [], set(['a']))
75
BadUrl, opener.checkAndFollowBranchReference, 'a')
71
self.assertRaises(BadUrl, opener.checkAndFollowBranchReference, 'a')
77
73
def testNotReference(self):
78
74
# When branch references are forbidden, checkAndFollowBranchReference
79
75
# does not raise on non-references.
80
76
opener = self.makeBranchOpener(False, ['a', None])
82
'a', opener.checkAndFollowBranchReference('a'))
77
self.assertEquals('a', opener.checkAndFollowBranchReference('a'))
83
78
self.assertEquals(['a'], opener.follow_reference_calls)
85
80
def testBranchReferenceForbidden(self):
97
92
# is allowed and the source URL points to a branch reference to a
98
93
# permitted location.
99
94
opener = self.makeBranchOpener(True, ['a', 'b', None])
101
'b', opener.checkAndFollowBranchReference('a'))
95
self.assertEquals('b', opener.checkAndFollowBranchReference('a'))
102
96
self.assertEquals(['a', 'b'], opener.follow_reference_calls)
104
98
def testCheckReferencedURLs(self):
107
101
opener = self.makeBranchOpener(
108
102
True, ['a', 'b', None], unsafe_urls=set('b'))
110
BadUrl, opener.checkAndFollowBranchReference, 'a')
103
self.assertRaises(BadUrl, opener.checkAndFollowBranchReference, 'a')
111
104
self.assertEquals(['a'], opener.follow_reference_calls)
113
106
def testSelfReferencingBranch(self):
130
123
self.assertEquals(['a', 'b'], opener.follow_reference_calls)
133
class TrackingProber(BzrProber):
134
"""Subclass of BzrProber which tracks URLs it has been asked to open."""
139
def probe_transport(klass, transport):
140
klass.seen_urls.append(transport.base)
141
return BzrProber.probe_transport(transport)
144
126
class TestSafeBranchOpenerStacking(TestCaseWithTransport):
147
129
super(TestSafeBranchOpenerStacking, self).setUp()
148
130
SafeBranchOpener.install_hook()
150
def makeBranchOpener(self, allowed_urls, probers=None):
132
def makeBranchOpener(self, allowed_urls):
151
133
policy = WhitelistPolicy(True, allowed_urls, True)
152
return SafeBranchOpener(policy, probers)
134
return SafeBranchOpener(policy)
154
136
def makeBranch(self, path, branch_format, repository_format):
155
137
"""Make a Bazaar branch at 'path' with the given formats."""
159
141
repository_format.initialize(bzrdir)
160
142
return bzrdir.create_branch()
162
def testProbers(self):
163
# Only the specified probers should be used
164
b = self.make_branch('branch')
165
opener = self.makeBranchOpener([b.base], probers=[])
166
self.assertRaises(NotBranchError, opener.open, b.base)
167
opener = self.makeBranchOpener([b.base], probers=[BzrProber])
168
self.assertEquals(b.base, opener.open(b.base).base)
170
def testDefaultProbers(self):
171
# If no probers are specified to the constructor
172
# of SafeBranchOpener, then a safe set will be used,
173
# rather than all probers registered in bzr.
174
self.addCleanup(ControlDirFormat.unregister_prober, TrackingProber)
175
ControlDirFormat.register_prober(TrackingProber)
176
# Open a location without any branches, so that all probers are
178
# First, check that the TrackingProber tracks correctly.
179
TrackingProber.seen_urls = []
180
opener = self.makeBranchOpener(["."], probers=[TrackingProber])
181
self.assertRaises(NotBranchError, opener.open, ".")
182
self.assertEquals(1, len(TrackingProber.seen_urls))
183
TrackingProber.seen_urls = []
184
# And make sure it's registered in such a way that BzrDir.open would
186
self.assertRaises(NotBranchError, BzrDir.open, ".")
187
self.assertEquals(1, len(TrackingProber.seen_urls))
188
TrackingProber.seen_urls = []
189
# Make sure that SafeBranchOpener doesn't use it if no
190
# probers were specified
191
opener = self.makeBranchOpener(["."])
192
self.assertRaises(NotBranchError, opener.open, ".")
193
self.assertEquals(0, len(TrackingProber.seen_urls))
195
144
def testAllowedURL(self):
196
145
# checkSource does not raise an exception for branches stacked on
197
146
# branches with allowed URLs.
286
235
a = self.make_branch('a', format='2a')
287
236
b = self.make_branch('b', format='2a')
288
237
b.set_stacked_on_url(a.base)
290
TrackingProber.seen_urls = []
291
opener = self.makeBranchOpener(
292
[a.base, b.base], probers=[TrackingProber])
295
set(TrackingProber.seen_urls), set([b.base, a.base]))
242
return BzrDir.open(url)
244
opener = self.makeBranchOpener([a.base, b.base])
245
opener.open(b.base, open_dir=open_dir)
246
self.assertEquals(seen_urls, set([b.base, a.base]))
297
248
def testCustomOpenerWithBranchReference(self):
298
249
# A custom function for opening a control dir can be specified.
299
250
a = self.make_branch('a', format='2a')
300
251
b_dir = self.make_bzrdir('b')
301
252
b = BranchReferenceFormat().initialize(b_dir, target_branch=a)
302
TrackingProber.seen_urls = []
303
opener = self.makeBranchOpener(
304
[a.base, b.base], probers=[TrackingProber])
307
set(TrackingProber.seen_urls), set([b.base, a.base]))
257
return BzrDir.open(url)
259
opener = self.makeBranchOpener([a.base, b.base])
260
opener.open(b.base, open_dir=open_dir)
261
self.assertEquals(seen_urls, set([b.base, a.base]))
310
264
class TestSafeOpen(TestCaseWithTransport):
added
removed
lib/lp/codehosting/tests/test_safe_open.py
