Skip to content
Draft
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
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@
"asar": false,
"target": "nsis",
"icon": "src/electron/frontend/assets/app-icon/logo-guide-draft.ico",
"requestedExecutionLevel": "requireAdministrator"
"requestedExecutionLevel": "asInvoker"
},
"mac": {
"asar": true,
Expand Down
89 changes: 71 additions & 18 deletions src/pyflask/manageNeuroconv/manage_neuroconv.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import math
import os
import re
import sys
import traceback
import zoneinfo
from datetime import datetime, timedelta
Expand All @@ -28,6 +29,60 @@

progress_handler = TQDMProgressHandler()


def create_link(target, link, target_is_dir=False):
"""Create a filesystem link, using junctions/hard links on Windows when symlinks aren't available."""
target = str(target)
link = str(link)

if sys.platform != "win32":
os.symlink(target, link, target_is_directory=target_is_dir)
return

try:
os.symlink(target, link, target_is_directory=target_is_dir)
return
except OSError:
pass

link_error = OSError(
f"Cannot create a link from '{link}' to '{target}'. On Windows, enable Developer Mode "
"or run NWB GUIDE as Administrator to allow symbolic links. For files, the source and "
"destination must also be on the same drive."
)

if target_is_dir:
import _winapi

try:
# Junctions require an absolute target path.
_winapi.CreateJunction(os.path.abspath(target), link)
except OSError:
raise link_error
else:
try:
os.link(target, link)
except OSError:
raise link_error


def _is_link(path) -> bool:
"""Return True if path is a symlink or a Windows directory junction."""
return path.is_symlink() or os.path.isjunction(path)


def _remove_link(path) -> None:
"""Remove a symlink or junction at path, leaving the target it points to intact.

Directory symlinks and junctions are removed with rmdir; file symlinks and hard links
are removed with unlink.
"""
try:
os.rmdir(path)
except NotADirectoryError:
os.unlink(path)


EXCLUDED_RECORDING_INTERFACE_PROPERTIES = ["contact_vector", "contact_shapes", "group", "location"]

EXTRA_INTERFACE_PROPERTIES = {
Expand Down Expand Up @@ -1209,9 +1264,9 @@ def get_conversion_info(info: dict) -> dict:
resolved_output_directory = path_info["directory"]
default_output_directory = path_info["default"]

# Remove symlink placed at the default_output_directory if this will hold real data
if resolved_output_directory == default_output_directory and default_output_directory.is_symlink():
default_output_directory.unlink()
# Remove a link placed at the default_output_directory if this will hold real data
if resolved_output_directory == default_output_directory and _is_link(default_output_directory):
_remove_link(default_output_directory)

resolved_output_path.parent.mkdir(exist_ok=True, parents=True) # Ensure all parent directories exist

Expand Down Expand Up @@ -1324,24 +1379,22 @@ def convert_to_nwb(

create_file(info, log_url=log_url)

# Create a symlink between the fake data and custom data
# Create a link between the fake data and custom data
if not resolved_output_directory == default_output_directory:
if default_output_directory.exists():
# If default default_output_directory is not a symlink, delete all contents and create a symlink there
if not default_output_directory.is_symlink():
rmtree(default_output_directory)

# If the location is already a symlink, but points to a different output location
# remove the existing symlink before creating a new one
elif (
default_output_directory.is_symlink()
and default_output_directory.readlink() is not resolved_output_directory
):
default_output_directory.unlink()
# A symlink or junction (reported here regardless of whether its target still exists) may
# point to a stale location. Remove the link itself, leaving its target contents intact,
# so it can be recreated pointing at the current output directory.
if _is_link(default_output_directory):
_remove_link(default_output_directory)

# A real directory holds outputs from a previous conversion. Delete it so a link can
# take its place.
elif default_output_directory.exists():
rmtree(default_output_directory)

# Create a pointer to the actual conversion outputs
if not default_output_directory.exists():
os.symlink(resolved_output_directory, default_output_directory)
create_link(resolved_output_directory, default_output_directory, target_is_dir=True)

return dict(file=str(output_path))

Expand Down Expand Up @@ -1675,7 +1728,7 @@ def _aggregate_symlinks_in_new_directory(paths, reason="", folder_path=None) ->
list(map(lambda name: os.path.join(path, name), os.listdir(path))), None, new_path
)
else:
new_path.symlink_to(path, path.is_dir())
create_link(path, new_path, target_is_dir=path.is_dir())

return folder_path

Expand Down
129 changes: 129 additions & 0 deletions src/pyflask/tests/test_link_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
"""Unit tests for the cross-platform filesystem link helpers in manage_neuroconv.

These cover the fallbacks used on Windows once the app no longer runs as Administrator:
symlink, directory junction, and hard link, plus link removal that must never delete the
data the link points to.
"""

import os
import sys
import types

import pytest
from manageNeuroconv.manage_neuroconv import _is_link, _remove_link, create_link


def _make_dir_with_file(base, name="data"):
target = base / name
target.mkdir()
(target / "keep.txt").write_text("important")
return target


def _raise_oserror(*args, **kwargs):
raise OSError("unavailable")


def test_create_link_creates_working_directory_link(tmp_path):
target = _make_dir_with_file(tmp_path)
link = tmp_path / "link"

create_link(target, link, target_is_dir=True)

assert _is_link(link)
assert (link / "keep.txt").read_text() == "important"


def test_create_link_creates_working_file_link(tmp_path):
target = tmp_path / "file.txt"
target.write_text("payload")
link = tmp_path / "link.txt"

create_link(target, link, target_is_dir=False)

assert link.read_text() == "payload"


def test_is_link_false_for_real_paths(tmp_path):
real_dir = _make_dir_with_file(tmp_path)
real_file = tmp_path / "plain.txt"
real_file.write_text("x")

assert not _is_link(real_dir)
assert not _is_link(real_file)


def test_remove_link_preserves_directory_target(tmp_path):
"""Removing a directory link must leave the data it points to intact (data-loss regression)."""
target = _make_dir_with_file(tmp_path)
link = tmp_path / "link"
create_link(target, link, target_is_dir=True)

_remove_link(link)

assert not link.exists()
assert not _is_link(link)
assert target.exists()
assert (target / "keep.txt").read_text() == "important"


def test_remove_link_removes_file_link(tmp_path):
target = tmp_path / "file.txt"
target.write_text("payload")
link = tmp_path / "link.txt"
create_link(target, link, target_is_dir=False)

_remove_link(link)

assert not link.exists()
assert target.read_text() == "payload"


def test_create_link_hardlink_fallback_for_files(tmp_path, monkeypatch):
"""On Windows without symlink privileges, files fall back to hard links."""
target = tmp_path / "file.txt"
target.write_text("payload")
link = tmp_path / "link.txt"

monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setattr(os, "symlink", _raise_oserror)

create_link(target, link, target_is_dir=False)

assert link.read_text() == "payload"
assert os.stat(link).st_ino == os.stat(target).st_ino # a hard link shares the inode


def test_create_link_junction_fallback_for_directories(tmp_path, monkeypatch):
"""On Windows without symlink privileges, directories fall back to junctions with an absolute target."""
original_symlink = os.symlink
target = _make_dir_with_file(tmp_path)
link = tmp_path / "link"

recorded = {}

def fake_create_junction(src, dst):
recorded["target"] = src
original_symlink(src, dst, target_is_directory=True)

monkeypatch.setitem(sys.modules, "_winapi", types.SimpleNamespace(CreateJunction=fake_create_junction))
monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setattr(os, "symlink", _raise_oserror)

create_link(target, link, target_is_dir=True)

assert os.path.isabs(recorded["target"])
assert (link / "keep.txt").read_text() == "important"


def test_create_link_raises_helpful_error_when_all_fallbacks_fail(tmp_path, monkeypatch):
target = tmp_path / "file.txt"
target.write_text("payload")
link = tmp_path / "link.txt"

monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setattr(os, "symlink", _raise_oserror)
monkeypatch.setattr(os, "link", _raise_oserror)

with pytest.raises(OSError, match="Administrator"):
create_link(target, link, target_is_dir=False)
Loading