本文整理汇总了Python中synnefo.lib.join_urls函数的典型用法代码示例。如果您正苦于以下问题:Python join_urls函数的具体用法?Python join_urls怎么用?Python join_urls使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了join_urls函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_set_unset_set
def test_set_unset_set(self):
cname = get_random_name()
self.create_container(cname)
oname, odata = self.upload_object(cname)[:-1]
self._assert_not_public_object(cname, oname)
# set public
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
self.assertEqual(r.status_code, 202)
public = self._assert_public_object(cname, oname, odata)
# unset public
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='false')
self.assertEqual(r.status_code, 202)
self._assert_not_public_object(cname, oname)
# set public
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
self.assertEqual(r.status_code, 202)
public2 = self._assert_public_object(cname, oname, odata)
self.assertTrue(public != public2)
# unset public
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='false')
self.assertEqual(r.status_code, 202)
self._assert_not_public_object(cname, oname)
开发者ID:antonis-m,项目名称:synnefo,代码行数:35,代码来源:public.py
示例2: test_not_allowed
def test_not_allowed(self):
self.view_path = join_urls(get_service_path(
pithos_settings.pithos_services, 'pithos_ui'), 'view')
self.view_url = join_urls(self.view_path, self.user, get_random_name(),
get_random_name())
r = self.delete(self.view_url)
self.assertEqual(r.status_code, 405)
self.assertTrue('Allow' in r)
self.assertEqual(sorted(r['Allow'].split(', ')), ['GET', 'HEAD'])
r = self.post(self.view_url)
self.assertEqual(r.status_code, 405)
self.assertTrue('Allow' in r)
self.assertEqual(sorted(r['Allow'].split(', ')), ['GET', 'HEAD'])
r = self.put(self.view_url)
self.assertEqual(r.status_code, 405)
self.assertTrue('Allow' in r)
self.assertEqual(sorted(r['Allow'].split(', ')), ['GET', 'HEAD'])
r = self.copy(self.view_url)
self.assertEqual(r.status_code, 405)
self.assertTrue('Allow' in r)
self.assertEqual(sorted(r['Allow'].split(', ')), ['GET', 'HEAD'])
r = self.move(self.view_url)
self.assertEqual(r.status_code, 405)
self.assertTrue('Allow' in r)
self.assertEqual(sorted(r['Allow'].split(', ')), ['GET', 'HEAD'])
开发者ID:apyrgio,项目名称:synnefo,代码行数:30,代码来源:views.py
示例3: test_write_directory
def test_write_directory(self):
url = join_urls(
self.pithos_path, self.user, self.container, self.object)
for type in ('application/directory', 'application/folder'):
# change content type
r = self.put(url, data='', content_type=type,
HTTP_X_MOVE_FROM='/%s/%s' % (
self.container, self.object))
self.assertEqual(r.status_code, 201)
info = self.get_object_info(self.container, self.object)
self.assertEqual(info['Content-Type'], type)
url = join_urls(
self.pithos_path, self.user, self.container, self.object)
r = self.post(
url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
HTTP_X_OBJECT_SHARING='write=*')
self._assert_write(self.object, self.users)
url = join_urls(
self.pithos_path, self.user, self.container, self.object)
r = self.post(
url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
HTTP_X_OBJECT_SHARING='write=%s' % ','.join(
self.users[:2]))
self._assert_write(self.object, self.users[:2])
group = self.groups.keys()[0]
members = self.groups[group]
url = join_urls(
self.pithos_path, self.user, self.container, self.object)
r = self.post(
url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
HTTP_X_OBJECT_SHARING='write=%s:%s' % (self.user, group))
self._assert_write(self.object, members)
开发者ID:antonis-m,项目名称:synnefo,代码行数:35,代码来源:permissions.py
示例4: test_not_allowed
def test_not_allowed(self):
cname = self.create_container()[0]
oname, odata = self.upload_object(cname)[:-1]
url = join_urls(self.pithos_path, self.user)
r = self.head(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.get(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.post(url, user='chuck', data=get_random_data())
self.assertEqual(r.status_code, 403)
url = join_urls(self.pithos_path, self.user, cname)
r = self.head(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.get(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.put(url, user='chuck', data=get_random_data())
self.assertEqual(r.status_code, 403)
r = self.post(url, user='chuck', data=get_random_data())
self.assertEqual(r.status_code, 403)
r = self.delete(url, user='chuck')
self.assertEqual(r.status_code, 403)
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.head(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.get(url, user='chuck')
self.assertEqual(r.status_code, 403)
r = self.put(url, user='chuck', data=get_random_data())
self.assertEqual(r.status_code, 403)
r = self.post(url, user='chuck', data=get_random_data())
self.assertEqual(r.status_code, 403)
r = self.delete(url, user='chuck')
self.assertEqual(r.status_code, 403)
开发者ID:antonis-m,项目名称:synnefo,代码行数:35,代码来源:permissions.py
示例5: test_rud
def test_rud(self, mrapi):
vol = VolumeFactory(status="IN_USE")
user = vol.userid
# READ
r = self.get(join_urls(VOLUMES_URL, "detail"), user)
api_vols = json.loads(r.content)["volumes"]
self.assertEqual(len(api_vols), 1)
api_vol = api_vols[0]
self.assertEqual(api_vol["id"], str(vol.id))
self.assertEqual(api_vol["display_name"], vol.name)
self.assertEqual(api_vol["display_description"], vol.description)
volume_url = join_urls(VOLUMES_URL, str(vol.id))
r = self.get(volume_url, user)
self.assertSuccess(r)
# UPDATE
data = {
"volume": {
"display_name": "lolo",
"display_description": "lala"
}
}
r = self.put(volume_url, user, json.dumps(data), "json")
self.assertSuccess(r)
api_vol = json.loads(r.content)["volume"]
self.assertEqual(api_vol["display_name"], "lolo")
self.assertEqual(api_vol["display_description"], "lala")
# DELETE
mrapi().ModifyInstance.return_value = 42
r = self.delete(volume_url, user)
self.assertSuccess(r)
开发者ID:grnet,项目名称:synnefo,代码行数:34,代码来源:api.py
示例6: test_get
def test_get(self):
response = self.get(join_urls(EXTENSIONS_URL, "SNF"), "user")
self.assertEqual(response.status_code, 404)
response = self.get(join_urls(EXTENSIONS_URL, "SNF_asfas_da"), "user")
self.assertEqual(response.status_code, 404)
response = self.get(join_urls(EXTENSIONS_URL, "SNF-AD"), "user")
self.assertEqual(response.status_code, 404)
开发者ID:AthinaB,项目名称:synnefo,代码行数:7,代码来源:extensions.py
示例7: test_manifestation
def test_manifestation(self):
self.create_container('κουβάς')
prefix = 'μέρη/'
data = ''
for i in range(5):
part = '%s%d' % (prefix, i)
data += self.upload_object('κουβάς', part)[1]
self.create_container('φάκελος')
url = join_urls(self.pithos_path, self.user, 'φάκελος', 'άπαντα')
r = self.put(url, data='', HTTP_X_OBJECT_MANIFEST='κουβάς/%s' % prefix)
self.assertEqual(r.status_code, 201)
r = self.head(url)
self.assertEqual(r.status_code, 200)
r = self.get(url)
self.assertEqual(r.status_code, 200)
self.assertEqual("".join(r.streaming_content), data)
# wrong manifestation
url = join_urls(self.pithos_path, self.user, 'φάκελος', 'άπαντα')
r = self.put(url, data='', HTTP_X_OBJECT_MANIFEST='κουβάς/λάθος')
self.assertEqual(r.status_code, 201)
r = self.get(url)
self.assertEqual(r.status_code, 200)
self.assertTrue("".join(r.streaming_content) != data)
开发者ID:grnet,项目名称:synnefo,代码行数:28,代码来源:unicode.py
示例8: test_manifestation
def test_manifestation(self):
self.create_container("κουβάς")
prefix = "μέρη/"
data = ""
for i in range(5):
part = "%s%d" % (prefix, i)
data += self.upload_object("κουβάς", part)[1]
self.create_container("φάκελος")
url = join_urls(self.pithos_path, self.user, "φάκελος", "άπαντα")
r = self.put(url, data="", HTTP_X_OBJECT_MANIFEST="κουβάς/%s" % prefix)
self.assertEqual(r.status_code, 201)
r = self.head(url)
self.assertEqual(r.status_code, 200)
r = self.get(url)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.content, data)
# wrong manifestation
url = join_urls(self.pithos_path, self.user, "φάκελος", "άπαντα")
r = self.put(url, data="", HTTP_X_OBJECT_MANIFEST="κουβάς/λάθος")
self.assertEqual(r.status_code, 201)
r = self.get(url)
self.assertEqual(r.status_code, 200)
self.assertTrue(r.content != data)
开发者ID:Erethon,项目名称:synnefo,代码行数:28,代码来源:unicode.py
示例9: test_groups
def test_groups(self):
# create a group
headers = {"HTTP_X_ACCOUNT_GROUP_γκρουπ": "chazapis,διογένης"}
url = join_urls(self.pithos_path, self.user)
r = self.post(url, **headers)
self.assertEqual(r.status_code, 202)
groups = self.get_account_groups()
self.assertTrue("γκρουπ" in groups)
self.assertEqual(groups["γκρουπ"], "chazapis,διογένης")
# check read access
self.create_container("φάκελος")
odata = self.upload_object("φάκελος", "ο1")[1]
r = self.head(url, user="διογένης")
self.assertEqual(r.status_code, 403)
r = self.get(url, user="διογένης")
self.assertEqual(r.status_code, 403)
# share for read
url = join_urls(self.pithos_path, self.user, "φάκελος", "ο1")
r = self.post(url, content_type="", HTTP_X_OBJECT_SHARING="read=%s:γκρουπ" % self.user)
self.assertEqual(r.status_code, 202)
r = self.head(url, user="διογένης")
self.assertEqual(r.status_code, 200)
r = self.get(url, user="διογένης")
self.assertEqual(r.status_code, 200)
# check write access
appended_data = get_random_data()
r = self.post(
url,
user="διογένης",
data=appended_data,
content_type="application/octet-stream",
HTTP_CONTENT_LENGTH=str(len(appended_data)),
HTTP_CONTENT_RANGE="bytes */*",
)
self.assertEqual(r.status_code, 403)
# share for write
url = join_urls(self.pithos_path, self.user, "φάκελος", "ο1")
r = self.post(url, content_type="", HTTP_X_OBJECT_SHARING="write=%s:γκρουπ" % self.user)
self.assertEqual(r.status_code, 202)
r = self.post(
url,
user="διογένης",
data=appended_data,
content_type="application/octet-stream",
HTTP_CONTENT_LENGTH=str(len(appended_data)),
HTTP_CONTENT_RANGE="bytes */*",
)
self.assertEqual(r.status_code, 204)
r = self.get(url, user="διογένης")
self.assertEqual(r.status_code, 200)
self.assertEqual(r.content, odata + appended_data)
开发者ID:Erethon,项目名称:synnefo,代码行数:60,代码来源:unicode.py
示例10: test_public_multiple_range
def test_public_multiple_range(self):
cname = self.create_container()[0]
oname, odata = self.upload_object(cname)[:-1]
url = join_urls(self.pithos_path, self.user, cname, oname)
# set public
url = join_urls(self.pithos_path, self.user, cname, oname)
r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
self.assertEqual(r.status_code, 202)
info = self.get_object_info(cname, oname)
public_url = info['X-Object-Public']
l = ['0-499', '-500', '1000-']
ranges = 'bytes=%s' % ','.join(l)
r = self.get(public_url, HTTP_RANGE=ranges)
self.assertEqual(r.status_code, 206)
self.assertTrue('content-type' in r)
p = re.compile(
'multipart/byteranges; boundary=(?P<boundary>[0-9a-f]{32}\Z)',
re.I)
m = p.match(r['content-type'])
if m is None:
self.fail('Invalid multiple range content type')
boundary = m.groupdict()['boundary']
cparts = r.content.split('--%s' % boundary)[1:-1]
# assert content parts length
self.assertEqual(len(cparts), len(l))
# for each content part assert headers
i = 0
for cpart in cparts:
content = cpart.split('\r\n')
headers = content[1:3]
content_range = headers[0].split(': ')
self.assertEqual(content_range[0], 'Content-Range')
r = l[i].split('-')
if not r[0] and not r[1]:
pass
elif not r[0]:
start = len(odata) - int(r[1])
end = len(odata)
elif not r[1]:
start = int(r[0])
end = len(odata)
else:
start = int(r[0])
end = int(r[1]) + 1
fdata = odata[start:end]
sdata = '\r\n'.join(content[4:-1])
self.assertEqual(len(fdata), len(sdata))
self.assertEquals(fdata, sdata)
i += 1
开发者ID:antonis-m,项目名称:synnefo,代码行数:55,代码来源:public.py
示例11: test_create_container
def test_create_container(self):
cname = 'φάκελος'
self.create_container(cname)
url = join_urls(self.pithos_path, self.user, cname)
r = self.head(url)
self.assertEqual(r.status_code, 204)
url = join_urls(self.pithos_path, self.user)
r = self.get(url)
self.assertEqual(r.status_code, 200)
containers = r.content.split('\n')
self.assertTrue('φάκελος' in containers)
开发者ID:grnet,项目名称:synnefo,代码行数:12,代码来源:unicode.py
示例12: setUp
def setUp(self):
PithosAPITest.setUp(self)
self.cname = self.create_container()[0]
self.oname, self.odata = self.upload_object(self.cname,
'φωτογραφία.JPG')[:-1]
self.view_path = join_urls(get_service_path(
pithos_settings.pithos_services, 'pithos_ui'), 'view')
self.view_url = join_urls(self.view_path, self.user, self.cname,
self.oname)
self.api_url = join_urls(self.pithos_path, self.user, self.cname,
self.oname)
开发者ID:apyrgio,项目名称:synnefo,代码行数:12,代码来源:views.py
示例13: test_volume_metadata
def test_volume_metadata(self):
vol = VolumeFactory()
volume_metadata_url = join_urls(join_urls(VOLUMES_URL, str(vol.id)),
"metadata")
# Empty metadata
response = self.get(volume_metadata_url, vol.userid)
self.assertSuccess(response)
metadata = json.loads(response.content)["metadata"]
self.assertEqual(metadata, {})
# Create metadata items
meta1 = {"metadata": {"key1": "val1", "\u2601": "\u2602"}}
response = self.post(volume_metadata_url, vol.userid,
json.dumps(meta1), "json")
self.assertSuccess(response)
response = self.get(volume_metadata_url, vol.userid)
self.assertSuccess(response)
metadata = json.loads(response.content)
self.assertEqual(metadata, meta1)
# Update existing metadata and add new
meta2 = {"metadata": {"\u2601": "unicode_val_2", "key3": "val3"}}
meta_db = {"metadata": {"key1": "val1",
"\u2601": "unicode_val_2",
"key3": "val3"}}
response = self.post(volume_metadata_url, vol.userid,
json.dumps(meta2), "json")
self.assertSuccess(response)
metadata = json.loads(response.content)
self.assertEqual(metadata, meta_db)
response = self.get(volume_metadata_url, vol.userid)
self.assertSuccess(response)
metadata = json.loads(response.content)
self.assertEqual(metadata, meta_db)
# Replace all metadata
meta3 = {"metadata": {"key4": "val4"}}
response = self.put(volume_metadata_url, vol.userid,
json.dumps(meta3), "json")
self.assertSuccess(response)
response = self.get(volume_metadata_url, vol.userid)
self.assertSuccess(response)
metadata = json.loads(response.content)
self.assertEqual(metadata, meta3)
# Delete metadata key
response = self.delete(join_urls(volume_metadata_url, "key4"),
vol.userid)
self.assertSuccess(response)
response = self.get(volume_metadata_url, vol.userid)
self.assertSuccess(response)
metadata = json.loads(response.content)["metadata"]
self.assertEqual(metadata, {})
开发者ID:grnet,项目名称:synnefo,代码行数:52,代码来源:api.py
示例14: get_url_patterns
def get_url_patterns(self):
_patterns = patterns(
'',
url(r'^%s/?$' % join_urls(self.endpoints_prefix,
self.authorization_endpoint.rstrip('/')),
self.auth_view,
name='%s_authenticate' % self.id),
url(r'^%s/?$' % join_urls(self.endpoints_prefix,
self.token_endpoint.rstrip('/')),
self.token_view,
name='%s_token' % self.id),
)
return _patterns
开发者ID:AthinaB,项目名称:synnefo,代码行数:13,代码来源:djangobackend.py
示例15: test_shared_with_me
def test_shared_with_me(self):
container = self.create_container(user='alice')[0]
obj, data = self.upload_object(container, user='alice')[:-1]
# share object
url = join_urls(self.pithos_path, 'alice', container, obj)
self.post(url, user='alice', content_type='',
HTTP_CONTENT_RANGE='bytes */*',
HTTP_X_OBJECT_SHARING='read=user')
url = join_urls(self.view_path, 'alice', container, obj)
r = self.view(url)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.content, data)
开发者ID:apyrgio,项目名称:synnefo,代码行数:14,代码来源:views.py
示例16: test_if_modified_since_invalid_date
def test_if_modified_since_invalid_date(self):
cname = 'apples'
url = join_urls(self.pithos_path, self.user, cname)
r = self.get(url, HTTP_IF_MODIFIED_SINCE='Monday')
self.assertEqual(r.status_code, 200)
self.assertEqual(r.content.split('\n')[:-1],
sorted(self.objects['apples'].keys()))
开发者ID:AthinaB,项目名称:synnefo,代码行数:7,代码来源:containers.py
示例17: test_list_objects_with_limit_marker
def test_list_objects_with_limit_marker(self):
cname = self.cnames[0]
url = join_urls(self.pithos_path, self.user, cname)
r = self.get('%s?limit=qwert' % url)
self.assertTrue(r.status_code != 500)
r = self.get('%s?limit=2' % url)
self.assertEqual(r.status_code, 200)
objects = r.content.split('\n')
if '' in objects:
objects.remove('')
onames = sorted(self.objects[cname].keys())
self.assertEqual(objects, onames[:2])
markers = ['How To Win Friends And Influence People.pdf',
'moms_birthday.jpg']
limit = 4
for m in markers:
r = self.get('%s?limit=%s&marker=%s' % (url, limit, m))
objects = r.content.split('\n')
if '' in objects:
objects.remove('')
start = onames.index(m) + 1
end = start + limit
end = end if len(onames) >= end else len(onames)
self.assertEqual(objects, onames[start:end])
开发者ID:AthinaB,项目名称:synnefo,代码行数:27,代码来源:containers.py
示例18: test_list_objects_containing_slash
def test_list_objects_containing_slash(self):
self.create_container('test')
self.upload_object('test', quote('/objectname', ''))
url = join_urls(self.pithos_path, self.user, 'test')
r = self.get(url)
objects = r.content.split('\n')
if '' in objects:
objects.remove('')
self.assertEqual(objects, ['/objectname'])
r = self.get('%s?format=json' % url)
try:
objects = json.loads(r.content)
except:
self.fail('json format expected')
self.assertEqual([o['name'] for o in objects], ['/objectname'])
r = self.get('%s?format=xml' % url)
try:
objects = minidom.parseString(r.content)
except:
self.fail('xml format expected')
self.assertEqual(
[n.firstChild.data for n in objects.getElementsByTagName('name')],
['/objectname'])
开发者ID:AthinaB,项目名称:synnefo,代码行数:27,代码来源:containers.py
示例19: test_get_meta
def test_get_meta(self):
self.create_container('apples')
# populate with objects
objects = {}
for i in range(random.randint(1, 100)):
# upload object
meta = {'foo%s' % i: 'bar'}
name, data, resp = self.upload_object('apples', **meta)
objects[name] = data
t1 = datetime.datetime.utcnow()
url = join_urls(self.pithos_path, self.user, 'apples')
r = self.head(url)
self.assertEqual(int(r['X-Container-Object-Count']), len(objects))
self.assertEqual(int(r['X-Container-Bytes-Used']),
sum([len(i) for i in objects.values()]))
self.assertTrue('X-Container-Block-Size' in r)
self.assertTrue('X-Container-Block-Hash' in r)
self.assertTrue('X-Container-Until-Timestamp' not in r)
self.assertEqual(r['X-Container-Policy-Versioning'], 'auto')
self.assertEqual(int(r['X-Container-Policy-Quota']), 0)
t2 = datetime.datetime.strptime(r['Last-Modified'], DATE_FORMATS[2])
delta = (t2 - t1)
threashold = datetime.timedelta(seconds=1)
self.assertTrue(delta < threashold)
self.assertTrue(r['X-Container-Object-Meta'])
(self.assertTrue('foo%s' % i in r['X-Container-Object-Meta'])
for i in range(len(objects)))
开发者ID:AthinaB,项目名称:synnefo,代码行数:30,代码来源:containers.py
示例20: test_list_until
def test_list_until(self):
account_info = self.get_account_info()
t = datetime.datetime.strptime(account_info['Last-Modified'],
DATE_FORMATS[2])
t1 = t + datetime.timedelta(seconds=1)
until = int(_time.mktime(t1.timetuple()))
_time.sleep(2)
cname = self.cnames[0]
self.upload_object(cname)
oname = self.objects[cname].keys()[-1]
self.delete_object(cname, oname)
url = join_urls(self.pithos_path, self.user, cname)
r = self.get('%s?until=%s' % (url, until))
self.assertTrue(r.status_code, 200)
objects = r.content.split('\n')
if '' in objects:
objects.remove('')
self.assertEqual(objects,
sorted(self.objects[cname].keys()))
r = self.get('%s?until=%s&format=json' % (url, until))
self.assertTrue(r.status_code, 200)
try:
objects = json.loads(r.content)
except:
self.fail('json format expected')
self.assertEqual([o['name'] for o in objects],
sorted(self.objects[cname].keys()))
开发者ID:AthinaB,项目名称:synnefo,代码行数:31,代码来源:containers.py
注:本文中的synnefo.lib.join_urls函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论