xiaoxia-saas/tests/unit/test_path_security.py

"""path_security 单元测试."""

import os
import tempfile

import pytest

from apps.worker.video_processing.path_security import (
    LOCAL_SCHEMA_PREFIX,
    MAX_PATH_LENGTH,
    PathSecurityError,
    is_in_allowed_dirs,
    is_path_safe,
    safe_resolve_path,
    sanitize_filename,
    validate_local_schema_path,
)


@pytest.fixture
def base_dir():
    with tempfile.TemporaryDirectory() as tmpdir:
        # 创建一个子文件用于测试
        with open(os.path.join(tmpdir, "test.mp4"), "w") as f:
            f.write("test")
        subdir = os.path.join(tmpdir, "subdir")
        os.makedirs(subdir)
        with open(os.path.join(subdir, "audio.mp3"), "w") as f:
            f.write("test")
        yield tmpdir


# ── safe_resolve_path ────────────────────────────────────────────────────────


class TestSafeResolvePath:
    def test_none_path_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="不能为空"):
            safe_resolve_path(None, base_dir)

    def test_empty_string_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="不能为空"):
            safe_resolve_path("", base_dir)

    def test_whitespace_path_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="不能为空"):
            safe_resolve_path("   ", base_dir)

    def test_too_long_path_raises(self, base_dir):
        long_path = "a" * (MAX_PATH_LENGTH + 1)
        with pytest.raises(PathSecurityError, match="路径过长"):
            safe_resolve_path(long_path, base_dir)

    def test_null_byte_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="空字节"):
            safe_resolve_path("file\x00.mp4", base_dir)

    def test_relative_path_within_base(self, base_dir):
        result = safe_resolve_path("test.mp4", base_dir)
        assert result.name == "test.mp4"
        assert str(result).startswith(str(os.path.realpath(base_dir)))

    def test_subdirectory_path(self, base_dir):
        result = safe_resolve_path("subdir/audio.mp3", base_dir)
        assert result.name == "audio.mp3"
        assert "subdir" in str(result)

    def test_parent_traversal_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="路径遍历"):
            safe_resolve_path("../etc/passwd", base_dir)

    def test_nested_parent_traversal_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="路径遍历"):
            safe_resolve_path("subdir/../../etc/passwd", base_dir)

    def test_absolute_path_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="绝对路径"):
            safe_resolve_path("/etc/passwd", base_dir)

    def test_absolute_path_with_allow_outside(self, base_dir):
        # allow_outside=True 时允许绝对路径（但会被危险路径模式检查）
        with pytest.raises(PathSecurityError, match="系统路径"):
            safe_resolve_path("/etc/passwd", base_dir, allow_outside=True)

    def test_local_schema_relative(self, base_dir):
        result = safe_resolve_path("local://test.mp4", base_dir)
        assert result.name == "test.mp4"
        assert str(result).startswith(str(os.path.realpath(base_dir)))

    def test_local_schema_absolute_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="绝对路径"):
            safe_resolve_path("local:///etc/passwd", base_dir)

    def test_local_schema_traversal_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="路径遍历"):
            safe_resolve_path("local://../secret", base_dir)

    def test_invalid_base_dir_raises(self):
        with pytest.raises(PathSecurityError, match="基路径"):
            safe_resolve_path("file.txt", "/nonexistent/dir")

    def test_allowed_extensions_valid(self, base_dir):
        result = safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp4"})
        assert result.suffix.lower() == ".mp4"

    def test_allowed_extensions_invalid_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="文件类型"):
            safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp3"})

    def test_no_extension_restriction(self, base_dir):
        # allowed_extensions=None 时不检查
        result = safe_resolve_path("test.mp4", base_dir, allowed_extensions=None)
        assert result is not None

    def test_path_object_input(self, base_dir):
        from pathlib import Path

        result = safe_resolve_path(Path("test.mp4"), base_dir)
        assert result.name == "test.mp4"

    def test_path_object_base_dir(self, base_dir):
        from pathlib import Path

        result = safe_resolve_path("test.mp4", Path(base_dir))
        assert result.name == "test.mp4"


# ── is_path_safe ────────────────────────────────────────────────────────────


