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
13 changes: 11 additions & 2 deletions src/wholocks/_linux.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ def add(label):
continue

name, cmdline, user = _proc_identity(pid_dir)
matched_paths = _matched_paths_for(pid_dir, match)
matched_paths = _matched_paths_for(pid_dir, match, target_inodes)
holders.append(
Holder(
pid=pid,
Expand Down Expand Up @@ -223,9 +223,10 @@ def add(label):
)


def _matched_paths_for(pid_dir, match):
def _matched_paths_for(pid_dir, match, target_inodes=None):
"""Collect the concrete paths this process holds (for display)."""
found = []
target_inodes = target_inodes or set()

def note(path):
if path and path not in found:
Expand All @@ -249,4 +250,12 @@ def note(path):
target, _ = _strip_deleted(target)
if match(target):
note(target)
continue
if target_inodes and target.startswith("/"):
try:
st = os.stat(os.path.join(fd_dir, fd))
except OSError:
continue
if (st.st_dev, st.st_ino) in target_inodes:
note(target)
return found
49 changes: 49 additions & 0 deletions tests/test_hardlink_paths.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
"""Linux hardlink holders should list the alternate path in holds:."""

import os
import sys

import pytest

pytestmark = pytest.mark.skipif(
sys.platform.startswith("win"), reason="requires POSIX hardlinks"
)


@pytest.fixture
def fake_proc(tmp_path, monkeypatch):
import wholocks._linux as _linux

proc = tmp_path / "proc"
proc.mkdir()
(proc / "stat").write_text("cpu 1 2 3\nbtime 1700000000\n")
monkeypatch.setattr(_linux, "PROC_ROOT", str(proc))
return proc


def _make_pid(proc_dir, pid, comm="logger"):
d = proc_dir / str(pid)
d.mkdir()
(d / "comm").write_text(comm + "\n")
(d / "cmdline").write_bytes(b"logger\0")
(d / "stat").write_text("%d (%s) S 1 1 1 0 -1 4194304 0 0 0 0 0 0 0 0 20 0 1 0 777 100\n" % (pid, comm))
(d / "maps").write_text("")
(d / "fd").mkdir()
return d


def test_hardlink_fd_lists_alternate_path(fake_proc, tmp_path):
import wholocks._linux as _linux

target = tmp_path / "canonical.log"
target.write_text("data")
alias = tmp_path / "rotated.log"
os.link(target, alias)

pid_dir = _make_pid(fake_proc, 9100)
os.symlink(os.path.realpath(str(alias)), pid_dir / "fd" / "3")

result = _linux.scan([str(target)], [], recursive=False)
assert len(result.holders) == 1
assert "hardlink" in result.holders[0].access[0]
assert os.path.realpath(str(alias)) in result.holders[0].paths
Loading