Qubes/core-admin-client
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tools: finish 'qvm-volume revert' implementation, add tests

Browse Source 
It wasn't aware of snapshot identifier, fix that. Also update to use
VM:VOLUME syntax, not POOL:VID.
This commit is contained in:
Marek Marczykowski-Górecki 2017-10-02 21:11:14 +02:00
parent c87820fba8
commit 8476afc306
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
2 changed files with 85 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
qubesadmin/tests/tools/qvm_volume.py
View File

@ -179,3 +179,74 @@ class TC_00_qvm_volume(qubesadmin.tests.QubesTestCase):
app=self.app))
self.assertIn('shrink not allowed', stderr.getvalue())
self.assertAllCalled()
def test_020_revert(self):
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00testvm class=AppVM state=Running\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.ListSnapshots', 'private', None)] = \
b'0\x00200101010000\n200201010000\n200301010000\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.Revert', 'private', b'200301010000')] = \
b'0\x00'
self.assertEqual(0,
qubesadmin.tools.qvm_volume.main(
['revert', 'testvm:private'],
app=self.app))
self.assertAllCalled()
def test_021_revert_error(self):
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00testvm class=AppVM state=Running\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.ListSnapshots', 'private', None)] = \
b'0\x00200101010000\n200201010000\n200301010000\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.Revert', 'private', b'200301010000')] = \
b'2\x00StoragePoolException\x00\x00Failed to revert volume: ' \
b'some error\x00'
with qubesadmin.tests.tools.StderrBuffer() as stderr:
self.assertEqual(1,
qubesadmin.tools.qvm_volume.main(
['revert', 'testvm:private'],
app=self.app))
self.assertIn('some error', stderr.getvalue())
self.assertAllCalled()
def test_022_revert_no_snapshots(self):
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00testvm class=AppVM state=Running\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.ListSnapshots', 'private', None)] = \
b'0\x00'
with qubesadmin.tests.tools.StderrBuffer() as stderr:
self.assertEqual(1,
qubesadmin.tools.qvm_volume.main(
['revert', 'testvm:private'],
app=self.app))
self.assertIn('No snapshots', stderr.getvalue())
self.assertAllCalled()
def test_023_revert_specific(self):
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00testvm class=AppVM state=Running\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\n'
self.app.expected_calls[
('testvm', 'admin.vm.volume.Revert', 'private', b'20050101')] = \
b'0\x00'
self.assertEqual(0,
qubesadmin.tools.qvm_volume.main(
['revert', 'testvm:private', '20050101'],
app=self.app))
self.assertAllCalled()

21
qubesadmin/tools/qvm_volume.py
View File

@ -134,13 +134,16 @@ def list_volumes(args):
def revert_volume(args):
''' Revert volume to previous state '''
volume = args.volume
app = args.app
try:
pool = app.pools[volume.pool]
pool.revert(volume)
except qubesadmin.exc.StoragePoolException as e:
print(str(e), file=sys.stderr)
sys.exit(1)
if args.revision:
revision = args.revision
else:
revisions = volume.revisions
if not revisions:
raise qubesadmin.exc.StoragePoolException(
'No snapshots available')
revision = volume.revisions[-1]
volume.revert(revision)
def extend_volumes(args):
@ -177,6 +180,10 @@ def init_revert_parser(sub_parsers):
help='revert volume to previous revision')
revert_parser.add_argument(metavar='VM:VOLUME', dest='volume',
action=qubesadmin.tools.VMVolumeAction)
revert_parser.add_argument(metavar='REVISION', dest='revision',
help='Optional revision to revert to;'
'if not specified, latest one is assumed',
action='store', nargs='?')
revert_parser.set_defaults(func=revert_volume)