Fix patching when shell=True in subprocess (#434)

* Set macos deployment target

.github/workflows/release.yml

@@ -85,6 +85,7 @@ jobs:
  run: python -m cibuildwheel --output-dir wheelhouse
  env:
  CIBW_BUILD: 'cp${{ matrix.python-version }}-*'
+ CIBW_ENVIRONMENT_MACOS: MACOSX_DEPLOYMENT_TARGET=10.15
 
  - name: Publish wheels to PyPI Unix
  if: matrix.os != 'windows-latest' 

.github/workflows/wheels.yml

@@ -81,3 +81,4 @@ jobs:
  run: python -m cibuildwheel --output-dir wheelhouse
  env:
  CIBW_BUILD: 'cp${{ matrix.python-version }}-*'
+ CIBW_ENVIRONMENT_MACOS: MACOSX_DEPLOYMENT_TARGET=10.15

src/viztracer/patch.py

@@ -66,10 +66,15 @@ def build_command(args: Sequence[str]) -> list[str] | None:
  def subprocess_init(self: subprocess.Popen[Any], args: Union[str, Sequence[Any], Any], **kwargs: Any) -> None:
  new_args = args
  if isinstance(new_args, str):
- new_args = shlex.split(new_args)
+ new_args = shlex.split(new_args, posix=sys.platform != "win32")
  if isinstance(new_args, Sequence):
  if "python" in os.path.basename(new_args[0]):
  new_args = build_command(new_args)
+ if new_args is not None and kwargs.get("shell") and isinstance(args, str):
+ # For shell=True, we should convert the commands back to string
+ # if it was passed as string
+ # This is mostly for Unix shell
+ new_args = " ".join(new_args)
  else:
  new_args = None
 

tests/test_multiprocess.py

@@ -51,6 +51,15 @@ def fib(n):
 p.wait()
 """
 
+file_subprocess_shell = """
+import subprocess
+import os
+with open(os.path.join(os.path.dirname(__file__), "sub.py"), "w") as f:
+ f.write("print('hello')")
+path = os.path.join(os.path.dirname(__file__), "sub.py")
+print(subprocess.call(f"python {path}", shell=True))
+"""
+
 file_fork = """
 import os
 import time
@@ -295,6 +304,21 @@ def test_code_string(self):
  with open(os.path.join(tmpdir, os.listdir(tmpdir)[0])) as f:
  self.assertSubprocessName("python -c", json.load(f))
 
+ def test_subprocess_shell_true(self):
+ def check_func(data):
+ pids = set()
+ for entry in data["traceEvents"]:
+ pids.add(entry["pid"])
+ self.assertEqual(len(pids), 2)
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ output_path = os.path.join(tmpdir, "result.json")
+ self.template(["viztracer", "-o", output_path, "cmdline_test.py"],
+ expected_output_file=output_path,
+ script=file_subprocess_shell,
+ expected_stdout=".*hello.*",
+ check_func=check_func)
+
  def test_nested(self):
  def check_func(data):
  pids = set()