Merge 56137b8 into master

Author: github-actions[bot]

tests/test_bash_runner.py

@@ -1,47 +1,19 @@
 # SPDX-FileCopyrightText: (c) 2021-2022 Artëm IG <github.com/rtmigo>
 # SPDX-License-Identifier: BSD-3-Clause
+from __future__ import annotations
 
-import subprocess
+import os
 import time
+from pathlib import Path
 from subprocess import Popen, TimeoutExpired, CalledProcessError, \
     CompletedProcess, PIPE
-from typing import Optional, Union, List
+from tempfile import NamedTemporaryFile
+from typing import Optional, List, Dict
 
-from vien._common import need_posix
-
-
-def run_as_bash_script(args: Union[str, List[str]],
-                       #                       commands_before: List[str],
-                       timeout: float = None,
-                       input_delay: float = None,
-                       capture_output: bool = False,
-                       input: bytes = None,
-                       executable: Optional[str] = None,
-                       **kwargs) -> subprocess.CompletedProcess:
-    """Runs the provided string as a .sh script."""
-
-    need_posix()
-
-    # print("Running", script)
-
-    # we need executable='/bin/bash' for Ubuntu 18.04, it will run '/bin/sh'
-    # otherwise. For MacOS 10.13 it seems to be optional
-    return _run_with_input_delay(args,
-                                 # shell=True,
-                                 executable=executable,
-                                 timeout=timeout,
-                                 input=input,
-                                 capture_output=capture_output,
-                                 input_delay=input_delay,
-                                 **kwargs)
-
-
-#subprocess.run()
 
 def _run_with_input_delay(*popenargs,
                           input_delay: float = None,
                           input: Optional[bytes] = None,
-                          # stdin: Optional[bytes] = None,
                           timeout: float = None,
                           check: bool = False,
                           capture_output: bool = False,
@@ -113,3 +85,34 @@ def _run_with_input_delay(*popenargs,
                                      output=stdout, stderr=stderr)
 
     return CompletedProcess(process.args, exit_code, stdout, stderr)
+
+
+def start_bash_shell(init_commands: List[str],
+                     input: Optional[str] = None,
+                     input_delay: Optional[float] = None,
+                     env: Optional[Dict] = None) -> CompletedProcess:
+    ubuntu_bashrc_path = Path(os.path.expanduser("~/.bashrc")).absolute()
+
+    if ubuntu_bashrc_path.exists():
+        # Ubuntu
+        with NamedTemporaryFile('r', suffix=".rc") as ntf:
+            init_commands = [f"source {ubuntu_bashrc_path}"] + init_commands
+
+            # creating temporary init script (like bash.rc)
+            temp_bash_rc = Path(ntf.name)
+            temp_bash_rc.write_text('\n'.join(init_commands))
+
+            return _run_with_input_delay(
+                ["/bin/bash", "--rcfile", str(temp_bash_rc), "-i"],
+                executable=None,
+                input=input.encode() if input else None,
+                input_delay=input_delay,
+                env=env)
+    else:
+        # MacOS
+        return _run_with_input_delay(
+            "\n".join(init_commands),
+            executable="/bin/bash",
+            input=input.encode() if input else None,
+            input_delay=input_delay,
+            env=env)

vien/_bash_runner.py

@@ -1,3 +1,3 @@
-__version__ = "8.1.3"
+__version__ = "8.1.4"
 __copyright__ = "(c) 2020-2022 Artem IG <github.com/rtmigo>"
 __license__ = "BSD-3-Clause"

vien/_constants.py

@@ -9,11 +9,10 @@
 import subprocess
 import sys
 from pathlib import Path
-from tempfile import TemporaryDirectory
 from typing import *
 
 from vien import is_posix
-from vien._bash_runner import run_as_bash_script
+from vien._bash_runner import start_bash_shell
 from vien._call_funcs import relative_fn_to_module_name, relative_inner_path
 from vien._cmdexe_escape_args import cmd_escape_arg
 from vien._colors import Colors
@@ -246,85 +245,45 @@ def _quoted(txt: str) -> str:
     return shlex.quote(txt)
 
 
-class OptionalTempDir:
-    def __init__(self):
-        self._temp_dir: Optional[TemporaryDirectory] = None
-
-    @property
-    def path(self) -> Path:
-        if self._temp_dir is None:
-            self._temp_dir = TemporaryDirectory()
-        return Path(self._temp_dir.name)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self._temp_dir is not None:
-            self._temp_dir.cleanup()
-            self._temp_dir = None
-
-
 def main_shell(dirs: Dirs, input: Optional[str], input_delay: Optional[float]):
     dirs.venv_must_exist()
 
-    with OptionalTempDir() as opt_temp_dir:
-        activate_path = dirs.venv_dir / "bin" / "activate"
-        old_ps1 = os.environ.get("PS1") or guess_bash_ps1()
-
-        if not old_ps1:
-            old_ps1 = r"\h:\W \u\$"  # default from MacOS
-
-        color_start = Colors.YELLOW
-        color_end = Colors.NOCOLOR
+    # with OptionalTempDir() as opt_temp_dir:
+    activate_path = dirs.venv_dir / "bin" / "activate"
+    old_ps1 = os.environ.get("PS1") or guess_bash_ps1()
 
-        venv_name = dirs.project_dir.name
-        new_ps1 = f"{color_start}({venv_name}){color_end}:{old_ps1} "
+    if not old_ps1:
+        old_ps1 = r"\h:\W \u\$"  # default from MacOS
 
-        bashrc_file = Path(os.path.expanduser("~/.bashrc")).absolute()
+    color_start = Colors.YELLOW
+    color_end = Colors.NOCOLOR
 
-        executable: Optional[str] = None
-        args: Union[str, List[str]]
+    venv_name = dirs.project_dir.name
+    new_ps1 = f"{color_start}({venv_name}){color_end}:{old_ps1} "
 
-        if bashrc_file.exists():
-            # Ubuntu
-            temp_bash_rc = opt_temp_dir.path / "bash.rc"
-            temp_bash_rc.write_bytes(
-                b'\n'.join([
-                    f"source {bashrc_file}".encode(),
-                    f"source {activate_path}".encode(),
-                    f'PS1={_quoted(new_ps1)}'.encode()]))
-            args = ["/bin/bash", "--rcfile", str(temp_bash_rc), "-i"]
-
-        else:
-            # MacOS
-            executable = "/bin/bash"
-            args = "\n".join([
-                f'source {shlex.quote(str(activate_path))}',
-                f"PS1={_quoted(new_ps1)}"])
-
-        # we will use [input] for testing: we will send a command to the stdin
-        # of the interactive sub-shell and later check whether the command was
-        # executed.
-        #
-        # We will also provide [input_delay] parameter. This allows the check
-        # whether
-        # the sub-shell was really interactive: did it wait for the input
-        #
-        # Surprisingly, the sub-shell will immediately close after executing
-        # the command. It seems it closes immediately after the subprocess.
-        # Popen closes the stdin. So it will not wait for "exit". But it serves
-        # the task well
-
-        cp = run_as_bash_script(
-            args,
-            executable=executable,
-            input=input.encode() if input else None,
-            input_delay=input_delay,
-            env=child_env(dirs.project_dir))
-
-        # the vien will return the same exit code as the shell returned
-        raise ChildExit(cp.returncode)
+    # we use [input] for testing: we send a command to the stdin of the
+    # interactive sub-shell and later check whether the command was
+    # executed.
+    #
+    # We will also provide [input_delay] parameter. This allows the check
+    # whether the sub-shell was really interactive: did it wait for
+    # the input
+    #
+    # Surprisingly, the sub-shell will immediately close after executing
+    # the command. It seems it closes immediately after the subprocess.
+    # Popen closes the stdin. So it will not wait for "exit". But it serves
+    # the task well
+
+    cp = start_bash_shell(init_commands=[
+        f'source {shlex.quote(str(activate_path))}',
+        f"PS1={_quoted(new_ps1)}"],
+        input=input,
+        input_delay=input_delay,
+        env=child_env(dirs.project_dir)
+    )
+
+    # the vien will return the same exit code as the shell returned
+    raise ChildExit(cp.returncode)
 
 
 def bash_args_to_str(args: List[str]) -> str: