Skip to content
Open
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
92 changes: 92 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from __future__ import annotations

import os
import socket
import tempfile
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -47,6 +49,96 @@ def requires_symlinks(_can_symlink) -> None:
)


@pytest.fixture(scope="session")
def _can_mkfifo() -> bool:
"""Whether this machine can create a FIFO (#2919).

Probed, not inferred from ``sys.platform``, for the same reason as
``_can_symlink``: ``os.mkfifo`` is absent on Windows, but it can also fail
on a POSIX host whose temp dir sits on a filesystem that has no FIFOs
(some network and container mounts). Either way an ``AttributeError`` or
``OSError`` raised while *building the fixture* says nothing about the code
under test.
"""
if not hasattr(os, "mkfifo"):
return False
with tempfile.TemporaryDirectory() as d:
try:
os.mkfifo(Path(d) / "probe-fifo")
except (OSError, NotImplementedError):
return False
return True


@pytest.fixture
def requires_fifo(_can_mkfifo) -> None:
"""Skip a test whose fixture is a named pipe when the platform has none."""
if not _can_mkfifo:
pytest.skip("named pipes (os.mkfifo) unavailable on this platform")


@pytest.fixture(scope="session")
def _can_bind_unix_socket() -> bool:
"""Whether this machine can bind an ``AF_UNIX`` socket to a path (#2919).

Windows 10+ does support ``AF_UNIX``, and CPython exposes it on some
builds, so this cannot be decided from the platform name either — probe and
let the hosts that can do it keep the coverage.
"""
if not hasattr(socket, "AF_UNIX"):
return False
with tempfile.TemporaryDirectory() as d:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
sock.bind(str(Path(d) / "probe.sock"))
except (OSError, NotImplementedError):
return False
finally:
sock.close()
return True


@pytest.fixture
def requires_unix_socket(_can_bind_unix_socket) -> None:
"""Skip a test that must bind an AF_UNIX socket where that is unsupported."""
if not _can_bind_unix_socket:
pytest.skip("AF_UNIX sockets unavailable on this platform")


@pytest.fixture(scope="session")
def _can_delete_cwd() -> bool:
"""Whether a directory can be removed while it is a process's CWD (#2919).

POSIX unlinks the directory entry and leaves the process sitting on an
orphaned inode — the state a detached hook inherits, and the one
``_rebuild_code`` has to survive. Windows keeps an open handle on the CWD,
so ``rmdir`` raises ``PermissionError: [WinError 32]`` and the scenario
cannot be constructed at all. The failure is in the fixture, not the code.
"""
old = Path.cwd()
with tempfile.TemporaryDirectory() as d:
probe = Path(d) / "probe-cwd"
probe.mkdir()
try:
os.chdir(probe)
probe.rmdir()
except OSError:
return False
finally:
os.chdir(old)
return True


@pytest.fixture
def requires_deletable_cwd(_can_delete_cwd) -> None:
"""Skip a test that must delete its own CWD where the OS forbids it."""
if not _can_delete_cwd:
pytest.skip(
"cannot remove a directory that is the process CWD on this platform "
"(Windows holds an open handle: WinError 32)"
)


@pytest.fixture(autouse=True)
def _sandbox_home(tmp_path_factory, monkeypatch):
"""Every test gets a throwaway HOME so installers/uninstallers can never
Expand Down
10 changes: 5 additions & 5 deletions tests/test_non_regular_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,15 @@ def test_regular_source_file_is_accepted(tree):
assert _is_regular_file(tree / "src" / "module.py") is True


def test_fifo_is_rejected(tree):
def test_fifo_is_rejected(tree, requires_fifo):
"""The shape that hangs the whole run."""
fifo = tree / "src" / "pipe.py"
os.mkfifo(fifo)
assert stat.S_ISFIFO(os.stat(fifo).st_mode)
assert _is_regular_file(fifo) is False


def test_unix_socket_is_rejected(tree):
def test_unix_socket_is_rejected(tree, requires_unix_socket):
sock_path = tree / "src" / "sock.py"
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
Expand All @@ -58,14 +58,14 @@ def test_directory_named_like_a_source_file_is_rejected(tree):
assert _is_regular_file(d) is False


def test_symlink_to_a_regular_file_is_accepted(tree):
def test_symlink_to_a_regular_file_is_accepted(tree, requires_symlinks):
target = tree / "src" / "module.py"
link = tree / "src" / "alias.py"
link.symlink_to(target)
assert _is_regular_file(link) is True


def test_symlink_pointing_at_a_fifo_is_rejected(tree):
def test_symlink_pointing_at_a_fifo_is_rejected(tree, requires_fifo, requires_symlinks):
"""A link to a FIFO blocks exactly like the FIFO, so stat must follow it."""
fifo = tree / "src" / "real.py"
os.mkfifo(fifo)
Expand All @@ -74,7 +74,7 @@ def test_symlink_pointing_at_a_fifo_is_rejected(tree):
assert _is_regular_file(link) is False


def test_broken_symlink_is_rejected_without_raising(tree):
def test_broken_symlink_is_rejected_without_raising(tree, requires_symlinks):
link = tree / "src" / "dangling.py"
link.symlink_to(tree / "src" / "does-not-exist.py")
assert _is_regular_file(link) is False
Expand Down
8 changes: 6 additions & 2 deletions tests/test_watch.py
Original file line number Diff line number Diff line change
Expand Up @@ -651,7 +651,9 @@ def test_graphify_root_preserves_absolute_when_user_supplied(tmp_path):
)


def test_rebuild_code_deleted_cwd_without_repo_root_returns_false(tmp_path, monkeypatch, capsys):
def test_rebuild_code_deleted_cwd_without_repo_root_returns_false(
tmp_path, monkeypatch, capsys, requires_deletable_cwd
):
"""Detached hooks can inherit a CWD that no longer exists.

Without GRAPHIFY_REPO_ROOT, the rebuild should fail cleanly before creating
Expand All @@ -675,7 +677,9 @@ def test_rebuild_code_deleted_cwd_without_repo_root_returns_false(tmp_path, monk
assert "current working directory no longer exists" in out


def test_rebuild_code_deleted_cwd_uses_graphify_repo_root(tmp_path, monkeypatch):
def test_rebuild_code_deleted_cwd_uses_graphify_repo_root(
tmp_path, monkeypatch, requires_deletable_cwd
):
"""GRAPHIFY_REPO_ROOT lets detached hook rebuilds recover from a deleted CWD."""
from graphify.watch import _rebuild_code

Expand Down