Revision 0c6ab9df snf-pithos-app/pithos/api/test/objects.py
| b/snf-pithos-app/pithos/api/test/objects.py | ||
|---|---|---|
| 41 | 41 |
from pithos.api.test import (PithosAPITest, pithos_settings, |
| 42 | 42 |
AssertMappingInvariant, AssertUUidInvariant, |
| 43 | 43 |
TEST_BLOCK_SIZE, TEST_HASH_ALGORITHM, |
| 44 |
DATE_FORMATS) |
|
| 44 |
DATE_FORMATS, pithos_test_settings)
|
|
| 45 | 45 |
from pithos.api.test.util import (md5_hash, merkle, strnextling, |
| 46 | 46 |
get_random_data, get_random_name) |
| 47 | 47 |
|
| ... | ... | |
| 924 | 924 |
get_random_name(), self.object)) |
| 925 | 925 |
self.assertEqual(r.status_code, 404) |
| 926 | 926 |
|
| 927 |
@pithos_test_settings(API_LIST_LIMIT=10) |
|
| 927 | 928 |
def test_copy_dir(self): |
| 928 | 929 |
folder = self.create_folder(self.container)[0] |
| 929 | 930 |
subfolder = self.create_folder( |
| 930 | 931 |
self.container, oname='%s/%s' % (folder, get_random_name()))[0] |
| 931 | 932 |
objects = [subfolder] |
| 932 | 933 |
append = objects.append |
| 933 |
append(self.upload_object(self.container, |
|
| 934 |
'%s/%s' % (folder, get_random_name()), |
|
| 935 |
depth='1')[0]) |
|
| 936 |
append(self.upload_object(self.container, |
|
| 937 |
'%s/%s' % (subfolder, get_random_name()), |
|
| 938 |
depth='2')[0]) |
|
| 934 |
for i in range(11): |
|
| 935 |
append(self.upload_object(self.container, |
|
| 936 |
'%s/%d' % (folder, i), |
|
| 937 |
depth='1')[0]) |
|
| 939 | 938 |
other = self.upload_object(self.container, strnextling(folder))[0] |
| 940 | 939 |
|
| 941 | 940 |
# copy dir |
| ... | ... | |
| 1006 | 1005 |
r = self.head(url) |
| 1007 | 1006 |
self.assertEqual(r.status_code, 404) |
| 1008 | 1007 |
|
| 1008 |
@pithos_test_settings(API_LIST_LIMIT=10) |
|
| 1009 | 1009 |
def test_move_dir(self): |
| 1010 | 1010 |
folder = self.create_folder(self.container)[0] |
| 1011 | 1011 |
subfolder = self.create_folder( |
| 1012 | 1012 |
self.container, oname='%s/%s' % (folder, get_random_name()))[0] |
| 1013 | 1013 |
objects = [subfolder] |
| 1014 | 1014 |
append = objects.append |
| 1015 |
append(self.upload_object(self.container, |
|
| 1016 |
'%s/%s' % (folder, get_random_name()), |
|
| 1017 |
depth='1')[0]) |
|
| 1018 |
append(self.upload_object(self.container, |
|
| 1019 |
'%s/%s' % (subfolder, get_random_name()), |
|
| 1020 |
depth='1')[0]) |
|
| 1015 |
for i in range(11): |
|
| 1016 |
append(self.upload_object(self.container, |
|
| 1017 |
'%s/%d' % (folder, i), |
|
| 1018 |
depth='1')[0]) |
|
| 1021 | 1019 |
other = self.upload_object(self.container, strnextling(folder))[0] |
| 1022 | 1020 |
|
| 1023 | 1021 |
# move dir |
| ... | ... | |
| 1212 | 1210 |
self.assertTrue('X-Object-Hash' in r)
|
| 1213 | 1211 |
self.assertEqual(r['X-Object-Hash'], self.etag) |
| 1214 | 1212 |
|
| 1215 |
# assert source object still exists
|
|
| 1213 |
# assert source object still exists |
|
| 1216 | 1214 |
url = join_urls(self.pithos_path, self.user, self.container, |
| 1217 | 1215 |
self.object) |
| 1218 | 1216 |
r = self.head(url) |
| ... | ... | |
| 1229 | 1227 |
self.assertTrue('X-Object-Hash' in r)
|
| 1230 | 1228 |
self.assertEqual(r['X-Object-Hash'], self.etag) |
| 1231 | 1229 |
|
| 1230 |
@pithos_test_settings(API_LIST_LIMIT=10) |
|
| 1231 |
def test_copy_dir_contents(self): |
|
| 1232 |
folder = self.create_folder(self.container)[0] |
|
| 1233 |
subfolder = self.create_folder( |
|
| 1234 |
self.container, oname='%s/%s' % (folder, get_random_name()))[0] |
|
| 1235 |
objects = [subfolder] |
|
| 1236 |
append = objects.append |
|
| 1237 |
for i in range(11): |
|
| 1238 |
append(self.upload_object(self.container, |
|
| 1239 |
'%s/%d' % (folder, i), |
|
| 1240 |
depth='1')[0]) |
|
| 1241 |
other = self.upload_object(self.container, strnextling(folder))[0] |
|
| 1242 |
|
|
| 1243 |
# copy dir |
|
| 1244 |
url = join_urls(self.pithos_path, self.user, self.container, |
|
| 1245 |
folder) |
|
| 1246 |
copy_folder = self.create_folder(self.container)[0] |
|
| 1247 |
r = self.copy('%s?delimiter=/' % url,
|
|
| 1248 |
HTTP_X_OBJECT_META_TEST='testcopy', |
|
| 1249 |
HTTP_DESTINATION='/%s/%s' % (self.container, |
|
| 1250 |
copy_folder)) |
|
| 1251 |
self.assertEqual(r.status_code, 201) |
|
| 1252 |
|
|
| 1253 |
for obj in objects: |
|
| 1254 |
# assert object exists |
|
| 1255 |
url = join_urls(self.pithos_path, self.user, self.container, |
|
| 1256 |
obj.replace(folder, copy_folder)) |
|
| 1257 |
r = self.head(url) |
|
| 1258 |
self.assertEqual(r.status_code, 200) |
|
| 1259 |
|
|
| 1260 |
# assert other has not been created under copy folder |
|
| 1261 |
url = join_urls(self.pithos_path, self.user, self.container, |
|
| 1262 |
'%s/%s' % (copy_folder, |
|
| 1263 |
other.replace(folder, copy_folder))) |
|
| 1264 |
r = self.head(url) |
|
| 1265 |
self.assertEqual(r.status_code, 404) |
|
| 1266 |
|
|
| 1232 | 1267 |
def test_copy_to_other_account(self): |
| 1233 | 1268 |
# create a container under alice account |
| 1234 | 1269 |
cname = self.create_container(user='alice')[0] |
| ... | ... | |
| 1347 | 1382 |
r = self.head(url) |
| 1348 | 1383 |
self.assertEqual(r.status_code, 404) |
| 1349 | 1384 |
|
| 1385 |
@pithos_test_settings(API_LIST_LIMIT=10) |
|
| 1386 |
def test_move_dir_contents(self): |
|
| 1387 |
folder = self.create_folder(self.container)[0] |
|
| 1388 |
subfolder = self.create_folder( |
|
| 1389 |
self.container, oname='%s/%s' % (folder, get_random_name()))[0] |
|
| 1390 |
objects = [subfolder] |
|
| 1391 |
append = objects.append |
|
| 1392 |
for i in range(11): |
|
| 1393 |
append(self.upload_object(self.container, |
|
| 1394 |
'%s/%d' % (folder, i), |
|
| 1395 |
depth='1')[0]) |
|
| 1396 |
other = self.upload_object(self.container, strnextling(folder))[0] |
|
| 1397 |
|
|
| 1398 |
# copy dir |
|
| 1399 |
url = join_urls(self.pithos_path, self.user, self.container, folder) |
|
| 1400 |
copy_folder = self.create_folder(self.container)[0] |
|
| 1401 |
r = self.move('%s?delimiter=/' % url,
|
|
| 1402 |
HTTP_X_OBJECT_META_TEST='testcopy', |
|
| 1403 |
HTTP_DESTINATION='/%s/%s' % (self.container, |
|
| 1404 |
copy_folder)) |
|
| 1405 |
self.assertEqual(r.status_code, 201) |
|
| 1406 |
|
|
| 1407 |
for obj in objects: |
|
| 1408 |
# assert object exists |
|
| 1409 |
url = join_urls(self.pithos_path, self.user, self.container, |
|
| 1410 |
obj.replace(folder, copy_folder)) |
|
| 1411 |
r = self.head(url) |
|
| 1412 |
self.assertEqual(r.status_code, 200) |
|
| 1413 |
|
|
| 1414 |
# assert other has not been created under copy folder |
|
| 1415 |
url = join_urls(self.pithos_path, self.user, self.container, |
|
| 1416 |
'%s/%s' % (copy_folder, |
|
| 1417 |
other.replace(folder, copy_folder))) |
|
| 1418 |
r = self.head(url) |
|
| 1419 |
self.assertEqual(r.status_code, 404) |
|
| 1420 |
|
|
| 1350 | 1421 |
def test_move_to_other_container(self): |
| 1351 | 1422 |
# move object to other container (not existing) |
| 1352 | 1423 |
cname = get_random_name() |
| ... | ... | |
| 1920 | 1991 |
r = self.delete(url) |
| 1921 | 1992 |
self.assertEqual(r.status_code, 404) |
| 1922 | 1993 |
|
| 1994 |
@pithos_test_settings(API_LIST_LIMIT=10) |
|
| 1923 | 1995 |
def test_delete_dir(self): |
| 1924 | 1996 |
folder = self.create_folder(self.container)[0] |
| 1925 | 1997 |
subfolder = self.create_folder( |
| 1926 | 1998 |
self.container, oname='%s/%s' % (folder, get_random_name()))[0] |
| 1927 | 1999 |
objects = [subfolder] |
| 1928 | 2000 |
append = objects.append |
| 1929 |
append(self.upload_object(self.container, |
|
| 1930 |
'%s/%s' % (folder, get_random_name()), |
|
| 1931 |
depth='1')[0]) |
|
| 1932 |
append(self.upload_object(self.container, |
|
| 1933 |
'%s/%s' % (subfolder, get_random_name()), |
|
| 1934 |
depth='2')[0]) |
|
| 2001 |
for i in range(11): |
|
| 2002 |
append(self.upload_object(self.container, |
|
| 2003 |
'%s/%d' % (folder, i), |
|
| 2004 |
depth='1')[0]) |
|
| 1935 | 2005 |
other = self.upload_object(self.container, strnextling(folder))[0] |
| 1936 | 2006 |
|
| 1937 | 2007 |
# move dir |
Also available in: Unified diff