openstack
/
python-heatclient
Code Issues Proposed changes

Merge "Implement snapshot operations"

This commit is contained in:
Jenkins 2014-11-24 02:31:16 +00:00 committed by Gerrit Code Review
parent e6949d0c48 49ab7b36df
commit f5f7eab85e
4 changed files with 278 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
heatclient/tests/test_shell.py
View File

@ -1750,6 +1750,146 @@ class ShellTestUserPass(ShellBase):
for r in required:
self.assertRegexpMatches(build_info_text, r)
@httpretty.activate
def test_stack_snapshot(self):
self.register_keystone_auth_fixture()
stack_dict = {"stack": {
"id": "1",
"stack_name": "teststack",
"stack_status": 'CREATE_COMPLETE',
"creation_time": "2012-10-25T01:58:47Z"
}}
resp_dict = {"snapshot": {
"id": "1",
"creation_time": "2012-10-25T01:58:47Z"
}}
resp = fakes.FakeHTTPResponse(
200,
'OK',
{'content-type': 'application/json'},
jsonutils.dumps(resp_dict))
http.HTTPClient.json_request(
'GET', '/stacks/teststack/1').AndReturn((resp, stack_dict))
http.HTTPClient.json_request(
'POST',
'/stacks/teststack/1/snapshots',
data={}).AndReturn((resp, resp_dict))
self.m.ReplayAll()
resp = self.shell('stack-snapshot teststack/1')
self.assertEqual(resp_dict, jsonutils.loads(resp))
@httpretty.activate
def test_snapshot_show(self):
self.register_keystone_auth_fixture()
stack_dict = {"stack": {
"id": "1",
"stack_name": "teststack",
"stack_status": 'CREATE_COMPLETE',
"creation_time": "2012-10-25T01:58:47Z"
}}
resp_dict = {"snapshot": {
"id": "2",
"creation_time": "2012-10-25T01:58:47Z"
}}
resp = fakes.FakeHTTPResponse(
200,
'OK',
{'content-type': 'application/json'},
jsonutils.dumps(resp_dict))
http.HTTPClient.json_request(
'GET', '/stacks/teststack/1').AndReturn((resp, stack_dict))
http.HTTPClient.json_request(
'GET',
'/stacks/teststack/1/snapshots/2').AndReturn((resp, resp_dict))
self.m.ReplayAll()
resp = self.shell('snapshot-show teststack/1 2')
self.assertEqual(resp_dict, jsonutils.loads(resp))
@httpretty.activate
def test_snapshot_delete(self):
self.register_keystone_auth_fixture()
stack_dict = {"stack": {
"id": "1",
"stack_name": "teststack",
"stack_status": 'CREATE_COMPLETE',
"creation_time": "2012-10-25T01:58:47Z"
}}
resp_dict = {"snapshot": {
"id": "2",
"creation_time": "2012-10-25T01:58:47Z"
}}
resp = fakes.FakeHTTPResponse(
204,
'No Content',
{},
None)
http.HTTPClient.json_request(
'GET', '/stacks/teststack/1').AndReturn((resp, stack_dict))
http.HTTPClient.json_request(
'DELETE',
'/stacks/teststack/1/snapshots/2').AndReturn((resp, resp_dict))
self.m.ReplayAll()
resp = self.shell('snapshot-delete teststack/1 2')
self.assertEqual("", resp)
@httpretty.activate
def test_snapshot_list(self):
self.register_keystone_auth_fixture()
stack_dict = {"stack": {
"id": "1",
"stack_name": "teststack",
"stack_status": 'CREATE_COMPLETE',
"creation_time": "2012-10-25T01:58:47Z"
}}
resp_dict = {"snapshots": [{
"id": "2",
"name": "snap1",
"status": "COMPLETE",
"status_reason": "",
"data": {}
}]}
resp = fakes.FakeHTTPResponse(
200,
'OK',
{'content-type': 'application/json'},
jsonutils.dumps(resp_dict))
http.HTTPClient.json_request(
'GET', '/stacks/teststack/1').AndReturn((resp, stack_dict))
http.HTTPClient.json_request(
'GET',
'/stacks/teststack/1/snapshots').AndReturn((resp, resp_dict))
self.m.ReplayAll()
list_text = self.shell('snapshot-list teststack/1')
required = [
'id',
'name',
'status',
'status_reason',
'data',
'2',
'COMPLETE',
'{}',
]
for r in required:
self.assertRegexpMatches(list_text, r)
class ShellTestEvents(ShellBase):

26
heatclient/tests/test_stacks.py
View File

@ -105,6 +105,32 @@ class StackOperationsTest(testtools.TestCase):
stack = stack.preview()
manager.preview.assert_called_once_with()
def test_snapshot(self):
manager = MagicMock()
stack = mock_stack(manager, 'the_stack', 'abcd1234')
stack.snapshot('foo')
manager.snapshot.assert_called_once_with('the_stack/abcd1234', 'foo')
def test_snapshot_show(self):
manager = MagicMock()
stack = mock_stack(manager, 'the_stack', 'abcd1234')
stack.snapshot_show('snap1234')
manager.snapshot_show.assert_called_once_with(
'the_stack/abcd1234', 'snap1234')
def test_snapshot_delete(self):
manager = MagicMock()
stack = mock_stack(manager, 'the_stack', 'abcd1234')
stack.snapshot_delete('snap1234')
manager.snapshot_delete.assert_called_once_with(
'the_stack/abcd1234', 'snap1234')
def test_snapshot_list(self):
manager = MagicMock()
stack = mock_stack(manager, 'the_stack', 'abcd1234')
stack.snapshot_list()
manager.snapshot_list.assert_called_once_with('the_stack/abcd1234')
class StackManagerNoPaginationTest(testtools.TestCase):

