[CLI] reverse to x11 rocker extension, use logger instead of print

rtwcli/rtw_cmds/rtw_cmds/commands.py

@@ -12,21 +12,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import argparse
 from rtwcli.command import add_subparsers_on_demand
 from rtwcli.command import CommandExtension
 
 
 class BaseCommand(CommandExtension):
-    def __init__(self, verbs_group):
+    def __init__(self, verbs_group: str):
         super().__init__()
         self.verbs_group = verbs_group
 
-    def add_arguments(self, parser, cli_name):
+    def add_arguments(self, parser: argparse.ArgumentParser, cli_name: str):
         self._subparser = parser
         # add arguments and sub-commands of verbs
         add_subparsers_on_demand(parser, cli_name, "_verb", self.verbs_group, required=False)
 
-    def main(self, *, parser, args):
+    def main(self, *, parser: argparse.ArgumentParser, args: argparse.Namespace):
         if not hasattr(args, "_verb"):
             # in case no verb was passed
             self._subparser.print_help()

rtwcli/rtw_cmds/rtw_cmds/docker/verbs.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from rtwcli import logger
 from rtwcli.docker_utils import (
     docker_exec_interactive_bash,
     docker_start,
@@ -28,24 +29,30 @@ class EnterVerb(VerbExtension):
     def main(self, *, args):
         ws = get_current_workspace()
         if not ws:
-            print("No workspace is active.")
+            logger.info("No workspace is active.")
             return
 
         if not ws.ws_docker_support:
-            print("The workspace does not support docker.")
+            logger.info("The workspace does not support docker.")
+            return
+
+        if not ws.docker_container_name:
+            logger.error("The workspace is missing the docker_container_name attribute.")
             return
 
         if not is_docker_container_running(ws.docker_container_name):
-            print(
+            logger.info(
                 f"The docker container '{ws.docker_container_name}' is not running, starting it now."
             )
             # fix missing .xauth file if it is not present
             if not fix_missing_xauth_file(ws.docker_container_name):
-                print(f"Failed to fix missing .xauth file for '{ws.docker_container_name}'.")
+                logger.error(
+                    f"Failed to fix missing .xauth file for '{ws.docker_container_name}'."
+                )
                 return
 
             if not docker_start(ws.docker_container_name):
-                print(f"Failed to start docker container '{ws.docker_container_name}'.")
+                logger.error(f"Failed to start docker container '{ws.docker_container_name}'.")
                 return
 
         docker_exec_interactive_bash(ws.docker_container_name)

rtwcli/rtw_cmds/rtw_cmds/workspace/create_verb.py

@@ -17,11 +17,12 @@
 from dataclasses import dataclass, field, fields
 import os
 import pathlib
-from pprint import pprint
 import shutil
 import textwrap
 from typing import Any, List
 import questionary
+import rich
+from rtwcli import logger
 from rtwcli.constants import (
     BASHRC_PATH,
     DISPLAY_MANAGER_WAYLAND,
@@ -221,7 +222,7 @@ def handle_upstream_ws_repos(self):
             )
 
         self.has_upstream_ws = True
-        print(
+        logger.info(
             f"Imported upstream ws repos from '{self.upstream_ws_repos_file_abs_path}' "
             f"to '{self.upstream_ws_src_abs_path}'"
         )
@@ -235,7 +236,7 @@ def handle_upstream_ws_repos(self):
                 "and create only the main workspace?"
             ).ask()
             if remove_upstream_ws:
-                print(f"Removing upstream workspace folder '{self.upstream_ws_abs_path}'")
+                logger.info(f"Removing upstream workspace folder '{self.upstream_ws_abs_path}'")
                 shutil.rmtree(self.upstream_ws_abs_path)
                 self.has_upstream_ws = False
 
@@ -257,31 +258,31 @@ def handle_repos(self):
                 f"branch='{self.repos_branch}', path='{self.repos_clone_abs_path}'"
             )
 
-        print(
+        logger.info(
             f"Successfully cloned repo='{self.repos_containing_repository_url}', "
             f"branch='{self.repos_branch}', path='{self.repos_clone_abs_path}'"
         )
 
         # main ws repos
         if os.path.exists(self.ws_repos_file_abs_path):
-            print(f"Found ws repos file '{self.ws_repos_file_abs_path}'")
+            logger.info(f"Found ws repos file '{self.ws_repos_file_abs_path}'")
             self.handle_main_ws_repos()
-            print(
+            logger.info(
                 f"Imported repos from '{self.ws_repos_file_abs_path}' to '{self.ws_src_abs_path}'"
             )
         else:
-            print(f"Main ws repos file '{self.ws_repos_file_abs_path}' does not exist.")
+            logger.info(f"Main ws repos file '{self.ws_repos_file_abs_path}' does not exist.")
 
         # upstream ws repos
         if os.path.isfile(self.upstream_ws_repos_file_abs_path):
-            print(f"Found upstream ws repos file '{self.upstream_ws_repos_file_abs_path}'")
+            logger.info(f"Found upstream ws repos file '{self.upstream_ws_repos_file_abs_path}'")
             self.handle_upstream_ws_repos()
-            print(
+            logger.info(
                 f"Imported upstream ws repos from '{self.upstream_ws_repos_file_abs_path}' "
                 f"to '{self.upstream_ws_src_abs_path}'"
             )
         else:
-            print(
+            logger.info(
                 f"Upstream ws repos file '{self.upstream_ws_repos_file_abs_path}' does not exist."
             )
 
@@ -334,7 +335,7 @@ def __post_init__(self):
         if self.repos_containing_repository_url:
             self.handle_repos()
         else:
-            print("No repos containing repository URL provided. Not importing any repos.")
+            logger.info("No repos containing repository URL provided. Not importing any repos.")
 
         if self.user_override_name:
             self.rtw_docker_clone_abs_path = replace_user_name_in_path(
@@ -649,7 +650,7 @@ def run_intermediate_container(self, create_args: CreateVerbArgs) -> Any:
                     f"{create_args.upstream_ws_abs_path}:{create_args.upstream_ws_abs_path}"
                 )
 
-        print(f"Creating intermediate docker container with volumes: {volumes}")
+        logger.info(f"Creating intermediate docker container with volumes: {volumes}")
         try:
             docker_client = docker.from_env()
             intermediate_container = docker_client.containers.run(
@@ -728,27 +729,27 @@ def execute_ws_cmds(
         intermediate_container: Any = None,
     ) -> None:
         for ws_cmd in ws_cmds:
-            print(f"Running ws command: {ws_cmd}")
+            logger.info(f"Running ws command: {ws_cmd}")
             ws_cmd_str = " ".join(ws_cmd)
             error_msg = f"Failed to execute ws command '{ws_cmd_str}'"
             if create_args.docker:
                 if not docker_exec_bash_cmd(intermediate_container.id, ws_cmd_str):
                     if create_args.ignore_ws_cmd_error:
-                        print(error_msg)
+                        logger.warn(error_msg)
                     else:
                         docker_stop(intermediate_container.id)
                         raise RuntimeError(error_msg)
             else:
                 if not run_bash_command(ws_cmd_str):
                     if create_args.ignore_ws_cmd_error:
-                        print(error_msg)
+                        logger.warn(error_msg)
                     else:
                         raise RuntimeError(error_msg)
 
     def change_ws_folder_permissions(
         self, create_args: CreateVerbArgs, intermediate_container: Any
     ):
-        print("Changing workspace folder permissions in the intermediate container.")
+        logger.info("Changing workspace folder permissions in the intermediate container.")
         if not change_docker_path_permissions(
             intermediate_container.id, create_args.ws_abs_path_in_docker
         ):
@@ -855,7 +856,7 @@ def setup_rtw_in_intermediate_image(
         ):
             raise RuntimeError("Failed to change permissions for the home folder.")
 
-        print(f"Committing container '{intermediate_container.id}'")
+        logger.info(f"Committing container '{intermediate_container.id}'")
         try:
             intermediate_container.commit(create_args.final_image_name)
         except docker.errors.APIError as e:  # type: ignore
@@ -874,15 +875,15 @@ def main(self, *, args):
             )
 
         if get_display_manager() == DISPLAY_MANAGER_WAYLAND:
-            print(f"Wayland display manager detected: '{DISPLAY_MANAGER_WAYLAND}'.")
+            logger.info(f"Wayland display manager detected: '{DISPLAY_MANAGER_WAYLAND}'.")
 
         filtered_args = get_filtered_args(args, list(fields(CreateVerbArgs)))
         filtered_args["ws_abs_path"] = os.path.normpath(os.path.abspath(args.ws_folder))
 
         create_args = CreateVerbArgs(**filtered_args)
-        print("### CREATE ARGS ###")
-        pprint(create_args)
-        print("### CREATE ARGS ###")
+        logger.info("### CREATE ARGS ###")
+        rich.print(create_args)
+        logger.info("### CREATE ARGS ###")
 
         if create_args.docker:
             self.build_intermediate_docker_image(create_args)
@@ -892,9 +893,9 @@ def main(self, *, args):
             intermediate_container = self.run_intermediate_container(create_args)
 
         ws_cmds = self.get_ws_cmds(create_args)
-        print("### WS CMDS ###")
-        pprint(ws_cmds)
-        print("### WS CMDS ###")
+        logger.info("### WS CMDS ###")
+        rich.print(ws_cmds)
+        logger.info("### WS CMDS ###")
         self.execute_ws_cmds(create_args, ws_cmds, intermediate_container)
 
         if create_args.docker:
@@ -960,11 +961,11 @@ def main(self, *, args):
 
         # remove the local files if the standalone flag is set
         if create_args.standalone:
-            print("Standalone flag is set. Removing the local workspace files.")
+            logger.info("Standalone flag is set. Removing the local workspace files.")
             shutil.rmtree(create_args.ws_abs_path)
-            print(f"Removed the workspace folder '{create_args.ws_abs_path}'")
+            logger.info(f"Removed the workspace folder '{create_args.ws_abs_path}'")
             if create_args.has_upstream_ws:
                 shutil.rmtree(create_args.upstream_ws_abs_path)
-                print(
+                logger.info(
                     f"Removed the upstream workspace folder '{create_args.upstream_ws_abs_path}'"
                 )

rtwcli/rtw_cmds/rtw_cmds/workspace/use_verb.py

@@ -15,6 +15,7 @@
 import argparse
 import os
 import questionary
+from rtwcli import logger
 from rtwcli.constants import (
     USE_WORKSPACE_SCRIPT_PATH,
     WORKSPACES_PATH,
@@ -47,7 +48,7 @@ def add_arguments(self, parser: argparse.ArgumentParser, cli_name: str):
     def main(self, *, args):
         workspaces_config = load_workspaces_config_from_yaml_file(WORKSPACES_PATH)
         if not workspaces_config.workspaces:
-            print(f"No workspaces found in config file '{WORKSPACES_PATH}'")
+            logger.info(f"No workspaces found in config file '{WORKSPACES_PATH}'")
             return
 
         ws_names = workspaces_config.get_ws_names()
@@ -68,16 +69,18 @@ def main(self, *, args):
 
         workspace = workspaces_config.workspaces.get(ws_name)
         if not workspace:
-            return f"Workspace '{ws_name}' not found."
+            logger.error(f"Workspace '{ws_name}' not found.")
+            return
 
-        print(f"Workspace data: {workspace}")
+        logger.info(f"Workspace data: {workspace}")
 
         script_content = create_bash_script_content_for_using_ws(
             workspace, USE_WORKSPACE_SCRIPT_PATH
         )
         tmp_file = WS_USE_BASH_FILE_PATH_FORMAT.format(ppid=os.getppid())
-        print(f"Following text will be written into file '{tmp_file}':\n{script_content}")
+        logger.info(f"Following content will be written into file '{tmp_file}':\n{script_content}")
         if not create_file_and_write(tmp_file, content=script_content):
-            return f"Failed to write workspace data to a file {tmp_file}."
+            logger.error(f"Failed to write workspace data to a file {tmp_file}.")
+            return
 
-        print(f"Using workspace '{ws_name}'")
+        logger.info(f"Using workspace '{ws_name}'")

rtwcli/rtw_rocker_extensions/rtw_rocker_extensions/x11tmp.py → rtwcli/rtw_rocker_extensions/rtw_rocker_extensions/tmpfs.py

@@ -19,30 +19,25 @@
 from rocker.core import RockerExtension
 
 
-class X11Tmp(RockerExtension):
+class Tmpfs(RockerExtension):
     @staticmethod
     def get_name() -> str:
-        return "x11tmp"
+        return "tmpfs"
 
     def __init__(self):
-        self.name = X11Tmp.get_name()
+        self.name = Tmpfs.get_name()
 
     def get_docker_args(self, cliargs: dict) -> str:
-        args = [" "]  # To separate from the previous argument
-        args.extend(["-e", "DISPLAY"])
-        args.extend(["-e", "TERM"])
-        args.extend(["-e", "QT_X11_NO_MITSHM=1"])
+        args = [" "]  # To separate from the previous
         args.extend(["--tmpfs", "/tmp"])
-        args.extend(["-v", "/tmp/.X11-unix:/tmp/.X11-unix"])
-        args.extend(["-v", "/etc/localtime:/etc/localtime:ro"])
         args.append(" ")  # To separate from the next argument
         return " ".join(args)
 
     @staticmethod
     def register_arguments(parser: argparse.ArgumentParser, defaults: dict = {}) -> None:
         parser.add_argument(
-            name_to_argument(X11Tmp.get_name()),
+            name_to_argument(Tmpfs.get_name()),
             action="store_true",
-            default=defaults.get(X11Tmp.get_name(), None),
-            help="Enable x11tmp extension",
+            default=defaults.get(Tmpfs.get_name(), None),
+            help="Enable tmpfs for /tmp in the container.",
         )

rtwcli/rtw_rocker_extensions/setup.py

@@ -35,7 +35,7 @@
     tests_require=["pytest"],
     entry_points={
         "rocker.extensions": [
-            "rtw_x11tmp = rtw_rocker_extensions.x11tmp:X11Tmp",
+            "rtw_tmpfs = rtw_rocker_extensions.tmpfs:Tmpfs",
         ],
     },
 )

rtwcli/rtwcli/rtwcli/command/__init__.py

@@ -18,6 +18,7 @@
 import inspect
 import os
 import types
+from typing import Optional
 
 from rtwcli.entry_points import get_entry_points
 from rtwcli.entry_points import get_first_line_doc
@@ -47,14 +48,14 @@ def __init__(self):
         super().__init__()
         satisfies_version(PLUGIN_SYSTEM_VERSION, "^0.1")
 
-    def add_arguments(self, parser, cli_name, *, argv=None):
+    def add_arguments(self, parser: argparse.ArgumentParser, cli_name: str, *, argv=None):
         pass
 
-    def main(self, *, parser, args):
+    def main(self, *, parser: argparse.ArgumentParser, args: argparse.Namespace):
         raise NotImplementedError()
 
 
-def get_command_extensions(group_name, *, exclude_names=None):
+def get_command_extensions(group_name: str, *, exclude_names=None):
     extensions = instantiate_extensions(group_name, exclude_names=exclude_names)
     for name, extension in extensions.items():
         extension.NAME = name
@@ -75,7 +76,13 @@ def __iter__(self):
 
 
 def add_subparsers_on_demand(
-    parser, cli_name, dest, group_name, hide_extensions=None, required=True, argv=None
+    parser: argparse.ArgumentParser,
+    cli_name: str,
+    dest: str,
+    group_name: str,
+    hide_extensions: Optional[list] = None,
+    required: bool = True,
+    argv: Optional[list] = None,
 ):
     """
     Create argparse subparser for each extension on demand.

rtwcli/rtwcli/rtwcli/command/extension_points.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import argparse
 from rtwcli.command import CommandExtension
 from rtwcli.entry_points import EXTENSION_POINT_GROUP_NAME
 from rtwcli.entry_points import get_entry_points
@@ -23,7 +24,7 @@
 class ExtensionPointsCommand(CommandExtension):
     """List extension points."""
 
-    def add_arguments(self, parser, cli_name):
+    def add_arguments(self, parser: argparse.ArgumentParser, cli_name: str):
         parser.add_argument(
             "--all",
             "-a",