tools: qvm-clone

The tool and tests.

qubesmgmt/tests/tools/qvm_clone.py

@@ -0,0 +1,62 @@
+# -*- encoding: utf8 -*-
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2017 Marek Marczykowski-Górecki
+#                               <marmarek@invisiblethingslab.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, see <http://www.gnu.org/licenses/>.
+import qubesmgmt.tests
+import qubesmgmt.tests.tools
+import qubesmgmt.tools.qvm_clone
+
+
+class TC_00_qvm_clone(qubesmgmt.tests.QubesTestCase):
+    def test_000_simple(self):
+        self.app.expected_calls[('test-vm', 'mgmt.vm.Clone', None,
+            b'name=new-vm')] = b'0\x00'
+        self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
+            b'0\x00new-vm class=AppVM state=Halted\n' \
+            b'test-vm class=AppVM state=Halted\n'
+        qubesmgmt.tools.qvm_clone.main(['test-vm', 'new-vm'], app=self.app)
+        self.assertAllCalled()
+
+    def test_001_missing_vm(self):
+        with self.assertRaises(SystemExit):
+            with qubesmgmt.tests.tools.StderrBuffer() as stderr:
+                qubesmgmt.tools.qvm_clone.main(['test-vm'], app=self.app)
+        self.assertIn('NAME', stderr.getvalue())
+        self.assertAllCalled()
+
+    def test_004_pool(self):
+        self.app.expected_calls[('test-vm', 'mgmt.vm.CloneInPool',
+            None, b'name=new-vm pool=some-pool')] = b'0\x00'
+        self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
+            b'0\x00new-vm class=AppVM state=Halted\n' \
+            b'test-vm class=AppVM state=Halted\n'
+        qubesmgmt.tools.qvm_clone.main(['-P', 'some-pool', 'test-vm', 'new-vm'],
+            app=self.app)
+        self.assertAllCalled()
+
+    def test_005_pools(self):
+        self.app.expected_calls[('test-vm', 'mgmt.vm.CloneInPool',
+            None, b'name=new-vm pool:private=some-pool '
+                  b'pool:volatile=other-pool')] = b'0\x00'
+        self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
+            b'0\x00new-vm class=AppVM state=Halted\n' \
+            b'test-vm class=AppVM state=Halted\n'
+        qubesmgmt.tools.qvm_clone.main(['--pool', 'private=some-pool',
+            '--pool', 'volatile=other-pool', 'test-vm', 'new-vm'],
+            app=self.app)
+        self.assertAllCalled()

qubesmgmt/tools/qvm_clone.py

@@ -0,0 +1,72 @@
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
+# Copyright (C) 2017 Marek Marczykowski-Górecki
+#                               <marmarek@invisiblethingslab.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+#
+
+''' Clone a domain '''
+
+import sys
+
+from qubesmgmt.tools import QubesArgumentParser
+
+parser = QubesArgumentParser(description=__doc__, vmname_nargs=1)
+parser.add_argument('new_name',
+                    metavar='NEWVM',
+                    action='store',
+                    help='name of the domain to create')
+
+group = parser.add_mutually_exclusive_group()
+group.add_argument('-P',
+                    metavar='POOL',
+                    dest='one_pool',
+                    default='',
+                    help='pool to use for the new domain')
+
+group.add_argument('-p',
+                    '--pool',
+                    action='append',
+                    metavar='VOLUME=POOL',
+                    help='specify the pool to use for the specific volume')
+
+
+def main(args=None, app=None):
+    ''' Clones an existing VM by copying all its disk files '''
+    args = parser.parse_args(args, app=app)
+    app = args.app
+    src_vm = args.domains[0]
+    new_name = args.new_name
+
+    pool = None
+    pools = {}
+    if args.one_pool:
+        pool = args.one_pool
+    elif hasattr(args, 'pool') and args.pool:
+        for pool_vol in args.pool:
+            try:
+                volume_name, pool_name = pool_vol.split('=')
+                pools[volume_name] = pool_name
+            except ValueError:
+                parser.error(
+                    'Pool argument must be of form: -P volume_name=pool_name')
+
+    app.clone_vm(src_vm, new_name, pool=pool, pools=pools)
+
+if __name__ == '__main__':
+    sys.exit(main())