~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransport.py

  • Committer: Martin Pool
  • Date: 2005-03-15 05:19:54 UTC
  • Revision ID: mbp@sourcefrog.net-20050315051954-e4bdd6dfd26f8ecf
witty comment

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2004, 2005 by Canonical Ltd
2
 
 
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.
7
 
 
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.
12
 
 
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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
 
18
 
from bzrlib.selftest import TestCaseInTempDir
19
 
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
20
 
from bzrlib.transport import NoSuchFile, FileExists
21
 
 
22
 
def test_transport(tester, t, readonly=False):
23
 
    """Test a transport object. Basically, it assumes that the
24
 
    Transport object is connected to the current working directory.
25
 
    So that whatever is done through the transport, should show
26
 
    up in the working directory, and vice-versa.
27
 
 
28
 
    This also tests to make sure that the functions work with both
29
 
    generators and lists (assuming iter(list) is effectively a generator)
30
 
    """
31
 
    import tempfile, os
32
 
    from bzrlib.transport.local import LocalTransport
33
 
 
34
 
    # Test has
35
 
    files = ['a', 'b', 'e', 'g']
36
 
    tester.build_tree(files)
37
 
    tester.assertEqual(t.has('a'), True)
38
 
    tester.assertEqual(t.has('c'), False)
39
 
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
40
 
            [True, True, False, False, True, False, True, False])
41
 
    tester.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
42
 
            [True, True, False, False, True, False, True, False])
43
 
 
44
 
    # Test get
45
 
    tester.assertEqual(t.get('a').read(), open('a').read())
46
 
    content_f = t.get_multi(files)
47
 
    for path,f in zip(files, content_f):
48
 
        tester.assertEqual(open(path).read(), f.read())
49
 
 
50
 
    content_f = t.get_multi(iter(files))
51
 
    for path,f in zip(files, content_f):
52
 
        tester.assertEqual(open(path).read(), f.read())
53
 
 
54
 
    tester.assertRaises(NoSuchFile, t.get, 'c')
55
 
    try:
56
 
        files = list(t.get_multi(['a', 'b', 'c']))
57
 
    except NoSuchFile:
58
 
        pass
59
 
    else:
60
 
        tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
61
 
    try:
62
 
        files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
63
 
    except NoSuchFile:
64
 
        pass
65
 
    else:
66
 
        tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
67
 
 
68
 
    # Test put
69
 
    if readonly:
70
 
        open('c', 'wb').write('some text for c\n')
71
 
    else:
72
 
        t.put('c', 'some text for c\n')
73
 
    tester.assert_(os.path.exists('c'))
74
 
    tester.assertEqual(open('c').read(), 'some text for c\n')
75
 
    tester.assertEqual(t.get('c').read(), 'some text for c\n')
76
 
    # Make sure 'has' is updated
77
 
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
78
 
            [True, True, True, False, True, False, True, False])
79
 
    if readonly:
80
 
        open('a', 'wb').write('new\ncontents for\na\n')
81
 
        open('d', 'wb').write('contents\nfor d\n')
82
 
    else:
83
 
        # Put also replaces contents
84
 
        tester.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
85
 
                                      ('d', 'contents\nfor d\n')]),
86
 
                         2)
87
 
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
88
 
            [True, True, True, True, True, False, True, False])
89
 
    tester.assertEqual(open('a').read(), 'new\ncontents for\na\n')
90
 
    tester.assertEqual(open('d').read(), 'contents\nfor d\n')
91
 
 
92
 
    if readonly:
93
 
        open('a', 'wb').write('diff\ncontents for\na\n')
94
 
        open('d', 'wb').write('another contents\nfor d\n')
95
 
    else:
96
 
        tester.assertEqual(
97
 
            t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
98
 
                              ('d', 'another contents\nfor d\n')]))
99
 
                         , 2)
100
 
    tester.assertEqual(open('a').read(), 'diff\ncontents for\na\n')
101
 
    tester.assertEqual(open('d').read(), 'another contents\nfor d\n')
102
 
 
103
 
    if not readonly:
104
 
        tester.assertRaises(NoSuchFile, t.put, 'path/doesnt/exist/c', 'contents')
105
 
 
106
 
    # Test mkdir
107
 
    os.mkdir('dir_a')
108
 
    tester.assertEqual(t.has('dir_a'), True)
109
 
    tester.assertEqual(t.has('dir_b'), False)
110
 
 
111
 
    if readonly:
112
 
        os.mkdir('dir_b')
113
 
    else:
114
 
        t.mkdir('dir_b')
115
 
    tester.assertEqual(t.has('dir_b'), True)
116
 
    tester.assert_(os.path.isdir('dir_b'))
117
 
 
118
 
    if readonly:
119
 
        os.mkdir('dir_c')
120
 
        os.mkdir('dir_d')
121
 
    else:
122
 
        t.mkdir_multi(['dir_c', 'dir_d'])
123
 
    tester.assertEqual(list(t.has_multi(['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_b'])),
124
 
            [True, True, True, True, False, True])
125
 
    for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d']:
126
 
        tester.assert_(os.path.isdir(d))
127
 
 
128
 
    if not readonly:
129
 
        tester.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
130
 
        tester.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
131
 
 
132
 
    # Make sure the transport recognizes when a
133
 
    # directory is created by other means
134
 
    # Caching Transports will fail, because dir_e was already seen not
135
 
    # to exist. So instead, we will search for a new directory
136
 
    #os.mkdir('dir_e')
137
 
    #if not readonly:
138
 
    #    tester.assertRaises(FileExists, t.mkdir, 'dir_e')
139
 
 
140
 
    os.mkdir('dir_f')
141
 
    if not readonly:
142
 
        tester.assertRaises(FileExists, t.mkdir, 'dir_f')
143
 
 
144
 
    # Test get/put in sub-directories
145
 
    if readonly:
146
 
        open('dir_a/a', 'wb').write('contents of dir_a/a')
147
 
        open('dir_b/b', 'wb').write('contents of dir_b/b')
148
 
    else:
149
 
        tester.assertEqual(
150
 
            t.put_multi([('dir_a/a', 'contents of dir_a/a'),
151
 
                         ('dir_b/b', 'contents of dir_b/b')])
152
 
                      , 2)
153
 
    for f in ('dir_a/a', 'dir_b/b'):
154
 
        tester.assertEqual(t.get(f).read(), open(f).read())
155
 
 
156
 
    # Test copy_to
157
 
    dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
158
 
    dtmp_base = os.path.basename(dtmp)
159
 
    local_t = LocalTransport(dtmp)
160
 
 
161
 
    files = ['a', 'b', 'c', 'd']
162
 
    t.copy_to(files, local_t)
163
 
    for f in files:
164
 
        tester.assertEquals(open(f).read(), open(os.path.join(dtmp_base, f)).read())
165
 
 
166
 
    # TODO: Test append
167
 
    # TODO: Make sure all entries support file-like objects as well as strings.
168
 
    # TODO: Test get_partial()
169
 
 
170
 
class LocalTransportTest(TestCaseInTempDir):
171
 
    def test_local_transport(self):
172
 
        from bzrlib.transport.local import LocalTransport
173
 
 
174
 
        t = LocalTransport('.')
175
 
        test_transport(self, t)
176
 
 
177
 
class HttpTransportTest(TestCaseWithWebserver):
178
 
    def test_http_transport(self):
179
 
        from bzrlib.transport.http import HttpTransport
180
 
 
181
 
        t = HttpTransport(self.get_remote_url('.'))
182
 
        test_transport(self, t, readonly=True)
183