Migrate action to utils to their own module

Author: ben-grande

qubesadmin/tools/qvm_kill.py

@@ -24,11 +24,10 @@
 import asyncio
 import sys
 
-import qubesadmin.exc
-import qubesadmin.tools
-import qubesadmin.tools.qvm_shutdown
+from qubesadmin.tools import QubesArgumentParser
+from qubesadmin.utils.kill import kill
 
-parser = qubesadmin.tools.QubesArgumentParser(
+parser = QubesArgumentParser(
     description='immediately terminate a qube without a graceful shutdown'
                 ' sequence',
     vmname_nargs='+')
@@ -37,7 +36,7 @@
 async def run_async(args=None, app=None):
     # pylint: disable=missing-docstring
     args = parser.parse_args(args, app=app)
-    failed = await qubesadmin.utils.kill(domains=args.domains)
+    failed = await kill(domains=args.domains)
     if not failed:
         return 0
     for qube, exc in failed.items():

qubesadmin/tools/qvm_pause.py

@@ -22,10 +22,13 @@
 
 import asyncio
 import sys
-import qubesadmin
 
+from qubesadmin.tools import QubesArgumentParser
+from qubesadmin.utils.pause import pause
+from qubesadmin.utils.suspend import suspend
 
-parser = qubesadmin.tools.QubesArgumentParser(vmname_nargs='+',
+
+parser = QubesArgumentParser(vmname_nargs='+',
     description='pause a qube',
     epilog='Paused qubes will be killed on system shutdown.')
 
@@ -48,10 +51,10 @@ def main(args=None, app=None):
     exit_code = 0
     if args.suspend:
         action = "suspend"
-        failed = asyncio.run(qubesadmin.utils.suspend(domains=args.domains))
+        failed = asyncio.run(suspend(domains=args.domains))
     else:
         action = "pause"
-        failed = asyncio.run(qubesadmin.utils.pause(domains=args.domains))
+        failed = asyncio.run(pause(domains=args.domains))
     if failed:
         exit_code = 1
         for qube, exc in failed.items():

qubesadmin/tools/qvm_shutdown.py

@@ -28,12 +28,12 @@
 import sys
 from warnings import warn
 
-import qubesadmin.tools
-import qubesadmin.exc
+from qubesadmin.tools import QubesArgumentParser
+from qubesadmin.exc import QubesVMInUseError, QubesVMShutdownTimeoutError
+from qubesadmin.utils.shutdown import shutdown
+from qubesadmin.utils.kill import kill
 
-parser = qubesadmin.tools.QubesArgumentParser(
-    description=__doc__, vmname_nargs="+"
-)
+parser = QubesArgumentParser(description=__doc__, vmname_nargs="+")
 
 parser.add_argument(
     "--wait",
@@ -67,18 +67,18 @@
 )
 
 
-async def shutdown(domains, **shutdown_kwargs):
+async def shutdown_group_exc(domains, **shutdown_kwargs):
     # pylint: disable=missing-function-docstring
-    failed = await qubesadmin.utils.shutdown(domains=domains, **shutdown_kwargs)
+    failed = await shutdown(domains=domains, **shutdown_kwargs)
     used = {
         qube: exc
         for qube, exc in failed.items()
-        if isinstance(exc, qubesadmin.exc.QubesVMInUseError)
+        if isinstance(exc, QubesVMInUseError)
     }
     timedout = {
         qube: exc
         for qube, exc in failed.items()
-        if isinstance(exc, qubesadmin.exc.QubesVMShutdownTimeoutError)
+        if isinstance(exc, QubesVMShutdownTimeoutError)
     }
     unhandled = {
         qube: exc
@@ -105,7 +105,7 @@ async def run_async(args=None, app=None):
         "wait": args.wait,
     }
 
-    unhandled, used, timedout = await shutdown(
+    unhandled, used, timedout = await shutdown_group_exc(
         domains=args.domains, **shutdown_kwargs
     )
     unhandled_retry = {}
@@ -123,7 +123,7 @@ async def run_async(args=None, app=None):
                 ", ".join(qube.name for qube in used)
             )
         )
-        failed = await shutdown(domains=used, **shutdown_kwargs)
+        failed = await shutdown_group_exc(domains=used, **shutdown_kwargs)
         unhandled_retry, used, timedout_retry = failed
         if not failed:
             break
@@ -144,7 +144,7 @@ async def run_async(args=None, app=None):
                 ", ".join(qube.name for qube in timedout)
             )
         )
-        unhandled, used, timedout = await shutdown(
+        unhandled, used, timedout = await shutdown_group_exc(
             domains=timedout, **shutdown_kwargs
         )
 
@@ -154,7 +154,7 @@ async def run_async(args=None, app=None):
                 ", ".join(qube.name for qube in timedout)
             )
         )
-        timedout = await qubesadmin.utils.kill(domains=timedout)
+        timedout = await kill(domains=timedout)
 
     for item in [unhandled, used, timedout]:
         for qube, exc in item.items():

qubesadmin/tools/qvm_start.py

@@ -19,12 +19,13 @@
 # with this program; if not, see <http://www.gnu.org/licenses/>.
 
 """qvm-start - start a domain"""
+
 import asyncio
 import sys
 
 import qubesadmin.exc
 import qubesadmin.tools
-import qubesadmin.utils
+from qubesadmin.utils.start import DriveAction, start_expert
 
 
 parser = qubesadmin.tools.QubesArgumentParser(
@@ -49,7 +50,7 @@
 
 parser_drive.add_argument(
     "--hddisk",
-    action=qubesadmin.utils.DriveAction,
+    action=DriveAction,
     dest="drive",
     prefix="hd:",
     help="temporarily attach specified drive as hard disk",
@@ -58,7 +59,7 @@
 parser_drive.add_argument(
     "--cdrom",
     metavar="IMAGE",
-    action=qubesadmin.utils.DriveAction,
+    action=DriveAction,
     dest="drive",
     prefix="cdrom:",
     help="temporarily attach specified drive as CD/DVD",
@@ -79,7 +80,7 @@ async def run_async(args=None, app=None):
     args = parser.parse_args(args, app=app)
     tasks = [
         asyncio.to_thread(
-            qubesadmin.utils.start_expert,
+            start_expert,
             domain=qube,
             skip_if_running=args.skip_if_running,
             drive=args.drive,

qubesadmin/tools/qvm_unpause.py

@@ -22,10 +22,12 @@
 
 import asyncio
 import sys
-import qubesadmin
 
+from qubesadmin.tools import QubesArgumentParser
+from qubesadmin.utils.unpause import unpause
 
-parser = qubesadmin.tools.QubesArgumentParser(
+
+parser = QubesArgumentParser(
     vmname_nargs='+',
     description='unpause a domain')
 
@@ -47,7 +49,7 @@ def main(args=None, app=None):
             if not vm.features.get("internal")
         ]
 
-    failed = asyncio.run(qubesadmin.utils.unpause(domains=domains))
+    failed = asyncio.run(unpause(domains=domains))
     action = "unpause/resume"
     if failed:
         exit_code = 1