~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

  • Committer: John Arbash Meinel
  • Date: 2006-09-06 02:57:51 UTC
  • mto: (1946.2.10 reduce-knit-churn)
  • mto: This revision was merged to the branch mainline in revision 1988.
  • Revision ID: john@arbash-meinel.com-20060906025751-686364ac616b8e91
rename non_atomic_put_* to put_*non_atomic, and re-order the functions

Show diffs side-by-side

added added

removed removed

Lines of Context:
145
145
                            t.put, 'b', StringIO('file-like\ncontents\n'))
146
146
        self.check_transport_contents('file-like\ncontents\n', t, 'b')
147
147
 
148
 
    def test_put_multi(self):
149
 
        t = self.get_transport()
150
 
 
151
 
        if t.is_readonly():
152
 
            return
153
 
        deprecation_msg = symbol_versioning.deprecation_string(
154
 
            t.put_multi, symbol_versioning.zero_eleven)
155
 
        self.assertEqual(2, self.callDeprecated([deprecation_msg],
156
 
            t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
157
 
                          ('d', StringIO('contents\nfor d\n'))]
158
 
            ))
159
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
160
 
                [True, False, False, True])
161
 
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
162
 
        self.check_transport_contents('contents\nfor d\n', t, 'd')
163
 
 
164
 
        self.assertEqual(2, self.callDeprecated([deprecation_msg],
165
 
            t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
166
 
                              ('d', StringIO('another contents\nfor d\n'))])
167
 
            ))
168
 
        self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
169
 
        self.check_transport_contents('another contents\nfor d\n', t, 'd')
170
 
 
171
 
    def test_put_file(self):
172
 
        t = self.get_transport()
173
 
 
174
 
        if t.is_readonly():
175
 
            self.assertRaises(TransportNotPossible,
176
 
                    t.put_file, 'a', StringIO('some text for a\n'))
177
 
            return
178
 
 
179
 
        t.put_file('a', StringIO('some text for a\n'))
180
 
        self.failUnless(t.has('a'))
181
 
        self.check_transport_contents('some text for a\n', t, 'a')
182
 
        # Put also replaces contents
183
 
        t.put_file('a', StringIO('new\ncontents for\na\n'))
184
 
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
185
 
        self.assertRaises(NoSuchFile,
186
 
                          t.put_file, 'path/doesnt/exist/c',
187
 
                              StringIO('contents'))
188
 
 
189
148
    def test_put_bytes(self):
190
149
        t = self.get_transport()
191
150
 
205
164
        self.assertRaises(NoSuchFile,
206
165
                          t.put_bytes, 'path/doesnt/exist/c', 'contents')
207
166
 
208
 
    def test_non_atomic_put_file(self):
209
 
        t = self.get_transport()
210
 
 
211
 
        if t.is_readonly():
212
 
            self.assertRaises(TransportNotPossible,
213
 
                    t.non_atomic_put_file, 'a', StringIO('some text for a\n'))
214
 
            return
215
 
 
216
 
        self.failIf(t.has('a'))
217
 
        t.non_atomic_put_file('a', StringIO('some text for a\n'))
218
 
        self.failUnless(t.has('a'))
219
 
        self.check_transport_contents('some text for a\n', t, 'a')
220
 
        # Put also replaces contents
221
 
        t.non_atomic_put_file('a', StringIO('new\ncontents for\na\n'))
222
 
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
223
 
 
224
 
        # Make sure we can create another file
225
 
        t.non_atomic_put_file('d', StringIO('contents for\nd\n'))
226
 
        # And overwrite 'a' with empty contents
227
 
        t.non_atomic_put_file('a', StringIO(''))
228
 
        self.check_transport_contents('contents for\nd\n', t, 'd')
229
 
        self.check_transport_contents('', t, 'a')
230
 
 
231
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_file, 'no/such/path',
232
 
                                       StringIO('contents\n'))
233
 
        # Now test the create_parent flag
234
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_file, 'dir/a',
235
 
                                       StringIO('contents\n'))
236
 
        self.failIf(t.has('dir/a'))
