198 |
198 |
# share an object
|
199 |
199 |
cname = self.cnames[0]
|
200 |
200 |
onames = self.objects[cname].keys()
|
201 |
|
oname = onames.pop()
|
202 |
|
url = join_urls(self.pithos_path, self.user, cname, oname)
|
|
201 |
shared1 = onames.pop()
|
|
202 |
url = join_urls(self.pithos_path, self.user, cname, shared1)
|
203 |
203 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')
|
204 |
204 |
self.assertEqual(r.status_code, 202)
|
205 |
205 |
|
|
206 |
# share another object
|
|
207 |
shared2 = onames.pop()
|
|
208 |
url = join_urls(self.pithos_path, self.user, cname, shared2)
|
|
209 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')
|
|
210 |
self.assertEqual(r.status_code, 202)
|
|
211 |
|
|
212 |
# publish another object
|
|
213 |
public1 = onames.pop()
|
|
214 |
url = join_urls(self.pithos_path, self.user, cname, public1)
|
|
215 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
|
216 |
self.assertEqual(r.status_code, 202)
|
|
217 |
|
206 |
218 |
# publish another object
|
207 |
|
other = onames.pop()
|
208 |
|
url = join_urls(self.pithos_path, self.user, cname, other)
|
|
219 |
public2 = onames.pop()
|
|
220 |
url = join_urls(self.pithos_path, self.user, cname, public2)
|
209 |
221 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
210 |
222 |
self.assertEqual(r.status_code, 202)
|
211 |
223 |
|
... | ... | |
216 |
228 |
objects = r.content.split('\n')
|
217 |
229 |
if '' in objects:
|
218 |
230 |
objects.remove('')
|
219 |
|
self.assertEqual([oname], objects)
|
|
231 |
self.assertEqual(sorted([shared1, shared2]), objects)
|
220 |
232 |
|
221 |
233 |
# list detailed shared and assert only the shared object is returned
|
222 |
234 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
226 |
238 |
objects = json.loads(r.content)
|
227 |
239 |
except:
|
228 |
240 |
self.fail('json format expected')
|
229 |
|
self.assertEqual([oname], [o['name'] for o in objects])
|
230 |
|
self.assertTrue('x_object_sharing' in objects[0])
|
231 |
|
self.assertTrue('x_object_public' not in objects[0])
|
|
241 |
l = sorted([shared1, shared2])
|
|
242 |
i = 0
|
|
243 |
for name in l:
|
|
244 |
self.assertEqual(objects[i]['name'], name)
|
|
245 |
self.assertEqual(objects[i]['bytes'],
|
|
246 |
len(self.objects[cname][name]))
|
|
247 |
self.assertTrue('x_object_sharing' in objects[i])
|
|
248 |
self.assertTrue('x_object_public' not in objects[i])
|
|
249 |
i += 1
|
232 |
250 |
|
233 |
251 |
# publish the shared object and assert it is still listed in the
|
234 |
252 |
# shared objects
|
235 |
|
url = join_urls(self.pithos_path, self.user, cname, oname)
|
|
253 |
url = join_urls(self.pithos_path, self.user, cname, shared1)
|
236 |
254 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
237 |
255 |
self.assertEqual(r.status_code, 202)
|
238 |
256 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
242 |
260 |
objects = json.loads(r.content)
|
243 |
261 |
except:
|
244 |
262 |
self.fail('json format expected')
|
245 |
|
self.assertEqual([oname], [o['name'] for o in objects])
|
246 |
|
self.assertTrue('x_object_sharing' in objects[0])
|
247 |
|
self.assertTrue('x_object_public' in objects[0])
|
|
263 |
i = 0
|
|
264 |
for name in l:
|
|
265 |
self.assertEqual(objects[i]['name'], name)
|
|
266 |
self.assertEqual(objects[i]['bytes'],
|
|
267 |
len(self.objects[cname][name]))
|
|
268 |
self.assertTrue('x_object_sharing' in objects[i])
|
|
269 |
if name == shared1:
|
|
270 |
self.assertTrue('x_object_public' in objects[i])
|
|
271 |
else:
|
|
272 |
self.assertTrue('x_object_public' not in objects[i])
|
|
273 |
i += 1
|
248 |
274 |
|
249 |
275 |
# create child object
|
250 |
|
descendant = strnextling(oname)
|
|
276 |
descendant = strnextling(shared1)
|
251 |
277 |
self.upload_object(cname, descendant)
|
252 |
278 |
# request shared and assert child obejct is not listed
|
253 |
279 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
256 |
282 |
objects = r.content.split('\n')
|
257 |
283 |
if '' in objects:
|
258 |
284 |
objects.remove('')
|
259 |
|
self.assertTrue(oname in objects)
|
|
285 |
self.assertTrue(shared1 in objects)
|
260 |
286 |
self.assertTrue(descendant not in objects)
|
261 |
287 |
|
262 |
288 |
# check folder inheritance
|
263 |
|
oname, _ = self.create_folder(cname, HTTP_X_OBJECT_SHARING='read=*')
|
|
289 |
folder, _ = self.create_folder(cname, HTTP_X_OBJECT_SHARING='read=*')
|
264 |
290 |
# create child object
|
265 |
|
descendant = '%s/%s' % (oname, get_random_name())
|
|
291 |
descendant = '%s/%s' % (folder, get_random_name())
|
266 |
292 |
self.upload_object(cname, descendant)
|
267 |
293 |
# request shared
|
268 |
294 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
271 |
297 |
objects = r.content.split('\n')
|
272 |
298 |
if '' in objects:
|
273 |
299 |
objects.remove('')
|
274 |
|
self.assertTrue(oname in objects)
|
|
300 |
self.assertTrue(folder in objects)
|
275 |
301 |
self.assertTrue(descendant in objects)
|
276 |
302 |
|
277 |
303 |
def test_list_public(self):
|
278 |
|
# publish an object
|
279 |
304 |
cname = self.cnames[0]
|
280 |
305 |
onames = self.objects[cname].keys()
|
281 |
|
oname = onames.pop()
|
282 |
|
other = onames.pop()
|
283 |
306 |
|
284 |
307 |
# publish an object
|
285 |
|
url = join_urls(self.pithos_path, self.user, cname, oname)
|
|
308 |
public1 = onames.pop()
|
|
309 |
url = join_urls(self.pithos_path, self.user, cname, public1)
|
286 |
310 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
287 |
311 |
self.assertEqual(r.status_code, 202)
|
288 |
312 |
|
|
313 |
# publish another
|
|
314 |
public2 = onames.pop()
|
|
315 |
url = join_urls(self.pithos_path, self.user, cname, public2)
|
|
316 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
|
317 |
self.assertEqual(r.status_code, 202)
|
|
318 |
|
|
319 |
# share an object
|
|
320 |
shared1 = onames.pop()
|
|
321 |
url = join_urls(self.pithos_path, self.user, cname, shared1)
|
|
322 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
|
|
323 |
self.assertEqual(r.status_code, 202)
|
|
324 |
|
289 |
325 |
# share another
|
290 |
|
url = join_urls(self.pithos_path, self.user, cname, other)
|
|
326 |
shared2 = onames.pop()
|
|
327 |
url = join_urls(self.pithos_path, self.user, cname, shared2)
|
291 |
328 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
|
292 |
329 |
self.assertEqual(r.status_code, 202)
|
293 |
330 |
|
294 |
331 |
# list public and assert only the public object is returned
|
295 |
332 |
url = join_urls(self.pithos_path, self.user, cname)
|
296 |
333 |
r = self.get('%s?public=' % url)
|
297 |
|
objects = r.content.split('\n')
|
298 |
334 |
self.assertEqual(r.status_code, 200)
|
299 |
|
self.assertTrue(oname in r.content.split('\n'))
|
300 |
|
(self.assertTrue(object not in objects) for object in o_names[1:])
|
|
335 |
objects = r.content.split('\n')
|
|
336 |
if '' in objects:
|
|
337 |
objects.remove('')
|
|
338 |
self.assertEqual(sorted([public1, public2]), objects)
|
301 |
339 |
|
302 |
340 |
# list detailed public and assert only the public object is returned
|
303 |
341 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
307 |
345 |
objects = json.loads(r.content)
|
308 |
346 |
except:
|
309 |
347 |
self.fail('json format expected')
|
310 |
|
self.assertEqual([oname], [obj['name'] for obj in objects])
|
311 |
|
self.assertTrue('x_object_sharing' not in objects[0])
|
312 |
|
self.assertTrue('x_object_public' in objects[0])
|
|
348 |
l = sorted([public1, public2])
|
|
349 |
i = 0
|
|
350 |
for name in l:
|
|
351 |
self.assertEqual(objects[i]['name'], name)
|
|
352 |
self.assertEqual(objects[i]['bytes'],
|
|
353 |
len(self.objects[cname][name]))
|
|
354 |
self.assertTrue('x_object_sharing' not in objects[i])
|
|
355 |
self.assertTrue('x_object_public' in objects[i])
|
|
356 |
i += 1
|
313 |
357 |
|
314 |
358 |
# share the public object and assert it is still listed in the
|
315 |
359 |
# public objects
|
316 |
|
url = join_urls(self.pithos_path, self.user, cname, oname)
|
|
360 |
url = join_urls(self.pithos_path, self.user, cname, public1)
|
317 |
361 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
|
318 |
362 |
self.assertEqual(r.status_code, 202)
|
319 |
363 |
url = join_urls(self.pithos_path, self.user, cname)
|
... | ... | |
323 |
367 |
objects = json.loads(r.content)
|
324 |
368 |
except:
|
325 |
369 |
self.fail('json format expected')
|
326 |
|
self.assertEqual([oname], [obj['name'] for obj in objects])
|
327 |
|
self.assertTrue('x_object_sharing' in objects[0])
|
328 |
|
self.assertTrue('x_object_public' in objects[0])
|
|
370 |
i = 0
|
|
371 |
for name in l:
|
|
372 |
self.assertEqual(objects[i]['name'], name)
|
|
373 |
self.assertEqual(objects[i]['bytes'],
|
|
374 |
len(self.objects[cname][name]))
|
|
375 |
if name == public1:
|
|
376 |
self.assertTrue('x_object_sharing' in objects[i])
|
|
377 |
else:
|
|
378 |
self.assertTrue('x_object_sharing' not in objects[i])
|
|
379 |
i += 1
|
329 |
380 |
|
330 |
381 |
url = join_urls(self.pithos_path, self.user, cname)
|
331 |
382 |
|
... | ... | |
339 |
390 |
self.assertEqual(r.status_code, 403)
|
340 |
391 |
|
341 |
392 |
# create child object
|
342 |
|
descendant = strnextling(oname)
|
|
393 |
descendant = strnextling(public1)
|
343 |
394 |
self.upload_object(cname, descendant)
|
344 |
395 |
# request public and assert child obejct is not listed
|
345 |
396 |
r = self.get('%s?public=' % url)
|
... | ... | |
347 |
398 |
if '' in objects:
|
348 |
399 |
objects.remove('')
|
349 |
400 |
self.assertEqual(r.status_code, 200)
|
350 |
|
self.assertTrue(oname in objects)
|
351 |
|
(self.assertTrue(o not in objects) for o in o_names[1:])
|
|
401 |
self.assertTrue(public1 in objects)
|
|
402 |
self.assertTrue(descendant not in objects)
|
352 |
403 |
|
353 |
404 |
# test folder inheritance
|
354 |
|
oname, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
|
|
405 |
folder, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
|
355 |
406 |
# create child object
|
356 |
|
descendant = '%s/%s' % (oname, get_random_name())
|
|
407 |
descendant = '%s/%s' % (folder, get_random_name())
|
357 |
408 |
self.upload_object(cname, descendant)
|
358 |
409 |
# request public
|
359 |
410 |
r = self.get('%s?public=' % url)
|
360 |
411 |
self.assertEqual(r.status_code, 200)
|
361 |
412 |
objects = r.content.split('\n')
|
362 |
|
self.assertTrue(oname in objects)
|
|
413 |
self.assertTrue(folder in objects)
|
363 |
414 |
self.assertTrue(descendant not in objects)
|
364 |
415 |
|
365 |
416 |
def test_list_shared_public(self):
|
366 |
|
# publish an object
|
367 |
417 |
cname = self.cnames[0]
|
368 |
418 |
container_url = join_urls(self.pithos_path, self.user, cname)
|
369 |
419 |
onames = self.objects[cname].keys()
|
370 |
|
oname = onames.pop()
|
371 |
|
url = join_urls(container_url, oname)
|
|
420 |
|
|
421 |
# publish an object
|
|
422 |
public1 = onames.pop()
|
|
423 |
url = join_urls(container_url, public1)
|
372 |
424 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
373 |
425 |
self.assertEqual(r.status_code, 202)
|
374 |
426 |
|
|
427 |
# publish another
|
|
428 |
public2 = onames.pop()
|
|
429 |
url = join_urls(container_url, public2)
|
|
430 |
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
|
|
431 |
self.assertEqual(r.status_code, 202)
|
|
432 |
|
|
433 |
# share an object
|
|
434 |
shared1 = onames.pop()
|
|
435 |
url = join_urls(container_url, shared1)
|
|
436 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
|
|
437 |
self.assertEqual(r.status_code, 202)
|
|
438 |
|
375 |
439 |
# share another
|
376 |
|
other = onames.pop()
|
377 |
|
url = join_urls(container_url, other)
|
|
440 |
shared2 = onames.pop()
|
|
441 |
url = join_urls(container_url, shared2)
|
378 |
442 |
r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
|
379 |
443 |
self.assertEqual(r.status_code, 202)
|
380 |
444 |
|
... | ... | |
382 |
446 |
r = self.get('%s?shared=&public=&format=json' % container_url)
|
383 |
447 |
self.assertEqual(r.status_code, 200)
|
384 |
448 |
objects = json.loads(r.content)
|
385 |
|
self.assertEqual([o['name'] for o in objects], sorted([oname, other]))
|
386 |
|
for o in objects:
|
387 |
|
if o['name'] == oname:
|
388 |
|
self.assertTrue('x_object_public' in o.keys())
|
389 |
|
elif o['name'] == other:
|
390 |
|
self.assertTrue('x_object_sharing' in o.keys())
|
|
449 |
l = sorted([public1, public2, shared1, shared2])
|
|
450 |
i = 0
|
|
451 |
for name in l:
|
|
452 |
self.assertEqual(objects[i]['name'], name)
|
|
453 |
self.assertEqual(objects[i]['bytes'],
|
|
454 |
len(self.objects[cname][name]))
|
|
455 |
self.assertTrue('x_object_sharing' in objects[i] or
|
|
456 |
'x_object_public' in objects[i])
|
|
457 |
i += 1
|
391 |
458 |
|
392 |
459 |
# assert not listing shared and public to a not shared user
|
393 |
460 |
r = self.get('%s?shared=&public=&format=json' % container_url,
|
... | ... | |
400 |
467 |
self.assertEqual(r.status_code, 403)
|
401 |
468 |
|
402 |
469 |
# create child object
|
403 |
|
descendant = strnextling(oname)
|
|
470 |
descendant = strnextling(public1)
|
404 |
471 |
self.upload_object(cname, descendant)
|
405 |
472 |
# request public and assert child obejct is not listed
|
406 |
473 |
r = self.get('%s?shared=&public=' % container_url)
|
... | ... | |
408 |
475 |
if '' in objects:
|
409 |
476 |
objects.remove('')
|
410 |
477 |
self.assertEqual(r.status_code, 200)
|
411 |
|
self.assertTrue(oname in objects)
|
412 |
|
(self.assertTrue(o not in objects) for o in o_names[1:])
|
|
478 |
self.assertEqual(objects, l)
|
413 |
479 |
|
414 |
480 |
# test folder inheritance
|
415 |
|
oname, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
|
|
481 |
folder, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
|
416 |
482 |
# create child object
|
417 |
|
descendant = '%s/%s' % (oname, get_random_name())
|
|
483 |
descendant = '%s/%s' % (folder, get_random_name())
|
418 |
484 |
self.upload_object(cname, descendant)
|
419 |
485 |
# request public
|
420 |
486 |
r = self.get('%s?shared=&public=' % container_url)
|
... | ... | |
422 |
488 |
objects = r.content.split('\n')
|
423 |
489 |
if '' in objects:
|
424 |
490 |
objects.remove('')
|
425 |
|
self.assertTrue(oname in objects)
|
|
491 |
self.assertTrue(folder in objects)
|
426 |
492 |
self.assertTrue(descendant not in objects)
|
427 |
493 |
|
428 |
494 |
def test_list_objects(self):
|