Skip to content

Commit 6c13566

Browse files
rtmigortmigo
committed
upd
1 parent 5b881ba commit 6c13566

File tree

3 files changed

+132
-165
lines changed

3 files changed

+132
-165
lines changed

‎tests/test_bash_runner.py

Lines changed: 63 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1,59 +1,63 @@
1-
# SPDX-FileCopyrightText: (c) 2021 Artëm IG <github.com/rtmigo>
2-
# SPDX-License-Identifier: BSD-3-Clause
3-
4-
5-
import unittest
6-
from pathlib import Path
7-
from tempfile import TemporaryDirectory
8-
from timeit import default_timer as timer
9-
10-
from tests.time_limited import TimeLimited
11-
from vien._bash_runner import *
12-
13-
14-
# @unittest.skipUnless(is_posix, "not POSIX")
15-
@unittest.skip("temp")
16-
class TestRunAsBash(unittest.TestCase):
17-
18-
# python3 -m unittest svet.bash_runner_test
19-
20-
def test_good_command_code_zero(self):
21-
bash_lines = [
22-
f'set -e',
23-
f"ls"]
24-
code = run_as_bash_script("\n".join(bash_lines), capture_output=True)
25-
self.assertEqual(code.returncode, 0) # ok
26-
27-
def test_bad_command_error_code(self):
28-
bash_lines = [
29-
f'set -e',
30-
f"ok_computer_make_me_happy"]
31-
code = run_as_bash_script("\n".join(bash_lines), capture_output=True)
32-
self.assertEqual(code.returncode, 127) # error
33-
34-
def test_alias_expansion(self):
35-
with TemporaryDirectory() as td:
36-
file_to_create = Path(td) / "to_be_or_not_to_be.txt"
37-
file_to_create_quoted = repr(str(file_to_create))
38-
bash_lines = [
39-
f'set -e',
40-
f"shopt -s expand_aliases",
41-
f'alias ohoho="echo"', # this will work in bash, but not in sh
42-
f'ohoho "that is the answer" > {file_to_create_quoted}']
43-
self.assertFalse(file_to_create.exists())
44-
code = run_as_bash_script("\n".join(bash_lines),
45-
capture_output=True)
46-
self.assertEqual(code.returncode, 0)
47-
self.assertTrue(file_to_create.exists())
48-
self.assertEqual(file_to_create.read_text().strip(),
49-
"that is the answer")
50-
51-
def test_input_delay(self):
52-
start = timer()
53-
# run interactive shell end type "exit" after small delay
54-
with TimeLimited(seconds=10): # safety net
55-
run_as_bash_script("exec bash", input="exit\n".encode(),
56-
input_delay=1, timeout=10, capture_output=True)
57-
end = timer()
58-
self.assertGreater(end - start, 0.9)
59-
self.assertLess(end - start, 5)
1+
# # SPDX-FileCopyrightText: (c) 2021 Artëm IG <github.com/rtmigo>
2+
# # SPDX-License-Identifier: BSD-3-Clause
3+
#
4+
#
5+
# import unittest
6+
# from pathlib import Path
7+
# from tempfile import TemporaryDirectory
8+
# from timeit import default_timer as timer
9+
#
10+
# from tests.time_limited import TimeLimited
11+
# from vien._bash_runner import *
12+
#
13+
#
14+
# # @unittest.skipUnless(is_posix, "not POSIX")
15+
# @unittest.skip("temp")
16+
# class TestRunAsBash(unittest.TestCase):
17+
#
18+
# # с 2022-03-20 эти тесты не проходят. Но они и не очень нужны:
19+
# # назначение функции run_as_bash_script изменилось. Важнее тесты фукнций,
20+
# # которые ее используют
21+
#
22+
# # python3 -m unittest svet.bash_runner_test
23+
#
24+
# def test_good_command_code_zero(self):
25+
# bash_lines = [
26+
# f'set -e',
27+
# f"ls"]
28+
# code = run_as_bash_script("\n".join(bash_lines), capture_output=True)
29+
# self.assertEqual(code.returncode, 0) # ok
30+
#
31+
# def test_bad_command_error_code(self):
32+
# bash_lines = [
33+
# f'set -e',
34+
# f"ok_computer_make_me_happy"]
35+
# code = run_as_bash_script("\n".join(bash_lines), capture_output=True)
36+
# self.assertEqual(code.returncode, 127) # error
37+
#
38+
# def test_alias_expansion(self):
39+
# with TemporaryDirectory() as td:
40+
# file_to_create = Path(td) / "to_be_or_not_to_be.txt"
41+
# file_to_create_quoted = repr(str(file_to_create))
42+
# bash_lines = [
43+
# f'set -e',
44+
# f"shopt -s expand_aliases",
45+
# f'alias ohoho="echo"', # this will work in bash, but not in sh
46+
# f'ohoho "that is the answer" > {file_to_create_quoted}']
47+
# self.assertFalse(file_to_create.exists())
48+
# code = run_as_bash_script("\n".join(bash_lines),
49+
# capture_output=True)
50+
# self.assertEqual(code.returncode, 0)
51+
# self.assertTrue(file_to_create.exists())
52+
# self.assertEqual(file_to_create.read_text().strip(),
53+
# "that is the answer")
54+
#
55+
# def test_input_delay(self):
56+
# start = timer()
57+
# # run interactive shell end type "exit" after small delay
58+
# with TimeLimited(seconds=10): # safety net
59+
# run_as_bash_script("exec bash", input="exit\n".encode(),
60+
# input_delay=1, timeout=10, capture_output=True)
61+
# end = timer()
62+
# self.assertGreater(end - start, 0.9)
63+
# self.assertLess(end - start, 5)