237
 
        t.non_atomic_put_file('dir/a', StringIO('contents for dir/a\n'),
 
167
    def test_put_bytes_non_atomic(self):
 
168
        t = self.get_transport()
 
169
 
 
170
        if t.is_readonly():
 
171
            self.assertRaises(TransportNotPossible,
 
172
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
 
173
            return
 
174
 
 
175
        self.failIf(t.has('a'))
 
176
        t.put_bytes_non_atomic('a', 'some text for a\n')
 
177
        self.failUnless(t.has('a'))
 
178
        self.check_transport_contents('some text for a\n', t, 'a')
 
179
        # Put also replaces contents
 
180
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
 
181
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
 
182
 
 
183
        # Make sure we can create another file
 
184
        t.put_bytes_non_atomic('d', 'contents for\nd\n')
 
185
        # And overwrite 'a' with empty contents
 
186
        t.put_bytes_non_atomic('a', '')
 
187
        self.check_transport_contents('contents for\nd\n', t, 'd')
 
188
        self.check_transport_contents('', t, 'a')
 
189
 
 
190
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
 
191
                                       'contents\n')
 
192
        # Now test the create_parent flag
 
193
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
 
194
                                       'contents\n')
 
195
        self.failIf(t.has('dir/a'))
 
196
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
 
197
                               create_parent_dir=True)
 
198
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
 
199
        
 
200
        # But we still get NoSuchFile if we can't make the parent dir
 
201
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
 
202
                                       'contents\n',
 
203
                                       create_parent_dir=True)
 
204
 
 
205
    def test_put_bytes_permissions(self):
 
206
        t = self.get_transport()
 
207
 
 
208
        if t.is_readonly():
 
209
            return
 
210
        if not t._can_roundtrip_unix_modebits():
 
211
            # Can't roundtrip, so no need to run this test
 
212
            return
 
213
        t.put_bytes('mode644', 'test text\n', mode=0644)
 
214
        self.assertTransportMode(t, 'mode644', 0644)
 
215
        t.put_bytes('mode666', 'test text\n', mode=0666)
 
216
        self.assertTransportMode(t, 'mode666', 0666)
 
217
        t.put_bytes('mode600', 'test text\n', mode=0600)
 
218
        self.assertTransportMode(t, 'mode600', 0600)
 
219
        # Yes, you can put_bytes a file such that it becomes readonly
 
220
        t.put_bytes('mode400', 'test text\n', mode=0400)
 
221
        self.assertTransportMode(t, 'mode400', 0400)
 
222
 
 
223
        # The default permissions should be based on the current umask
 
224
        umask = osutils.get_umask()
 
225
        t.put_bytes('nomode', 'test text\n', mode=None)
 
226
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
 
227
        
 
228
    def test_put_bytes_non_atomic_permissions(self):
 
229
        t = self.get_transport()
 
230
 
 
231
        if t.is_readonly():
 
232
            return
 
233
        if not t._can_roundtrip_unix_modebits():
 
234
            # Can't roundtrip, so no need to run this test
 
235
            return
 
236
        t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
 
237
        self.assertTransportMode(t, 'mode644', 0644)
 
238
        t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
 
239
        self.assertTransportMode(t, 'mode666', 0666)
 
240
        t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
 
241
        self.assertTransportMode(t, 'mode600', 0600)
 
242
        t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
 
243
        self.assertTransportMode(t, 'mode400', 0400)
 
244
 
 
245
        # The default permissions should be based on the current umask
 
246
        umask = osutils.get_umask()
 
247
        t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
 
248
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
 
249
        
 
250
    def test_put_file(self):
 
251
        t = self.get_transport()
 
252
 
 
253
        if t.is_readonly():
 
254
            self.assertRaises(TransportNotPossible,
 
255
                    t.put_file, 'a', StringIO('some text for a\n'))
 
256
            return
 
257
 
 
258
        t.put_file('a', StringIO('some text for a\n'))
 
259
        self.failUnless(t.has('a'))
 
260
        self.check_transport_contents('some text for a\n', t, 'a')
 
261
        # Put also replaces contents
 
262
        t.put_file('a', StringIO('new\ncontents for\na\n'))
 
263
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
 
264
        self.assertRaises(NoSuchFile,
 
265
                          t.put_file, 'path/doesnt/exist/c',
 
266
                              StringIO('contents'))
 
267
 
 
268
    def test_put_file_non_atomic(self):
 
269
        t = self.get_transport()
 
270
 
 
271
        if t.is_readonly():
 
