[PATCH v1 2/4] dts: add context manager for interactive shells

[Luca Vizzarro]

On 14/05/2024 21:14, jspewock at iol.unh.edu wrote:
> +class CriticalInteractiveShell(InteractiveShell):
<snip>
> +    _get_priviledged_command: Callable[[str], str] | None
typo: privileged
> +
> +    def __init__(
> +        self,
> +        interactive_session: SSHClient,
> +        logger: DTSLogger,
> +        get_privileged_command: Callable[[str], str] | None,
> +        app_args: str = "",
> +        timeout: float = SETTINGS.timeout,
> +    ) -> None:
> +        """Store parameters for creating an interactive shell, but do not start the application.
> +
> +        Note that this method also does not create the channel for the application, as this is
> +        something that isn't needed until the application starts.
> +
> +        Args:
> +            interactive_session: The SSH session dedicated to interactive shells.
> +            logger: The logger instance this session will use.
> +            get_privileged_command: A method for modifying a command to allow it to use
> +                elevated privileges. If :data:`None`, the application will not be started
> +                with elevated privileges.
> +            app_args: The command line arguments to be passed to the application on startup.
> +            timeout: The timeout used for the SSH channel that is dedicated to this interactive
> +                shell. This timeout is for collecting output, so if reading from the buffer
> +                and no output is gathered within the timeout, an exception is thrown. The default
> +                value for this argument may be modified using the :option:`--timeout` command-line
> +                argument or the :envvar:`DTS_TIMEOUT` environment variable.
> +        """
> +        self._interactive_session = interactive_session
> +        self._logger = logger
> +        self._timeout = timeout
> +        self._app_args = app_args
> +        self._get_priviledged_command = get_privileged_command
> +
> +    def __enter__(self: CriticalInteractiveShellType) -> CriticalInteractiveShellType:

This kind of type hinting is achievable with Python's own `Self`, which 
you can already add to this patch because mypy installs 
typing_extensions. Nevertheless, `Self` is also being introduced 
properly in my mypy update patch series.

> +
> +    def __exit__(self, type: BaseException, value: BaseException, traceback: TracebackType) -> None:
Since you are not using the arguments I'd just ignore them with `_`.
> +        """Exit the context block.
> +
> +        Upon exiting a context block with this class, we want to ensure that the instance of the
> +        application is explicitly closed and properly cleaned up using it's close method. Note that
> +        because this method returns :data:`None` if an exception was raised within the block, it is
> +        not handled and will be re-raised after the application is closed.
> +
> +        Args:
> +            type: Type of exception that was thrown in the context block if there was one.
> +            value: Value of the exception thrown in the context block if there was one.
> +            traceback: Traceback of the exception thrown in the context block if there was one.
> +        """
> +        self.close()

> diff --git a/dts/framework/remote_session/interactive_shell.py b/dts/framework/remote_session/interactive_shell.py
> index d1a9d8a6d2..08b8ba6a3e 100644
> --- a/dts/framework/remote_session/interactive_shell.py
> +++ b/dts/framework/remote_session/interactive_shell.py
> @@ -89,16 +89,19 @@ def __init__(
>                   and no output is gathered within the timeout, an exception is thrown.
>           """
>           self._interactive_session = interactive_session
> -        self._ssh_channel = self._interactive_session.invoke_shell()
> -        self._stdin = self._ssh_channel.makefile_stdin("w")
> -        self._stdout = self._ssh_channel.makefile("r")
> -        self._ssh_channel.settimeout(timeout)
> -        self._ssh_channel.set_combine_stderr(True)  # combines stdout and stderr streams
>           self._logger = logger
>           self._timeout = timeout
>           self._app_args = app_args
> +        self._init_channel()
>           self._start_application(get_privileged_command)
>   
> +    def _init_channel(self):
> +        self._ssh_channel = self._interactive_session.invoke_shell()
> +        self._stdin = self._ssh_channel.makefile_stdin("w")
> +        self._stdout = self._ssh_channel.makefile("r")
> +        self._ssh_channel.settimeout(self._timeout)
> +        self._ssh_channel.set_combine_stderr(True)  # combines stdout and stderr streams
> +
>       def _start_application(self, get_privileged_command: Callable[[str], str] | None) -> None:
>           """Starts a new interactive application based on the path to the app.
>   
Hilariously I've made the same exact change in my testpmd params patch 
series!
> diff --git a/dts/tests/TestSuite_pmd_buffer_scatter.py b/dts/tests/TestSuite_pmd_buffer_scatter.py
> index 3701c47408..41f6090a7e 100644
> --- a/dts/tests/TestSuite_pmd_buffer_scatter.py
> +++ b/dts/tests/TestSuite_pmd_buffer_scatter.py
> @@ -112,17 +112,21 @@ def pmd_scatter(self, mbsize: int) -> None:
>               ),
>               privileged=True,
>           )
> -        testpmd.set_forward_mode(TestPmdForwardingModes.mac)
> -        testpmd.start()
> -
> -        for offset in [-1, 0, 1, 4, 5]:
> -            recv_payload = self.scatter_pktgen_send_packet(mbsize + offset)
> -            self._logger.debug(f"Payload of scattered packet after forwarding: \n{recv_payload}")
> -            self.verify(
> -                ("58 " * 8).strip() in recv_payload,
> -                f"Payload of scattered packet did not match expected payload with offset {offset}.",
> -            )
> -        testpmd.stop()
> +        with testpmd_shell as testpmd:
> +            testpmd.set_forward_mode(TestPmdForwardingModes.mac)
> +            testpmd.start()
> +
> +            for offset in [-1, 0, 1, 4, 5]:
> +                recv_payload = self.scatter_pktgen_send_packet(mbsize + offset)
> +                self._logger.debug(
> +                    f"Payload of scattered packet after forwarding: \n{recv_payload}"
> +                )
> +                self.verify(
> +                    ("58 " * 8).strip() in recv_payload,
> +                    "Payload of scattered packet did not match expected payload with offset "
> +                    f"{offset}.",
> +                )
> +            testpmd.stop()

Since we are exiting the context, implicitly it means we want to stop 
and close. Can't we also implicit call stop when closing?