本文整理汇总了Python中swift.common.constraints.check_metadata函数的典型用法代码示例。如果您正苦于以下问题:Python check_metadata函数的具体用法?Python check_metadata怎么用?Python check_metadata使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了check_metadata函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_check_metadata_size
def test_check_metadata_size(self):
headers = {}
size = 0
chunk = constraints.MAX_META_NAME_LENGTH + \
constraints.MAX_META_VALUE_LENGTH
x = 0
while size + chunk < constraints.MAX_META_OVERALL_SIZE:
headers['X-Object-Meta-%04d%s' %
(x, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \
'v' * constraints.MAX_META_VALUE_LENGTH
size += chunk
x += 1
self.assertIsNone(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'))
# add two more headers in case adding just one falls exactly on the
# limit (eg one header adds 1024 and the limit is 2048)
headers['X-Object-Meta-%04d%s' %
(x, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \
'v' * constraints.MAX_META_VALUE_LENGTH
headers['X-Object-Meta-%04d%s' %
(x + 1, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \
'v' * constraints.MAX_META_VALUE_LENGTH
resp = constraints.check_metadata(Request.blank(
'/', headers=headers), 'object')
self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
self.assertIn(b'Total metadata too large', resp.body)
开发者ID:jgmerritt,项目名称:swift,代码行数:26,代码来源:test_constraints.py
示例2: test_check_metadata_count
def test_check_metadata_count(self):
headers = {}
for x in range(constraints.MAX_META_COUNT):
headers["X-Object-Meta-%d" % x] = "v"
self.assertEquals(constraints.check_metadata(Request.blank("/", headers=headers), "object"), None)
headers["X-Object-Meta-Too-Many"] = "v"
self.assertEquals(
constraints.check_metadata(Request.blank("/", headers=headers), "object").status_int, HTTP_BAD_REQUEST
)
开发者ID:heemanshu,项目名称:swift_liberty,代码行数:9,代码来源:test_constraints.py
示例3: test_check_metadata_count
def test_check_metadata_count(self):
headers = {}
for x in xrange(constraints.MAX_META_COUNT):
headers['X-Object-Meta-%d' % x] = 'v'
self.assertEquals(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'), None)
headers['X-Object-Meta-Too-Many'] = 'v'
self.assertEquals(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object').status_int, HTTP_BAD_REQUEST)
开发者ID:HugoKuo,项目名称:swift,代码行数:9,代码来源:test_constraints.py
示例4: test_validate_bad_meta
def test_validate_bad_meta(self):
req = Request.blank(
'/v/a/c/o',
headers={'x-object-meta-hello':
'ab' * constraints.MAX_HEADER_SIZE})
self.assertEquals(constraints.check_metadata(req, 'object').status_int,
HTTP_BAD_REQUEST)
self.assertIn('x-object-meta-hello', constraints.check_metadata(req,
'object').body.lower())
开发者ID:HugoKuo,项目名称:swift,代码行数:9,代码来源:test_constraints.py
示例5: test_check_metadata_value_length
def test_check_metadata_value_length(self):
value = 'a' * constraints.MAX_META_VALUE_LENGTH
headers = {'X-Object-Meta-Name': value}
self.assertEquals(constraints.check_metadata(Request.blank('/',
headers=headers), 'object'), None)
value = 'a' * (constraints.MAX_META_VALUE_LENGTH + 1)
headers = {'X-Object-Meta-Name': value}
self.assertEquals(constraints.check_metadata(Request.blank('/',
headers=headers), 'object').status_int, HTTP_BAD_REQUEST)
开发者ID:ArikaChen,项目名称:swift,代码行数:9,代码来源:test_constraints.py
示例6: test_check_metadata_name_length
def test_check_metadata_name_length(self):
name = 'a' * constraints.MAX_META_NAME_LENGTH
headers = {'X-Object-Meta-%s' % name: 'v'}
self.assertEquals(constraints.check_metadata(Request.blank('/',
headers=headers), 'object'), None)
name = 'a' * (constraints.MAX_META_NAME_LENGTH + 1)
headers = {'X-Object-Meta-%s' % name: 'v'}
self.assertEquals(constraints.check_metadata(Request.blank('/',
headers=headers), 'object').status_int, HTTP_BAD_REQUEST)
开发者ID:ArikaChen,项目名称:swift,代码行数:9,代码来源:test_constraints.py
示例7: test_check_metadata_non_utf8
def test_check_metadata_non_utf8(self):
headers = {'X-Account-Meta-Foo': b'\xff'}
self.assertEqual(constraints.check_metadata(Request.blank(
'/', headers=headers), 'account').status_int, HTTP_BAD_REQUEST)
headers = {b'X-Container-Meta-\xff': 'foo'}
self.assertEqual(constraints.check_metadata(Request.blank(
'/', headers=headers), 'container').status_int, HTTP_BAD_REQUEST)
# Object's OK; its metadata isn't serialized as JSON
headers = {'X-Object-Meta-Foo': b'\xff'}
self.assertIsNone(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'))
开发者ID:prashanthpai,项目名称:swift,代码行数:11,代码来源:test_constraints.py
示例8: test_check_metadata_count
def test_check_metadata_count(self):
headers = {}
for x in range(constraints.MAX_META_COUNT):
headers['X-Object-Meta-%d' % x] = 'v'
self.assertIsNone(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'))
headers['X-Object-Meta-Too-Many'] = 'v'
resp = constraints.check_metadata(Request.blank(
'/', headers=headers), 'object')
self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
self.assertIn(b'Too many metadata items', resp.body)
开发者ID:jgmerritt,项目名称:swift,代码行数:12,代码来源:test_constraints.py
示例9: test_check_metadata_name_length
def test_check_metadata_name_length(self):
name = "a" * constraints.MAX_META_NAME_LENGTH
headers = {"X-Object-Meta-%s" % name: "v"}
self.assertEquals(constraints.check_metadata(Request.blank("/", headers=headers), "object"), None)
name = "a" * (constraints.MAX_META_NAME_LENGTH + 1)
headers = {"X-Object-Meta-%s" % name: "v"}
self.assertEquals(
constraints.check_metadata(Request.blank("/", headers=headers), "object").status_int, HTTP_BAD_REQUEST
)
self.assertIn(
("X-Object-Meta-%s" % name).lower(),
constraints.check_metadata(Request.blank("/", headers=headers), "object").body.lower(),
)
开发者ID:heemanshu,项目名称:swift_liberty,代码行数:13,代码来源:test_constraints.py
示例10: test_check_metadata_name_length
def test_check_metadata_name_length(self):
name = 'a' * constraints.MAX_META_NAME_LENGTH
headers = {'X-Object-Meta-%s' % name: 'v'}
self.assertIsNone(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'))
name = 'a' * (constraints.MAX_META_NAME_LENGTH + 1)
headers = {'X-Object-Meta-%s' % name: 'v'}
resp = constraints.check_metadata(Request.blank(
'/', headers=headers), 'object')
self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
self.assertIn(
b'x-object-meta-%s' % name.encode('ascii'), resp.body.lower())
self.assertIn(b'Metadata name too long', resp.body)
开发者ID:jgmerritt,项目名称:swift,代码行数:14,代码来源:test_constraints.py
示例11: test_check_metadata_value_length
def test_check_metadata_value_length(self):
value = 'a' * constraints.MAX_META_VALUE_LENGTH
headers = {'X-Object-Meta-Name': value}
self.assertIsNone(constraints.check_metadata(Request.blank(
'/', headers=headers), 'object'))
value = 'a' * (constraints.MAX_META_VALUE_LENGTH + 1)
headers = {'X-Object-Meta-Name': value}
resp = constraints.check_metadata(Request.blank(
'/', headers=headers), 'object')
self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
self.assertIn(b'x-object-meta-name', resp.body.lower())
self.assertIn(
str(constraints.MAX_META_VALUE_LENGTH).encode('ascii'), resp.body)
self.assertIn(b'Metadata value longer than 256', resp.body)
开发者ID:jgmerritt,项目名称:swift,代码行数:15,代码来源:test_constraints.py
示例12: POST
def POST(self, req):
"""HTTP POST request handler."""
error_response = \
self.clean_acls(req) or check_metadata(req, 'container')
if error_response:
return error_response
if not req.environ.get('swift_owner'):
for key in self.app.swift_owner_headers:
req.headers.pop(key, None)
if req.environ.get('reseller_request', False) and \
'X-Container-Sharding' in req.headers:
req.headers[get_sys_meta_prefix('container') + 'Sharding'] = \
str(config_true_value(req.headers['X-Container-Sharding']))
account_partition, accounts, container_count = \
self.account_info(self.account_name, req)
if not accounts:
return HTTPNotFound(request=req)
container_partition, containers = self.app.container_ring.get_nodes(
self.account_name, self.container_name)
headers = self.generate_request_headers(req, transfer=True)
clear_info_cache(self.app, req.environ,
self.account_name, self.container_name)
resp = self.make_requests(
req, self.app.container_ring, container_partition, 'POST',
req.swift_entity_path, [headers] * len(containers))
return resp
开发者ID:openstack,项目名称:swift,代码行数:26,代码来源:container.py
示例13: PUT
def PUT(self, req):
"""HTTP PUT request handler."""
error_response = \
self.clean_acls(req) or check_metadata(req, 'container')
if error_response:
return error_response
if len(self.container_name) > MAX_CONTAINER_NAME_LENGTH:
resp = HTTPBadRequest(request=req)
resp.body = 'Container name length of %d longer than %d' % \
(len(self.container_name), MAX_CONTAINER_NAME_LENGTH)
return resp
account_partition, accounts, container_count = \
self.account_info(self.account_name,
autocreate=self.app.account_autocreate)
if self.app.max_containers_per_account > 0 and \
container_count >= self.app.max_containers_per_account and \
self.account_name not in self.app.max_containers_whitelist:
resp = HTTPForbidden(request=req)
resp.body = 'Reached container limit of %s' % \
self.app.max_containers_per_account
return resp
if not accounts:
return HTTPNotFound(request=req)
container_partition, containers = self.app.container_ring.get_nodes(
self.account_name, self.container_name)
headers = self._backend_requests(req, len(containers),
account_partition, accounts)
if self.app.memcache:
cache_key = get_container_memcache_key(self.account_name,
self.container_name)
self.app.memcache.delete(cache_key)
resp = self.make_requests(
req, self.app.container_ring,
container_partition, 'PUT', req.path_info, headers)
return resp
开发者ID:Neil-Jubinville,项目名称:swift,代码行数:35,代码来源:container.py
示例14: POST
def POST(self, req):
"""HTTP POST request handler."""
error_response = check_metadata(req, 'account')
if error_response:
return error_response
account_partition, accounts = \
self.app.account_ring.get_nodes(self.account_name)
headers = self.generate_request_headers(req, transfer=True)
if self.app.memcache:
self.app.memcache.delete(
get_account_memcache_key(self.account_name))
resp = self.make_requests(
req, self.app.account_ring, account_partition, 'POST',
req.path_info, [headers] * len(accounts))
if resp.status_int == HTTP_NOT_FOUND and self.app.account_autocreate:
if len(self.account_name) > MAX_ACCOUNT_NAME_LENGTH:
resp = HTTPBadRequest(request=req)
resp.body = 'Account name length of %d longer than %d' % \
(len(self.account_name), MAX_ACCOUNT_NAME_LENGTH)
return resp
resp = self.make_requests(
Request.blank('/v1/' + self.account_name),
self.app.account_ring, account_partition, 'PUT',
'/' + self.account_name, [headers] * len(accounts))
if not is_success(resp.status_int):
self.app.logger.warning('Could not autocreate account %r' %
self.account_name)
return resp
return resp
开发者ID:blapid,项目名称:swift,代码行数:29,代码来源:account.py
示例15: PUT
def PUT(self, req):
"""HTTP PUT request handler."""
print 'in PUT function of accountcontroller class'
if not self.app.allow_account_management:
return HTTPMethodNotAllowed(
request=req,
headers={'Allow': ', '.join(self.allowed_methods)})
error_response = check_metadata(req, 'account')
print 'error_response'
if error_response:
return error_response
if len(self.account_name) > constraints.MAX_ACCOUNT_NAME_LENGTH:
resp = HTTPBadRequest(request=req)
resp.body = 'Account name length of %d longer than %d' % \
(len(self.account_name),
constraints.MAX_ACCOUNT_NAME_LENGTH)
return resp
account_partition, accounts = \
self.app.account_ring.get_nodes(self.account_name)
print ' account_partition, accounts',account_partion,accounts
headers = self.generate_request_headers(req, transfer=True)
print 'headers',headers
clear_info_cache(self.app, req.environ, self.account_name)
resp = self.make_requests(
req, self.app.account_ring, account_partition, 'PUT',
req.swift_entity_path, [headers] * len(accounts))
print 'resp',resp
self.add_acls_from_sys_metadata(resp)
print 'in PUT function of accountcontroller class END'
return resp
开发者ID:jannatunnoor,项目名称:test_swift,代码行数:30,代码来源:account.py
示例16: PUT
def PUT(self, req):
"""HTTP PUT request handler."""
error_response = self.clean_acls(req) or check_metadata(req, "container")
if error_response:
return error_response
if len(self.container_name) > MAX_CONTAINER_NAME_LENGTH:
resp = HTTPBadRequest(request=req)
resp.body = "Container name length of %d longer than %d" % (
len(self.container_name),
MAX_CONTAINER_NAME_LENGTH,
)
return resp
account_partition, accounts, container_count = self.account_info(self.account_name, req)
if not accounts and self.app.account_autocreate:
self.autocreate_account(req.environ, self.account_name)
account_partition, accounts, container_count = self.account_info(self.account_name, req)
if not accounts:
return HTTPNotFound(request=req)
if (
self.app.max_containers_per_account > 0
and container_count >= self.app.max_containers_per_account
and self.account_name not in self.app.max_containers_whitelist
):
resp = HTTPForbidden(request=req)
resp.body = "Reached container limit of %s" % self.app.max_containers_per_account
return resp
container_partition, containers = self.app.container_ring.get_nodes(self.account_name, self.container_name)
headers = self._backend_requests(req, len(containers), account_partition, accounts)
clear_info_cache(self.app, req.environ, self.account_name, self.container_name)
resp = self.make_requests(req, self.app.container_ring, container_partition, "PUT", req.path_info, headers)
return resp
开发者ID:zaitcev,项目名称:swift-lfs,代码行数:31,代码来源:container.py
示例17: POST
def POST(self, req):
"""HTTP POST request handler."""
error_response = \
self.clean_acls(req) or check_metadata(req, 'container')
if error_response:
return error_response
if not req.environ.get('swift_owner'):
for key in self.app.swift_owner_headers:
req.headers.pop(key, None)
headers = self.generate_request_headers(req, transfer=True)
clear_info_cache(self.app, req.environ,
self.account_name, self.container_name)
storage = self.app.storage
metadata = {}
metadata.update(("user.%s" % k, v) for k, v in req.headers.iteritems()
if k.lower() in self.pass_through_headers or
is_sys_or_user_meta('container', k))
try:
storage.container_update(self.account_name, self.container_name,
metadata, headers=headers)
resp = HTTPNoContent(request=req)
except exceptions.NoSuchContainer:
resp = self.PUT(req)
return resp
开发者ID:GuillaumeDelaporte,项目名称:oio-swift,代码行数:27,代码来源:container.py
示例18: POST
def POST(self, req):
"""HTTP POST request handler."""
error_response = \
self.clean_acls(req) or check_metadata(req, 'container')
if error_response:
return error_response
account_partition, accounts, container_count = \
self.account_info(self.account_name,
autocreate=self.app.account_autocreate)
if not accounts:
return HTTPNotFound(request=req)
container_partition, containers = self.app.container_ring.get_nodes(
self.account_name, self.container_name)
headers = {'X-Timestamp': normalize_timestamp(time.time()),
'x-trans-id': self.trans_id,
'Connection': 'close'}
self.transfer_headers(req.headers, headers)
if self.app.memcache:
cache_key = get_container_memcache_key(self.account_name,
self.container_name)
self.app.memcache.delete(cache_key)
resp = self.make_requests(req, self.app.container_ring,
container_partition, 'POST', req.path_info,
[headers] * len(containers))
return resp
开发者ID:DmitryMezhensky,项目名称:Hadoop-and-Swift-integration,代码行数:25,代码来源:container.py
示例19: test_check_metadata_value_length
def test_check_metadata_value_length(self):
value = "a" * constraints.MAX_META_VALUE_LENGTH
headers = {"X-Object-Meta-Name": value}
self.assertEquals(constraints.check_metadata(Request.blank("/", headers=headers), "object"), None)
value = "a" * (constraints.MAX_META_VALUE_LENGTH + 1)
headers = {"X-Object-Meta-Name": value}
self.assertEquals(
constraints.check_metadata(Request.blank("/", headers=headers), "object").status_int, HTTP_BAD_REQUEST
)
self.assertIn(
"x-object-meta-name", constraints.check_metadata(Request.blank("/", headers=headers), "object").body.lower()
)
self.assertIn(
str(constraints.MAX_META_VALUE_LENGTH),
constraints.check_metadata(Request.blank("/", headers=headers), "object").body,
)
开发者ID:heemanshu,项目名称:swift_liberty,代码行数:16,代码来源:test_constraints.py
示例20: POST
def POST(self, req):
"""HTTP POST request handler."""
error_response = check_metadata(req, "account")
if error_response:
return error_response
account_partition, accounts = self.app.account_ring.get_nodes(self.account_name)
headers = {"X-Timestamp": normalize_timestamp(time.time()), "X-Trans-Id": self.trans_id, "Connection": "close"}
self.transfer_headers(req.headers, headers)
if self.app.memcache:
self.app.memcache.delete("account%s" % req.path_info.rstrip("/"))
resp = self.make_requests(
req, self.app.account_ring, account_partition, "POST", req.path_info, [headers] * len(accounts)
)
if resp.status_int == HTTP_NOT_FOUND and self.app.account_autocreate:
if len(self.account_name) > MAX_ACCOUNT_NAME_LENGTH:
resp = HTTPBadRequest(request=req)
resp.body = "Account name length of %d longer than %d" % (
len(self.account_name),
MAX_ACCOUNT_NAME_LENGTH,
)
return resp
resp = self.make_requests(
Request.blank("/v1/" + self.account_name),
self.app.account_ring,
account_partition,
"PUT",
"/" + self.account_name,
[headers] * len(accounts),
)
if not is_success(resp.status_int):
self.app.logger.warning("Could not autocreate account %r" % self.account_name)
return resp
return resp
开发者ID:hortonworkstest,项目名称:Hadoop-and-Swift-integration,代码行数:33,代码来源:account.py
注:本文中的swift.common.constraints.check_metadata函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论