本文整理汇总了Python中swift.common.direct_client.direct_get_account函数的典型用法代码示例。如果您正苦于以下问题：Python direct_get_account函数的具体用法？Python direct_get_account怎么用？Python direct_get_account使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了direct_get_account函数的12个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_direct_client_exception
def test_direct_client_exception(self):
stub_headers = {'X-Trans-Id': 'txb5f59485c578460f8be9e-0053478d09'}
body = 'a server error has occurred'
with mocked_http_conn(500, stub_headers, body):
try:
direct_client.direct_get_account(self.node, self.part,
self.account)
except ClientException as err:
pass
else:
self.fail('ClientException not raised')
self.assertEqual(err.http_status, 500)
expected_err_msg_parts = (
'Account server %s:%s' % (self.node['ip'], self.node['port']),
'GET %r' % self.account_path,
'status 500',
)
for item in expected_err_msg_parts:
self.assertTrue(
item in str(err), '%r was not in "%s"' % (item, err))
self.assertEqual(err.http_host, self.node['ip'])
self.assertEqual(err.http_port, self.node['port'])
self.assertEqual(err.http_device, self.node['device'])
self.assertEqual(err.http_status, 500)
self.assertEqual(err.http_reason, 'Internal Error')
self.assertEqual(err.http_headers, stub_headers)
开发者ID:aureliengoulon，项目名称:swift，代码行数:26，代码来源:test_direct_client.py
示例2: test_direct_get_account
def test_direct_get_account(self):
node = {'ip': '1.2.3.4', 'port': '6000', 'device': 'sda'}
part = '0'
account = 'a'
headers = {
'X-Account-Container-Count': '1',
'X-Account-Object-Count': '1',
'X-Account-Bytes-Used': '1',
'X-Timestamp': '1234567890',
'X-PUT-Timestamp': '1234567890'}
body = '[{"count": 1, "bytes": 20971520, "name": "c1"}]'
fake_headers = {}
for header, value in headers.items():
fake_headers[header.lower()] = value
was_http_connector = direct_client.http_connect
direct_client.http_connect = mock_http_connect(200, fake_headers, body)
resp_headers, resp = direct_client.direct_get_account(node, part, account)
fake_headers.update({'user-agent': 'direct-client %s' % os.getpid()})
self.assertEqual(fake_headers, resp_headers)
self.assertEqual(json_loads(body), resp)
direct_client.http_connect = mock_http_connect(204, fake_headers, body)
resp_headers, resp = direct_client.direct_get_account(node, part, account)
fake_headers.update({'user-agent': 'direct-client %s' % os.getpid()})
self.assertEqual(fake_headers, resp_headers)
self.assertEqual([], resp)
direct_client.http_connect = was_http_connector
开发者ID:zhouyuan，项目名称:swift，代码行数:35，代码来源:test_direct_client.py
示例3: test_direct_get_account_error
def test_direct_get_account_error(self):
with mocked_http_conn(500) as conn:
with self.assertRaises(ClientException) as raised:
direct_client.direct_get_account(
self.node, self.part, self.account)
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(raised.exception.http_status, 500)
self.assertTrue('GET' in str(raised.exception))
开发者ID:jgmerritt，项目名称:swift，代码行数:9，代码来源:test_direct_client.py
示例4: test_direct_get_account_error
def test_direct_get_account_error(self):
with mocked_http_conn(500) as conn:
try:
direct_client.direct_get_account(
self.node, self.part, self.account)
except ClientException as err:
pass
else:
self.fail('ClientException not raised')
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(err.http_status, 500)
self.assertTrue('GET' in str(err))
开发者ID:aureliengoulon，项目名称:swift，代码行数:13，代码来源:test_direct_client.py
示例5: test_direct_get_account
def test_direct_get_account(self):
stub_headers = HeaderKeyDict({
'X-Account-Container-Count': '1',
'X-Account-Object-Count': '1',
'X-Account-Bytes-Used': '1',
'X-Timestamp': '1234567890',
'X-PUT-Timestamp': '1234567890'})
body = '[{"count": 1, "bytes": 20971520, "name": "c1"}]'
with mocked_http_conn(200, stub_headers, body) as conn:
resp_headers, resp = direct_client.direct_get_account(
self.node, self.part, self.account, marker='marker',
prefix='prefix', delimiter='delimiter', limit=1000,
end_marker='endmarker', reverse='on')
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(conn.req_headers['user-agent'], self.user_agent)
self.assertEqual(resp_headers, stub_headers)
self.assertEqual(json.loads(body), resp)
self.assertTrue('marker=marker' in conn.query_string)
self.assertTrue('delimiter=delimiter' in conn.query_string)
self.assertTrue('limit=1000' in conn.query_string)
self.assertTrue('prefix=prefix' in conn.query_string)
self.assertTrue('format=json' in conn.query_string)
self.assertTrue('end_marker=endmarker' in conn.query_string)
self.assertTrue('reverse=on' in conn.query_string)
开发者ID:aureliengoulon，项目名称:swift，代码行数:28，代码来源:test_direct_client.py
示例6: do_test
def do_test(req_params):
stub_headers = HeaderKeyDict({
'X-Account-Container-Count': '1',
'X-Account-Object-Count': '1',
'X-Account-Bytes-Used': '1',
'X-Timestamp': '1234567890',
'X-PUT-Timestamp': '1234567890'})
body = '[{"count": 1, "bytes": 20971520, "name": "c1"}]'
with mocked_http_conn(200, stub_headers, body) as conn:
resp_headers, resp = direct_client.direct_get_account(
self.node, self.part, self.account, **req_params)
try:
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(conn.req_headers['user-agent'],
self.user_agent)
self.assertEqual(resp_headers, stub_headers)
self.assertEqual(json.loads(body), resp)
self.assertIn('format=json', conn.query_string)
for k, v in req_params.items():
if v is None:
self.assertNotIn('&%s' % k, conn.query_string)
else:
self.assertIn('&%s=%s' % (k, v), conn.query_string)
except AssertionError as err:
self.fail('Failed with params %s: %s' % (req_params, err))
开发者ID:psachin，项目名称:swift，代码行数:29，代码来源:test_direct_client.py
示例7: test_direct_client_exception
def test_direct_client_exception(self):
stub_headers = {'X-Trans-Id': 'txb5f59485c578460f8be9e-0053478d09'}
body = 'a server error has occurred'
with mocked_http_conn(500, stub_headers, body):
with self.assertRaises(ClientException) as raised:
direct_client.direct_get_account(self.node, self.part,
self.account)
self.assertEqual(raised.exception.http_status, 500)
expected_err_msg_parts = (
'Account server %s:%s' % (self.node['ip'], self.node['port']),
'GET %r' % self.account_path,
'status 500',
)
for item in expected_err_msg_parts:
self.assertIn(item, str(raised.exception))
self.assertEqual(raised.exception.http_host, self.node['ip'])
self.assertEqual(raised.exception.http_port, self.node['port'])
self.assertEqual(raised.exception.http_device, self.node['device'])
self.assertEqual(raised.exception.http_status, 500)
self.assertEqual(raised.exception.http_reason, 'Internal Error')
self.assertEqual(raised.exception.http_headers, stub_headers)
开发者ID:jgmerritt，项目名称:swift，代码行数:21，代码来源:test_direct_client.py
示例8: test_direct_get_account_no_content_does_not_parse_body
def test_direct_get_account_no_content_does_not_parse_body(self):
headers = {
'X-Account-Container-Count': '1',
'X-Account-Object-Count': '1',
'X-Account-Bytes-Used': '1',
'X-Timestamp': '1234567890',
'X-PUT-Timestamp': '1234567890'}
with mocked_http_conn(204, headers) as conn:
resp_headers, resp = direct_client.direct_get_account(
self.node, self.part, self.account)
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(conn.req_headers['user-agent'], self.user_agent)
self.assertEqual(resp_headers, resp_headers)
self.assertEqual([], resp)
开发者ID:aureliengoulon，项目名称:swift，代码行数:16，代码来源:test_direct_client.py
示例9: test_direct_get_account
def test_direct_get_account(self):
stub_headers = HeaderKeyDict({
'X-Account-Container-Count': '1',
'X-Account-Object-Count': '1',
'X-Account-Bytes-Used': '1',
'X-Timestamp': '1234567890',
'X-PUT-Timestamp': '1234567890'})
body = '[{"count": 1, "bytes": 20971520, "name": "c1"}]'
with mocked_http_conn(200, stub_headers, body) as conn:
resp_headers, resp = direct_client.direct_get_account(
self.node, self.part, self.account)
self.assertEqual(conn.method, 'GET')
self.assertEqual(conn.path, self.account_path)
self.assertEqual(conn.req_headers['user-agent'], self.user_agent)
self.assertEqual(resp_headers, stub_headers)
self.assertEqual(json.loads(body), resp)
开发者ID:701，项目名称:swift，代码行数:19，代码来源:test_direct_client.py
示例10: test_main
#.........这里部分代码省略.........
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assertTrue(found1)
self.assertTrue(found2)
apart, anodes = self.account_ring.get_nodes(self.account)
kill_nonprimary_server(anodes, self.ipport2server, self.pids)
kill_server((anodes[0]['ip'], anodes[0]['port']),
self.ipport2server, self.pids)
# Kill account servers excepting two of the primaries
# Delete container1
client.delete_container(self.url, self.token, container1)
# Put container2/object2
client.put_object(self.url, self.token, container2, 'object2', '12345')
# Assert account level knows container1 is gone but doesn't know about
# container2/object2 yet
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assertTrue(not found1)
self.assertTrue(found2)
# Run container updaters
Manager(['container-updater']).once()
# Assert account level now knows about container2/object2
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 2)
self.assertEquals(container['bytes'], 9)
self.assertTrue(not found1)
self.assertTrue(found2)
# Restart other primary account server
start_server((anodes[0]['ip'], anodes[0]['port']),
self.ipport2server, self.pids)
# Assert that server doesn't know about container1's deletion or the
# new container2/object2 yet
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '2')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assertTrue(found1)
self.assertTrue(found2)
# Get to final state
self.get_to_final_state()
# Assert that server is now up to date
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 2)
self.assertEquals(container['bytes'], 9)
self.assertTrue(not found1)
self.assertTrue(found2)
开发者ID:renanalan，项目名称:swift，代码行数:101，代码来源:test_account_failures.py
示例11: test_main
#.........这里部分代码省略.........
if c['name'] == container1:
found1 = True
self.assertEquals(c['count'], 0)
self.assertEquals(c['bytes'], 0)
elif c['name'] == container2:
found2 = True
self.assertEquals(c['count'], 1)
self.assertEquals(c['bytes'], 4)
self.assert_(found1)
self.assert_(found2)
apart, anodes = self.account_ring.get_nodes(self.account)
kill(self.pids[self.port2server[anodes[0]['port']]], SIGTERM)
client.delete_container(self.url, self.token, container1)
client.put_object(self.url, self.token, container2, 'object2', '12345')
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for c in containers:
if c['name'] == container1:
found1 = True
elif c['name'] == container2:
found2 = True
self.assertEquals(c['count'], 1)
self.assertEquals(c['bytes'], 4)
self.assert_(not found1)
self.assert_(found2)
ps = []
for n in xrange(1, 5):
ps.append(Popen(['swift-container-updater',
'/etc/swift/container-server/%d.conf' % n,
'once']))
for p in ps:
p.wait()
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for c in containers:
if c['name'] == container1:
found1 = True
elif c['name'] == container2:
found2 = True
self.assertEquals(c['count'], 2)
self.assertEquals(c['bytes'], 9)
self.assert_(not found1)
self.assert_(found2)
self.pids[self.port2server[anodes[0]['port']]] = \
Popen(['swift-account-server',
'/etc/swift/account-server/%d.conf' %
((anodes[0]['port'] - 6002) / 10)]).pid
sleep(2)
# This is the earlier counts and bytes because the first node doesn't
# have the newest udpates yet.
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '2')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for c in containers:
if c['name'] == container1:
found1 = True
elif c['name'] == container2:
found2 = True
# This is the earlier count and bytes because the first node
# doesn't have the newest udpates yet.
self.assertEquals(c['count'], 1)
self.assertEquals(c['bytes'], 4)
# This okay because the first node hasn't got the update that
# container1 was deleted yet.
self.assert_(found1)
self.assert_(found2)
get_to_final_state()
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for c in containers:
if c['name'] == container1:
found1 = True
elif c['name'] == container2:
found2 = True
self.assertEquals(c['count'], 2)
self.assertEquals(c['bytes'], 9)
self.assert_(not found1)
self.assert_(found2)
开发者ID:HodongHwang，项目名称:swift，代码行数:101，代码来源:test_account_failures.py
示例12: test_main
#.........这里部分代码省略.........
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '2')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
self.assertEquals(container['count'], 0)
self.assertEquals(container['bytes'], 0)
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assert_(found1)
self.assert_(found2)
apart, anodes = self.account_ring.get_nodes(self.account)
kill_nonprimary_server(anodes, self.port2server, self.pids)
kill_server(anodes[0]['port'], self.port2server, self.pids)
client.delete_container(self.url, self.token, container1)
client.put_object(self.url, self.token, container2, 'object2', '12345')
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assert_(not found1)
self.assert_(found2)
processes = []
for node in xrange(1, 5):
processes.append(Popen([
'swift-container-updater',
self.configs['container'] % node,
'once']))
for process in processes:
process.wait()
headers, containers = client.get_account(self.url, self.token)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 2)
self.assertEquals(container['bytes'], 9)
self.assert_(not found1)
self.assert_(found2)
start_server(anodes[0]['port'], self.port2server, self.pids)
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '2')
self.assertEquals(headers['x-account-object-count'], '1')
self.assertEquals(headers['x-account-bytes-used'], '4')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 1)
self.assertEquals(container['bytes'], 4)
self.assert_(found1)
self.assert_(found2)
get_to_final_state()
headers, containers = \
direct_client.direct_get_account(anodes[0], apart, self.account)
self.assertEquals(headers['x-account-container-count'], '1')
self.assertEquals(headers['x-account-object-count'], '2')
self.assertEquals(headers['x-account-bytes-used'], '9')
found1 = False
found2 = False
for container in containers:
if container['name'] == container1:
found1 = True
elif container['name'] == container2:
found2 = True
self.assertEquals(container['count'], 2)
self.assertEquals(container['bytes'], 9)
self.assert_(not found1)
self.assert_(found2)
开发者ID:10389030，项目名称:swift，代码行数:101，代码来源:test_account_failures.py
注：本文中的swift.common.direct_client.direct_get_account函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论