本文整理汇总了Python中zaqar.openstack.common.jsonutils.loads函数的典型用法代码示例。如果您正苦于以下问题:Python loads函数的具体用法?Python loads怎么用?Python loads使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_update_metadata
def test_update_metadata(self):
self.skip("This should use patch instead")
xyz_queue_path = self.url_prefix + '/queues/xyz'
xyz_queue_path_metadata = xyz_queue_path
# Create
self.simulate_put(xyz_queue_path, headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_201)
# Set meta
doc1 = '{"messages": {"ttl": 600}}'
self.simulate_put(xyz_queue_path_metadata,
headers=self.headers,
body=doc1)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Update
doc2 = '{"messages": {"ttl": 100}}'
self.simulate_put(xyz_queue_path_metadata,
headers=self.headers,
body=doc2)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Get
result = self.simulate_get(xyz_queue_path_metadata,
headers=self.headers)
result_doc = jsonutils.loads(result[0])
self.assertEqual(result_doc, jsonutils.loads(doc2))
开发者ID:gashe5363,项目名称:zaqar,代码行数:29,代码来源:test_queue_lifecycle.py
示例2: test_update_metadata
def test_update_metadata(self):
xyz_queue_path = self.url_prefix + '/queues/xyz'
xyz_queue_path_metadata = xyz_queue_path + '/metadata'
# Create
project_id = '480924'
self.simulate_put(xyz_queue_path, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_201)
# Set meta
doc1 = '{"messages": {"ttl": 600}}'
self.simulate_put(xyz_queue_path_metadata, project_id, body=doc1)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Update
doc2 = '{"messages": {"ttl": 100}}'
self.simulate_put(xyz_queue_path_metadata, project_id, body=doc2)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Get
result = self.simulate_get(xyz_queue_path_metadata, project_id)
result_doc = jsonutils.loads(result[0])
self.assertEqual(result_doc, jsonutils.loads(doc2))
self.assertEqual(self.srmock.headers_dict['Content-Location'],
xyz_queue_path_metadata)
开发者ID:AsherBond,项目名称:marconi,代码行数:26,代码来源:test_queue_lifecycle.py
示例3: test_simple
def test_simple(self):
self.headers = {
'Client-ID': str(uuid.uuid4()),
'X-Project-ID': '338730984abc_1'
}
gumshoe_queue_path = self.url_prefix + '/queues/gumshoe'
doc = '{"messages": {"ttl": 600}}'
self.simulate_put(gumshoe_queue_path,
headers=self.headers,
body=doc)
self.assertEqual(self.srmock.status, falcon.HTTP_503)
location = ('Location', gumshoe_queue_path)
self.assertNotIn(location, self.srmock.headers)
result = self.simulate_get(gumshoe_queue_path,
headers=self.headers)
result_doc = jsonutils.loads(result[0])
self.assertEqual(self.srmock.status, falcon.HTTP_503)
self.assertNotEqual(result_doc, jsonutils.loads(doc))
self.simulate_get(gumshoe_queue_path + '/stats',
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_503)
self.simulate_get(self.url_prefix + '/queues',
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_503)
self.simulate_delete(gumshoe_queue_path, headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_503)
开发者ID:gashe5363,项目名称:zaqar,代码行数:32,代码来源:test_queue_lifecycle.py
示例4: _listing_test
def _listing_test(self, count=10, limit=10,
marker=None, detailed=False):
# NOTE(cpp-cabrera): delete initial pool - it will interfere
# with listing tests
self.simulate_delete(self.pool)
query = '?limit={0}&detailed={1}'.format(limit, detailed)
if marker:
query += '&marker={0}'.format(marker)
with pools(self, count, self.doc['uri'], 'my-group') as expected:
result = self.simulate_get(self.url_prefix + '/pools',
query_string=query)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
results = jsonutils.loads(result[0])
self.assertIsInstance(results, dict)
self.assertIn('pools', results)
self.assertIn('links', results)
pool_list = results['pools']
link = results['links'][0]
self.assertEqual('next', link['rel'])
href = falcon.uri.parse_query_string(link['href'])
self.assertIn('marker', href)
self.assertEqual(href['limit'], str(limit))
self.assertEqual(href['detailed'], str(detailed).lower())
next_query_string = ('?marker={marker}&limit={limit}'
'&detailed={detailed}').format(**href)
next_result = self.simulate_get(link['href'].split('?')[0],
query_string=next_query_string)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
next_pool = jsonutils.loads(next_result[0])
next_pool_list = next_pool['pools']
self.assertIn('links', next_pool)
if limit < count:
self.assertEqual(len(next_pool_list),
min(limit, count-limit))
else:
# NOTE(jeffrey4l): when limit >= count, there will be no
# pools in the 2nd page.
self.assertTrue(len(next_pool_list) == 0)
self.assertEqual(len(pool_list), min(limit, count))
for s in pool_list + next_pool_list:
# NOTE(flwang): It can't assumed that both sqlalchemy and
# mongodb can return query result with the same order. Just
# like the order they're inserted. Actually, sqlalchemy can't
# guarantee that. So we're leveraging the relationship between
# pool weight and the index of pools fixture to get the
# right pool to verify.
expect = expected[s['weight']]
path, weight, group = expect[:3]
self._pool_expect(s, path, weight, self.doc['uri'])
if detailed:
self.assertIn('options', s)
self.assertEqual(s['options'], expect[-1])
else:
self.assertNotIn('options', s)
开发者ID:rose,项目名称:zaqar,代码行数:60,代码来源:test_pools.py
示例5: test_basics_thoroughly
def test_basics_thoroughly(self, project_id):
gumshoe_queue_path_metadata = self.gumshoe_queue_path + '/metadata'
gumshoe_queue_path_stats = self.gumshoe_queue_path + '/stats'
# Stats not found - queue not created yet
self.simulate_get(gumshoe_queue_path_stats, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
# Metadata not found - queue not created yet
self.simulate_get(gumshoe_queue_path_metadata, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
# Create
self.simulate_put(self.gumshoe_queue_path, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_201)
location = self.srmock.headers_dict['Location']
self.assertEqual(location, self.gumshoe_queue_path)
# Ensure queue existence
self.simulate_head(self.gumshoe_queue_path, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Add metadata
doc = '{"messages": {"ttl": 600}}'
self.simulate_put(gumshoe_queue_path_metadata,
project_id, body=doc)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Fetch metadata
result = self.simulate_get(gumshoe_queue_path_metadata,
project_id)
result_doc = jsonutils.loads(result[0])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(result_doc, jsonutils.loads(doc))
# Stats empty queue
self.simulate_get(gumshoe_queue_path_stats, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
# Delete
self.simulate_delete(self.gumshoe_queue_path, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Get non-existent queue
self.simulate_get(self.gumshoe_queue_path, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
# Get non-existent stats
self.simulate_get(gumshoe_queue_path_stats, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
# Get non-existent metadata
self.simulate_get(gumshoe_queue_path_metadata, project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
开发者ID:AsherBond,项目名称:marconi,代码行数:55,代码来源:test_queue_lifecycle.py
示例6: test_list
def test_list(self):
path = self.queue_path + '/messages'
self._post_messages(path, repeat=10)
query_string = 'limit=3&echo=true'
body = self.simulate_get(path,
query_string=query_string,
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(self.srmock.headers_dict['Content-Location'],
path + '?' + query_string)
cnt = 0
while jsonutils.loads(body[0])['messages'] != []:
contents = jsonutils.loads(body[0])
[target, params] = contents['links'][0]['href'].split('?')
for msg in contents['messages']:
self.simulate_get(msg['href'], headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
body = self.simulate_get(target,
query_string=params,
headers=self.headers)
cnt += 1
self.assertEqual(cnt, 4)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self._empty_message_list(body)
# Stats
body = self.simulate_get(self.queue_path + '/stats',
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
message_stats = jsonutils.loads(body[0])['messages']
self.assertEqual(self.srmock.headers_dict['Content-Location'],
self.queue_path + '/stats')
# NOTE(kgriffs): The other parts of the stats are tested
# in tests.storage.base and so are not repeated here.
expected_pattern = self.queue_path + '/messages/[^/]+$'
for message_stat_name in ('oldest', 'newest'):
self.assertThat(message_stats[message_stat_name]['href'],
matchers.MatchesRegex(expected_pattern))
# NOTE(kgriffs): Try to get messages for a missing queue
body = self.simulate_get(self.url_prefix +
'/queues/nonexistent/messages',
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self._empty_message_list(body)
开发者ID:AsherBond,项目名称:marconi,代码行数:53,代码来源:test_messages.py
示例7: _listing_test
def _listing_test(self, count=10, limit=10,
marker=None, detailed=False):
# NOTE(cpp-cabrera): delete initial flavor - it will interfere
# with listing tests
self.simulate_delete(self.flavor_path)
query = '?limit={0}&detailed={1}'.format(limit, detailed)
if marker:
query += '&marker={2}'.format(marker)
with flavors(self, count, self.doc['pool']) as expected:
result = self.simulate_get(self.url_prefix + '/flavors',
query_string=query)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
results = jsonutils.loads(result[0])
self.assertIsInstance(results, dict)
self.assertIn('flavors', results)
self.assertIn('links', results)
flavors_list = results['flavors']
link = results['links'][0]
self.assertEqual('next', link['rel'])
href = falcon.uri.parse_query_string(link['href'])
self.assertIn('marker', href)
self.assertEqual(href['limit'], str(limit))
self.assertEqual(href['detailed'], str(detailed).lower())
next_query_string = ('?marker={marker}&limit={limit}'
'&detailed={detailed}').format(**href)
next_result = self.simulate_get(link['href'].split('?')[0],
query_string=next_query_string)
next_flavors = jsonutils.loads(next_result[0])
next_flavors_list = next_flavors['flavors']
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertIn('links', next_flavors)
if limit < count:
self.assertEqual(len(next_flavors_list),
min(limit, count-limit))
else:
self.assertTrue(len(next_flavors_list) == 0)
self.assertEqual(len(flavors_list), min(limit, count))
for i, s in enumerate(flavors_list + next_flavors_list):
expect = expected[i]
path, capabilities = expect[:2]
self._flavor_expect(s, path, self.doc['pool'])
if detailed:
self.assertIn('capabilities', s)
self.assertEqual(s['capabilities'], capabilities)
else:
self.assertNotIn('capabilities', s)
开发者ID:jeffrey4l,项目名称:zaqar,代码行数:51,代码来源:test_flavors.py
示例8: test_get_claimed_contains_claim_id_in_href
def test_get_claimed_contains_claim_id_in_href(self):
path = self.queue_path
res = self._post_messages(path + '/messages', repeat=5)
for url in jsonutils.loads(res[0])['resources']:
message = self.simulate_get(url)
self.assertNotIn('claim_id', jsonutils.loads(message[0])['href'])
self.simulate_post(path + '/claims',
body='{"ttl": 100, "grace": 100}',
headers=self.headers)
self.assertEqual(self.srmock.status, falcon.HTTP_201)
for url in jsonutils.loads(res[0])['resources']:
message = self.simulate_get(url)
self.assertIn('claim_id', jsonutils.loads(message[0])['href'])
开发者ID:jeffrey4l,项目名称:zaqar,代码行数:14,代码来源:test_messages.py
示例9: _listing_test
def _listing_test(self, count=10, limit=10,
marker=None, detailed=False):
# NOTE(cpp-cabrera): delete initial flavor - it will interfere
# with listing tests
self.simulate_delete(self.flavor_path)
query = '?limit={0}&detailed={1}'.format(limit, detailed)
if marker:
query += '&marker={2}'.format(marker)
with flavors(self, count, self.doc['pool']) as expected:
result = self.simulate_get(self.url_prefix + '/flavors',
query_string=query)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
results = jsonutils.loads(result[0])
self.assertIsInstance(results, dict)
self.assertIn('flavors', results)
flavors_list = results['flavors']
self.assertEqual(len(flavors_list), min(limit, count))
for i, s in enumerate(flavors_list):
expect = expected[i]
path, capabilities = expect[:2]
self._flavor_expect(s, path, self.doc['pool'])
if detailed:
self.assertIn('capabilities', s)
self.assertEqual(s['capabilities'], capabilities)
else:
self.assertNotIn('capabilities', s)
开发者ID:gashe5363,项目名称:zaqar,代码行数:27,代码来源:test_flavors.py
示例10: _listing_test
def _listing_test(self, count=10, limit=10,
marker=None, detailed=False):
# NOTE(cpp-cabrera): delete initial pool - it will interfere
# with listing tests
self.simulate_delete(self.pool)
query = '?limit={0}&detailed={1}'.format(limit, detailed)
if marker:
query += '&marker={2}'.format(marker)
with pools(self, count, self.doc['uri']) as expected:
result = self.simulate_get(self.url_prefix + '/pools',
query_string=query)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
results = jsonutils.loads(result[0])
self.assertIsInstance(results, dict)
self.assertIn('pools', results)
pool_list = results['pools']
self.assertEqual(len(pool_list), min(limit, count))
for s in pool_list:
# NOTE(flwang): It can't assumed that both sqlalchemy and
# mongodb can return query result with the same order. Just
# like the order they're inserted. Actually, sqlalchemy can't
# guarantee that. So we're leveraging the relationship between
# pool weight and the index of pools fixture to get the
# right pool to verify.
expect = expected[s['weight']]
path, weight = expect[:2]
self._pool_expect(s, path, weight, self.doc['uri'])
if detailed:
self.assertIn('options', s)
self.assertEqual(s['options'], expect[-1])
else:
self.assertNotIn('options', s)
开发者ID:AsherBond,项目名称:marconi,代码行数:33,代码来源:test_pools.py
示例11: test_empty_listing
def test_empty_listing(self):
self.simulate_delete(self.flavor_path)
result = self.simulate_get(self.url_prefix + '/flavors')
results = jsonutils.loads(result[0])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertTrue(len(results['flavors']) == 0)
self.assertIn('links', results)
开发者ID:jeffrey4l,项目名称:zaqar,代码行数:7,代码来源:test_flavors.py
示例12: test_detailed_get_works
def test_detailed_get_works(self):
result = self.simulate_get(self.flavor_path,
query_string='?detailed=True')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
pool = jsonutils.loads(result[0])
self._flavor_expect(pool, self.flavor_path, self.doc['pool'])
self.assertIn('capabilities', pool)
self.assertEqual(pool['capabilities'], {})
开发者ID:jeffrey4l,项目名称:zaqar,代码行数:8,代码来源:test_flavors.py
示例13: test_detailed_get_works
def test_detailed_get_works(self):
result = self.simulate_get(self.pool,
query_string='?detailed=True')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
pool = jsonutils.loads(result[0])
self._pool_expect(pool, self.pool, self.doc['weight'],
self.doc['uri'])
self.assertIn('options', pool)
self.assertEqual(pool['options'], {})
开发者ID:rose,项目名称:zaqar,代码行数:9,代码来源:test_pools.py
示例14: test_pop_empty_queue
def test_pop_empty_queue(self):
query_string = 'pop=1'
result = self.simulate_delete(self.messages_path, self.project_id,
query_string=query_string)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
result_doc = jsonutils.loads(result[0])
self.assertEqual(result_doc['messages'], [])
开发者ID:AsherBond,项目名称:marconi,代码行数:9,代码来源:test_v1_1.py
示例15: test_basic
def test_basic(self):
path = self.url_prefix + '/health'
body = self.simulate_get(path)
health = jsonutils.loads(body[0])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertTrue(health['storage_reachable'])
self.assertIsNotNone(health['message_volume'])
for op in health['operation_status']:
self.assertTrue(health['operation_status'][op]['succeeded'])
开发者ID:AsherBond,项目名称:marconi,代码行数:9,代码来源:test_health.py
示例16: test_message_listing
def test_message_listing(self):
self._prepare_messages(storage.DEFAULT_MESSAGES_PER_PAGE + 1)
result = self.simulate_get(self.messages_path,
headers={'Client-ID': str(uuid.uuid4())})
self.assertEqual(self.srmock.status, falcon.HTTP_200)
messages = jsonutils.loads(result[0])['messages']
self.assertEqual(len(messages), storage.DEFAULT_MESSAGES_PER_PAGE)
开发者ID:AsherBond,项目名称:marconi,代码行数:10,代码来源:test_default_limits.py
示例17: test_delete_message_with_invalid_claim_doesnt_delete_message
def test_delete_message_with_invalid_claim_doesnt_delete_message(self):
path = self.queue_path
resp = self._post_messages(path + '/messages', 1)
location = jsonutils.loads(resp[0])['resources'][0]
self.simulate_delete(location, query_string='claim_id=invalid')
self.assertEqual(self.srmock.status, falcon.HTTP_204)
self.simulate_get(location, self.project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_200)
开发者ID:AsherBond,项目名称:marconi,代码行数:10,代码来源:test_messages.py
示例18: test_custom_metadata
def test_custom_metadata(self):
# Set
doc = '{{"messages": {{"ttl": 600}}, "padding": "{pad}"}}'
max_size = self.transport_cfg.max_queue_metadata
padding_len = max_size - (len(doc) - 2)
doc = doc.format(pad='x' * padding_len)
self.simulate_put(self.fizbat_queue_path,
headers=self.headers,
body=doc)
self.assertEqual(self.srmock.status, falcon.HTTP_201)
# Get
result = self.simulate_get(self.fizbat_queue_path,
headers=self.headers)
result_doc = jsonutils.loads(result[0])
self.assertEqual(result_doc, jsonutils.loads(doc))
self.assertEqual(self.srmock.status, falcon.HTTP_200)
开发者ID:gashe5363,项目名称:zaqar,代码行数:19,代码来源:test_queue_lifecycle.py
示例19: test_claim_creation
def test_claim_creation(self):
self._prepare_messages(storage.DEFAULT_MESSAGES_PER_CLAIM + 1)
result = self.simulate_post(self.claims_path,
body='{"ttl": 60, "grace": 60}')
self.assertEqual(self.srmock.status, falcon.HTTP_201)
messages = jsonutils.loads(result[0])
self.assertEqual(len(messages), storage.DEFAULT_MESSAGES_PER_CLAIM)
开发者ID:AsherBond,项目名称:marconi,代码行数:10,代码来源:test_default_limits.py
示例20: test_listing_marker_is_respected
def test_listing_marker_is_respected(self):
self.simulate_delete(self.flavor_path)
with flavors(self, 10, self.doc['pool']) as expected:
result = self.simulate_get(self.url_prefix + '/flavors',
query_string='?marker=3')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
flavor_list = jsonutils.loads(result[0])['flavors']
self.assertEqual(len(flavor_list), 6)
path, capabilities = expected[4][:2]
self._flavor_expect(flavor_list[0], path, self.doc['pool'])
开发者ID:jeffrey4l,项目名称:zaqar,代码行数:11,代码来源:test_flavors.py
注:本文中的zaqar.openstack.common.jsonutils.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论