272
            self.assertRaises(TransportNotPossible,
 
273
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
 
274
            return
 
275
 
 
276
        self.failIf(t.has('a'))
 
277
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
 
278
        self.failUnless(t.has('a'))
 
279
        self.check_transport_contents('some text for a\n', t, 'a')
 
280
        # Put also replaces contents
 
281
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
 
282
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
 
283
 
 
284
        # Make sure we can create another file
 
285
        t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
 
286
        # And overwrite 'a' with empty contents
 
287
        t.put_file_non_atomic('a', StringIO(''))
 
288
        self.check_transport_contents('contents for\nd\n', t, 'd')
 
289
        self.check_transport_contents('', t, 'a')
 
290
 
 
291
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
 
292
                                       StringIO('contents\n'))
 
293
        # Now test the create_parent flag
 
294
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
 
295
                                       StringIO('contents\n'))
 
296
        self.failIf(t.has('dir/a'))
 
297
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
238
298
                              create_parent_dir=True)
239
299
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
240
300
        
241
301
        # But we still get NoSuchFile if we can't make the parent dir
242
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_file, 'not/there/a',
 
302
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
243
303
                                       StringIO('contents\n'),
244
304
                                       create_parent_dir=True)
245
305
 
246
 
    def test_non_atomic_put_bytes(self):
247
 
        t = self.get_transport()
248
 
 
249
 
        if t.is_readonly():
250
 
            self.assertRaises(TransportNotPossible,
251
 
                    t.non_atomic_put_bytes, 'a', 'some text for a\n')
252
 
            return
253
 
 
254
 
        self.failIf(t.has('a'))
255
 
        t.non_atomic_put_bytes('a', 'some text for a\n')
256
 
        self.failUnless(t.has('a'))
257
 
        self.check_transport_contents('some text for a\n', t, 'a')
258
 
        # Put also replaces contents
259
 
        t.non_atomic_put_bytes('a', 'new\ncontents for\na\n')
260
 
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
261
 
 
262
 
        # Make sure we can create another file
263
 
        t.non_atomic_put_bytes('d', 'contents for\nd\n')
264
 
        # And overwrite 'a' with empty contents
265
 
        t.non_atomic_put_bytes('a', '')
266
 
        self.check_transport_contents('contents for\nd\n', t, 'd')
267
 
        self.check_transport_contents('', t, 'a')
268
 
 
269
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_bytes, 'no/such/path',
270
 
                                       'contents\n')
271
 
        # Now test the create_parent flag
272
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_bytes, 'dir/a',
273
 
                                       'contents\n')
274
 
        self.failIf(t.has('dir/a'))
275
 
        t.non_atomic_put_bytes('dir/a', 'contents for dir/a\n',
276
 
                               create_parent_dir=True)
277
 
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
278
 
        
279
 
        # But we still get NoSuchFile if we can't make the parent dir
280
 
        self.assertRaises(NoSuchFile, t.non_atomic_put_bytes, 'not/there/a',
281
 
                                       'contents\n',
282
 
                                       create_parent_dir=True)
283
 
 
284
306
    def test_put_file_permissions(self):
285
307
 
286
308
        t = self.get_transport()
312
334
        t.put_file('nomode', StringIO('test text\n'), mode=None)
313
335
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
314
336
        
315
 
    def test_put_bytes_permissions(self):
316
 
        t = self.get_transport()
317
 
 
318
 
        if t.is_readonly():
319
 
            return
320
 
        if not t._can_roundtrip_unix_modebits():
321
 
            # Can't roundtrip, so no need to run this test
322
 
            return
323
 
        t.put_bytes('mode644', 'test text\n', mode=0644)
324
 
        self.assertTransportMode(t, 'mode644', 0644)
325
 
        t.put_bytes('mode666', 'test text\n', mode=0666)
326
 
        self.assertTransportMode(t, 'mode666', 0666)
327
 
        t.put_bytes('mode600', 'test text\n', mode=0600)
328
 
        self.assertTransportMode(t, 'mode600', 0600)
329
 
        # Yes, you can put_bytes a file such that it becomes readonly
330
 
        t.put_bytes('mode400', 'test text\n', mode=0400)
331
 
        self.assertTransportMode(t, 'mode400', 0400)
332
 
 
333
 
        # The default permissions should be based on the current umask
334
 
        umask = osutils.get_umask()
335
 
        t.put_bytes('nomode', 'test text\n', mode=None)
336
 
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
337
 
        
338
 
    def test_non_atomic_put_file_permissions(self):
339
 
        t = self.get_transport()