67
heatclient/v1/shell.py
View File

@ -896,3 +896,70 @@ def do_build_info(hc, args):
'engine': utils.json_formatter,
}
utils.print_dict(result, formatters=formatters)
@utils.arg('id', metavar='<NAME or ID>',
help='Name or ID of stack to snapshot.')
@utils.arg('-n', '--name', metavar='<NAME>',
help='If specified, the name given to the snapshot.')
def do_stack_snapshot(hc, args):
'''Make a snapshot of a stack.'''
fields = {'stack_id': args.id}
if args.name:
fields['name'] = args.name
try:
snapshot = hc.stacks.snapshot(**fields)
except exc.HTTPNotFound:
raise exc.CommandError('Stack not found: %s' % args.id)
else:
print(jsonutils.dumps(snapshot, indent=2, ensure_ascii=False))
@utils.arg('id', metavar='<NAME or ID>',
help='Name or ID of the stack containing the snapshot.')
@utils.arg('snapshot', metavar='<SNAPSHOT>',
help='The ID of the snapshot to show.')
def do_snapshot_show(hc, args):
'''Show a snapshot of a stack.'''
fields = {'stack_id': args.id, 'snapshot_id': args.snapshot}
try:
snapshot = hc.stacks.snapshot_show(**fields)
except exc.HTTPNotFound:
raise exc.CommandError('Stack or snapshot not found')
else:
print(jsonutils.dumps(snapshot, indent=2, ensure_ascii=False))
@utils.arg('id', metavar='<NAME or ID>',
help='Name or ID of the stack containing the snapshot.')
@utils.arg('snapshot', metavar='<SNAPSHOT>',
help='The ID of the snapshot to delete.')
def do_snapshot_delete(hc, args):
'''Delete a snapshot of a stack.'''
fields = {'stack_id': args.id, 'snapshot_id': args.snapshot}
try:
hc.stacks.snapshot_delete(**fields)
except exc.HTTPNotFound:
raise exc.CommandError('Stack or snapshot not found')
@utils.arg('id', metavar='<NAME or ID>',
help='Name or ID of the stack containing the snapshots.')
def do_snapshot_list(hc, args):
'''List the snapshots of a stack.'''
fields = {'stack_id': args.id}
try:
snapshots = hc.stacks.snapshot_list(**fields)
except exc.HTTPNotFound:
raise exc.CommandError('Stack not found: %s' % args.id)
else:
fields = ['id', 'name', 'status', 'status_reason', 'data']
formatters = {
'id': lambda x: x['id'],
'name': lambda x: x['name'],
'status': lambda x: x['status'],
'status_reason': lambda x: x['status_reason'],
'data': lambda x: jsonutils.dumps(x['data'], indent=2,
ensure_ascii=False),
}
utils.print_list(snapshots["snapshots"], fields, formatters=formatters)

45
heatclient/v1/stacks.py
View File

@ -38,6 +38,18 @@ class Stack(base.Resource):
def abandon(self):
return self.manager.abandon(self.identifier)
def snapshot(self, name=None):
return self.manager.snapshot(self.identifier, name)
def snapshot_show(self, snapshot_id):
return self.manager.snapshot_show(self.identifier, snapshot_id)
def snapshot_delete(self, snapshot_id):
return self.manager.snapshot_delete(self.identifier, snapshot_id)
def snapshot_list(self):
return self.manager.snapshot_list(self.identifier)
def get(self):
# set_loaded() first ... so if we have to bail, we know we tried.
self._loaded = True
@ -138,6 +150,39 @@ class StackManager(base.BaseManager):
'/stacks/%s/abandon' % stack.identifier)
return body
def snapshot(self, stack_id, name=None):
"""Snapshot a stack."""
stack = self.get(stack_id)
data = {}
if name:
data['name'] = name
resp, body = self.client.json_request(
'POST',
'/stacks/%s/snapshots' % stack.identifier,
data=data)
return body
def snapshot_show(self, stack_id, snapshot_id):
stack = self.get(stack_id)
resp, body = self.client.json_request(
'GET',
'/stacks/%s/snapshots/%s' % (stack.identifier, snapshot_id))
return body
def snapshot_delete(self, stack_id, snapshot_id):
stack = self.get(stack_id)
resp, body = self.client.json_request(
'DELETE',
'/stacks/%s/snapshots/%s' % (stack.identifier, snapshot_id))
return body
def snapshot_list(self, stack_id):
stack = self.get(stack_id)
resp, body = self.client.json_request(
'GET',
'/stacks/%s/snapshots' % stack.identifier)
return body
def get(self, stack_id):
"""Get the metadata for a specific stack.