Skip to content
29 changes: 29 additions & 0 deletions kwave/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import logging
import os
import platform
import stat
from pathlib import Path
from typing import List
from urllib.request import urlretrieve
Expand Down Expand Up @@ -77,6 +78,31 @@ def _hash_file(filepath: str) -> str:
return md5.hexdigest()


def _ensure_executable(binary_filepath) -> None:
# Self-heal the executable bit on Linux/macOS. urlretrieve creates files
# at 0644, and prior versions of this package didn't fix that up, so users
# upgrading with a cached non-executable binary on disk would otherwise
# stay stuck (the cache check below returns True and skips re-download).
# Any OS-level failure here (broken symlink, read-only FS, wrong ownership,
# TOCTOU race) is degraded to a warning so it never aborts `import kwave`.
if PLATFORM == "windows":
return
try:
current_mode = os.stat(binary_filepath).st_mode
desired_mode = current_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
if current_mode == desired_mode:
return
os.chmod(binary_filepath, desired_mode)
except OSError: # pragma: no cover - defensive; degrades to warning, never fatal
# Don't abort import. The user can chmod +x manually or reinstall
# into a writable location.
logging.warning(
"kwave: cannot set executable bit on %s — backend='cpp' may fail with "
"Permission denied. Run `chmod +x` manually or reinstall.",
binary_filepath,
)


def _is_binary_present(binary_name: str, binary_type: str) -> bool:
binary_filepath = BINARY_PATH / binary_name
binary_file_exists = os.path.exists(binary_filepath)
Expand Down Expand Up @@ -106,6 +132,8 @@ def _is_binary_present(binary_name: str, binary_type: str) -> bool:
if existing_metadata["url"] not in latest_urls:
return False

_ensure_executable(binary_filepath)

# No need to check `version` field for now
# because we version is already present in the URL
return True
Expand Down Expand Up @@ -176,6 +204,7 @@ def download_binaries(system_os: str, bin_type: str):
try:
binary_filepath = os.path.join(BINARY_PATH, filename)
urlretrieve(url, binary_filepath)
_ensure_executable(binary_filepath)
_record_binary_metadata(binary_version=binary_version, binary_filepath=binary_filepath, binary_url=url, filename=filename)

except TimeoutError:
Expand Down
91 changes: 87 additions & 4 deletions tests/test__init__.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,98 @@
import importlib
import json
import os
import stat
import sys
from unittest.mock import patch

import pytest


def test__init():
with pytest.raises(NotImplementedError):
with patch("platform.system", lambda: "Unknown"):
import kwave
import kwave

importlib.reload(kwave)
try:
with pytest.raises(NotImplementedError):
with patch("platform.system", lambda: "Unknown"):
importlib.reload(kwave)
finally:
# The failed reload above left kwave.PLATFORM = "unknown" in module
# state (line where PLATFORM is set ran before the NotImplementedError).
# Reload once more without the patch so subsequent tests see a valid
# PLATFORM / URL_DICT pair.
importlib.reload(kwave)


def _seed_binary(binary_path, binary_name, url, *, mode=0o644):
"""Write a fake binary at the given path + matching metadata file."""
binary_path.mkdir(parents=True, exist_ok=True)
filepath = binary_path / binary_name
filepath.write_bytes(b"fake-binary-payload")
filepath.chmod(mode)
import kwave

metadata = {
"url": url,
"version": url.rsplit("/", 2)[-2],
"file_hash": kwave._hash_file(str(filepath)),
}
(binary_path / f"{binary_name}_metadata.json").write_text(json.dumps(metadata))
return filepath


@pytest.mark.skipif(sys.platform == "win32", reason="exec bit is meaningless on Windows")
def test_download_sets_executable_bit(tmp_path, monkeypatch):
"""Regression for #740 — urlretrieve creates files at 0644 and the C++
backend fails with Permission denied (exit 126) when the executor invokes
the binary."""
import kwave

monkeypatch.setattr(kwave, "BINARY_PATH", tmp_path)
Comment thread
waltsims marked this conversation as resolved.

test_url = next(urls[0] for urls in kwave.URL_DICT[kwave.PLATFORM].values() if urls)
filename = test_url.rsplit("/", 1)[-1]

def fake_urlretrieve(url, dest):
with open(dest, "wb") as f:
f.write(b"fake-binary-payload")
os.chmod(dest, 0o644)

monkeypatch.setattr(kwave, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(kwave, "URL_DICT", {kwave.PLATFORM: {"test": [test_url]}})

kwave.download_binaries(kwave.PLATFORM, "test")

binary_filepath = tmp_path / filename
assert binary_filepath.exists()
mode = os.stat(binary_filepath).st_mode
assert mode & stat.S_IXUSR, "owner exec bit not set after download"
assert mode & stat.S_IXGRP, "group exec bit not set after download"
assert mode & stat.S_IXOTH, "other exec bit not set after download"


@pytest.mark.skipif(sys.platform == "win32", reason="exec bit is meaningless on Windows")
def test_existing_non_executable_binary_is_healed(tmp_path, monkeypatch):
"""Regression for #740 — users with a cached non-executable binary from a
pre-fix install would otherwise stay broken across upgrades because
_is_binary_present skips re-download for valid cached binaries."""
import kwave

monkeypatch.setattr(kwave, "BINARY_PATH", tmp_path)

test_url = next(urls[0] for urls in kwave.URL_DICT[kwave.PLATFORM].values() if urls)
filename = test_url.rsplit("/", 1)[-1]
binary_filepath = _seed_binary(tmp_path, filename, test_url, mode=0o644)
pre_mode = os.stat(binary_filepath).st_mode
assert not (pre_mode & stat.S_IXUSR), "test fixture should start without exec bit"

monkeypatch.setattr(kwave, "URL_DICT", {kwave.PLATFORM: {"test": [test_url]}})

assert kwave._is_binary_present(filename, "test") is True

post_mode = os.stat(binary_filepath).st_mode
assert post_mode & stat.S_IXUSR, "owner exec bit not healed on cache hit"
assert post_mode & stat.S_IXGRP, "group exec bit not healed on cache hit"
assert post_mode & stat.S_IXOTH, "other exec bit not healed on cache hit"


if __name__ == "__main__":
Expand Down
Loading