‎vien/_bash_runner.py

Lines changed: 36 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,42 +1,15 @@
11
# SPDX-FileCopyrightText: (c) 2021-2022 Artëm IG <github.com/rtmigo>
22
# SPDX-License-Identifier: BSD-3-Clause
3+
from __future__ import annotations
34

4-
import subprocess
5+
import os
56
import time
7+
from pathlib import Path
68
from subprocess import Popen, TimeoutExpired, CalledProcessError, \
79
CompletedProcess, PIPE
8-
from typing import Optional, Union, List
10+
from tempfile import NamedTemporaryFile
11+
from typing import Optional, List, Dict
912

10-
from vien._common import need_posix
11-
12-
13-
def run_as_bash_script(args: Union[str, List[str]],
14-
# commands_before: List[str],
15-
timeout: float = None,
16-
input_delay: float = None,
17-
capture_output: bool = False,
18-
input: bytes = None,
19-
executable: Optional[str] = None,
20-
**kwargs) -> subprocess.CompletedProcess:
21-
"""Runs the provided string as a .sh script."""
22-
23-
need_posix()
24-
25-
# print("Running", script)
26-
27-
# we need executable='/bin/bash' for Ubuntu 18.04, it will run '/bin/sh'
28-
# otherwise. For MacOS 10.13 it seems to be optional
29-
return _run_with_input_delay(args,
30-
# shell=True,
31-
executable=executable,
32-
timeout=timeout,
33-
input=input,
34-
capture_output=capture_output,
35-
input_delay=input_delay,
36-
**kwargs)
37-
38-
39-
#subprocess.run()
4013