class TestIsPathSafe:
    def test_safe_path_returns_true(self, base_dir):
        assert is_path_safe("test.mp4", base_dir) is True

    def test_unsafe_path_returns_false(self, base_dir):
        assert is_path_safe("../etc/passwd", base_dir) is False

    def test_none_returns_false(self, base_dir):
        assert is_path_safe(None, base_dir) is False


# ── validate_local_schema_path ──────────────────────────────────────────────


class TestValidateLocalSchemaPath:
    def test_valid_local_path(self, base_dir):
        result = validate_local_schema_path("local://test.mp4", base_dir)
        assert result.name == "test.mp4"

    def test_missing_prefix_raises(self, base_dir):
        with pytest.raises(PathSecurityError, match="开头"):
            validate_local_schema_path("test.mp4", base_dir)

    def test_traversal_raises(self, base_dir):
        with pytest.raises(PathSecurityError):
            validate_local_schema_path("local://../secret", base_dir)

    def test_absolute_path_raises(self, base_dir):
        with pytest.raises(PathSecurityError):
            validate_local_schema_path("local:///etc/passwd", base_dir)


# ── sanitize_filename ───────────────────────────────────────────────────────


class TestSanitizeFilename:
    def test_normal_filename(self):
        assert sanitize_filename("hello.mp4") == "hello.mp4"

    def test_empty_returns_unnamed(self):
        assert sanitize_filename("") == "unnamed"

    def test_none_default(self):
        # 空字符串会返回unnamed
        assert sanitize_filename("") == "unnamed"

    def test_removes_path_separators(self):
        assert "/" not in sanitize_filename("path/to/file.mp4")
        assert "\\" not in sanitize_filename("path\\to\\file.mp4")

    def test_removes_control_characters(self):
        result = sanitize_filename("file\x01\x02name.mp4")
        assert "\x01" not in result
        assert "\x02" not in result

    def test_removes_dangerous_chars(self):
        result = sanitize_filename("file<name>.mp4")
        assert "<" not in result
        assert ">" not in result

    def test_removes_leading_dots(self):
        assert not sanitize_filename(".hidden").startswith(".")
        assert not sanitize_filename("..hidden").startswith(".")

    def test_chinese_characters_preserved(self):
        result = sanitize_filename("视频文件.mp4")
        assert "视频文件" in result

    def test_long_filename_truncated(self):
        long_name = "a" * 300 + ".mp4"
        result = sanitize_filename(long_name)
        assert len(result) <= 255
        assert result.endswith(".mp4")

    def test_spaces_preserved(self):
        result = sanitize_filename("my file.mp4")
        assert "my file.mp4" == result

    def test_underscores_hyphens_preserved(self):
        result = sanitize_filename("my_file-name.mp4")
        assert result == "my_file-name.mp4"

    def test_all_dots_returns_unnamed(self):
        assert sanitize_filename("...") == "unnamed"


# ── is_in_allowed_dirs ──────────────────────────────────────────────────────


class TestIsInAllowedDirs:
    def test_path_in_allowed_dir(self, base_dir):
        filepath = os.path.join(base_dir, "test.mp4")
        from pathlib import Path

        assert is_in_allowed_dirs(filepath, [Path(base_dir)]) is True

    def test_path_not_in_allowed_dir(self, base_dir):
        from pathlib import Path

        assert is_in_allowed_dirs("/etc/passwd", [Path(base_dir)]) is False

    def test_subdirectory_in_allowed(self, base_dir):
        from pathlib import Path

        sub = os.path.join(base_dir, "subdir", "audio.mp3")
        assert is_in_allowed_dirs(sub, [Path(base_dir)]) is True

    def test_none_allowed_dirs_uses_default(self):
        # None 使用默认配置（包含 /tmp）
        result = is_in_allowed_dirs("/tmp/test.mp4")
        assert isinstance(result, bool)

    def test_allowed_dirs_list_is_empty(self):
        from pathlib import Path

        assert is_in_allowed_dirs("/tmp/test", []) is False


# ── PathSecurityError class ─────────────────────────────────────────────────


class TestPathSecurityError:
    def test_is_value_error(self):
        assert issubclass(PathSecurityError, ValueError)

    def test_message_preserved(self):
        err = PathSecurityError("test message")
        assert str(err) == "test message"