From 1f0b73c1d0940c2d982ce0b79252f5869d282d24 Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Sun, 30 Mar 2025 10:38:23 +0200 Subject: [PATCH 1/8] clean up and handle arguments by splitting avoiding shell usage --- .justfile | 2 +- jbang/__init__.py | 74 +++------------------------------------------ setup.py | 2 +- tests/test_jbang.py | 9 ++++++ 4 files changed, 16 insertions(+), 71 deletions(-) diff --git a/.justfile b/.justfile index b3145bf..207a449 100644 --- a/.justfile +++ b/.justfile @@ -3,7 +3,7 @@ default: test: source venv/bin/activate - pip3 install -e . + #pip3 install -e . pip install -e ".[test]" python -m pytest diff --git a/jbang/__init__.py b/jbang/__init__.py index 98302bb..bb8cb9c 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -1,3 +1,4 @@ +import shlex import subprocess import os import platform @@ -216,75 +217,10 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: if 'process' in globals(): del globals()['process'] -def exec(*args: str, capture_output: bool = False) -> Any: - """Execute jbang command for library usage.""" - arg_line = " ".join(args) - jbang_path = _get_jbang_path() - installer_cmd = _get_installer_command() - - if not jbang_path and not installer_cmd: - raise JbangExecutionError( - f"Unable to pre-install jbang: {arg_line}. Please install jbang manually.", - 1 - ) - - subprocess_args = { - "shell": False, - "universal_newlines": True, - "stdout": subprocess.PIPE, - "stderr": subprocess.PIPE, - "stdin": subprocess.PIPE - } - - try: - if jbang_path: - process = subprocess.Popen( - [jbang_path] + list(args), - **subprocess_args - ) - else: - if "curl" in installer_cmd: - process = subprocess.Popen( - f"{installer_cmd} {arg_line}", - shell=True, - **{k: v for k, v in subprocess_args.items() if k != "shell"} - ) - else: - # PowerShell case - temp_script = os.path.join(os.environ.get('TEMP', '/tmp'), 'jbang.ps1') - with open(temp_script, 'w') as f: - f.write(installer_cmd) - process = subprocess.Popen( - ["powershell", "-Command", f"{temp_script} {arg_line}"], - **subprocess_args - ) - - stdout, stderr = process.communicate() - - if process.returncode != 0: - raise JbangExecutionError( - f"Command failed with code {process.returncode}: {arg_line}", - process.returncode - ) - - result = type('CommandResult', (), { - 'returncode': process.returncode, - 'stdout': stdout, - 'stderr': stderr - }) - - if not capture_output: - if stdout: - print(stdout, end='', flush=True) - if stderr: - print(stderr, end='', flush=True, file=sys.stderr) - - return result - - except Exception as e: - if isinstance(e, JbangExecutionError): - raise - raise JbangExecutionError(str(e), 1) +def exec(arg: str, capture_output: bool = False) -> Any: + """Execute jbang command simulating shell.""" + args = shlex.split(arg) + return _exec_library(*args, capture_output=capture_output) def main(): """Command-line entry point for jbang-python.""" diff --git a/setup.py b/setup.py index 12a2ad1..0d549b6 100755 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ long_description = (this_directory / "README.md").read_text() setup(name='jbang', - version='0.5.6', + version='0.5.7', description='Python for JBang - Java Script in your Python', long_description=long_description, long_description_content_type='text/markdown', diff --git a/tests/test_jbang.py b/tests/test_jbang.py index 47e212b..b243a54 100644 --- a/tests/test_jbang.py +++ b/tests/test_jbang.py @@ -25,4 +25,13 @@ def test_error_handling(): print("\nTesting error handling...") with pytest.raises(Exception): jbang.exec('nonexistent-script-name') + print("✓ Error handling works") + +def test_multiple_arguments(): + """Test multiple arguments.""" + print("\nTesting multiple arguments...") + out = jbang.exec('-D="funky bear" properties@jbangdev') + assert out.returncode == 0 + assert 'funky bear' in out.stdout + print("✓ Error handling works") \ No newline at end of file From c08887a61abd796fad63703453d7c396791e485d Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Sun, 30 Mar 2025 10:54:35 +0200 Subject: [PATCH 2/8] dont fail fast --- .github/workflows/build.yaml | 32 +++++++++----------------------- 1 file changed, 9 insertions(+), 23 deletions(-) diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 9b3e066..91b79f0 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -1,4 +1,4 @@ -name: Build and Test +name: Test on: push: @@ -10,36 +10,22 @@ jobs: test: runs-on: ${{ matrix.os }} strategy: + fail-fast: false # This ensures all matrix jobs run even if one fails matrix: - os: [ubuntu-latest, windows-latest] - python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13"] - exclude: - - os: windows-latest - python-version: "3.8" # Windows doesn't support Python 3.8 - - os: windows-latest - python-version: "3.9" # Windows doesn't support Python 3.9 + os: [ubuntu-latest, windows-latest, macos-latest] + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12'] steps: - - uses: actions/checkout@v4 - + - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - - - name: Install build dependencies + - name: Install dependencies run: | python -m pip install --upgrade pip - pip install setuptools wheel build - - - name: Install package and test dependencies - run: | - pip install -e ".[test]" - + pip install pytest pytest-cov + pip install -e . - name: Run tests run: | - pytest tests/ - - - name: Build package - run: | - python setup.py sdist bdist_wheel \ No newline at end of file + pytest --cov=jbang tests/ \ No newline at end of file From aad9bd41f2e2920364e6c5c0e574cd4327131534 Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Sun, 30 Mar 2025 10:56:24 +0200 Subject: [PATCH 3/8] better error message --- jbang/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/jbang/__init__.py b/jbang/__init__.py index bb8cb9c..9ad632b 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -130,7 +130,7 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: if process.returncode != 0: raise JbangExecutionError( - f"Command failed with code {process.returncode}: {arg_line}", + f"Command failed with code {process.returncode}: {process.args}", process.returncode ) @@ -197,7 +197,7 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: process.wait() if process.returncode != 0: raise JbangExecutionError( - f"Command failed with code {process.returncode}: {arg_line}", + f"Command failed with code {process.returncode}: {process.args}", process.returncode ) return type('CommandResult', (), {'returncode': process.returncode}) From 4011a657027ee1068b1a8b0e48b6528ff8b40a9f Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Mon, 31 Mar 2025 05:59:46 +0200 Subject: [PATCH 4/8] add logging for tests --- .github/workflows/build.yaml | 2 +- .gitignore | 1 + .justfile | 2 +- jbang/__init__.py | 50 +++++++++++++++++++++++++++++------- 4 files changed, 44 insertions(+), 11 deletions(-) diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 91b79f0..4f420b2 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -28,4 +28,4 @@ jobs: pip install -e . - name: Run tests run: | - pytest --cov=jbang tests/ \ No newline at end of file + pytest --cov=jbang tests/ -o log_cli_level=DEBUG -o log_cli=true \ No newline at end of file diff --git a/.gitignore b/.gitignore index 4948bdb..446fe4d 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ dist jbang.egg-info jbang/__pycache__ __pycache__ +.coverage diff --git a/.justfile b/.justfile index 207a449..c4a0f43 100644 --- a/.justfile +++ b/.justfile @@ -5,7 +5,7 @@ test: source venv/bin/activate #pip3 install -e . pip install -e ".[test]" - python -m pytest + python -m pytest -o log_cli_level=DEBUG # -o log_cli=true release: source venv/bin/activate diff --git a/jbang/__init__.py b/jbang/__init__.py index 9ad632b..c40bf2b 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -17,32 +17,42 @@ def __init__(self, message, exit_code): def _get_jbang_path() -> Optional[str]: """Get the path to jbang executable.""" + log.debug("Searching for jbang executable...") for cmd in ['jbang', './jbang.cmd' if platform.system() == 'Windows' else None, './jbang']: if cmd: + log.debug(f"Checking for command: {cmd}") result = subprocess.run(f"which {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: + log.debug(f"Found jbang at: {cmd}") return cmd + log.warning("No jbang executable found in PATH") return None def _get_installer_command() -> Optional[str]: """Get the appropriate installer command based on available tools.""" + log.debug("Checking for available installer tools...") if subprocess.run("which curl", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0 and \ subprocess.run("which bash", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0: + log.debug("Will use curl/bash installer if needed") return "curl -Ls https://sh.jbang.dev | bash -s -" elif subprocess.run("which powershell", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0: + log.debug("Will use PowerShell installer if needed") return 'iex "& { $(iwr -useb https://ps.jbang.dev) } $args"' + log.warning("No suitable installer found") return None def _setup_subprocess_args(capture_output: bool = False) -> Dict[str, Any]: """Setup subprocess arguments with proper terminal interaction.""" + log.debug(f"Setting up subprocess arguments (capture_output={capture_output})") args = { "shell": False, "universal_newlines": True, - "start_new_session": False, # Changed to False to ensure proper signal propagation - "preexec_fn": os.setpgrp if platform.system() != "Windows" else None # Create new process group + "start_new_session": False, + "preexec_fn": os.setpgrp if platform.system() != "Windows" else None } if capture_output: + log.debug("Configuring for captured output") args.update({ "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, @@ -72,13 +82,14 @@ def _setup_subprocess_args(capture_output: bool = False) -> Dict[str, Any]: def _handle_signal(signum, frame): """Handle signals and propagate them to child processes.""" + log.debug(f"Received signal {signum}") if hasattr(frame, 'f_globals') and 'process' in frame.f_globals: process = frame.f_globals['process'] - if process and process.poll() is None: # Process is still running + if process and process.poll() is None: + log.debug(f"Propagating signal {signum} to process group {os.getpgid(process.pid)}") if platform.system() == "Windows": process.terminate() else: - # Send signal to the entire process group os.killpg(os.getpgid(process.pid), signum) process.wait() sys.exit(0) @@ -86,10 +97,13 @@ def _handle_signal(signum, frame): def _exec_library(*args: str, capture_output: bool = False) -> Any: """Execute jbang command for library usage.""" arg_line = " ".join(args) + log.debug(f"Executing library command: {arg_line}") + jbang_path = _get_jbang_path() installer_cmd = _get_installer_command() if not jbang_path and not installer_cmd: + log.error("No jbang executable or installer found") raise JbangExecutionError( f"Unable to pre-install jbang: {arg_line}. Please install jbang manually.", 1 @@ -105,11 +119,13 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: try: if jbang_path: + log.debug(f"Using jbang executable: {jbang_path}") process = subprocess.Popen( [jbang_path] + list(args), **subprocess_args ) else: + log.debug(f"Using installer command: {installer_cmd}") if "curl" in installer_cmd: process = subprocess.Popen( f"{installer_cmd} {arg_line}", @@ -117,8 +133,8 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: **{k: v for k, v in subprocess_args.items() if k != "shell"} ) else: - # PowerShell case temp_script = os.path.join(os.environ.get('TEMP', '/tmp'), 'jbang.ps1') + log.debug(f"Creating temporary PowerShell script: {temp_script}") with open(temp_script, 'w') as f: f.write(installer_cmd) process = subprocess.Popen( @@ -126,9 +142,12 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: **subprocess_args ) + log.debug(f"Process started with PID: {process.pid}") stdout, stderr = process.communicate() if process.returncode != 0: + log.error(f"Command failed with code {process.returncode}") + log.error(f"stderr: {stderr}") raise JbangExecutionError( f"Command failed with code {process.returncode}: {process.args}", process.returncode @@ -149,6 +168,7 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: return result except Exception as e: + log.debug(f"Exception during execution: {str(e)}", exc_info=True) if isinstance(e, JbangExecutionError): raise raise JbangExecutionError(str(e), 1) @@ -156,10 +176,13 @@ def _exec_library(*args: str, capture_output: bool = False) -> Any: def _exec_cli(*args: str, capture_output: bool = False) -> Any: """Execute jbang command for CLI usage.""" arg_line = " ".join(args) + log.debug(f"Executing CLI command: {arg_line}") + jbang_path = _get_jbang_path() installer_cmd = _get_installer_command() if not jbang_path and not installer_cmd: + log.warn("No jbang executable or installer found") raise JbangExecutionError( f"Unable to pre-install jbang: {arg_line}. Please install jbang manually.", 1 @@ -169,11 +192,13 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: try: if jbang_path: + log.debug(f"Using jbang executable: {jbang_path}") process = subprocess.Popen( [jbang_path] + list(args), **subprocess_args ) else: + log.debug(f"Using installer command: {installer_cmd}") if "curl" in installer_cmd: process = subprocess.Popen( f"{installer_cmd} {arg_line}", @@ -181,8 +206,8 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: **{k: v for k, v in subprocess_args.items() if k != "shell"} ) else: - # PowerShell case temp_script = os.path.join(os.environ.get('TEMP', '/tmp'), 'jbang.ps1') + log.debug(f"Creating temporary PowerShell script: {temp_script}") with open(temp_script, 'w') as f: f.write(installer_cmd) process = subprocess.Popen( @@ -190,18 +215,20 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: **subprocess_args ) - # Store process in globals for signal handler + log.debug(f"Process started with PID: {process.pid}") globals()['process'] = process try: process.wait() if process.returncode != 0: + log.warn(f"Command failed with code {process.returncode}") raise JbangExecutionError( f"Command failed with code {process.returncode}: {process.args}", process.returncode ) return type('CommandResult', (), {'returncode': process.returncode}) except KeyboardInterrupt: + log.debug("Received keyboard interrupt") if platform.system() == "Windows": process.terminate() else: @@ -209,37 +236,42 @@ def _exec_cli(*args: str, capture_output: bool = False) -> Any: process.wait() raise except Exception as e: + log.warn(f"Exception during execution: {str(e)}", exc_info=True) if isinstance(e, JbangExecutionError): raise raise JbangExecutionError(str(e), 1) finally: - # Clean up globals if 'process' in globals(): del globals()['process'] def exec(arg: str, capture_output: bool = False) -> Any: """Execute jbang command simulating shell.""" + log.debug(f"Executing command: {arg}") args = shlex.split(arg) return _exec_library(*args, capture_output=capture_output) def main(): """Command-line entry point for jbang-python.""" + log.debug("Starting jbang-python CLI") # Register signal handlers signal.signal(signal.SIGINT, _handle_signal) signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGHUP, _handle_signal) signal.signal(signal.SIGQUIT, _handle_signal) + log.debug("Signal handlers registered") try: result = _exec_cli(*sys.argv[1:], capture_output=False) sys.exit(result.returncode) except KeyboardInterrupt: + log.debug("Received keyboard interrupt, exiting") sys.exit(0) except JbangExecutionError as e: + log.debug(f"Jbang execution error: {str(e)}") sys.exit(e.exit_code) except Exception as e: - print(f"Error: {str(e)}", file=sys.stderr) + log.debug(f"Unexpected error: {str(e)}", exc_info=True) sys.exit(1) if __name__ == "__main__": From 3ed7199baf92b40081405db815d9ae3049f60836 Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Mon, 31 Mar 2025 06:13:30 +0200 Subject: [PATCH 5/8] only search for installer if needed --- .justfile | 2 +- jbang/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.justfile b/.justfile index c4a0f43..4caf90c 100644 --- a/.justfile +++ b/.justfile @@ -5,7 +5,7 @@ test: source venv/bin/activate #pip3 install -e . pip install -e ".[test]" - python -m pytest -o log_cli_level=DEBUG # -o log_cli=true + python -m pytest -o log_cli_level=DEBUG -o log_cli=true release: source venv/bin/activate diff --git a/jbang/__init__.py b/jbang/__init__.py index c40bf2b..974a420 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -23,7 +23,7 @@ def _get_jbang_path() -> Optional[str]: log.debug(f"Checking for command: {cmd}") result = subprocess.run(f"which {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: - log.debug(f"Found jbang at: {cmd}") + log.debug(f"Found jbang at: " + result.stdout.decode('utf-8')) return cmd log.warning("No jbang executable found in PATH") return None From 8b338c9403d6bcaabeb1c2878498196e630cd8e6 Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Mon, 31 Mar 2025 07:54:32 +0200 Subject: [PATCH 6/8] ensure just don't have jbang in path --- .justfile | 6 +++--- tests/test_jbang.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.justfile b/.justfile index 4caf90c..4d62863 100644 --- a/.justfile +++ b/.justfile @@ -1,11 +1,11 @@ -default: - echo 'Hello, world!' test: source venv/bin/activate #pip3 install -e . pip install -e ".[test]" - python -m pytest -o log_cli_level=DEBUG -o log_cli=true + + echo Running tests with no jbang in PATH + PATH=$(echo $PATH | tr ':' '\n' | grep -v "\.jbang/bin" | tr '\n' ':' | sed 's/:$//') python -m pytest -o log_cli_level=DEBUG -o log_cli=true release: source venv/bin/activate diff --git a/tests/test_jbang.py b/tests/test_jbang.py index b243a54..e80e9ca 100644 --- a/tests/test_jbang.py +++ b/tests/test_jbang.py @@ -23,8 +23,8 @@ def test_catalog_script(): def test_error_handling(): """Test error handling.""" print("\nTesting error handling...") - with pytest.raises(Exception): - jbang.exec('nonexistent-script-name') + out = jbang.exec('nonexistent-script-name') + assert out.returncode == 2 print("✓ Error handling works") def test_multiple_arguments(): From 8caa9d50db3dcb3fe181c3b3721455c47c58649d Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Mon, 31 Mar 2025 19:23:44 +0200 Subject: [PATCH 7/8] port from .js --- .justfile | 6 +- jbang/__init__.py | 353 ++++++++++++++------------------------------ tests/test_jbang.py | 5 +- 3 files changed, 117 insertions(+), 247 deletions(-) diff --git a/.justfile b/.justfile index 4d62863..9201913 100644 --- a/.justfile +++ b/.justfile @@ -1,13 +1,13 @@ test: - source venv/bin/activate + source .venv/bin/activate #pip3 install -e . - pip install -e ".[test]" + uv pip install -e ".[test]" echo Running tests with no jbang in PATH PATH=$(echo $PATH | tr ':' '\n' | grep -v "\.jbang/bin" | tr '\n' ':' | sed 's/:$//') python -m pytest -o log_cli_level=DEBUG -o log_cli=true release: source venv/bin/activate - pip3 install setuptools + uv pip install setuptools gh release create `python3 setup.py --version` --generate-notes diff --git a/jbang/__init__.py b/jbang/__init__.py index 974a420..fde67c5 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -1,277 +1,146 @@ -import shlex -import subprocess +import logging import os import platform -import logging -import sys +import shutil import signal -from typing import Optional, Dict, Any +import subprocess +import sys +from typing import Any, Dict, Optional + +# Configure logging based on environment variable +debug_enabled = 'jbang' in os.environ.get('DEBUG', '') +if debug_enabled: + logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s %(levelname)s %(name)s: %(message)s' + ) + log = logging.getLogger(__name__) -class JbangExecutionError(Exception): - """Custom exception to capture Jbang execution errors with exit code.""" - def __init__(self, message, exit_code): - super().__init__(message) - self.exit_code = exit_code +def quote(xs): + def quote_string(s): + if s == '': + return "''" + if isinstance(s, dict) and 'op' in s: + return ''.join(['\\' + char for char in s['op']]) + + if any(char in s for char in ['"', ' ']) and "'" not in s: + return "'" + s.replace("'", "\\'").replace("\\", "\\\\") + "'" + + if any(char in s for char in ['"', "'", ' ']): + return '"' + s.replace('"', '\\"').replace('\\', '\\\\').replace('$', '\\$').replace('`', '\\`').replace('!', '\\!') + '"' + + return ''.join(['\\' + char if char in '#!"$&\'()*,:;<=>?[\\]^`{|}' else char for char in s]) + + return ' '.join(map(quote_string, xs)) -def _get_jbang_path() -> Optional[str]: - """Get the path to jbang executable.""" +def _getCommandLine(args: list) -> Optional[str]: + """Get the jbang command line with arguments, using no-install option if needed.""" log.debug("Searching for jbang executable...") - for cmd in ['jbang', './jbang.cmd' if platform.system() == 'Windows' else None, './jbang']: + + argLine = quote(args) + # Try different possible jbang locations + path = None + for cmd in ['jbang', + './jbang.cmd' if platform.system() == 'Windows' else None, + os.path.expanduser('~/.jbang/bin/jbang')]: if cmd: - log.debug(f"Checking for command: {cmd}") - result = subprocess.run(f"which {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - if result.returncode == 0: - log.debug(f"Found jbang at: " + result.stdout.decode('utf-8')) - return cmd - log.warning("No jbang executable found in PATH") - return None - -def _get_installer_command() -> Optional[str]: - """Get the appropriate installer command based on available tools.""" - log.debug("Checking for available installer tools...") - if subprocess.run("which curl", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0 and \ - subprocess.run("which bash", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0: - log.debug("Will use curl/bash installer if needed") - return "curl -Ls https://sh.jbang.dev | bash -s -" - elif subprocess.run("which powershell", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0: - log.debug("Will use PowerShell installer if needed") - return 'iex "& { $(iwr -useb https://ps.jbang.dev) } $args"' - log.warning("No suitable installer found") - return None - -def _setup_subprocess_args(capture_output: bool = False) -> Dict[str, Any]: - """Setup subprocess arguments with proper terminal interaction.""" - log.debug(f"Setting up subprocess arguments (capture_output={capture_output})") - args = { - "shell": False, - "universal_newlines": True, - "start_new_session": False, - "preexec_fn": os.setpgrp if platform.system() != "Windows" else None - } + if shutil.which(cmd): + path = cmd + break - if capture_output: - log.debug("Configuring for captured output") - args.update({ - "stdout": subprocess.PIPE, - "stderr": subprocess.PIPE, - "stdin": subprocess.PIPE - }) + if path: + log.debug(f"found existing jbang installation at: {path}") + return ' '.join([path, argLine]) + + # Try no-install options + if shutil.which('curl') and shutil.which('bash'): + log.debug("running jbang using curl and bash") + return " ".join(["curl", "-Ls", "https://sh.jbang.dev", "|", "bash", "-s", "-", argLine]) + elif shutil.which('powershell'): + log.debug("running jbang using PowerShell") + return " ".join(["powershell", "-Command", "iex \"& { $(iwr -useb https://ps.jbang.dev) } $argLine\""]) else: - # Try to connect to actual terminal if available - try: - if hasattr(sys.stdin, 'fileno'): - args["stdin"] = sys.stdin - except (IOError, OSError): - args["stdin"] = subprocess.PIPE - - try: - if hasattr(sys.stdout, 'fileno'): - args["stdout"] = sys.stdout - except (IOError, OSError): - args["stdout"] = subprocess.PIPE + log.debug("no jbang installation found") + return None - try: - if hasattr(sys.stderr, 'fileno'): - args["stderr"] = sys.stderr - except (IOError, OSError): - args["stderr"] = subprocess.PIPE +def exec(*args: str) -> Any: + log.debug(f"try to execute async command: {args}") + + cmdLine = _getCommandLine(list(args)) + + if cmdLine: + result = subprocess.run( + cmdLine, + shell=True, + capture_output=True, + text=True, + check=True + ) + return type('CommandResult', (), { + 'stdout': result.stdout, + 'stderr': result.stderr, + 'exitCode': result.returncode + }) + else: + print("Could not locate a way to run jbang. Try install jbang manually and try again.") + raise Exception( + "Could not locate a way to run jbang. Try install jbang manually and try again.", + 2 + ) - return args +def spawnSync(*args: str) -> Any: + log.debug(f"try to execute sync command: {args}") + + cmdLine = _getCommandLine(list(args)) + + if cmdLine: + result = subprocess.run( + cmdLine, + shell=True, + stdin=sys.stdin, + stdout=sys.stdout, + stderr=sys.stderr, + check=True + ) + return type('CommandResult', (), { + 'stdout': result.stdout, + 'stderr': result.stderr, + 'exitCode': result.returncode + }) + else: + print("Could not locate a way to run jbang. Try install jbang manually and try again.") + raise Exception( + "Could not locate a way to run jbang. Try install jbang manually and try again.", + 2 + ) def _handle_signal(signum, frame): """Handle signals and propagate them to child processes.""" - log.debug(f"Received signal {signum}") if hasattr(frame, 'f_globals') and 'process' in frame.f_globals: process = frame.f_globals['process'] - if process and process.poll() is None: - log.debug(f"Propagating signal {signum} to process group {os.getpgid(process.pid)}") + if process and process.poll() is None: # Process is still running if platform.system() == "Windows": process.terminate() else: + # Send signal to the entire process group os.killpg(os.getpgid(process.pid), signum) process.wait() sys.exit(0) -def _exec_library(*args: str, capture_output: bool = False) -> Any: - """Execute jbang command for library usage.""" - arg_line = " ".join(args) - log.debug(f"Executing library command: {arg_line}") - - jbang_path = _get_jbang_path() - installer_cmd = _get_installer_command() - - if not jbang_path and not installer_cmd: - log.error("No jbang executable or installer found") - raise JbangExecutionError( - f"Unable to pre-install jbang: {arg_line}. Please install jbang manually.", - 1 - ) - - subprocess_args = { - "shell": False, - "universal_newlines": True, - "stdout": subprocess.PIPE, - "stderr": subprocess.PIPE, - "stdin": subprocess.PIPE - } - - try: - if jbang_path: - log.debug(f"Using jbang executable: {jbang_path}") - process = subprocess.Popen( - [jbang_path] + list(args), - **subprocess_args - ) - else: - log.debug(f"Using installer command: {installer_cmd}") - if "curl" in installer_cmd: - process = subprocess.Popen( - f"{installer_cmd} {arg_line}", - shell=True, - **{k: v for k, v in subprocess_args.items() if k != "shell"} - ) - else: - temp_script = os.path.join(os.environ.get('TEMP', '/tmp'), 'jbang.ps1') - log.debug(f"Creating temporary PowerShell script: {temp_script}") - with open(temp_script, 'w') as f: - f.write(installer_cmd) - process = subprocess.Popen( - ["powershell", "-Command", f"{temp_script} {arg_line}"], - **subprocess_args - ) - - log.debug(f"Process started with PID: {process.pid}") - stdout, stderr = process.communicate() - - if process.returncode != 0: - log.error(f"Command failed with code {process.returncode}") - log.error(f"stderr: {stderr}") - raise JbangExecutionError( - f"Command failed with code {process.returncode}: {process.args}", - process.returncode - ) - - result = type('CommandResult', (), { - 'returncode': process.returncode, - 'stdout': stdout, - 'stderr': stderr - }) - - if not capture_output: - if stdout: - print(stdout, end='', flush=True) - if stderr: - print(stderr, end='', flush=True, file=sys.stderr) - - return result - - except Exception as e: - log.debug(f"Exception during execution: {str(e)}", exc_info=True) - if isinstance(e, JbangExecutionError): - raise - raise JbangExecutionError(str(e), 1) - -def _exec_cli(*args: str, capture_output: bool = False) -> Any: - """Execute jbang command for CLI usage.""" - arg_line = " ".join(args) - log.debug(f"Executing CLI command: {arg_line}") - - jbang_path = _get_jbang_path() - installer_cmd = _get_installer_command() - - if not jbang_path and not installer_cmd: - log.warn("No jbang executable or installer found") - raise JbangExecutionError( - f"Unable to pre-install jbang: {arg_line}. Please install jbang manually.", - 1 - ) - - subprocess_args = _setup_subprocess_args(capture_output) - - try: - if jbang_path: - log.debug(f"Using jbang executable: {jbang_path}") - process = subprocess.Popen( - [jbang_path] + list(args), - **subprocess_args - ) - else: - log.debug(f"Using installer command: {installer_cmd}") - if "curl" in installer_cmd: - process = subprocess.Popen( - f"{installer_cmd} {arg_line}", - shell=True, - **{k: v for k, v in subprocess_args.items() if k != "shell"} - ) - else: - temp_script = os.path.join(os.environ.get('TEMP', '/tmp'), 'jbang.ps1') - log.debug(f"Creating temporary PowerShell script: {temp_script}") - with open(temp_script, 'w') as f: - f.write(installer_cmd) - process = subprocess.Popen( - ["powershell", "-Command", f"{temp_script} {arg_line}"], - **subprocess_args - ) - - log.debug(f"Process started with PID: {process.pid}") - globals()['process'] = process - - try: - process.wait() - if process.returncode != 0: - log.warn(f"Command failed with code {process.returncode}") - raise JbangExecutionError( - f"Command failed with code {process.returncode}: {process.args}", - process.returncode - ) - return type('CommandResult', (), {'returncode': process.returncode}) - except KeyboardInterrupt: - log.debug("Received keyboard interrupt") - if platform.system() == "Windows": - process.terminate() - else: - os.killpg(os.getpgid(process.pid), signal.SIGINT) - process.wait() - raise - except Exception as e: - log.warn(f"Exception during execution: {str(e)}", exc_info=True) - if isinstance(e, JbangExecutionError): - raise - raise JbangExecutionError(str(e), 1) - finally: - if 'process' in globals(): - del globals()['process'] - -def exec(arg: str, capture_output: bool = False) -> Any: - """Execute jbang command simulating shell.""" - log.debug(f"Executing command: {arg}") - args = shlex.split(arg) - return _exec_library(*args, capture_output=capture_output) - def main(): """Command-line entry point for jbang-python.""" log.debug("Starting jbang-python CLI") - - # Register signal handlers - signal.signal(signal.SIGINT, _handle_signal) - signal.signal(signal.SIGTERM, _handle_signal) - signal.signal(signal.SIGHUP, _handle_signal) - signal.signal(signal.SIGQUIT, _handle_signal) - log.debug("Signal handlers registered") try: - result = _exec_cli(*sys.argv[1:], capture_output=False) - sys.exit(result.returncode) + result = spawnSync(*sys.argv[1:]) + sys.exit(result.exitCode) except KeyboardInterrupt: - log.debug("Received keyboard interrupt, exiting") - sys.exit(0) - except JbangExecutionError as e: - log.debug(f"Jbang execution error: {str(e)}") - sys.exit(e.exit_code) + log.debug("Keyboard interrupt") + sys.exit(130) except Exception as e: - log.debug(f"Unexpected error: {str(e)}", exc_info=True) + log.error(f"Unexpected error: {str(e)}", exc_info=True) sys.exit(1) if __name__ == "__main__": diff --git a/tests/test_jbang.py b/tests/test_jbang.py index e80e9ca..4be7429 100644 --- a/tests/test_jbang.py +++ b/tests/test_jbang.py @@ -5,7 +5,7 @@ def test_version_command(): """Test version command.""" print("\nTesting version command...") try: - out = jbang.exec('--version',capture_output=True) + out = jbang.exec('--version') assert out.returncode == 0 print("✓ Version command works") except Exception as e: @@ -15,7 +15,8 @@ def test_catalog_script(): """Test catalog script execution.""" print("\nTesting catalog script...") try: - jbang.exec('properties@jbangdev') + out = jbang.exec('properties@jbangdev') + assert out.returncode == 0 print("✓ Catalog script works") except Exception as e: pytest.fail(f"✗ Catalog script failed: {e}") From b522d2e19fc605920b545ed6241218e88aa3e6de Mon Sep 17 00:00:00 2001 From: Max Rydahl Andersen Date: Mon, 31 Mar 2025 19:40:23 +0200 Subject: [PATCH 8/8] update --- jbang/__init__.py | 6 +++--- tests/test_jbang.py | 8 ++++---- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/jbang/__init__.py b/jbang/__init__.py index fde67c5..65a480b 100644 --- a/jbang/__init__.py +++ b/jbang/__init__.py @@ -35,11 +35,11 @@ def quote_string(s): return ' '.join(map(quote_string, xs)) -def _getCommandLine(args: list) -> Optional[str]: +def _getCommandLine(args) -> Optional[str]: """Get the jbang command line with arguments, using no-install option if needed.""" log.debug("Searching for jbang executable...") - argLine = quote(args) + argLine = args # Try different possible jbang locations path = None for cmd in ['jbang', @@ -93,7 +93,7 @@ def exec(*args: str) -> Any: def spawnSync(*args: str) -> Any: log.debug(f"try to execute sync command: {args}") - cmdLine = _getCommandLine(list(args)) + cmdLine = _getCommandLine(args) if cmdLine: result = subprocess.run( diff --git a/tests/test_jbang.py b/tests/test_jbang.py index 4be7429..8b9006d 100644 --- a/tests/test_jbang.py +++ b/tests/test_jbang.py @@ -6,7 +6,7 @@ def test_version_command(): print("\nTesting version command...") try: out = jbang.exec('--version') - assert out.returncode == 0 + assert out.exitCode == 0 print("✓ Version command works") except Exception as e: pytest.fail(f"✗ Version command failed: {e}") @@ -16,7 +16,7 @@ def test_catalog_script(): print("\nTesting catalog script...") try: out = jbang.exec('properties@jbangdev') - assert out.returncode == 0 + assert out.exitCode == 0 print("✓ Catalog script works") except Exception as e: pytest.fail(f"✗ Catalog script failed: {e}") @@ -25,13 +25,13 @@ def test_error_handling(): """Test error handling.""" print("\nTesting error handling...") out = jbang.exec('nonexistent-script-name') - assert out.returncode == 2 + assert out.exitCode == 2 print("✓ Error handling works") def test_multiple_arguments(): """Test multiple arguments.""" print("\nTesting multiple arguments...") - out = jbang.exec('-D="funky bear" properties@jbangdev') + out = jbang.exec('-Dx="funky bear" properties@jbangdev') assert out.returncode == 0 assert 'funky bear' in out.stdout