4114
def _run_with_input_delay(*popenargs,
4215
input_delay: float = None,
@@ -113,3 +86,34 @@ def _run_with_input_delay(*popenargs,
11386
output=stdout, stderr=stderr)
11487

11588
return CompletedProcess(process.args, exit_code, stdout, stderr)
89+
90+
91+
def start_bash_shell(init_commands: List[str],
92+
input: Optional[str] = None,
93+
input_delay: Optional[float] = None,
94+
env: Optional[Dict] = None) -> CompletedProcess:
95+
ubuntu_bashrc_path = Path(os.path.expanduser("~/.bashrc")).absolute()
96+
97+
if ubuntu_bashrc_path.exists():
98+
# Ubuntu
99+
with NamedTemporaryFile('r', suffix=".rc") as tdf:
100+
init_commands = [f"source {ubuntu_bashrc_path}"] + init_commands
101+
102+
# creating temporary init script (like bash.rc)
103+
temp_bash_rc = Path(tdf.name)
104+
temp_bash_rc.write_text('\n'.join(init_commands))
105+
106+
return _run_with_input_delay(
107+
["/bin/bash", "--rcfile", str(temp_bash_rc), "-i"],
108+
executable=None,
109+
input=input.encode() if input else None,
110+
input_delay=input_delay,
111+
env=env)
112+
else:
113+
# MacOS
114+
return _run_with_input_delay(
115+
"\n".join(init_commands),
116+
executable="/bin/bash",
117+
input=input.encode() if input else None,
118+
input_delay=input_delay,
119+
env=env)

‎vien/_main.py

Lines changed: 33 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,10 @@
99
import subprocess
1010
import sys
1111
from pathlib import Path
12-
from tempfile import TemporaryDirectory
1312
from typing import *
1413

1514
from vien import is_posix
16-
from vien._bash_runner import run_as_bash_script
15+
from vien._bash_runner import start_bash_shell
1716
from vien._call_funcs import relative_fn_to_module_name, relative_inner_path
1817
from vien._cmdexe_escape_args import cmd_escape_arg
1918
from vien._colors import Colors
@@ -246,85 +245,45 @@ def _quoted(txt: str) -> str:
246245
return shlex.quote(txt)
247246

248247

249-
class OptionalTempDir:
250-
def __init__(self):
251-
self._temp_dir: Optional[TemporaryDirectory] = None
252-
253-
@property
254-
def path(self) -> Path:
255-
if self._temp_dir is None:
256-
self._temp_dir = TemporaryDirectory()
257-
return Path(self._temp_dir.name)
258-
259-
def __enter__(self):
260-
return self
261-
262-
def __exit__(self, exc_type, exc_val, exc_tb):
263-
if self._temp_dir is not None:
264-
self._temp_dir.cleanup()
265-
self._temp_dir = None
266-
267-
268248
def main_shell(dirs: Dirs, input: Optional[str], input_delay: Optional[float]):
269249
dirs.venv_must_exist()
270250

271-
with OptionalTempDir() as opt_temp_dir:
272-
activate_path = dirs.venv_dir / "bin" / "activate"
273-
old_ps1 = os.environ.get("PS1") or guess_bash_ps1()
274-
275-
if not old_ps1:
276-
old_ps1 = r"\h:\W \u\$" # default from MacOS
277-
278-
color_start = Colors.YELLOW
279-
color_end = Colors.NOCOLOR
251+
# with OptionalTempDir() as opt_temp_dir:
252+
activate_path = dirs.venv_dir / "bin" / "activate"
253+
old_ps1 = os.environ.get("PS1") or guess_bash_ps1()
280254

281-
venv_name = dirs.project_dir.name
282-
new_ps1 = f"{color_start}({venv_name}){color_end}:{old_ps1} "
255+
if not old_ps1:
256+
old_ps1 = r"\h:\W \u\$" # default from MacOS
283257

284-
bashrc_file = Path(os.path.expanduser("~/.bashrc")).absolute()
258+
color_start = Colors.YELLOW
259+
color_end = Colors.NOCOLOR
285260

286-
executable: Optional[str] = None
287-
args: Union[str, List[str]]
261+
venv_name = dirs.project_dir.name
262+
new_ps1 = f"{color_start}({venv_name}){color_end}:{old_ps1} "
288263

289-
if bashrc_file.exists():
290-
# Ubuntu
291-
temp_bash_rc = opt_temp_dir.path / "bash.rc"
292-
temp_bash_rc.write_bytes(
293-
b'\n'.join([
294-
f"source {bashrc_file}".encode(),
295-
f"source {activate_path}".encode(),
296-
f'PS1={_quoted(new_ps1)}'.encode()]))
297-
args = ["/bin/bash", "--rcfile", str(temp_bash_rc), "-i"]
298-
299-
else:
300-
# MacOS
301-
executable = "/bin/bash"
302-
args = "\n".join([
303-
f'source {shlex.quote(str(activate_path))}',
304-
f"PS1={_quoted(new_ps1)}"])
305-
306-
# we will use [input] for testing: we will send a command to the stdin
307-
# of the interactive sub-shell and later check whether the command was
308-
# executed.
309-
#
310-
# We will also provide [input_delay] parameter. This allows the check
311-
# whether
312-
# the sub-shell was really interactive: did it wait for the input
313-
#
314-
# Surprisingly, the sub-shell will immediately close after executing
315-
# the command. It seems it closes immediately after the subprocess.
316-
# Popen closes the stdin. So it will not wait for "exit". But it serves
317-
# the task well
318-
319-
cp = run_as_bash_script(
320-
args,
321-
executable=executable,
322-
input=input.encode() if input else None,
323-
input_delay=input_delay,
324-
env=child_env(dirs.project_dir))
325-
326-
# the vien will return the same exit code as the shell returned
327-
raise ChildExit(cp.returncode)
264+
# we use [input] for testing: we send a command to the stdin of the
265+
# interactive sub-shell and later check whether the command was
266+
# executed.
267+
#
268+
# We will also provide [input_delay] parameter. This allows the check
269+
# whether the sub-shell was really interactive: did it wait for
270+
# the input
271+
#
272+
# Surprisingly, the sub-shell will immediately close after executing
273+
# the command. It seems it closes immediately after the subprocess.
274+
# Popen closes the stdin. So it will not wait for "exit". But it serves
275+
# the task well
276+
277+
cp = start_bash_shell(init_commands=[
278+
f'source {shlex.quote(str(activate_path))}',
279+
f"PS1={_quoted(new_ps1)}"],
280+
input=input,
281+
input_delay=input_delay,
282+
env=child_env(dirs.project_dir)
283+
)
284+
285+
# the vien will return the same exit code as the shell returned
286+
raise ChildExit(cp.returncode)
328287

329288

330289
def bash_args_to_str(args: List[str]) -> str:

0 commit comments

Comments
 (0)