1
# Copyright (C) 2005-2011, 2015, 2016 Canonical Ltd
1
# Copyright (C) 2005-2011 Canonical Ltd
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
520
520
self.assertTransportMode(t, 'dir777', 0777)
522
522
def test_put_bytes_unicode(self):
523
# Expect put_bytes to raise AssertionError or UnicodeEncodeError if
524
# given unicode "bytes". UnicodeEncodeError doesn't really make sense
525
# (we don't want to encode unicode here at all, callers should be
526
# strictly passing bytes to put_bytes), but we allow it for backwards
527
# compatibility. At some point we should use a specific exception.
528
# See https://bugs.launchpad.net/bzr/+bug/106898.
523
529
t = self.get_transport()
524
530
if t.is_readonly():
526
532
unicode_string = u'\u1234'
527
self.assertRaises(TypeError, t.put_bytes, 'foo', unicode_string)
534
(AssertionError, UnicodeEncodeError),
535
t.put_bytes, 'foo', unicode_string)
537
def test_put_file_unicode(self):
538
# Like put_bytes, except with a StringIO.StringIO of a unicode string.
539
# This situation can happen (and has) if code is careless about the type
540
# of "string" they initialise/write to a StringIO with. We cannot use
541
# cStringIO, because it never returns unicode from read.
542
# Like put_bytes, UnicodeEncodeError isn't quite the right exception to
543
# raise, but we raise it for hysterical raisins.
544
t = self.get_transport()
547
unicode_file = pyStringIO(u'\u1234')
548
self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
529
550
def test_mkdir(self):
530
551
t = self.get_transport()
609
630
def test_opening_a_file_stream_can_set_mode(self):
610
631
t = self.get_transport()
611
632
if t.is_readonly():
612
self.assertRaises((TransportNotPossible, NotImplementedError),
613
t.open_write_stream, 'foo')
615
634
if not t._can_roundtrip_unix_modebits():
616
635
# Can't roundtrip, so no need to run this test
619
637
def check_mode(name, mode, expected):
620
638
handle = t.open_write_stream(name, mode=mode)
988
1002
# perhaps all of this could be done in a subdirectory
990
1004
t.put_bytes('a', 'a first file\n')
991
self.assertEqual([True, False], list(t.has_multi(['a', 'b'])))
1005
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
993
1007
t.move('a', 'b')
994
1008
self.assertTrue(t.has('b'))
995
1009
self.assertFalse(t.has('a'))
997
1011
self.check_transport_contents('a first file\n', t, 'b')
998
self.assertEqual([False, True], list(t.has_multi(['a', 'b'])))
1012
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1000
1014
# Overwrite a file
1001
1015
t.put_bytes('c', 'c this file\n')
1194
1208
raise TestSkipped("not a connected transport")
1196
1210
t2 = t1.clone('subdir')
1197
self.assertEqual(t1._parsed_url.scheme, t2._parsed_url.scheme)
1198
self.assertEqual(t1._parsed_url.user, t2._parsed_url.user)
1199
self.assertEqual(t1._parsed_url.password, t2._parsed_url.password)
1200
self.assertEqual(t1._parsed_url.host, t2._parsed_url.host)
1201
self.assertEqual(t1._parsed_url.port, t2._parsed_url.port)
1211
self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
1212
self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
1213
self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
1214
self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
1215
self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1203
1217
def test__reuse_for(self):
1204
1218
t = self.get_transport()
1768
1782
"""Segment parameters should be stripped and stored in
1769
1783
transport.segment_parameters."""
1770
1784
transport = self.get_transport("foo")
1771
self.assertEqual({}, transport.get_segment_parameters())
1785
self.assertEquals({}, transport.get_segment_parameters())
1773
1787
def test_segment_parameters(self):
1774
1788
"""Segment parameters should be stripped and stored in
1777
1791
parameters = {"key1": "val1", "key2": "val2"}
1778
1792
url = urlutils.join_segment_parameters(base_url, parameters)
1779
1793
transport = _mod_transport.get_transport_from_url(url)
1780
self.assertEqual(parameters, transport.get_segment_parameters())
1794
self.assertEquals(parameters, transport.get_segment_parameters())
1782
1796
def test_set_segment_parameters(self):
1783
1797
"""Segment parameters can be set and show up in base."""
1784
1798
transport = self.get_transport("foo")
1785
1799
orig_base = transport.base
1786
1800
transport.set_segment_parameter("arm", "board")
1787
self.assertEqual("%s,arm=board" % orig_base, transport.base)
1788
self.assertEqual({"arm": "board"}, transport.get_segment_parameters())
1801
self.assertEquals("%s,arm=board" % orig_base, transport.base)
1802
self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
1789
1803
transport.set_segment_parameter("arm", None)
1790
1804
transport.set_segment_parameter("nonexistant", None)
1791
self.assertEqual({}, transport.get_segment_parameters())
1792
self.assertEqual(orig_base, transport.base)
1805
self.assertEquals({}, transport.get_segment_parameters())
1806
self.assertEquals(orig_base, transport.base)
1794
1808
def test_stat_symlink(self):
1795
1809
# if a transport points directly to a symlink (and supports symlinks