1
# Copyright (C) 2005-2011, 2015, 2016 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
31
from ..directory_service import directories
32
from ..sixish import (
35
from ..transport import (
45
import breezy.transport.trace
52
# TODO: Should possibly split transport-specific tests into their own files.
55
class TestTransport(tests.TestCase):
56
"""Test the non transport-concrete class functionality."""
58
def test__get_set_protocol_handlers(self):
59
handlers = transport._get_protocol_handlers()
60
self.assertNotEqual([], handlers.keys())
61
transport._clear_protocol_handlers()
62
self.addCleanup(transport._set_protocol_handlers, handlers)
63
self.assertEqual([], transport._get_protocol_handlers().keys())
65
def test_get_transport_modules(self):
66
handlers = transport._get_protocol_handlers()
67
self.addCleanup(transport._set_protocol_handlers, handlers)
68
# don't pollute the current handlers
69
transport._clear_protocol_handlers()
71
class SampleHandler(object):
72
"""I exist, isnt that enough?"""
73
transport._clear_protocol_handlers()
74
transport.register_transport_proto('foo')
75
transport.register_lazy_transport('foo',
76
'breezy.tests.test_transport',
77
'TestTransport.SampleHandler')
78
transport.register_transport_proto('bar')
79
transport.register_lazy_transport('bar',
80
'breezy.tests.test_transport',
81
'TestTransport.SampleHandler')
82
self.assertEqual([SampleHandler.__module__,
83
'breezy.transport.chroot',
84
'breezy.transport.pathfilter'],
85
transport._get_transport_modules())
87
def test_transport_dependency(self):
88
"""Transport with missing dependency causes no error"""
89
saved_handlers = transport._get_protocol_handlers()
90
self.addCleanup(transport._set_protocol_handlers, saved_handlers)
91
# don't pollute the current handlers
92
transport._clear_protocol_handlers()
93
transport.register_transport_proto('foo')
94
transport.register_lazy_transport(
95
'foo', 'breezy.tests.test_transport', 'BadTransportHandler')
97
transport.get_transport_from_url('foo://fooserver/foo')
98
except errors.UnsupportedProtocol as e:
99
self.assertEqual('Unsupported protocol'
100
' for url "foo://fooserver/foo":'
101
' Unable to import library "some_lib":'
102
' testing missing dependency', str(e))
104
self.fail('Did not raise UnsupportedProtocol')
106
def test_transport_fallback(self):
107
"""Transport with missing dependency causes no error"""
108
saved_handlers = transport._get_protocol_handlers()
109
self.addCleanup(transport._set_protocol_handlers, saved_handlers)
110
transport._clear_protocol_handlers()
111
transport.register_transport_proto('foo')
112
transport.register_lazy_transport(
113
'foo', 'breezy.tests.test_transport', 'BackupTransportHandler')
114
transport.register_lazy_transport(
115
'foo', 'breezy.tests.test_transport', 'BadTransportHandler')
116
t = transport.get_transport_from_url('foo://fooserver/foo')
117
self.assertTrue(isinstance(t, BackupTransportHandler))
119
def test_ssh_hints(self):
120
"""Transport ssh:// should raise an error pointing out bzr+ssh://"""
122
transport.get_transport_from_url('ssh://fooserver/foo')
123
except errors.UnsupportedProtocol as e:
124
self.assertEqual('Unsupported protocol'
125
' for url "ssh://fooserver/foo":'
126
' bzr supports bzr+ssh to operate over ssh,'
127
' use "bzr+ssh://fooserver/foo".',
130
self.fail('Did not raise UnsupportedProtocol')
132
def test_LateReadError(self):
133
"""The LateReadError helper should raise on read()."""
134
a_file = transport.LateReadError('a path')
137
except errors.ReadError as error:
138
self.assertEqual('a path', error.path)
139
self.assertRaises(errors.ReadError, a_file.read, 40)
142
def test_local_abspath_non_local_transport(self):
143
# the base implementation should throw
144
t = memory.MemoryTransport()
145
e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
146
self.assertEqual('memory:///t is not a local path.', str(e))
149
class TestCoalesceOffsets(tests.TestCase):
151
def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
152
coalesce = transport.Transport._coalesce_offsets
153
exp = [transport._CoalescedOffset(*x) for x in expected]
154
out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
156
self.assertEqual(exp, out)
158
def test_coalesce_empty(self):
161
def test_coalesce_simple(self):
162
self.check([(0, 10, [(0, 10)])], [(0, 10)])
164
def test_coalesce_unrelated(self):
165
self.check([(0, 10, [(0, 10)]),
167
], [(0, 10), (20, 10)])
169
def test_coalesce_unsorted(self):
170
self.check([(20, 10, [(0, 10)]),
172
], [(20, 10), (0, 10)])
174
def test_coalesce_nearby(self):
175
self.check([(0, 20, [(0, 10), (10, 10)])],
178
def test_coalesce_overlapped(self):
179
self.assertRaises(ValueError,
180
self.check, [(0, 15, [(0, 10), (5, 10)])],
183
def test_coalesce_limit(self):
184
self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
185
(30, 10), (40, 10)]),
186
(60, 50, [(0, 10), (10, 10), (20, 10),
187
(30, 10), (40, 10)]),
188
], [(10, 10), (20, 10), (30, 10), (40, 10),
189
(50, 10), (60, 10), (70, 10), (80, 10),
190
(90, 10), (100, 10)],
193
def test_coalesce_no_limit(self):
194
self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
195
(30, 10), (40, 10), (50, 10),
196
(60, 10), (70, 10), (80, 10),
198
], [(10, 10), (20, 10), (30, 10), (40, 10),
199
(50, 10), (60, 10), (70, 10), (80, 10),
200
(90, 10), (100, 10)])
202
def test_coalesce_fudge(self):
203
self.check([(10, 30, [(0, 10), (20, 10)]),
204
(100, 10, [(0, 10)]),
205
], [(10, 10), (30, 10), (100, 10)],
208
def test_coalesce_max_size(self):
209
self.check([(10, 20, [(0, 10), (10, 10)]),
211
# If one range is above max_size, it gets its own coalesced
213
(100, 80, [(0, 80)]), ],
214
[(10, 10), (20, 10), (30, 50), (100, 80)],
217
def test_coalesce_no_max_size(self):
218
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
219
[(10, 10), (20, 10), (30, 50), (80, 100)],
222
def test_coalesce_default_limit(self):
223
# By default we use a 100MB max size.
224
ten_mb = 10 * 1024 * 1024
226
[(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
227
(10 * ten_mb, ten_mb, [(0, ten_mb)])],
228
[(i * ten_mb, ten_mb) for i in range(11)])
230
[(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
231
[(i * ten_mb, ten_mb) for i in range(11)],
232
max_size=1 * 1024 * 1024 * 1024)
235
class TestMemoryServer(tests.TestCase):
237
def test_create_server(self):
238
server = memory.MemoryServer()
239
server.start_server()
240
url = server.get_url()
241
self.assertTrue(url in transport.transport_list_registry)
242
t = transport.get_transport_from_url(url)
245
self.assertFalse(url in transport.transport_list_registry)
246
self.assertRaises(errors.UnsupportedProtocol,
247
transport.get_transport, url)
250
class TestMemoryTransport(tests.TestCase):
252
def test_get_transport(self):
253
memory.MemoryTransport()
255
def test_clone(self):
256
t = memory.MemoryTransport()
257
self.assertTrue(isinstance(t, memory.MemoryTransport))
258
self.assertEqual("memory:///", t.clone("/").base)
260
def test_abspath(self):
261
t = memory.MemoryTransport()
262
self.assertEqual("memory:///relpath", t.abspath('relpath'))
264
def test_abspath_of_root(self):
265
t = memory.MemoryTransport()
266
self.assertEqual("memory:///", t.base)
267
self.assertEqual("memory:///", t.abspath('/'))
269
def test_abspath_of_relpath_starting_at_root(self):
270
t = memory.MemoryTransport()
271
self.assertEqual("memory:///foo", t.abspath('/foo'))
273
def test_append_and_get(self):
274
t = memory.MemoryTransport()
275
t.append_bytes('path', b'content')
276
self.assertEqual(t.get('path').read(), b'content')
277
t.append_file('path', BytesIO(b'content'))
278
with t.get('path') as f:
279
self.assertEqual(f.read(), b'contentcontent')
281
def test_put_and_get(self):
282
t = memory.MemoryTransport()
283
t.put_file('path', BytesIO(b'content'))
284
self.assertEqual(t.get('path').read(), b'content')
285
t.put_bytes('path', b'content')
286
self.assertEqual(t.get('path').read(), b'content')
288
def test_append_without_dir_fails(self):
289
t = memory.MemoryTransport()
290
self.assertRaises(errors.NoSuchFile,
291
t.append_bytes, 'dir/path', b'content')
293
def test_put_without_dir_fails(self):
294
t = memory.MemoryTransport()
295
self.assertRaises(errors.NoSuchFile,
296
t.put_file, 'dir/path', BytesIO(b'content'))
298
def test_get_missing(self):
299
transport = memory.MemoryTransport()
300
self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
302
def test_has_missing(self):
303
t = memory.MemoryTransport()
304
self.assertEqual(False, t.has('foo'))
306
def test_has_present(self):
307
t = memory.MemoryTransport()
308
t.append_bytes('foo', b'content')
309
self.assertEqual(True, t.has('foo'))
311
def test_list_dir(self):
312
t = memory.MemoryTransport()
313
t.put_bytes('foo', b'content')
315
t.put_bytes('dir/subfoo', b'content')
316
t.put_bytes('dirlike', b'content')
318
self.assertEqual(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
319
self.assertEqual(['subfoo'], sorted(t.list_dir('dir')))
321
def test_mkdir(self):
322
t = memory.MemoryTransport()
324
t.append_bytes('dir/path', b'content')
325
with t.get('dir/path') as f:
326
self.assertEqual(f.read(), b'content')
328
def test_mkdir_missing_parent(self):
329
t = memory.MemoryTransport()
330
self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
332
def test_mkdir_twice(self):
333
t = memory.MemoryTransport()
335
self.assertRaises(errors.FileExists, t.mkdir, 'dir')
337
def test_parameters(self):
338
t = memory.MemoryTransport()
339
self.assertEqual(True, t.listable())
340
self.assertEqual(False, t.is_readonly())
342
def test_iter_files_recursive(self):
343
t = memory.MemoryTransport()
345
t.put_bytes('dir/foo', b'content')
346
t.put_bytes('dir/bar', b'content')
347
t.put_bytes('bar', b'content')
348
paths = set(t.iter_files_recursive())
349
self.assertEqual({'dir/foo', 'dir/bar', 'bar'}, paths)
352
t = memory.MemoryTransport()
353
t.put_bytes('foo', b'content')
354
t.put_bytes('bar', b'phowar')
355
self.assertEqual(7, t.stat('foo').st_size)
356
self.assertEqual(6, t.stat('bar').st_size)
359
class ChrootDecoratorTransportTest(tests.TestCase):
360
"""Chroot decoration specific tests."""
362
def test_abspath(self):
363
# The abspath is always relative to the chroot_url.
364
server = chroot.ChrootServer(
365
transport.get_transport_from_url('memory:///foo/bar/'))
366
self.start_server(server)
367
t = transport.get_transport_from_url(server.get_url())
368
self.assertEqual(server.get_url(), t.abspath('/'))
370
subdir_t = t.clone('subdir')
371
self.assertEqual(server.get_url(), subdir_t.abspath('/'))
373
def test_clone(self):
374
server = chroot.ChrootServer(
375
transport.get_transport_from_url('memory:///foo/bar/'))
376
self.start_server(server)
377
t = transport.get_transport_from_url(server.get_url())
378
# relpath from root and root path are the same
379
relpath_cloned = t.clone('foo')
380
abspath_cloned = t.clone('/foo')
381
self.assertEqual(server, relpath_cloned.server)
382
self.assertEqual(server, abspath_cloned.server)
384
def test_chroot_url_preserves_chroot(self):
385
"""Calling get_transport on a chroot transport's base should produce a
386
transport with exactly the same behaviour as the original chroot
389
This is so that it is not possible to escape a chroot by doing::
390
url = chroot_transport.base
391
parent_url = urlutils.join(url, '..')
392
new_t = transport.get_transport_from_url(parent_url)
394
server = chroot.ChrootServer(
395
transport.get_transport_from_url('memory:///path/subpath'))
396
self.start_server(server)
397
t = transport.get_transport_from_url(server.get_url())
398
new_t = transport.get_transport_from_url(t.base)
399
self.assertEqual(t.server, new_t.server)
400
self.assertEqual(t.base, new_t.base)
402
def test_urljoin_preserves_chroot(self):
403
"""Using urlutils.join(url, '..') on a chroot URL should not produce a
404
URL that escapes the intended chroot.
406
This is so that it is not possible to escape a chroot by doing::
407
url = chroot_transport.base
408
parent_url = urlutils.join(url, '..')
409
new_t = transport.get_transport_from_url(parent_url)
411
server = chroot.ChrootServer(
412
transport.get_transport_from_url('memory:///path/'))
413
self.start_server(server)
414
t = transport.get_transport_from_url(server.get_url())
416
urlutils.InvalidURLJoin, urlutils.join, t.base, '..')
419
class TestChrootServer(tests.TestCase):
421
def test_construct(self):
422
backing_transport = memory.MemoryTransport()
423
server = chroot.ChrootServer(backing_transport)
424
self.assertEqual(backing_transport, server.backing_transport)
426
def test_setUp(self):
427
backing_transport = memory.MemoryTransport()
428
server = chroot.ChrootServer(backing_transport)
429
server.start_server()
430
self.addCleanup(server.stop_server)
431
self.assertTrue(server.scheme
432
in transport._get_protocol_handlers().keys())
434
def test_stop_server(self):
435
backing_transport = memory.MemoryTransport()
436
server = chroot.ChrootServer(backing_transport)
437
server.start_server()
439
self.assertFalse(server.scheme
440
in transport._get_protocol_handlers().keys())
442
def test_get_url(self):
443
backing_transport = memory.MemoryTransport()
444
server = chroot.ChrootServer(backing_transport)
445
server.start_server()
446
self.addCleanup(server.stop_server)
447
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
450
class TestHooks(tests.TestCase):
451
"""Basic tests for transport hooks"""
453
def _get_connected_transport(self):
454
return transport.ConnectedTransport("bogus:nowhere")
456
def test_transporthooks_initialisation(self):
457
"""Check all expected transport hook points are set up"""
458
hookpoint = transport.TransportHooks()
459
self.assertTrue("post_connect" in hookpoint,
460
"post_connect not in %s" % (hookpoint,))
462
def test_post_connect(self):
463
"""Ensure the post_connect hook is called when _set_transport is"""
465
transport.Transport.hooks.install_named_hook("post_connect",
467
t = self._get_connected_transport()
468
self.assertLength(0, calls)
469
t._set_connection("connection", "auth")
470
self.assertEqual(calls, [t])
473
class PathFilteringDecoratorTransportTest(tests.TestCase):
474
"""Pathfilter decoration specific tests."""
476
def test_abspath(self):
477
# The abspath is always relative to the base of the backing transport.
478
server = pathfilter.PathFilteringServer(
479
transport.get_transport_from_url('memory:///foo/bar/'),
481
server.start_server()
482
t = transport.get_transport_from_url(server.get_url())
483
self.assertEqual(server.get_url(), t.abspath('/'))
485
subdir_t = t.clone('subdir')
486
self.assertEqual(server.get_url(), subdir_t.abspath('/'))
489
def make_pf_transport(self, filter_func=None):
490
"""Make a PathFilteringTransport backed by a MemoryTransport.
492
:param filter_func: by default this will be a no-op function. Use this
493
parameter to override it."""
494
if filter_func is None:
497
server = pathfilter.PathFilteringServer(
498
transport.get_transport_from_url('memory:///foo/bar/'),
500
server.start_server()
501
self.addCleanup(server.stop_server)
502
return transport.get_transport_from_url(server.get_url())
504
def test__filter(self):
505
# _filter (with an identity func as filter_func) always returns
506
# paths relative to the base of the backing transport.
507
t = self.make_pf_transport()
508
self.assertEqual('foo', t._filter('foo'))
509
self.assertEqual('foo/bar', t._filter('foo/bar'))
510
self.assertEqual('', t._filter('..'))
511
self.assertEqual('', t._filter('/'))
512
# The base of the pathfiltering transport is taken into account too.
513
t = t.clone('subdir1/subdir2')
514
self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
515
self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
516
self.assertEqual('subdir1', t._filter('..'))
517
self.assertEqual('', t._filter('/'))
519
def test_filter_invocation(self):
523
filter_log.append(path)
525
t = self.make_pf_transport(filter)
527
self.assertEqual(['abc'], filter_log)
529
t.clone('abc').has('xyz')
530
self.assertEqual(['abc/xyz'], filter_log)
533
self.assertEqual(['abc'], filter_log)
535
def test_clone(self):
536
t = self.make_pf_transport()
537
# relpath from root and root path are the same
538
relpath_cloned = t.clone('foo')
539
abspath_cloned = t.clone('/foo')
540
self.assertEqual(t.server, relpath_cloned.server)
541
self.assertEqual(t.server, abspath_cloned.server)
543
def test_url_preserves_pathfiltering(self):
544
"""Calling get_transport on a pathfiltered transport's base should
545
produce a transport with exactly the same behaviour as the original
546
pathfiltered transport.
548
This is so that it is not possible to escape (accidentally or
549
otherwise) the filtering by doing::
550
url = filtered_transport.base
551
parent_url = urlutils.join(url, '..')
552
new_t = transport.get_transport_from_url(parent_url)
554
t = self.make_pf_transport()
555
new_t = transport.get_transport_from_url(t.base)
556
self.assertEqual(t.server, new_t.server)
557
self.assertEqual(t.base, new_t.base)
560
class ReadonlyDecoratorTransportTest(tests.TestCase):
561
"""Readonly decoration specific tests."""
563
def test_local_parameters(self):
564
# connect to . in readonly mode
565
t = readonly.ReadonlyTransportDecorator('readonly+.')
566
self.assertEqual(True, t.listable())
567
self.assertEqual(True, t.is_readonly())
569
def test_http_parameters(self):
570
from breezy.tests.http_server import HttpServer
571
# connect to '.' via http which is not listable
572
server = HttpServer()
573
self.start_server(server)
574
t = transport.get_transport_from_url('readonly+' + server.get_url())
575
self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
576
self.assertEqual(False, t.listable())
577
self.assertEqual(True, t.is_readonly())
580
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
581
"""NFS decorator specific tests."""
583
def get_nfs_transport(self, url):
584
# connect to url with nfs decoration
585
return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
587
def test_local_parameters(self):
588
# the listable and is_readonly parameters
589
# are not changed by the fakenfs decorator
590
t = self.get_nfs_transport('.')
591
self.assertEqual(True, t.listable())
592
self.assertEqual(False, t.is_readonly())
594
def test_http_parameters(self):
595
# the listable and is_readonly parameters
596
# are not changed by the fakenfs decorator
597
from breezy.tests.http_server import HttpServer
598
# connect to '.' via http which is not listable
599
server = HttpServer()
600
self.start_server(server)
601
t = self.get_nfs_transport(server.get_url())
602
self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
603
self.assertEqual(False, t.listable())
604
self.assertEqual(True, t.is_readonly())
606
def test_fakenfs_server_default(self):
607
# a FakeNFSServer() should bring up a local relpath server for itself
608
server = test_server.FakeNFSServer()
609
self.start_server(server)
610
# the url should be decorated appropriately
611
self.assertStartsWith(server.get_url(), 'fakenfs+')
612
# and we should be able to get a transport for it
613
t = transport.get_transport_from_url(server.get_url())
614
# which must be a FakeNFSTransportDecorator instance.
615
self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
617
def test_fakenfs_rename_semantics(self):
618
# a FakeNFS transport must mangle the way rename errors occur to
619
# look like NFS problems.
620
t = self.get_nfs_transport('.')
621
self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
623
self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
626
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
627
"""Tests for simulation of VFAT restrictions"""
629
def get_vfat_transport(self, url):
630
"""Return vfat-backed transport for test directory"""
631
from breezy.transport.fakevfat import FakeVFATTransportDecorator
632
return FakeVFATTransportDecorator('vfat+' + url)
634
def test_transport_creation(self):
635
from breezy.transport.fakevfat import FakeVFATTransportDecorator
636
t = self.get_vfat_transport('.')
637
self.assertIsInstance(t, FakeVFATTransportDecorator)
639
def test_transport_mkdir(self):
640
t = self.get_vfat_transport('.')
642
self.assertTrue(t.has('hello'))
643
self.assertTrue(t.has('Hello'))
645
def test_forbidden_chars(self):
646
t = self.get_vfat_transport('.')
647
self.assertRaises(ValueError, t.has, "<NU>")
650
class BadTransportHandler(transport.Transport):
651
def __init__(self, base_url):
652
raise errors.DependencyNotPresent('some_lib',
653
'testing missing dependency')
656
class BackupTransportHandler(transport.Transport):
657
"""Test transport that works as a backup for the BadTransportHandler"""
661
class TestTransportImplementation(tests.TestCaseInTempDir):
662
"""Implementation verification for transports.
664
To verify a transport we need a server factory, which is a callable
665
that accepts no parameters and returns an implementation of
666
breezy.transport.Server.
668
That Server is then used to construct transport instances and test
669
the transport via loopback activity.
671
Currently this assumes that the Transport object is connected to the
672
current working directory. So that whatever is done
673
through the transport, should show up in the working
674
directory, and vice-versa. This is a bug, because its possible to have
675
URL schemes which provide access to something that may not be
676
result in storage on the local disk, i.e. due to file system limits, or
677
due to it being a database or some other non-filesystem tool.
679
This also tests to make sure that the functions work with both
680
generators and lists (assuming iter(list) is effectively a generator)
684
super(TestTransportImplementation, self).setUp()
685
self._server = self.transport_server()
686
self.start_server(self._server)
688
def get_transport(self, relpath=None):
689
"""Return a connected transport to the local directory.
691
:param relpath: a path relative to the base url.
693
base_url = self._server.get_url()
694
url = self._adjust_url(base_url, relpath)
695
# try getting the transport via the regular interface:
696
t = transport.get_transport_from_url(url)
697
# vila--20070607 if the following are commented out the test suite
698
# still pass. Is this really still needed or was it a forgotten
700
if not isinstance(t, self.transport_class):
701
# we did not get the correct transport class type. Override the
702
# regular connection behaviour by direct construction.
703
t = self.transport_class(url)
707
class TestTransportFromPath(tests.TestCaseInTempDir):
709
def test_with_path(self):
710
t = transport.get_transport_from_path(self.test_dir)
711
self.assertIsInstance(t, local.LocalTransport)
712
self.assertEqual(t.base.rstrip("/"),
713
urlutils.local_path_to_url(self.test_dir))
715
def test_with_url(self):
716
t = transport.get_transport_from_path("file:")
717
self.assertIsInstance(t, local.LocalTransport)
720
urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
723
class TestTransportFromUrl(tests.TestCaseInTempDir):
725
def test_with_path(self):
726
self.assertRaises(urlutils.InvalidURL, transport.get_transport_from_url,
729
def test_with_url(self):
730
url = urlutils.local_path_to_url(self.test_dir)
731
t = transport.get_transport_from_url(url)
732
self.assertIsInstance(t, local.LocalTransport)
733
self.assertEqual(t.base.rstrip("/"), url)
735
def test_with_url_and_segment_parameters(self):
736
url = urlutils.local_path_to_url(self.test_dir) + ",branch=foo"
737
t = transport.get_transport_from_url(url)
738
self.assertIsInstance(t, local.LocalTransport)
739
self.assertEqual(t.base.rstrip("/"), url)
740
with open(os.path.join(self.test_dir, "afile"), 'w') as f:
742
self.assertTrue(t.has("afile"))
745
class TestLocalTransports(tests.TestCase):
747
def test_get_transport_from_abspath(self):
748
here = osutils.abspath('.')
749
t = transport.get_transport(here)
750
self.assertIsInstance(t, local.LocalTransport)
751
self.assertEqual(t.base, urlutils.local_path_to_url(here) + '/')
753
def test_get_transport_from_relpath(self):
754
t = transport.get_transport('.')
755
self.assertIsInstance(t, local.LocalTransport)
756
self.assertEqual(t.base, urlutils.local_path_to_url('.') + '/')
758
def test_get_transport_from_local_url(self):
759
here = osutils.abspath('.')
760
here_url = urlutils.local_path_to_url(here) + '/'
761
t = transport.get_transport(here_url)
762
self.assertIsInstance(t, local.LocalTransport)
763
self.assertEqual(t.base, here_url)
765
def test_local_abspath(self):
766
here = osutils.abspath('.')
767
t = transport.get_transport(here)
768
self.assertEqual(t.local_abspath(''), here)
771
class TestLocalTransportMutation(tests.TestCaseInTempDir):
773
def test_local_transport_mkdir(self):
774
here = osutils.abspath('.')
775
t = transport.get_transport(here)
777
self.assertTrue(os.path.exists('test'))
779
def test_local_transport_mkdir_permission_denied(self):
780
# See https://bugs.launchpad.net/bzr/+bug/606537
781
here = osutils.abspath('.')
782
t = transport.get_transport(here)
784
def fake_chmod(path, mode):
785
e = OSError('permission denied')
786
e.errno = errno.EPERM
788
self.overrideAttr(os, 'chmod', fake_chmod)
790
t.mkdir('test2', mode=0o707)
791
self.assertTrue(os.path.exists('test'))
792
self.assertTrue(os.path.exists('test2'))
795
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
797
def test_local_fdatasync_calls_fdatasync(self):
798
"""Check fdatasync on a stream tries to flush the data to the OS.
800
We can't easily observe the external effect but we can at least see
804
fdatasync = getattr(os, 'fdatasync', sentinel)
805
if fdatasync is sentinel:
806
raise tests.TestNotApplicable('fdatasync not supported')
807
t = self.get_transport('.')
808
calls = self.recordCalls(os, 'fdatasync')
809
w = t.open_write_stream('out')
812
with open('out', 'rb') as f:
813
# Should have been flushed.
814
self.assertEqual(f.read(), b'foo')
815
self.assertEqual(len(calls), 1, calls)
817
def test_missing_directory(self):
818
t = self.get_transport('.')
819
self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
822
class TestWin32LocalTransport(tests.TestCase):
824
def test_unc_clone_to_root(self):
825
self.requireFeature(features.win32_feature)
826
# Win32 UNC path like \\HOST\path
827
# clone to root should stop at least at \\HOST part
829
t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
832
self.assertEqual(t.base, 'file://HOST/')
833
# make sure we reach the root
835
self.assertEqual(t.base, 'file://HOST/')
838
class TestConnectedTransport(tests.TestCase):
839
"""Tests for connected to remote server transports"""
841
def test_parse_url(self):
842
t = transport.ConnectedTransport(
843
'http://simple.example.com/home/source')
844
self.assertEqual(t._parsed_url.host, 'simple.example.com')
845
self.assertEqual(t._parsed_url.port, None)
846
self.assertEqual(t._parsed_url.path, '/home/source/')
847
self.assertTrue(t._parsed_url.user is None)
848
self.assertTrue(t._parsed_url.password is None)
850
self.assertEqual(t.base, 'http://simple.example.com/home/source/')
852
def test_parse_url_with_at_in_user(self):
854
t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
855
self.assertEqual(t._parsed_url.user, 'user@host.com')
857
def test_parse_quoted_url(self):
858
t = transport.ConnectedTransport(
859
'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
860
self.assertEqual(t._parsed_url.host, 'exAmple.com')
861
self.assertEqual(t._parsed_url.port, 2222)
862
self.assertEqual(t._parsed_url.user, 'robey')
863
self.assertEqual(t._parsed_url.password, 'h@t')
864
self.assertEqual(t._parsed_url.path, '/path/')
866
# Base should not keep track of the password
867
self.assertEqual(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
869
def test_parse_invalid_url(self):
870
self.assertRaises(urlutils.InvalidURL,
871
transport.ConnectedTransport,
872
'sftp://lily.org:~janneke/public/bzr/gub')
874
def test_relpath(self):
875
t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
877
self.assertEqual(t.relpath('sftp://user@host.com/abs/path/sub'),
879
self.assertRaises(errors.PathNotChild, t.relpath,
880
'http://user@host.com/abs/path/sub')
881
self.assertRaises(errors.PathNotChild, t.relpath,
882
'sftp://user2@host.com/abs/path/sub')
883
self.assertRaises(errors.PathNotChild, t.relpath,
884
'sftp://user@otherhost.com/abs/path/sub')
885
self.assertRaises(errors.PathNotChild, t.relpath,
886
'sftp://user@host.com:33/abs/path/sub')
887
# Make sure it works when we don't supply a username
888
t = transport.ConnectedTransport('sftp://host.com/abs/path')
889
self.assertEqual(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
891
# Make sure it works when parts of the path will be url encoded
892
t = transport.ConnectedTransport('sftp://host.com/dev/%path')
893
self.assertEqual(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
895
def test_connection_sharing_propagate_credentials(self):
896
t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
897
self.assertEqual('user', t._parsed_url.user)
898
self.assertEqual('host.com', t._parsed_url.host)
899
self.assertIs(None, t._get_connection())
900
self.assertIs(None, t._parsed_url.password)
901
c = t.clone('subdir')
902
self.assertIs(None, c._get_connection())
903
self.assertIs(None, t._parsed_url.password)
905
# Simulate the user entering a password
907
connection = object()
908
t._set_connection(connection, password)
909
self.assertIs(connection, t._get_connection())
910
self.assertIs(password, t._get_credentials())
911
self.assertIs(connection, c._get_connection())
912
self.assertIs(password, c._get_credentials())
914
# credentials can be updated
915
new_password = 'even more secret'
916
c._update_credentials(new_password)
917
self.assertIs(connection, t._get_connection())
918
self.assertIs(new_password, t._get_credentials())
919
self.assertIs(connection, c._get_connection())
920
self.assertIs(new_password, c._get_credentials())
923
class TestReusedTransports(tests.TestCase):
924
"""Tests for transport reuse"""
926
def test_reuse_same_transport(self):
927
possible_transports = []
928
t1 = transport.get_transport_from_url(
929
'http://foo/', possible_transports=possible_transports)
930
self.assertEqual([t1], possible_transports)
931
t2 = transport.get_transport_from_url('http://foo/',
932
possible_transports=[t1])
933
self.assertIs(t1, t2)
935
# Also check that final '/' are handled correctly
936
t3 = transport.get_transport_from_url('http://foo/path/')
937
t4 = transport.get_transport_from_url('http://foo/path',
938
possible_transports=[t3])
939
self.assertIs(t3, t4)
941
t5 = transport.get_transport_from_url('http://foo/path')
942
t6 = transport.get_transport_from_url('http://foo/path/',
943
possible_transports=[t5])
944
self.assertIs(t5, t6)
946
def test_don_t_reuse_different_transport(self):
947
t1 = transport.get_transport_from_url('http://foo/path')
948
t2 = transport.get_transport_from_url('http://bar/path',
949
possible_transports=[t1])
950
self.assertIsNot(t1, t2)
953
class TestTransportTrace(tests.TestCase):
955
def test_decorator(self):
956
t = transport.get_transport_from_url('trace+memory://')
957
self.assertIsInstance(
958
t, breezy.transport.trace.TransportTraceDecorator)
960
def test_clone_preserves_activity(self):
961
t = transport.get_transport_from_url('trace+memory://')
963
self.assertTrue(t is not t2)
964
self.assertTrue(t._activity is t2._activity)
966
# the following specific tests are for the operations that have made use of
967
# logging in tests; we could test every single operation but doing that
968
# still won't cause a test failure when the top level Transport API
969
# changes; so there is little return doing that.
971
t = transport.get_transport_from_url('trace+memory:///')
972
t.put_bytes('foo', b'barish')
975
# put_bytes records the bytes, not the content to avoid memory
977
expected_result.append(('put_bytes', 'foo', 6, None))
978
# get records the file name only.
979
expected_result.append(('get', 'foo'))
980
self.assertEqual(expected_result, t._activity)
982
def test_readv(self):
983
t = transport.get_transport_from_url('trace+memory:///')
984
t.put_bytes('foo', b'barish')
985
list(t.readv('foo', [(0, 1), (3, 2)],
986
adjust_for_latency=True, upper_limit=6))
988
# put_bytes records the bytes, not the content to avoid memory
990
expected_result.append(('put_bytes', 'foo', 6, None))
991
# readv records the supplied offset request
992
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
993
self.assertEqual(expected_result, t._activity)
996
class TestSSHConnections(tests.TestCaseWithTransport):
998
def test_bzr_connect_to_bzr_ssh(self):
999
"""get_transport of a bzr+ssh:// behaves correctly.
1001
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
1003
# This test actually causes a bzr instance to be invoked, which is very
1004
# expensive: it should be the only such test in the test suite.
1005
# A reasonable evolution for this would be to simply check inside
1006
# check_channel_exec_request that the command is appropriate, and then
1007
# satisfy requests in-process.
1008
self.requireFeature(features.paramiko)
1009
# SFTPFullAbsoluteServer has a get_url method, and doesn't
1010
# override the interface (doesn't change self._vendor).
1011
# Note that this does encryption, so can be slow.
1012
from breezy.tests import stub_sftp
1014
# Start an SSH server
1015
self.command_executed = []
1016
# XXX: This is horrible -- we define a really dumb SSH server that
1017
# executes commands, and manage the hooking up of stdin/out/err to the
1018
# SSH channel ourselves. Surely this has already been implemented
1022
class StubSSHServer(stub_sftp.StubServer):
1026
def check_channel_exec_request(self, channel, command):
1027
self.test.command_executed.append(command)
1028
proc = subprocess.Popen(
1029
command, shell=True, stdin=subprocess.PIPE,
1030
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
1033
# XXX: horribly inefficient, not to mention ugly.
1034
# Start a thread for each of stdin/out/err, and relay bytes
1035
# from the subprocess to channel and vice versa.
1036
def ferry_bytes(read, write, close):
1045
(channel.recv, proc.stdin.write, proc.stdin.close),
1046
(proc.stdout.read, channel.sendall, channel.close),
1047
(proc.stderr.read, channel.sendall_stderr, channel.close)]
1048
started.append(proc)
1049
for read, write, close in file_functions:
1050
t = threading.Thread(
1051
target=ferry_bytes, args=(read, write, close))
1057
ssh_server = stub_sftp.SFTPFullAbsoluteServer(StubSSHServer)
1058
# We *don't* want to override the default SSH vendor: the detected one
1059
# is the one to use.
1061
# FIXME: I don't understand the above comment, SFTPFullAbsoluteServer
1062
# inherits from SFTPServer which forces the SSH vendor to
1063
# ssh.ParamikoVendor(). So it's forced, not detected. --vila 20100623
1064
self.start_server(ssh_server)
1065
port = ssh_server.port
1067
if sys.platform == 'win32':
1068
bzr_remote_path = sys.executable + ' ' + self.get_brz_path()
1070
bzr_remote_path = self.get_brz_path()
1071
self.overrideEnv('BZR_REMOTE_PATH', bzr_remote_path)
1073
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
1074
# variable is used to tell bzr what command to run on the remote end.
1075
path_to_branch = osutils.abspath('.')
1076
if sys.platform == 'win32':
1077
# On Windows, we export all drives as '/C:/, etc. So we need to
1078
# prefix a '/' to get the right path.
1079
path_to_branch = '/' + path_to_branch
1080
url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
1081
t = transport.get_transport(url)
1082
self.permit_url(t.base)
1086
[b'%s serve --inet --directory=/ --allow-writes' %
1087
bzr_remote_path.encode()],
1088
self.command_executed)
1089
# Make sure to disconnect, so that the remote process can stop, and we
1090
# can cleanup. Then pause the test until everything is shutdown
1091
t._client._medium.disconnect()
1094
# First wait for the subprocess
1096
# And the rest are threads
1097
for t in started[1:]:
1101
class TestUnhtml(tests.TestCase):
1103
"""Tests for unhtml_roughly"""
1105
def test_truncation(self):
1106
fake_html = "<p>something!\n" * 1000
1107
result = http.unhtml_roughly(fake_html)
1108
self.assertEqual(len(result), 1000)
1109
self.assertStartsWith(result, " something!")
1112
class SomeDirectory(object):
1114
def look_up(self, name, url):
1118
class TestLocationToUrl(tests.TestCase):
1120
def get_base_location(self):
1121
path = osutils.abspath('/foo/bar')
1122
if path.startswith('/'):
1123
url = 'file://%s' % (path,)
1125
# On Windows, abspaths start with the drive letter, so we have to
1126
# add in the extra '/'
1127
url = 'file:///%s' % (path,)
1130
def test_regular_url(self):
1131
self.assertEqual("file://foo", location_to_url("file://foo"))
1133
def test_directory(self):
1134
directories.register("bar:", SomeDirectory, "Dummy directory")
1135
self.addCleanup(directories.remove, "bar:")
1136
self.assertEqual("http://bar", location_to_url("bar:"))
1138
def test_unicode_url(self):
1139
self.assertRaises(urlutils.InvalidURL, location_to_url,
1140
b"http://fo/\xc3\xaf".decode("utf-8"))
1142
def test_unicode_path(self):
1143
path, url = self.get_base_location()
1144
location = path + b"\xc3\xaf".decode("utf-8")
1146
self.assertEqual(url, location_to_url(location))
1148
def test_path(self):
1149
path, url = self.get_base_location()
1150
self.assertEqual(url, location_to_url(path))
1152
def test_relative_file_url(self):
1153
self.assertEqual(urlutils.local_path_to_url(".") + "/bar",
1154
location_to_url("file:bar"))
1156
def test_absolute_file_url(self):
1157
self.assertEqual("file:///bar", location_to_url("file:/bar"))
added
removed
breezy/tests/test_transport.py
