src/python/jw/pkg/App.py

@@ -6,13 +6,12 @@ from __future__ import annotations
 
 import os
 import re
-import sys
 
 from enum import Enum, auto
 from functools import cache
-from typing import TYPE_CHECKING, Any, override
+from typing import TYPE_CHECKING, override
 
-from .lib.App import App as Base
+from .lib.ExecApp import ExecApp as Base
 from .lib.Distro import Distro
 from .lib.log import DEBUG, ERR, log
 from .lib.ProjectConf import ProjectConf
@@ -23,7 +22,6 @@ if TYPE_CHECKING:
     from argparse import ArgumentParser
     from typing import TypeAlias
 
-    from .lib.ExecContext import ExecContext
     from .lib.PackageFilter import PackageFilter
 
 # Meaning of pkg.requires.xxx variables
@@ -299,16 +297,15 @@ class App(Base):
 
     def __init__(self, distro: Distro | None = None) -> None:
 
-        super().__init__('jw-pkg swiss army knife', modules = ['jw.pkg.cmds'])
+        super().__init__(
+            description = 'jw-pkg swiss army knife', modules = ['jw.pkg.cmds']
+        )
 
         # -- Members without default values
-        self.__opt_interactive: bool | None = None
-        self.__opt_verbose: bool | None = None
         self.__top_name: str | None = None
         self.__distro = distro
         self.___topdir: str | None = None
         self.___pretty_topdir: str | None = None
-        self.__exec_context: ExecContext | None = None
 
         # -- Members with default values
         self.__topdir_fmt = 'absolute'
@@ -332,12 +329,6 @@ class App(Base):
                 default_pkg_filter = pkg_filter,
             )
 
-    @override
-    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
-        if self.__exec_context is not None:
-            await self.__exec_context.close()
-            self.__exec_context = None
-
     @override
     def _add_arguments(self, parser: ArgumentParser) -> None:
         super()._add_arguments(parser)
@@ -361,21 +352,6 @@ class App(Base):
             default = None,
             help = 'Distribution ID (default is taken from /etc/os-release)',
         )
-        parser.add_argument(
-            '--interactive',
-            choices = ['true', 'false', 'auto'],
-            default = 'true',
-            help = 'Wait for user input or try to proceed unattended',
-        )
-        parser.add_argument(
-            '--verbose',
-            action = 'store_true',
-            default = False,
-            help = "Be verbose on stderr about what's being done on the distro level",
-        )
-        parser.add_argument(
-            '--target', default = 'local', help = 'Run commands on this host'
-        )
         parser.add_argument(
             '--pkg-filter',
             help = 'Default filter for all distribution package-related operations',
@@ -404,44 +380,6 @@ class App(Base):
         await self.__init_async()
         await super()._run(args)
 
-    @property
-    def interactive(self) -> bool:
-        if self.__opt_interactive is None:
-            match self.args.interactive:
-                case 'true':
-                    self.__opt_interactive = True
-                case 'false':
-                    self.__opt_interactive = False
-                case 'auto':
-                    self.__opt_interactive = sys.stdin.isatty()
-                case _:
-                    raise ValueError(
-                        f'Unknown --interactive value: {self.args.interactive}'
-                    )
-            # Not logically possible to fail, but this keeps pyright happy
-            assert self.__opt_interactive is not None
-        return self.__opt_interactive
-
-    @property
-    def verbose(self) -> bool:
-        if self.__opt_verbose is None:
-            self.__opt_verbose = self.args.verbose
-            # Not logically possible to fail, but this keeps pyright happy
-            assert self.__opt_verbose is not None
-        return self.__opt_verbose
-
-    @property
-    def exec_context(self) -> ExecContext:
-        if self.__exec_context is None:
-            from .lib.ExecContext import ExecContext
-
-            self.__exec_context = ExecContext.create(
-                self.args.target,
-                interactive = self.interactive,
-                verbose_default = self.verbose,
-            )
-        return self.__exec_context
-
     @property
     def top_name(self) -> str | None:
         return self.__top_name

src/python/jw/pkg/lib/ExecApp.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import sys
+
+from typing import TYPE_CHECKING, Any, override
+
+from .App import App as Base
+
+if TYPE_CHECKING:
+
+    from argparse import ArgumentParser
+
+    from .ExecContext import ExecContext
+
+class ExecApp(Base):  # export
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self.__opt_interactive: bool | None = None
+        self.__opt_verbose: bool | None = None
+        self.__exec_context: ExecContext | None = None
+
+    @override
+    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
+        if self.__exec_context is not None:
+            await self.__exec_context.close()
+            self.__exec_context = None
+        # -- Close the app (async runner, event loop) after the exec context,
+        # which may still need them.
+        await super().__aexit__(exc_type, exc, tb)
+
+    @override
+    def _add_arguments(self, parser: ArgumentParser) -> None:
+        super()._add_arguments(parser)
+        parser.add_argument(
+            '--interactive',
+            choices = ['true', 'false', 'auto'],
+            default = 'true',
+            help = 'Wait for user input or try to proceed unattended',
+        )
+        parser.add_argument(
+            '--verbose',
+            action = 'store_true',
+            default = False,
+            help = "Be verbose on stderr about what's being done on the distro level",
+        )
+        parser.add_argument(
+            '--target', default = 'local', help = 'Run commands on this host'
+        )
+
+    @property
+    def interactive(self) -> bool:
+        if self.__opt_interactive is None:
+            match self.args.interactive:
+                case 'true':
+                    self.__opt_interactive = True
+                case 'false':
+                    self.__opt_interactive = False
+                case 'auto':
+                    self.__opt_interactive = sys.stdin.isatty()
+                case _:
+                    raise ValueError(
+                        f'Unknown --interactive value: {self.args.interactive}'
+                    )
+            # Not logically possible to fail, but this keeps pyright happy
+            assert self.__opt_interactive is not None
+        return self.__opt_interactive
+
+    @property
+    def verbose(self) -> bool:
+        if self.__opt_verbose is None:
+            self.__opt_verbose = self.args.verbose
+            # Not logically possible to fail, but this keeps pyright happy
+            assert self.__opt_verbose is not None
+        return self.__opt_verbose
+
+    @property
+    def exec_context(self) -> ExecContext:
+        if self.__exec_context is None:
+            from .ExecContext import ExecContext
+
+            self.__exec_context = ExecContext.create(
+                self.args.target,
+                interactive = self.interactive,
+                verbose_default = self.verbose,
+            )
+        return self.__exec_context

test/integration/jw-pkg/help/test-expected.txt

@@ -1,10 +1,11 @@
 ============= Running: jw-pkg.py -t ../../../.. --log-level info --help
 usage: jw-pkg.py [--log-flags LOG_FLAGS] [--log-level LOG_LEVEL]
                  [--log-file LOG_FILE] [--backtrace]
-                 [--write-profile WRITE_PROFILE] [-t TOPDIR]
+                 [--write-profile WRITE_PROFILE]
-                 [--topdir-format TOPDIR_FORMAT] [-p PREFIX]
+                 [--interactive {true,false,auto}] [--verbose]
-                 [--distro-id DISTRO_ID] [--interactive {true,false,auto}]
+                 [--target TARGET] [-t TOPDIR] [--topdir-format TOPDIR_FORMAT]
-                 [--verbose] [--target TARGET] [--pkg-filter PKG_FILTER] [-h]
+                 [-p PREFIX] [--distro-id DISTRO_ID] [--pkg-filter PKG_FILTER]
+                 [-h]
                   ...
 
 jw-pkg swiss army knife
@@ -18,6 +19,12 @@ options:
   --backtrace           Show exception backtraces (default: False)
   --write-profile WRITE_PROFILE
                         Profile code and store output to file (default: None)
+  --interactive {true,false,auto}
+                        Wait for user input or try to proceed unattended
+                        (default: true)
+  --verbose             Be verbose on stderr about what's being done on the
+                        distro level (default: False)
+  --target TARGET       Run commands on this host (default: local)
   -t, --topdir TOPDIR   Project Path (default: None)
   --topdir-format TOPDIR_FORMAT
                         Output references to topdir as one of "make:<var-
@@ -28,12 +35,6 @@ options:
   --distro-id DISTRO_ID
                         Distribution ID (default is taken from /etc/os-
                         release) (default: None)
-  --interactive {true,false,auto}
-                        Wait for user input or try to proceed unattended
-                        (default: true)
-  --verbose             Be verbose on stderr about what's being done on the
-                        distro level (default: False)
-  --target TARGET       Run commands on this host (default: local)
   --pkg-filter PKG_FILTER
                         Default filter for all distribution package-related
                         operations (default: None)

test/unit/python/jw/pkg/lib/ExecApp/Makefile

@@ -0,0 +1,7 @@
+TOPDIR = ../../../../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+all:
+test: run

test/unit/python/jw/pkg/lib/ExecApp/test.py

@@ -0,0 +1,57 @@
+import asyncio
+
+import sys
+
+from jw.pkg.lib.Cmd import Cmd
+from jw.pkg.lib.ExecApp import ExecApp
+
+# -- A minimal root command so the app can be built without a command tree.
+
+class RootCmd(Cmd):
+
+    def __init__(self, parent):
+        super().__init__(parent, 'root', 'Root command')
+
+    async def _run(self, args):
+        pass
+
+# -- Counts the close() calls made by App.__aexit__() when the async context
+# manager exits.
+
+class RecordingExecApp(ExecApp):
+
+    closed = 0
+
+    def close(self):
+        RecordingExecApp.closed += 1
+        super().close()
+
+# -- App.__init__ builds the parser from sys.argv, so point it at a clean
+# command line while constructing the app.
+
+saved_argv = sys.argv
+sys.argv = ['jw-pkg-test']
+try:
+    app = RecordingExecApp(description = 'ExecApp test', root = RootCmd)
+finally:
+    sys.argv = saved_argv
+
+# -- ExecApp registers the exec-related options on the top-level parser.
+
+opts = [
+    o for a in app.parser._actions for o in getattr(a, 'option_strings', ())
+]
+for opt in ('--interactive', '--verbose', '--target'):
+    assert opt in opts, f'{opt} must be registered by ExecApp'
+
+# -- Exiting the async context must close the app, through the
+# App.__aexit__() that ExecApp.__aexit__() chains to.
+
+async def exit_context():
+    async with app:
+        pass
+
+asyncio.run(exit_context())
+assert RecordingExecApp.closed == 1, 'exiting the app context must close the app'
+
+print('All ExecApp tests passed')