123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""Path policy: supported extensions, exclusion, deterministic discovery, and
|
|
symlink-escape rejection."""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from photo_pipeline import path_policy
|
|
|
|
|
|
def test_is_supported_is_case_insensitive():
|
|
assert path_policy.is_supported("a.jpg")
|
|
assert path_policy.is_supported("a.JPG")
|
|
assert path_policy.is_supported("a.HEIC")
|
|
assert not path_policy.is_supported("a.txt")
|
|
assert not path_policy.is_supported("a.mp4")
|
|
|
|
|
|
def test_is_excluded_matches_ignore_and_thumb_anywhere():
|
|
assert path_policy.is_excluded("lib/_IGNORE/secret.jpg")
|
|
assert path_policy.is_excluded("lib/album/.@__thumb/t.jpg")
|
|
assert not path_policy.is_excluded("lib/album/photo.jpg")
|
|
|
|
|
|
def test_discovery_finds_supported_and_skips_excluded_and_unsupported(tmp_path):
|
|
(tmp_path / "album").mkdir()
|
|
(tmp_path / "_IGNORE").mkdir()
|
|
(tmp_path / "album" / ".@__thumb").mkdir()
|
|
keep = [tmp_path / "root.jpg", tmp_path / "album" / "a.PNG"]
|
|
for p in keep:
|
|
p.write_bytes(b"x")
|
|
(tmp_path / "notes.txt").write_bytes(b"x")
|
|
(tmp_path / "_IGNORE" / "secret.jpg").write_bytes(b"x")
|
|
(tmp_path / "album" / ".@__thumb" / "thumb.jpg").write_bytes(b"x")
|
|
|
|
found = path_policy.discover([tmp_path])
|
|
assert found == sorted(keep)
|
|
|
|
|
|
def test_discovery_is_deterministic(tmp_path):
|
|
for i in range(5):
|
|
(tmp_path / f"{i}.jpg").write_bytes(b"x")
|
|
assert path_policy.discover([tmp_path]) == path_policy.discover([tmp_path])
|
|
|
|
|
|
def test_resolve_within_accepts_inside_and_rejects_outside(tmp_path):
|
|
root = tmp_path / "lib"
|
|
root.mkdir()
|
|
inside = root / "a.jpg"
|
|
inside.write_bytes(b"x")
|
|
assert path_policy.resolve_within(root, inside) == inside.resolve()
|
|
with pytest.raises(path_policy.PathPolicyError):
|
|
path_policy.resolve_within(root, tmp_path / "outside.jpg")
|
|
|
|
|
|
@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks unsupported")
|
|
def test_symlink_escaping_root_fails_discovery(tmp_path):
|
|
root = tmp_path / "lib"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside.jpg"
|
|
outside.write_bytes(b"x")
|
|
link = root / "link.jpg"
|
|
try:
|
|
os.symlink(outside, link)
|
|
except (OSError, NotImplementedError):
|
|
pytest.skip("cannot create symlink on this platform")
|
|
with pytest.raises(path_policy.PathPolicyError):
|
|
path_policy.discover([root])
|
|
|
|
|
|
@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks unsupported")
|
|
def test_symlink_within_root_is_allowed(tmp_path):
|
|
root = tmp_path / "lib"
|
|
root.mkdir()
|
|
target = root / "real.jpg"
|
|
target.write_bytes(b"x")
|
|
link = root / "alias.jpg"
|
|
try:
|
|
os.symlink(target, link)
|
|
except (OSError, NotImplementedError):
|
|
pytest.skip("cannot create symlink on this platform")
|
|
found = path_policy.discover([root])
|
|
assert target in found and link in found
|
|
|
|
|
|
def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
|
|
"""The resolved path is the answer, not a yes/no: a caller that revalidates one
|
|
name and then opens another has an open symlink race (US07-02)."""
|
|
root = tmp_path / "lib"
|
|
(root / "sub").mkdir(parents=True)
|
|
target = root / "sub" / "real.jpg"
|
|
target.write_bytes(b"x")
|
|
link = root / "alias.jpg"
|
|
os.symlink(target, link)
|
|
|
|
assert path_policy.resolve_in_roots([root], link) == target.resolve()
|
|
assert path_policy.resolve_in_roots([root], target) == target.resolve()
|
|
|
|
|
|
def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
|
|
root = tmp_path / "lib"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside.jpg"
|
|
outside.write_bytes(b"x")
|
|
link = root / "alias.jpg"
|
|
os.symlink(outside, link)
|
|
|
|
for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
|
|
with pytest.raises(path_policy.PathPolicyError) as raised:
|
|
path_policy.resolve_in_roots([root], candidate)
|
|
assert str(outside) not in str(raised.value)
|
|
|
|
|
|
def test_resolve_in_roots_checks_every_configured_root(tmp_path):
|
|
first, second = tmp_path / "one", tmp_path / "two"
|
|
first.mkdir()
|
|
second.mkdir()
|
|
photo = second / "b.jpg"
|
|
photo.write_bytes(b"x")
|
|
assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
|
|
# No configured boundary means nothing to check against.
|
|
assert path_policy.resolve_in_roots([], photo) == photo
|