lib.ExecApp: Add class

ExecApp is a ready-made base class for applications that operate
through an ExecContext: it adds the --interactive, --verbose and
--target options, exposes interactive, verbose and exec_context
properties, and closes the exec context when the async context
manager exits.

The code for that has lived in jw.pkg.App code before, which now
inherits from ExecApp.

A fix along the way: __aexit__() closes the exec context and then chains
to super().__aexit__(), so App.close() runs when the async context
manager exits. Before the change, exiting the async context left the app
unclosed; close() ran only on the run() path. Add a unit test that
builds an ExecApp with a root command and asserts that close() runs on
context exit and that the exec options are registered.

The exec options are now registered before App's own options,
which moves them up in the rendered --help output. Update the
golden file of the help integration test to match.

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v0.84.2

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')