From 6c13566210a06fdced2d397644a14f045a6121de Mon Sep 17 00:00:00 2001 From: Artem IG Date: Sun, 20 Mar 2022 06:17:47 +0300 Subject: [PATCH 1/3] upd --- tests/test_bash_runner.py | 122 ++++++++++++++++++++------------------ vien/_bash_runner.py | 68 +++++++++++---------- vien/_main.py | 107 +++++++++++---------------------- 3 files changed, 132 insertions(+), 165 deletions(-) diff --git a/tests/test_bash_runner.py b/tests/test_bash_runner.py index f59da54..20226cd 100755 --- a/tests/test_bash_runner.py +++ b/tests/test_bash_runner.py @@ -1,59 +1,63 @@ -# SPDX-FileCopyrightText: (c) 2021 Artëm IG -# SPDX-License-Identifier: BSD-3-Clause - - -import unittest -from pathlib import Path -from tempfile import TemporaryDirectory -from timeit import default_timer as timer - -from tests.time_limited import TimeLimited -from vien._bash_runner import * - - -# @unittest.skipUnless(is_posix, "not POSIX") -@unittest.skip("temp") -class TestRunAsBash(unittest.TestCase): - - # python3 -m unittest svet.bash_runner_test - - def test_good_command_code_zero(self): - bash_lines = [ - f'set -e', - f"ls"] - code = run_as_bash_script("\n".join(bash_lines), capture_output=True) - self.assertEqual(code.returncode, 0) # ok - - def test_bad_command_error_code(self): - bash_lines = [ - f'set -e', - f"ok_computer_make_me_happy"] - code = run_as_bash_script("\n".join(bash_lines), capture_output=True) - self.assertEqual(code.returncode, 127) # error - - def test_alias_expansion(self): - with TemporaryDirectory() as td: - file_to_create = Path(td) / "to_be_or_not_to_be.txt" - file_to_create_quoted = repr(str(file_to_create)) - bash_lines = [ - f'set -e', - f"shopt -s expand_aliases", - f'alias ohoho="echo"', # this will work in bash, but not in sh - f'ohoho "that is the answer" > {file_to_create_quoted}'] - self.assertFalse(file_to_create.exists()) - code = run_as_bash_script("\n".join(bash_lines), - capture_output=True) - self.assertEqual(code.returncode, 0) - self.assertTrue(file_to_create.exists()) - self.assertEqual(file_to_create.read_text().strip(), - "that is the answer") - - def test_input_delay(self): - start = timer() - # run interactive shell end type "exit" after small delay - with TimeLimited(seconds=10): # safety net - run_as_bash_script("exec bash", input="exit\n".encode(), - input_delay=1, timeout=10, capture_output=True) - end = timer() - self.assertGreater(end - start, 0.9) - self.assertLess(end - start, 5) +# # SPDX-FileCopyrightText: (c) 2021 Artëm IG +# # SPDX-License-Identifier: BSD-3-Clause +# +# +# import unittest +# from pathlib import Path +# from tempfile import TemporaryDirectory +# from timeit import default_timer as timer +# +# from tests.time_limited import TimeLimited +# from vien._bash_runner import * +# +# +# # @unittest.skipUnless(is_posix, "not POSIX") +# @unittest.skip("temp") +# class TestRunAsBash(unittest.TestCase): +# +# # с 2022-03-20 эти тесты не проходят. Но они и не очень нужны: +# # назначение функции run_as_bash_script изменилось. Важнее тесты фукнций, +# # которые ее используют +# +# # python3 -m unittest svet.bash_runner_test +# +# def test_good_command_code_zero(self): +# bash_lines = [ +# f'set -e', +# f"ls"] +# code = run_as_bash_script("\n".join(bash_lines), capture_output=True) +# self.assertEqual(code.returncode, 0) # ok +# +# def test_bad_command_error_code(self): +# bash_lines = [ +# f'set -e', +# f"ok_computer_make_me_happy"] +# code = run_as_bash_script("\n".join(bash_lines), capture_output=True) +# self.assertEqual(code.returncode, 127) # error +# +# def test_alias_expansion(self): +# with TemporaryDirectory() as td: +# file_to_create = Path(td) / "to_be_or_not_to_be.txt" +# file_to_create_quoted = repr(str(file_to_create)) +# bash_lines = [ +# f'set -e', +# f"shopt -s expand_aliases", +# f'alias ohoho="echo"', # this will work in bash, but not in sh +# f'ohoho "that is the answer" > {file_to_create_quoted}'] +# self.assertFalse(file_to_create.exists()) +# code = run_as_bash_script("\n".join(bash_lines), +# capture_output=True) +# self.assertEqual(code.returncode, 0) +# self.assertTrue(file_to_create.exists()) +# self.assertEqual(file_to_create.read_text().strip(), +# "that is the answer") +# +# def test_input_delay(self): +# start = timer() +# # run interactive shell end type "exit" after small delay +# with TimeLimited(seconds=10): # safety net +# run_as_bash_script("exec bash", input="exit\n".encode(), +# input_delay=1, timeout=10, capture_output=True) +# end = timer() +# self.assertGreater(end - start, 0.9) +# self.assertLess(end - start, 5) diff --git a/vien/_bash_runner.py b/vien/_bash_runner.py index 540abd5..67ae883 100755 --- a/vien/_bash_runner.py +++ b/vien/_bash_runner.py @@ -1,42 +1,15 @@ # SPDX-FileCopyrightText: (c) 2021-2022 Artëm IG # 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, @@ -113,3 +86,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 tdf: + init_commands = [f"source {ubuntu_bashrc_path}"] + init_commands + + # creating temporary init script (like bash.rc) + temp_bash_rc = Path(tdf.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) diff --git a/vien/_main.py b/vien/_main.py index fe04abc..8cf5c17 100755 --- a/vien/_main.py +++ b/vien/_main.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: From 30fa166ede38a2000c061c4ef75ebbb3837a3681 Mon Sep 17 00:00:00 2001 From: Artem IG Date: Sun, 20 Mar 2022 06:35:50 +0300 Subject: [PATCH 2/3] upd --- tests/test_bash_runner.py | 63 --------------------------------------- vien/_bash_runner.py | 5 ++-- 2 files changed, 2 insertions(+), 66 deletions(-) delete mode 100755 tests/test_bash_runner.py diff --git a/tests/test_bash_runner.py b/tests/test_bash_runner.py deleted file mode 100755 index 20226cd..0000000 --- a/tests/test_bash_runner.py +++ /dev/null @@ -1,63 +0,0 @@ -# # SPDX-FileCopyrightText: (c) 2021 Artëm IG -# # SPDX-License-Identifier: BSD-3-Clause -# -# -# import unittest -# from pathlib import Path -# from tempfile import TemporaryDirectory -# from timeit import default_timer as timer -# -# from tests.time_limited import TimeLimited -# from vien._bash_runner import * -# -# -# # @unittest.skipUnless(is_posix, "not POSIX") -# @unittest.skip("temp") -# class TestRunAsBash(unittest.TestCase): -# -# # с 2022-03-20 эти тесты не проходят. Но они и не очень нужны: -# # назначение функции run_as_bash_script изменилось. Важнее тесты фукнций, -# # которые ее используют -# -# # python3 -m unittest svet.bash_runner_test -# -# def test_good_command_code_zero(self): -# bash_lines = [ -# f'set -e', -# f"ls"] -# code = run_as_bash_script("\n".join(bash_lines), capture_output=True) -# self.assertEqual(code.returncode, 0) # ok -# -# def test_bad_command_error_code(self): -# bash_lines = [ -# f'set -e', -# f"ok_computer_make_me_happy"] -# code = run_as_bash_script("\n".join(bash_lines), capture_output=True) -# self.assertEqual(code.returncode, 127) # error -# -# def test_alias_expansion(self): -# with TemporaryDirectory() as td: -# file_to_create = Path(td) / "to_be_or_not_to_be.txt" -# file_to_create_quoted = repr(str(file_to_create)) -# bash_lines = [ -# f'set -e', -# f"shopt -s expand_aliases", -# f'alias ohoho="echo"', # this will work in bash, but not in sh -# f'ohoho "that is the answer" > {file_to_create_quoted}'] -# self.assertFalse(file_to_create.exists()) -# code = run_as_bash_script("\n".join(bash_lines), -# capture_output=True) -# self.assertEqual(code.returncode, 0) -# self.assertTrue(file_to_create.exists()) -# self.assertEqual(file_to_create.read_text().strip(), -# "that is the answer") -# -# def test_input_delay(self): -# start = timer() -# # run interactive shell end type "exit" after small delay -# with TimeLimited(seconds=10): # safety net -# run_as_bash_script("exec bash", input="exit\n".encode(), -# input_delay=1, timeout=10, capture_output=True) -# end = timer() -# self.assertGreater(end - start, 0.9) -# self.assertLess(end - start, 5) diff --git a/vien/_bash_runner.py b/vien/_bash_runner.py index 67ae883..643fa44 100755 --- a/vien/_bash_runner.py +++ b/vien/_bash_runner.py @@ -14,7 +14,6 @@ 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, @@ -96,11 +95,11 @@ def start_bash_shell(init_commands: List[str], if ubuntu_bashrc_path.exists(): # Ubuntu - with NamedTemporaryFile('r', suffix=".rc") as tdf: + 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(tdf.name) + temp_bash_rc = Path(ntf.name) temp_bash_rc.write_text('\n'.join(init_commands)) return _run_with_input_delay( From 56137b8522a5b56a0207380f5d54167f9dcb3e99 Mon Sep 17 00:00:00 2001 From: Artem IG Date: Mon, 21 Mar 2022 05:50:38 +0300 Subject: [PATCH 3/3] publish --- vien/_constants.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vien/_constants.py b/vien/_constants.py index 99d4317..4d7ae15 100755 --- a/vien/_constants.py +++ b/vien/_constants.py @@ -1,3 +1,3 @@ -__version__ = "8.1.3" +__version__ = "8.1.4" __copyright__ = "(c) 2020-2022 Artem IG " __license__ = "BSD-3-Clause"