Add Windows-specific file handle helper for testing sharing violations

Author: Евгений Блинов

tests/helpers.py

@@ -1,12 +1,56 @@
+from contextlib import ExitStack, contextmanager
 from io import BytesIO
 from pathlib import Path
+from sys import platform
 from tarfile import TarInfo
 from tarfile import open as open_tar
 from typing import Dict, Iterable, Optional
 
 from dirstree import Crawler
 from emptylog.call_data import LoggerCallData
 
+if platform == 'win32':
+    from ctypes import (  # type: ignore[attr-defined]  # POSIX typeshed omits the Windows-only API.
+        WinDLL,
+        c_uint32,
+        c_void_p,
+        c_wchar_p,
+        get_last_error,
+    )
+
+
+@contextmanager
+def hold_windows_path_open(path: Path, *, share_mode: int, flags: int):
+    """
+    Keep a Windows filesystem object open with explicit sharing permissions.
+
+    This helper is called only by Windows-only tests.  It uses the native API
+    because Python's regular ``open`` does not let tests choose delete sharing.
+    """
+    if platform != 'win32':
+        raise RuntimeError('Windows handles can only be held open on Windows.')
+
+    kernel32 = WinDLL('kernel32', use_last_error=True)
+    create_file = kernel32.CreateFileW
+    create_file.argtypes = [c_wchar_p, c_uint32, c_uint32, c_void_p, c_uint32, c_uint32, c_void_p]
+    create_file.restype = c_void_p
+    close_handle = kernel32.CloseHandle
+    close_handle.argtypes = [c_void_p]
+    close_handle.restype = c_uint32
+
+    generic_read = 0x80000000
+    open_existing = 3
+    invalid_handle = c_void_p(-1).value
+    handle = create_file(str(path), generic_read, share_mode, None, open_existing, flags, None)
+
+    if handle == invalid_handle:
+        raise OSError(get_last_error(), f'Could not open Windows lock target: {path}')
+
+    with ExitStack() as resources:
+        resources.callback(close_handle, handle)
+
+        yield
+
 
 def assert_any_message_contains(calls: Iterable[LoggerCallData], *chunks: str) -> None:
     """Assert that at least one logged message contains all chunks, case-insensitively."""

tests/plugins/test_directory_isolate.py

@@ -21,6 +21,7 @@
 
 from tests.helpers import (
     assert_any_message_contains,
+    hold_windows_path_open,
     make_tar_bytes,
     read_tree,
 )
@@ -42,6 +43,10 @@
 )
 
 VENV_PYTHON_RELATIVE_PATH = Path('Scripts') / 'python.exe' if os_name == 'nt' else Path('bin') / 'python'
+WINDOWS_FILE_ATTRIBUTE_NORMAL = 0x00000080
+WINDOWS_FILE_SHARE_READ = 0x00000001
+WINDOWS_FILE_SHARE_WRITE = 0x00000002
+WINDOWS_SHARING_VIOLATION_REASON = 'The process cannot access the file because it is being used by another process'
 
 
 class TemporaryIsolateFactory(Protocol):
@@ -2397,6 +2402,36 @@ def test_load_permission_denied_write(request, temporary_isolate):
     assert_any_message_contains(logger.data.exception, 'archive', 'failed')
 
 
+@pytest.mark.skipif(os_name != 'nt', reason='Windows sharing violations are not available on POSIX')
+def test_load_windows_locked_file_reports_backup_failure_and_retains_old_data(temporary_isolate):
+    """
+    Verify that a native Windows commit failure logs an archive failure and preserves live data.
+
+    A readable handle that denies delete sharing lets rollback copy the existing
+    file into backup before deletion fails, exercising the common rollback path
+    that must ignore the incomplete backup copy.
+    """
+    isolate = temporary_isolate()
+    existing_file = isolate.directory / 'old.txt'
+    existing_file.write_text('old')
+    logger = MemoryLogger()
+
+    with hold_windows_path_open(
+        existing_file,
+        share_mode=WINDOWS_FILE_SHARE_READ | WINDOWS_FILE_SHARE_WRITE,
+        flags=WINDOWS_FILE_ATTRIBUTE_NORMAL,
+    ), pytest.raises(
+        ArchiveUnpackError,
+        match=match(f'Archive commit failed; rollback attempted. Cause: {WINDOWS_SHARING_VIOLATION_REASON}.'),
+    ) as raised:
+        isolate.load(make_tar_bytes({'new.txt': b'new'}), logger=logger)
+
+    assert 'Unrestored paths:' not in str(raised.value)
+    assert existing_file.read_text() == 'old'
+    assert not (isolate.directory / 'new.txt').exists()
+    assert_any_message_contains(logger.data.exception, 'archive', 'failed')
+
+
 @pytest.mark.skipif(os_name == 'nt', reason='permission mode semantics differ on Windows')
 def test_dump_permission_denied_read(request, temporary_isolate):
     """Verify that a read failure during dump propagates the read error and logs the failure."""
@@ -2412,3 +2447,25 @@ def test_dump_permission_denied_read(request, temporary_isolate):
         isolate.dump(logger=logger)
 
     assert_any_message_contains(logger.data.exception, 'dump', 'failed')
