1
# Copyright (C) 2005-2011 Canonical Ltd
1
# Copyright (C) 2005-2011, 2015, 2016 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.
529
523
t = self.get_transport()
530
524
if t.is_readonly():
532
526
unicode_string = u'\u1234'
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)
527
self.assertRaises(TypeError, t.put_bytes, 'foo', unicode_string)
550
529
def test_mkdir(self):
551
530
t = self.get_transport()
630
609
def test_opening_a_file_stream_can_set_mode(self):
631
610
t = self.get_transport()
632
611
if t.is_readonly():
612
self.assertRaises((TransportNotPossible, NotImplementedError),
613
t.open_write_stream, 'foo')
634
615
if not t._can_roundtrip_unix_modebits():
635
616
# Can't roundtrip, so no need to run this test
637
619
def check_mode(name, mode, expected):
638
620
handle = t.open_write_stream(name, mode=mode)
1002
988
# perhaps all of this could be done in a subdirectory
1004
990
t.put_bytes('a', 'a first file\n')
1005
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
991
self.assertEqual([True, False], list(t.has_multi(['a', 'b'])))
1007
993
t.move('a', 'b')
1008
994
self.assertTrue(t.has('b'))
1009
995
self.assertFalse(t.has('a'))
1011
997
self.check_transport_contents('a first file\n', t, 'b')
1012
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
998
self.assertEqual([False, True], list(t.has_multi(['a', 'b'])))
1014
1000
# Overwrite a file
1015
1001
t.put_bytes('c', 'c this file\n')
1208
1194
raise TestSkipped("not a connected transport")
1210
1196
t2 = t1.clone('subdir')
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)
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)
1217
1203
def test__reuse_for(self):
1218
1204
t = self.get_transport()
1782
1768
"""Segment parameters should be stripped and stored in
1783
1769
transport.segment_parameters."""
1784
1770
transport = self.get_transport("foo")
1785
self.assertEquals({}, transport.get_segment_parameters())
1771
self.assertEqual({}, transport.get_segment_parameters())
1787
1773
def test_segment_parameters(self):
1788
1774
"""Segment parameters should be stripped and stored in
1791
1777
parameters = {"key1": "val1", "key2": "val2"}
1792
1778
url = urlutils.join_segment_parameters(base_url, parameters)
1793
1779
transport = _mod_transport.get_transport_from_url(url)
1794
self.assertEquals(parameters, transport.get_segment_parameters())
1780
self.assertEqual(parameters, transport.get_segment_parameters())
1796
1782
def test_set_segment_parameters(self):
1797
1783
"""Segment parameters can be set and show up in base."""
1798
1784
transport = self.get_transport("foo")
1799
1785
orig_base = transport.base
1800
1786
transport.set_segment_parameter("arm", "board")
1801
self.assertEquals("%s,arm=board" % orig_base, transport.base)
1802
self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
1787
self.assertEqual("%s,arm=board" % orig_base, transport.base)
1788
self.assertEqual({"arm": "board"}, transport.get_segment_parameters())
1803
1789
transport.set_segment_parameter("arm", None)
1804
1790
transport.set_segment_parameter("nonexistant", None)
1805
self.assertEquals({}, transport.get_segment_parameters())
1806
self.assertEquals(orig_base, transport.base)
1791
self.assertEqual({}, transport.get_segment_parameters())
1792
self.assertEqual(orig_base, transport.base)
1808
1794
def test_stat_symlink(self):
1809
1795
# if a transport points directly to a symlink (and supports symlinks