Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 9 additions & 23 deletions .github/workflows/build.yaml
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
name: Build and Test
name: Test

on:
push:
Expand All @@ -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
pytest --cov=jbang tests/ -o log_cli_level=DEBUG -o log_cli=true
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,4 @@ dist
jbang.egg-info
jbang/__pycache__
__pycache__
.coverage
17 changes: 8 additions & 9 deletions .justfile
Original file line number Diff line number Diff line change
@@ -1,13 +1,12 @@
default:
echo 'Hello, world!'

test:
source venv/bin/activate
pip3 install -e .
pip install -e ".[test]"
python -m pytest
source .venv/bin/activate
#pip3 install -e .
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/:$//') .venv/bin/python -m pytest -o log_cli_level=DEBUG -o log_cli=true

release:
source venv/bin/activate
pip3 install setuptools
source .venv/bin/activate
uv pip install setuptools
gh release create `python3 setup.py --version` --generate-notes
308 changes: 2 additions & 306 deletions jbang/__init__.py
Original file line number Diff line number Diff line change
@@ -1,310 +1,6 @@
import subprocess
import os
import platform
import logging
import sys
import signal
from typing import Optional, Dict, Any
from .jbang import exec, spawnSync, main, quote

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 _get_jbang_path() -> Optional[str]:
"""Get the path to jbang executable."""
for cmd in ['jbang', './jbang.cmd' if platform.system() == 'Windows' else None, './jbang']:
if cmd:
result = subprocess.run(f"which {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if result.returncode == 0:
return cmd
return None

def _get_installer_command() -> Optional[str]:
"""Get the appropriate installer command based on available 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:
return "curl -Ls https://sh.jbang.dev | bash -s -"
elif subprocess.run("which powershell", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0:
return 'iex "& { $(iwr -useb https://ps.jbang.dev) } $args"'
return None

def _setup_subprocess_args(capture_output: bool = False) -> Dict[str, Any]:
"""Setup subprocess arguments with proper terminal interaction."""
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
}

if capture_output:
args.update({
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"stdin": subprocess.PIPE
})
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

try:
if hasattr(sys.stderr, 'fileno'):
args["stderr"] = sys.stderr
except (IOError, OSError):
args["stderr"] = subprocess.PIPE

return args

def _handle_signal(signum, frame):
"""Handle signals and propagate them to child processes."""
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 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)
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_cli(*args: str, capture_output: bool = False) -> Any:
"""Execute jbang command for CLI 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 = _setup_subprocess_args(capture_output)

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
)

# Store process in globals for signal handler
globals()['process'] = process

try:
process.wait()
if process.returncode != 0:
raise JbangExecutionError(
f"Command failed with code {process.returncode}: {arg_line}",
process.returncode
)
return type('CommandResult', (), {'returncode': process.returncode})
except KeyboardInterrupt:
if platform.system() == "Windows":
process.terminate()
else:
os.killpg(os.getpgid(process.pid), signal.SIGINT)
process.wait()
raise
except Exception as e:
if isinstance(e, JbangExecutionError):
raise
raise JbangExecutionError(str(e), 1)
finally:
# Clean up globals
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 main():
"""Command-line entry point for jbang-python."""

# 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)

try:
result = _exec_cli(*sys.argv[1:], capture_output=False)
sys.exit(result.returncode)
except KeyboardInterrupt:
sys.exit(0)
except JbangExecutionError as e:
sys.exit(e.exit_code)
except Exception as e:
print(f"Error: {str(e)}", file=sys.stderr)
sys.exit(1)
__all__ = ['exec', 'spawnSync', 'main', 'quote']

if __name__ == "__main__":
main()
Loading
Loading