本文整理汇总了Python中test.functional.tests.Utils类的典型用法代码示例。如果您正苦于以下问题：Python Utils类的具体用法？Python Utils怎么用？Python Utils使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Utils类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setUp
def setUp(cls):
if cls.tempurl_enabled is None:
cls.tempurl_enabled = 'tempurl' in cluster_info
if not cls.tempurl_enabled:
return
super(TestContainerTempurlEnv, cls).setUp()
cls.tempurl_key = Utils.create_name()
cls.tempurl_key2 = Utils.create_name()
# creating another account and connection
# for ACL tests
config2 = deepcopy(tf.config)
config2['account'] = tf.config['account2']
config2['username'] = tf.config['username2']
config2['password'] = tf.config['password2']
cls.conn2 = Connection(config2)
cls.conn2.authenticate()
cls.account2 = Account(
cls.conn2, config2.get('account', config2['username']))
cls.account2 = cls.conn2.get_account()
cls.container = cls.account.container(Utils.create_name())
if not cls.container.create({
'x-container-meta-temp-url-key': cls.tempurl_key,
'x-container-meta-temp-url-key-2': cls.tempurl_key2,
'x-container-read': cls.account2.name}):
raise ResponseError(cls.conn.response)
cls.obj = cls.container.file(Utils.create_name())
cls.obj.write("obj contents")
cls.other_obj = cls.container.file(Utils.create_name())
cls.other_obj.write("other obj contents")
开发者ID:matthewoliver，项目名称:swift，代码行数:34，代码来源:test_tempurl.py
示例2: setUp
def setUp(self):
if 'slo' not in cluster_info:
raise SkipTest("SLO not enabled")
self.conn = Connection(tf.config)
self.conn.authenticate()
self.account = Account(
self.conn, tf.config.get('account', tf.config['username']))
self.account.delete_containers()
# create a container with versioning
self.versions_container = self.account.container(Utils.create_name())
self.container = self.account.container(Utils.create_name())
self.segments_container = self.account.container(Utils.create_name())
if not self.container.create(
hdrs={'X-Versions-Location': self.versions_container.name}):
raise ResponseError(self.conn.response)
if 'versions' not in self.container.info():
raise SkipTest("Object versioning not enabled")
for cont in (self.versions_container, self.segments_container):
if not cont.create():
raise ResponseError(self.conn.response)
# create some segments
self.seg_info = {}
for letter, size in (('a', 1024 * 1024),
('b', 1024 * 1024)):
seg_name = letter
file_item = self.segments_container.file(seg_name)
file_item.write(letter * size)
self.seg_info[seg_name] = {
'size_bytes': size,
'etag': file_item.md5,
'path': '/%s/%s' % (self.segments_container.name, seg_name)}
开发者ID:psachin，项目名称:swift，代码行数:35，代码来源:test_versioned_writes.py
示例3: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.conn.authenticate()
if cls.tempurl_enabled is None:
cls.tempurl_enabled = 'tempurl' in cluster_info
if not cls.tempurl_enabled:
return
cls.tempurl_key = Utils.create_name()
cls.tempurl_key2 = Utils.create_name()
cls.account = Account(
cls.conn, tf.config.get('account', tf.config['username']))
cls.account.delete_containers()
cls.account.update_metadata({
'temp-url-key': cls.tempurl_key,
'temp-url-key-2': cls.tempurl_key2
})
cls.container = cls.account.container(Utils.create_name())
if not cls.container.create():
raise ResponseError(cls.conn.response)
cls.obj = cls.container.file(Utils.create_name())
cls.obj.write("obj contents")
cls.other_obj = cls.container.file(Utils.create_name())
cls.other_obj.write("other obj contents")
开发者ID:hurricanerix，项目名称:swift，代码行数:28，代码来源:test_tempurl.py
示例4: test_GET_DLO_outside_container
def test_GET_DLO_outside_container(self):
seg1 = self.env.container.file(
"get-dlo-outside-seg1" + Utils.create_name())
seg2 = self.env.container.file(
"get-dlo-outside-seg2" + Utils.create_name())
seg1.write("one fish two fish ")
seg2.write("red fish blue fish")
container2 = self.env.account.container(Utils.create_name())
container2.create()
manifest = container2.file("manifest" + Utils.create_name())
manifest.write(
'',
hdrs={"X-Object-Manifest": "%s/get-dlo-outside-seg" %
(self.env.container.name,)})
expires = int(time()) + 86400
parms = self.tempurl_parms(
'GET', expires, self.env.conn.make_path(manifest.path),
self.env.tempurl_key)
# cross container tempurl works fine for account tempurl key
contents = manifest.read(parms=parms, cfg={'no_auth_token': True})
self.assertEqual(contents, "one fish two fish red fish blue fish")
self.assert_status([200])
开发者ID:matthewoliver，项目名称:swift，代码行数:26，代码来源:test_tempurl.py
示例5: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.conn.authenticate()
config2 = tf.config.copy()
config2['username'] = tf.config['username3']
config2['password'] = tf.config['password3']
cls.conn2 = Connection(config2)
cls.conn2.authenticate()
cls.account = Account(cls.conn, tf.config.get('account',
tf.config['username']))
cls.account.delete_containers()
cls.container = cls.account.container(Utils.create_name())
cls.container2 = cls.account.container(Utils.create_name())
for cont in (cls.container, cls.container2):
if not cont.create():
raise ResponseError(cls.conn.response)
# avoid getting a prefix that stops halfway through an encoded
# character
prefix = Utils.create_name().decode("utf-8")[:10].encode("utf-8")
cls.segment_prefix = prefix
for letter in ('a', 'b', 'c', 'd', 'e'):
file_item = cls.container.file("%s/seg_lower%s" % (prefix, letter))
file_item.write(letter * 10)
file_item = cls.container.file("%s/seg_upper%s" % (prefix, letter))
file_item.write(letter.upper() * 10)
for letter in ('f', 'g', 'h', 'i', 'j'):
file_item = cls.container2.file("%s/seg_lower%s" %
(prefix, letter))
file_item.write(letter * 10)
man1 = cls.container.file("man1")
man1.write('man1-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_lower" %
(cls.container.name, prefix)})
man2 = cls.container.file("man2")
man2.write('man2-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_upper" %
(cls.container.name, prefix)})
manall = cls.container.file("manall")
manall.write('manall-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg" %
(cls.container.name, prefix)})
mancont2 = cls.container.file("mancont2")
mancont2.write(
'mancont2-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_lower" %
(cls.container2.name, prefix)})
开发者ID:hurricanerix，项目名称:swift，代码行数:58，代码来源:test_dlo.py
示例6: test_PUT_manifest_access
def test_PUT_manifest_access(self):
new_obj = self.env.container.file(Utils.create_name())
# give out a signature which allows a PUT to new_obj
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'PUT', expires, self.env.conn.make_path(new_obj.path),
self.env.tempurl_key)
put_parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
# try to create manifest pointing to some random container
try:
new_obj.write('', {
'x-object-manifest': '%s/foo' % 'some_random_container'
}, parms=put_parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
# create some other container
other_container = self.env.account.container(Utils.create_name())
if not other_container.create():
raise ResponseError(self.conn.response)
# try to create manifest pointing to new container
try:
new_obj.write('', {
'x-object-manifest': '%s/foo' % other_container
}, parms=put_parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
# try again using a tempurl POST to an already created object
new_obj.write('', {}, parms=put_parms, cfg={'no_auth_token': True})
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'POST', expires, self.env.conn.make_path(new_obj.path),
self.env.tempurl_key)
post_parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
try:
new_obj.post({'x-object-manifest': '%s/foo' % other_container},
parms=post_parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
开发者ID:hurricanerix，项目名称:swift，代码行数:51，代码来源:test_tempurl.py
示例7: setUp
def setUp(cls):
super(TestDloEnv, cls).setUp()
cls.container = cls.account.container(Utils.create_name())
cls.container2 = cls.account.container(Utils.create_name())
for cont in (cls.container, cls.container2):
if not cont.create():
raise ResponseError(cls.conn.response)
# avoid getting a prefix that stops halfway through an encoded
# character
prefix = Utils.create_name().decode("utf-8")[:10].encode("utf-8")
cls.segment_prefix = prefix
for letter in ('a', 'b', 'c', 'd', 'e'):
file_item = cls.container.file("%s/seg_lower%s" % (prefix, letter))
file_item.write(letter * 10)
file_item = cls.container.file(
"%s/seg_upper_%%ff%s" % (prefix, letter))
file_item.write(letter.upper() * 10)
for letter in ('f', 'g', 'h', 'i', 'j'):
file_item = cls.container2.file("%s/seg_lower%s" %
(prefix, letter))
file_item.write(letter * 10)
man1 = cls.container.file("man1")
man1.write('man1-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_lower" %
(cls.container.name, prefix)})
man2 = cls.container.file("man2")
man2.write('man2-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_upper_%%25ff" %
(cls.container.name, prefix)})
manall = cls.container.file("manall")
manall.write('manall-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg" %
(cls.container.name, prefix)})
mancont2 = cls.container.file("mancont2")
mancont2.write(
'mancont2-contents',
hdrs={"X-Object-Manifest": "%s/%s/seg_lower" %
(cls.container2.name, prefix)})
开发者ID:mahak，项目名称:swift，代码行数:48，代码来源:test_dlo.py
示例8: test_PUT_large_object
def test_PUT_large_object(self):
file_item = self.env.container.file(Utils.create_name())
data = File.random_data(1024 * 1024 * 2)
self.assertTrue(file_item.write(data))
self.assert_status(201)
self.assertTrue(data == file_item.read())
self.assert_status(200)
开发者ID:gluster，项目名称:gluster-swift，代码行数:7，代码来源:gluster_swift_tests.py
示例9: _test_versioning_check_acl_setup
def _test_versioning_check_acl_setup(self):
container = self.env.container
versions_container = self.env.versions_container
versions_container.create(hdrs={'X-Container-Read': '.r:*,.rlistings'})
obj_name = Utils.create_name()
versioned_obj = container.file(obj_name)
versioned_obj.write("aaaaa")
self.assertEqual("aaaaa", versioned_obj.read())
versioned_obj.write("bbbbb")
self.assertEqual("bbbbb", versioned_obj.read())
# Use token from second account and try to delete the object
org_token = self.env.account.conn.storage_token
self.env.account.conn.storage_token = self.env.conn2.storage_token
try:
with self.assertRaises(ResponseError) as cm:
versioned_obj.delete()
self.assertEqual(403, cm.exception.status)
finally:
self.env.account.conn.storage_token = org_token
# Verify with token from first account
self.assertEqual("bbbbb", versioned_obj.read())
return versioned_obj
开发者ID:nadeemsyed，项目名称:swift，代码行数:26，代码来源:test_versioned_writes.py
示例10: test_versioning_dlo
def test_versioning_dlo(self):
container = self.env.container
versions_container = self.env.versions_container
obj_name = Utils.create_name()
for i in ('1', '2', '3'):
time.sleep(.01) # guarantee that the timestamp changes
obj_name_seg = obj_name + '/' + i
versioned_obj = container.file(obj_name_seg)
versioned_obj.write(i)
versioned_obj.write(i + i)
self.assertEqual(3, versions_container.info()['object_count'])
man_file = container.file(obj_name)
man_file.write('', hdrs={"X-Object-Manifest": "%s/%s/" %
(self.env.container.name, obj_name)})
# guarantee that the timestamp changes
time.sleep(.01)
# write manifest file again
man_file.write('', hdrs={"X-Object-Manifest": "%s/%s/" %
(self.env.container.name, obj_name)})
self.assertEqual(3, versions_container.info()['object_count'])
self.assertEqual("112233", man_file.read())
开发者ID:bebule，项目名称:swift，代码行数:27，代码来源:test_versioned_writes.py
示例11: test_object_expiry_X_Delete_At_PUT
def test_object_expiry_X_Delete_At_PUT(self):
obj = self.env.container.file(Utils.create_name())
x_delete_at = str(int(time.time()) + 2)
obj.write_random(self.env.file_size,
hdrs={'X-Delete-At': x_delete_at})
# Object is not expired. Should still be accessible.
obj.read()
self.assert_status(200)
# Ensure X-Delete-At is saved as object metadata.
self.assertEqual(x_delete_at, str(obj.info()['x_delete_at']))
# Wait for object to be expired.
time.sleep(3)
# Object has expired. Should no longer be accessible.
self.assertRaises(ResponseError, obj.read)
self.assert_status(404)
# Object should still be present on filesystem.
self.assertTrue(os.path.isfile(os.path.join(self.env.root_dir,
self.env.container.name,
obj.name)))
# But, GET on container should list the expired object.
result = self.env.container.files()
self.assertTrue(obj.name in self.env.container.files())
# Check existence of corresponding tracker object in gsexpiring
# account.
enteredLoop = False
for c in self.env.client.iter_containers("gsexpiring"):
for o in self.env.client.iter_objects("gsexpiring", c['name']):
enteredLoop = True
l = o['name'].split('/')
self.assertTrue(l[0].endswith('AUTH_' + self.env.account.name))
self.assertEqual(l[1], self.env.container.name)
self.assertEqual(l[2], obj.name)
if not enteredLoop:
self.fail("Tracker object not found.")
# Run expirer daemon once.
self.env.expirer.once()
# Ensure object is physically deleted from filesystem.
self.assertFalse(os.path.exists(os.path.join(self.env.root_dir,
self.env.container.name,
obj.name)))
# Ensure tracker object is consumed.
try:
self.env.client.iter_containers("gsexpiring").next()
except StopIteration:
pass
else:
self.fail("Tracker object persists!")
# GET on container should no longer list the object.
self.assertFalse(obj.name in self.env.container.files())
开发者ID:gluster，项目名称:gluster-swift，代码行数:60，代码来源:test_object_expirer_swift.py
示例12: testObjectMetadataWhenFileModified
def testObjectMetadataWhenFileModified(self):
data = "I'm whatever Gotham needs me to be "
data_hash = hashlib.md5(data).hexdigest()
# Create an object through object interface
object_name = Utils.create_name()
object_item = self.env.container.file(object_name)
object_item.write(data)
# Make sure GET works
self.assertEqual(data, object_item.read())
self.assert_status(200)
# Check Etag is right
self.assertEqual(data_hash, object_item.info()['etag'])
self.assert_status(200)
# Extend/append more data to file from filesystem interface
file_path = os.path.join(self.env.root_dir,
'AUTH_' + self.env.account.name,
self.env.container.name,
object_name)
more_data = "- Batman"
with open(file_path, 'a') as f:
f.write(more_data)
total_data = data + more_data
total_data_hash = hashlib.md5(total_data).hexdigest()
# Make sure GET works
self.assertEqual(total_data, object_item.read())
self.assert_status(200)
# Check Etag and content-length is right
metadata = object_item.info()
self.assert_status(200)
self.assertEqual(total_data_hash, metadata['etag'])
self.assertEqual(len(total_data), int(metadata['content_length']))
# Re-write the file to be shorter
new_data = "I am Batman"
new_data_hash = hashlib.md5(new_data).hexdigest()
with open(file_path, 'w') as f:
f.write(new_data)
# Make sure GET works
self.assertEqual(new_data, object_item.read())
self.assert_status(200)
# Check Etag and content-length is right
metadata = object_item.info()
self.assert_status(200)
self.assertEqual(new_data_hash, metadata['etag'])
self.assertEqual(len(new_data), int(metadata['content_length']))
# Modify the file but let the length remain same
new_data = "I am Antman"
new_data_hash = hashlib.md5(new_data).hexdigest()
with open(file_path, 'w') as f:
f.write(new_data)
# Make sure GET works
self.assertEqual(new_data, object_item.read())
self.assert_status(200)
# Check Etag and content-length is right
metadata = object_item.info()
self.assert_status(200)
self.assertEqual(new_data_hash, metadata['etag'])
self.assertEqual(len(new_data), int(metadata['content_length']))
开发者ID:jeffrolland，项目名称:swiftonfile，代码行数:60，代码来源:swift_on_file_tests.py
示例13: test_dlo_post_with_manifest_regular_object
def test_dlo_post_with_manifest_regular_object(self):
# verify that performing a POST to a regular object
# with a manifest header will create a DLO.
# Put a regular object
file_item = self.env.container.file(Utils.create_name())
file_item.write('file contents', hdrs={})
# sanity checks
file_contents = file_item.read(parms={})
self.assertEqual('file contents', file_contents)
# get the path associated with man1
x_o_m = self.env.container.file('man1').info()['x_object_manifest']
# POST a x-object-manifest value to the regular object
file_item.post({'x-object-manifest': x_o_m})
# verify that the file is now a manifest
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('file contents', manifest_contents)
expected_contents = ''.join([(c * 10) for c in 'abcde'])
contents = file_item.read(parms={})
self.assertEqual(expected_contents, contents)
file_item.info()
resp_headers = file_item.conn.response.getheaders()
self.assertIn(('x-object-manifest', x_o_m), resp_headers)
开发者ID:hurricanerix，项目名称:swift，代码行数:27，代码来源:test_dlo.py
示例14: test_dlo_post_without_manifest_header
def test_dlo_post_without_manifest_header(self):
# verify that a POST to a DLO manifest object with no
# x-object-manifest header will cause the existing x-object-manifest
# header to be lost
# create a new manifest for this test to avoid test coupling.
x_o_m = self.env.container.file('man1').info()['x_object_manifest']
file_item = self.env.container.file(Utils.create_name())
file_item.write('manifest-contents', hdrs={"X-Object-Manifest": x_o_m})
# sanity checks
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('manifest-contents', manifest_contents)
expected_contents = ''.join([(c * 10) for c in 'abcde'])
contents = file_item.read(parms={})
self.assertEqual(expected_contents, contents)
# POST with no x-object-manifest header
file_item.post({})
# verify that existing x-object-manifest was removed
info = file_item.info()
self.assertNotIn('x_object_manifest', info)
# verify that object content was not changed
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('manifest-contents', manifest_contents)
# verify that object is no longer a manifest
contents = file_item.read(parms={})
self.assertEqual('manifest-contents', contents)
开发者ID:hurricanerix，项目名称:swift，代码行数:31，代码来源:test_dlo.py
示例15: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.conn.authenticate()
if cls.static_web_enabled is None:
cls.static_web_enabled = 'staticweb' in cluster_info
if not cls.static_web_enabled:
return
cls.account = Account(
cls.conn, tf.config.get('account', tf.config['username']))
cls.account.delete_containers()
cls.container = cls.account.container(Utils.create_name())
if not cls.container.create(
hdrs={'X-Container-Read': '.r:*,.rlistings'}):
raise ResponseError(cls.conn.response)
objects = ['index',
'error',
'listings_css',
'dir/',
'dir/obj',
'dir/subdir/',
'dir/subdir/obj']
cls.objects = {}
for item in sorted(objects):
parent = None
if '/' in item.rstrip('/'):
parent, _ = item.rstrip('/').rsplit('/', 1)
path = '%s/%s' % (cls.objects[parent + '/'].name,
Utils.create_name())
else:
path = Utils.create_name()
if item[-1] == '/':
cls.objects[item] = cls.container.file(path)
cls.objects[item].write(hdrs={
'Content-Type': 'application/directory'})
else:
cls.objects[item] = cls.container.file(path)
cls.objects[item].write('%s contents' % item)
开发者ID:matthewoliver，项目名称:swift，代码行数:43，代码来源:test_staticweb.py
示例16: test_dlo_post_with_manifest_header
def test_dlo_post_with_manifest_header(self):
# verify that performing a POST to a DLO manifest
# preserves the fact that it is a manifest file.
# verify that the x-object-manifest header may be updated.
# create a new manifest for this test to avoid test coupling.
x_o_m = self.env.container.file('man1').info()['x_object_manifest']
file_item = self.env.container.file(Utils.create_name())
file_item.write('manifest-contents', hdrs={"X-Object-Manifest": x_o_m})
# sanity checks
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('manifest-contents', manifest_contents)
expected_contents = ''.join([(c * 10) for c in 'abcde'])
contents = file_item.read(parms={})
self.assertEqual(expected_contents, contents)
# POST a modified x-object-manifest value
new_x_o_m = x_o_m.rstrip('lower') + 'upper'
file_item.post({'x-object-meta-foo': 'bar',
'x-object-manifest': new_x_o_m})
# verify that x-object-manifest was updated
file_item.info()
resp_headers = file_item.conn.response.getheaders()
self.assertIn(('x-object-manifest', new_x_o_m), resp_headers)
self.assertIn(('x-object-meta-foo', 'bar'), resp_headers)
# verify that manifest content was not changed
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('manifest-contents', manifest_contents)
# verify that updated manifest points to new content
expected_contents = ''.join([(c * 10) for c in 'ABCDE'])
contents = file_item.read(parms={})
self.assertEqual(expected_contents, contents)
# Now revert the manifest to point to original segments, including a
# multipart-manifest=get param just to check that has no effect
file_item.post({'x-object-manifest': x_o_m},
parms={'multipart-manifest': 'get'})
# verify that x-object-manifest was reverted
info = file_item.info()
self.assertIn('x_object_manifest', info)
self.assertEqual(x_o_m, info['x_object_manifest'])
# verify that manifest content was not changed
manifest_contents = file_item.read(parms={'multipart-manifest': 'get'})
self.assertEqual('manifest-contents', manifest_contents)
# verify that updated manifest points new content
expected_contents = ''.join([(c * 10) for c in 'abcde'])
contents = file_item.read(parms={})
self.assertEqual(expected_contents, contents)
开发者ID:hurricanerix，项目名称:swift，代码行数:55，代码来源:test_dlo.py
示例17: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.conn.authenticate()
cls.account = Account(cls.conn, tf.config.get('account',
tf.config['username']))
#cls.root_dir = os.path.join('/mnt/swiftonhpss/test')
cls.root_dir = os.path.join('/srv/hpss')
cls.account.delete_containers()
cls.file_size = 8
cls.container = cls.account.container(Utils.create_name())
if not cls.container.create(hdrs={'X-Storage-Policy': 'swiftonhpss'}):
raise ResponseError(cls.conn.response)
cls.dirs = [
'dir1',
'dir2',
'dir1/subdir1',
'dir1/subdir2',
'dir1/subdir1/subsubdir1',
'dir1/subdir1/subsubdir2',
'dir1/subdir with spaces',
'dir1/subdir+with{whatever',
]
cls.files = [
'file1',
'file A',
'dir1/file2',
'dir1/subdir1/file2',
'dir1/subdir1/file3',
'dir1/subdir1/file4',
'dir1/subdir1/subsubdir1/file5',
'dir1/subdir1/subsubdir1/file6',
'dir1/subdir1/subsubdir1/file7',
'dir1/subdir1/subsubdir1/file8',
'dir1/subdir1/subsubdir2/file9',
'dir1/subdir1/subsubdir2/file0',
'dir1/subdir with spaces/file B',
'dir1/subdir+with{whatever/file D',
]
stored_files = set()
for d in cls.dirs:
file = cls.container.file(d)
file.write(hdrs={'Content-Type': 'application/directory'})
for f in cls.files:
file = cls.container.file(f)
file.write_random(cls.file_size, hdrs={'Content-Type':
'application/octet-stream'})
stored_files.add(f)
cls.stored_files = sorted(stored_files)
cls.sorted_objects = sorted(set(cls.dirs + cls.files))
开发者ID:openstack，项目名称:swiftonhpss，代码行数:54，代码来源:swift_on_file_tests.py
示例18: testObjectContentFromMountPoint
def testObjectContentFromMountPoint(self):
file_name = Utils.create_name()
file_item = self.env.container.file(file_name)
data = file_item.write_random()
self.assert_status(201)
file_info = file_item.info()
fhOnMountPoint = open(os.path.join(self.env.root_dir,self.env.container.name,file_name),'r')
data_read_from_mountP = fhOnMountPoint.read()
md5_returned = hashlib.md5(data_read_from_mountP).hexdigest()
self.assertEquals(md5_returned,file_info['etag'])
fhOnMountPoint.close()
开发者ID:gluster，项目名称:gluster-swift，代码行数:11，代码来源:gluster_swift_tests.py
示例19: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.storage_url, cls.storage_token = cls.conn.authenticate()
cls.account = Account(cls.conn, tf.config.get('account',
tf.config['username']))
# Second connection for ACL tests
config2 = deepcopy(tf.config)
config2['account'] = tf.config['account2']
config2['username'] = tf.config['username2']
config2['password'] = tf.config['password2']
cls.conn2 = Connection(config2)
cls.conn2.authenticate()
# avoid getting a prefix that stops halfway through an encoded
# character
prefix = Utils.create_name().decode("utf-8")[:10].encode("utf-8")
cls.versions_container = cls.account.container(prefix + "-versions")
if not cls.versions_container.create():
raise ResponseError(cls.conn.response)
cls.container = cls.account.container(prefix + "-objs")
container_headers = {
cls.location_header_key: cls.versions_container.name}
if not cls.container.create(hdrs=container_headers):
if cls.conn.response.status == 412:
cls.versioning_enabled = False
return
raise ResponseError(cls.conn.response)
container_info = cls.container.info()
# if versioning is off, then cls.location_header_key won't persist
cls.versioning_enabled = 'versions' in container_info
# setup another account to test ACLs
config2 = deepcopy(tf.config)
config2['account'] = tf.config['account2']
config2['username'] = tf.config['username2']
config2['password'] = tf.config['password2']
cls.conn2 = Connection(config2)
cls.storage_url2, cls.storage_token2 = cls.conn2.authenticate()
cls.account2 = cls.conn2.get_account()
cls.account2.delete_containers()
# setup another account with no access to anything to test ACLs
config3 = deepcopy(tf.config)
config3['account'] = tf.config['account']
config3['username'] = tf.config['username3']
config3['password'] = tf.config['password3']
cls.conn3 = Connection(config3)
cls.storage_url3, cls.storage_token3 = cls.conn3.authenticate()
cls.account3 = cls.conn3.get_account()
开发者ID:hurricanerix，项目名称:swift，代码行数:54，代码来源:test_versioned_writes.py
示例20: setUp
def setUp(cls):
cls.conn = Connection(tf.config)
cls.conn.authenticate()
cls.account = Account(cls.conn, tf.config.get("account", tf.config["username"]))
# PGB - change hardcoded SoF mountpoint
# cls.root_dir = os.path.join('/mnt/swiftonhpss/test')
cls.root_dir = os.path.join("/srv/swift/hpss")
cls.account.delete_containers()
cls.file_size = 8
cls.container = cls.account.container(Utils.create_name())
if not cls.container.create(None, None):
raise ResponseError(cls.conn.response)
cls.dirs = [
"dir1",
"dir2",
"dir1/subdir1",
"dir1/subdir2",
"dir1/subdir1/subsubdir1",
"dir1/subdir1/subsubdir2",
"dir1/subdir with spaces",
"dir1/subdir+with{whatever",
]
cls.files = [
"file1",
"file A",
"dir1/file2",
"dir1/subdir1/file2",
"dir1/subdir1/file3",
"dir1/subdir1/file4",
"dir1/subdir1/subsubdir1/file5",
"dir1/subdir1/subsubdir1/file6",
"dir1/subdir1/subsubdir1/file7",
"dir1/subdir1/subsubdir1/file8",
"dir1/subdir1/subsubdir2/file9",
"dir1/subdir1/subsubdir2/file0",
"dir1/subdir with spaces/file B",
"dir1/subdir+with{whatever/file D",
]
stored_files = set()
for d in cls.dirs:
file = cls.container.file(d)
file.write(hdrs={"Content-Type": "application/directory"})
for f in cls.files:
file = cls.container.file(f)
file.write_random(cls.file_size, hdrs={"Content-Type": "application/octet-stream"})
stored_files.add(f)
cls.stored_files = sorted(stored_files)
cls.sorted_objects = sorted(set(cls.dirs + cls.files))
开发者ID:hpss-collaboration，项目名称:swiftonhpss，代码行数:53，代码来源:swift_on_file_tests.py
注：本文中的test.functional.tests.Utils类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论