1
# Copyright (C) 2004, 2005 by 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
from bzrlib.selftest import TestCaseInTempDir
20
def test_transport(tester, t, readonly=False):
21
"""Test a transport object. Basically, it assumes that the
22
Transport object is connected to the current working directory.
23
So that whatever is done through the transport, should show
24
up in the working directory, and vice-versa.
26
This also tests to make sure that the functions work with both
27
generators and lists (assuming iter(list) is effectively a generator)
30
from bzrlib.transport.local import LocalTransport
33
files = ['a', 'b', 'e', 'g']
34
tester.build_tree(files)
35
tester.assertEqual(t.has('a'), True)
36
tester.assertEqual(t.has('c'), False)
37
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
38
[True, True, False, False, True, False, True, False])
39
tester.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
40
[True, True, False, False, True, False, True, False])
43
tester.assertEqual(t.get('a').read(), open('a').read())
44
content_f = t.get_multi(files)
45
for path,f in zip(files, content_f):
46
tester.assertEqual(open(path).read(), f.read())
48
content_f = t.get_multi(iter(files))
49
for path,f in zip(files, content_f):
50
tester.assertEqual(open(path).read(), f.read())
52
tester.assertRaises(NoSuchFile, t.get, 'c')
54
files = list(t.get_multi(['a', 'b', 'c']))
58
tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
60
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
64
tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
68
open('c', 'wb').write('some text for c\n')
70
t.put('c', 'some text for c\n')
71
tester.assert_(os.path.exists('c'))
72
tester.assertEqual(open('c').read(), 'some text for c\n')
73
tester.assertEqual(t.get('c').read(), 'some text for c\n')
74
# Make sure 'has' is updated
75
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
76
[True, True, True, False, True, False, True, False])
78
open('a', 'wb').write('new\ncontents for\na\n')
79
open('d', 'wb').write('contents\nfor d\n')
81
# Put also replaces contents
82
tester.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
83
('d', 'contents\nfor d\n')]),
85
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
86
[True, True, True, True, True, False, True, False])
87
tester.assertEqual(open('a').read(), 'new\ncontents for\na\n')
88
tester.assertEqual(open('d').read(), 'contents\nfor d\n')
91
open('a', 'wb').write('diff\ncontents for\na\n')
92
open('d', 'wb').write('another contents\nfor d\n')
95
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
96
('d', 'another contents\nfor d\n')]))
98
tester.assertEqual(open('a').read(), 'diff\ncontents for\na\n')
99
tester.assertEqual(open('d').read(), 'another contents\nfor d\n')
102
tester.assertRaises(NoSuchFile, t.put, 'path/doesnt/exist/c', 'contents')
106
tester.assertEqual(t.has('dir_a'), True)
107
tester.assertEqual(t.has('dir_b'), False)
113
tester.assertEqual(t.has('dir_b'), True)
114
tester.assert_(os.path.isdir('dir_b'))
120
t.mkdir_multi(['dir_c', 'dir_d'])
121
tester.assertEqual(list(t.has_multi(['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_b'])),
122
[True, True, True, True, False, True])
123
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d']:
124
tester.assert_(os.path.isdir(d))
127
tester.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
128
tester.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
130
# Make sure the transport recognizes when a
131
# directory is created by other means
132
# Caching Transports will fail, because dir_e was already seen not
133
# to exist. So instead, we will search for a new directory
136
# tester.assertRaises(FileExists, t.mkdir, 'dir_e')
140
tester.assertRaises(FileExists, t.mkdir, 'dir_f')
142
# Test get/put in sub-directories
144
open('dir_a/a', 'wb').write('contents of dir_a/a')
145
open('dir_b/b', 'wb').write('contents of dir_b/b')
148
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
149
('dir_b/b', 'contents of dir_b/b')])
151
for f in ('dir_a/a', 'dir_b/b'):
152
tester.assertEqual(t.get(f).read(), open(f).read())
155
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
156
dtmp_base = os.path.basename(dtmp)
157
local_t = LocalTransport(dtmp)
159
files = ['a', 'b', 'c', 'd']
160
t.copy_to(files, local_t)
162
tester.assertEquals(open(f).read(), open(os.path.join(dtmp_base, f)).read())
165
# TODO: Make sure all entries support file-like objects as well as strings.
167
class LocalTransportTest(TestCaseInTempDir):
168
def test_local_transport(self):
169
from bzrlib.transport.local import LocalTransport
171
t = LocalTransport('.')
172
test_transport(self, t)
174
class HttpServer(object):
175
"""This just encapsulates spawning and stopping
179
"""This just spawns a separate process to serve files from
180
this directory. Call the .stop() function to kill the
183
from BaseHTTPServer import HTTPServer
184
from SimpleHTTPServer import SimpleHTTPRequestHandler
186
if hasattr(os, 'fork'):
190
else: # How do we handle windows, which doesn't have fork?
191
raise NotImplementedError('At present HttpServer cannot fork on Windows')
193
# We might be able to do something like os.spawn() for the
194
# python executable, and give it a simple script to run.
195
# but then how do we kill it?
198
self.s = HTTPServer(('', 9999), SimpleHTTPRequestHandler)
199
# TODO: Is there something nicer than killing the server when done?
200
self.s.serve_forever()
201
except KeyboardInterrupt:
209
if hasattr(os, 'kill'):
211
os.kill(self.pid, signal.SIGINT)
212
os.waitpid(self.pid, 0)
215
raise NotImplementedError('At present HttpServer cannot stop on Windows')
217
class HttpTransportTest(TestCaseInTempDir):
218
def test_http_transport(self):
219
from bzrlib.transport.http import HttpTransport
223
t = HttpTransport('http://localhost:9999/')
224
test_transport(self, t, readonly=True)
added
removed
bzrlib/selftest/testtransport.py