+
+
+@pytest.mark.skipif(os_name != 'nt', reason='Windows sharing violations are not available on POSIX')
+def test_dump_windows_locked_file_propagates_read_failure_and_logs_failure(temporary_isolate):
+    """
+    Verify that a native Windows read failure during dump is propagated and logged.
+
+    An exclusive native handle prevents the archive operation from reading a
+    regular isolate file, so dump must fail instead of returning partial bytes.
+    """
+    isolate = temporary_isolate()
+    unreadable = isolate.directory / 'unreadable.txt'
+    unreadable.write_text('secret')
+    logger = MemoryLogger()
+
+    with hold_windows_path_open(unreadable, share_mode=0, flags=WINDOWS_FILE_ATTRIBUTE_NORMAL), pytest.raises(
+        PermissionError,
+        match=match(f"[Errno 13] Permission denied: '{unreadable}'"),
+    ):
+        isolate.dump(logger=logger)
+
+    assert_any_message_contains(logger.data.exception, 'dump', 'failed')

tests/plugins/test_temporary_directory_throng.py

@@ -15,7 +15,7 @@
 from locklib import LockTraceWrapper
 from suby.subprocess_result import SubprocessResult
 
-from tests.helpers import make_tar_bytes
+from tests.helpers import hold_windows_path_open, make_tar_bytes
 from throng import (
     InvalidBaseDirectoryError,
     IsolateDeletedError,
@@ -28,6 +28,11 @@
     TemporaryDirectoryThrong,
 )
 
+WINDOWS_DIRECTORY_HANDLE_FLAGS = 0x02000000
+WINDOWS_FILE_SHARE_READ = 0x00000001
+WINDOWS_FILE_SHARE_WRITE = 0x00000002
+WINDOWS_SHARING_VIOLATION_REASON = 'The process cannot access the file because it is being used by another process'
+
 
 def assert_isolate_deleted(operation):
     with pytest.raises(IsolateDeletedError, match=match('Isolate has been deleted.')):
@@ -136,6 +141,29 @@ def test_temp_base_not_writable(tmp_path, request):
     ]
 
 
+@pytest.mark.skipif(os_name != 'nt', reason='Windows read-only directory attributes do not apply on POSIX')
+def test_temp_base_read_only_on_windows_is_rejected_and_logged(tmp_path, request):
+    """
+    Verify that a Windows read-only temporary base directory is rejected and logged.
+
+    Setting the Windows read-only attribute makes the configured base fail the
+    plugin's ``W_OK`` check, so no temporary isolate may be created inside it.
+    """
+    base_directory = tmp_path / 'base'
+    base_directory.mkdir()
+    request.addfinalizer(lambda: base_directory.chmod(S_IREAD | S_IWRITE | S_IEXEC))
+    base_directory.chmod(S_IREAD)
+    config = TemporaryDirectoryIsolationConfig(base_directory=str(base_directory))
+    logger = MemoryLogger()
+
+    with pytest.raises(InvalidBaseDirectoryError, match=match(f'Temporary base directory is not writable: {base_directory}')):
+        TemporaryDirectoryThrong(logger=logger, config=config).get_isolate()
+
+    assert [str(call.message) for call in logger.data.error] == [
+        f'Temporary base directory is not writable: {base_directory}',
+    ]
+
+
 def test_temp_isolates_are_distinct(tmp_path):
     """Verify that temporary isolates get separate directories and do not share files."""
     config = TemporaryDirectoryIsolationConfig(base_directory=str(tmp_path))
@@ -373,6 +401,38 @@ def test_temp_delete_failure_raises_and_does_not_log_success(tmp_path, request):
     assert [str(call.message) for call in logger.data.info].count('Delete completed successfully.') == 1
 
 
+@pytest.mark.skipif(os_name != 'nt', reason='Windows sharing violations are not available on POSIX')
+def test_temp_delete_locked_windows_directory_raises_and_can_be_retried(tmp_path):
+    """
+    Verify that a native Windows delete failure is logged and leaves deletion retryable.
+
+    An open directory handle without delete sharing blocks the first cleanup;
+    after the handle closes, the same isolate must be deletable successfully.
+    """
+    logger = MemoryLogger()
+    throng = TemporaryDirectoryThrong(logger=logger, config=TemporaryDirectoryIsolationConfig(base_directory=str(tmp_path)))
+    isolate = throng.get_isolate()
+    isolate_directory = isolate.directory
+    denied_path = isolate_directory if version_info >= (3, 12) else str(isolate_directory)
+    permission_error_message = f'[WinError 32] {WINDOWS_SHARING_VIOLATION_REASON}: {denied_path!r}'
+
+    with hold_windows_path_open(
+        isolate_directory,
+        share_mode=WINDOWS_FILE_SHARE_READ | WINDOWS_FILE_SHARE_WRITE,
+        flags=WINDOWS_DIRECTORY_HANDLE_FLAGS,
+    ), pytest.raises(PermissionError, match=match(permission_error_message)):
+        isolate.delete()
+
+    assert isolate_directory.exists()
+    assert any('Delete failed' in str(call.message) for call in logger.data.exception)
+    assert all(str(call.message) != 'Delete completed successfully.' for call in logger.data.info)
+
+    isolate.delete()
+
+    assert not isolate_directory.exists()
+    assert [str(call.message) for call in logger.data.info].count('Delete completed successfully.') == 1
+
+
 def test_temp_delete_after_delete_raises(tmp_path):
     """Verify that a second delete call is rejected and logged as a delete operation."""
     logger = MemoryLogger()