340
 
 
341
 
        if t.is_readonly():
342
 
            return
343
 
        if not t._can_roundtrip_unix_modebits():
344
 
            # Can't roundtrip, so no need to run this test
345
 
            return
346
 
        t.non_atomic_put_file('mode644', StringIO('test text\n'), mode=0644)
347
 
        self.assertTransportMode(t, 'mode644', 0644)
348
 
        t.non_atomic_put_file('mode666', StringIO('test text\n'), mode=0666)
349
 
        self.assertTransportMode(t, 'mode666', 0666)
350
 
        t.non_atomic_put_file('mode600', StringIO('test text\n'), mode=0600)
351
 
        self.assertTransportMode(t, 'mode600', 0600)
352
 
        # Yes, you can non_atomic_put_file a file such that it becomes readonly
353
 
        t.non_atomic_put_file('mode400', StringIO('test text\n'), mode=0400)
354
 
        self.assertTransportMode(t, 'mode400', 0400)
355
 
 
356
 
        # The default permissions should be based on the current umask
357
 
        umask = osutils.get_umask()
358
 
        t.non_atomic_put_file('nomode', StringIO('test text\n'), mode=None)
359
 
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
360
 
        
361
 
    def test_non_atomic_put_bytes_permissions(self):
362
 
        t = self.get_transport()
363
 
 
364
 
        if t.is_readonly():
365
 
            return
366
 
        if not t._can_roundtrip_unix_modebits():
367
 
            # Can't roundtrip, so no need to run this test
368
 
            return
369
 
        t.non_atomic_put_bytes('mode644', 'test text\n', mode=0644)
370
 
        self.assertTransportMode(t, 'mode644', 0644)
371
 
        t.non_atomic_put_bytes('mode666', 'test text\n', mode=0666)
372
 
        self.assertTransportMode(t, 'mode666', 0666)
373
 
        t.non_atomic_put_bytes('mode600', 'test text\n', mode=0600)
374
 
        self.assertTransportMode(t, 'mode600', 0600)
375
 
        t.non_atomic_put_bytes('mode400', 'test text\n', mode=0400)
376
 
        self.assertTransportMode(t, 'mode400', 0400)
377
 
 
378
 
        # The default permissions should be based on the current umask
379
 
        umask = osutils.get_umask()
380
 
        t.non_atomic_put_bytes('nomode', 'test text\n', mode=None)
381
 
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
382
 
        
 
337
    def test_put_file_non_atomic_permissions(self):
 
338
        t = self.get_transport()
 
339
 
 
340
        if t.is_readonly():
 
341
            return
 
342
        if not t._can_roundtrip_unix_modebits():
 
343
            # Can't roundtrip, so no need to run this test
 
344
            return
 
345
        t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
 
346
        self.assertTransportMode(t, 'mode644', 0644)
 
347
        t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
 
348
        self.assertTransportMode(t, 'mode666', 0666)
 
349
        t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
 
350
        self.assertTransportMode(t, 'mode600', 0600)
 
351
        # Yes, you can put_file_non_atomic a file such that it becomes readonly
 
352
        t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
 
353
        self.assertTransportMode(t, 'mode400', 0400)
 
354
 
 
355
        # The default permissions should be based on the current umask
 
356
        umask = osutils.get_umask()
 
357
        t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
 
358
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
 
359
        
 
360
    def test_put_multi(self):
 
361
        t = self.get_transport()
 
362
 
 
363
        if t.is_readonly():
 
364
            return
 
365
        deprecation_msg = symbol_versioning.deprecation_string(
 
366
            t.put_multi, symbol_versioning.zero_eleven)
 
367
        self.assertEqual(2, self.callDeprecated([deprecation_msg],
 
368
            t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
 
369
                          ('d', StringIO('contents\nfor d\n'))]
 
370
            ))
 
371
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
 
372
                [True, False, False, True])
 
373
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
 
374
        self.check_transport_contents('contents\nfor d\n', t, 'd')
 
375
 
 
376
        self.assertEqual(2, self.callDeprecated([deprecation_msg],
 
377
            t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
 
378
                              ('d', StringIO('another contents\nfor d\n'))])
 
379
            ))
 
380
        self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
 
381
        self.check_transport_contents('another contents\nfor d\n', t, 'd')
 
382
 
383
383
    def test_mkdir(self):
384
384
        t = self.get_transport()
385
385