xiaoxia-saas/tests/unit/test_oss_helpers_pure.py

"""OSS助手纯逻辑测试 — normalize_storage_key / resolve_asset_path 输入校验."""

from __future__ import annotations

import hashlib
import os
from pathlib import Path
from unittest.mock import patch

import pytest
from video_processing.oss_helpers import normalize_storage_key, resolve_asset_path


class TestNormalizeStorageKey:
    """normalize_storage_key 存储键标准化测试."""

    def test_plain_key_passthrough(self):
        """普通路径原样返回."""
        assert normalize_storage_key("path/to/file.mp4") == "path/to/file.mp4"

    def test_https_url_extracts_path(self):
        """HTTPS URL提取path部分."""
        result = normalize_storage_key("https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4")
        assert result == "path/to/file.mp4"

    def test_http_url_extracts_path(self):
        """HTTP URL提取path部分."""
        result = normalize_storage_key("http://example.com/assets/video.mp4")
        assert result == "assets/video.mp4"

    def test_url_with_query_params(self):
        """带query参数的URL只取path."""
        result = normalize_storage_key("https://bucket.oss-cn-hangzhou.aliyuncs.com/file.mp4?token=abc&expires=123")
        assert result == "file.mp4"

    def test_leading_slash_stripped(self):
        """开头斜杠被去掉."""
        assert normalize_storage_key("/path/to/file.mp4") == "path/to/file.mp4"

    def test_url_without_path(self):
        """URL没有path部分返回空字符串."""
        result = normalize_storage_key("https://example.com")
        assert result == ""

    def test_nested_path(self):
        """多层嵌套路径."""
        assert normalize_storage_key("a/b/c/d/file.mp4") == "a/b/c/d/file.mp4"

    def test_empty_string(self):
        """空字符串."""
        assert normalize_storage_key("") == ""

    def test_url_with_port(self):
        """带端口的URL."""
        result = normalize_storage_key("http://localhost:9000/bucket/file.mp4")
        assert result == "bucket/file.mp4"


class TestResolveAssetPathInputValidation:
    """resolve_asset_path 输入校验测试（不涉及真实下载）."""

    def test_empty_string_returns_none(self, tmp_path):
        """空字符串返回None."""
        assert resolve_asset_path("", tmp_path) is None

    def test_none_returns_none(self, tmp_path):
        """None返回None（类型检查）."""
        assert resolve_asset_path(None, tmp_path) is None  # type: ignore

    def test_non_string_returns_none(self, tmp_path):
        """非字符串返回None."""
        assert resolve_asset_path(123, tmp_path) is None  # type: ignore

    def test_null_byte_rejected(self, tmp_path):
        """包含空字节的asset_id被拒绝."""
        assert resolve_asset_path("file\x00.mp4", tmp_path) is None

    def test_path_traversal_rejected(self, tmp_path):
        """包含../的路径遍历攻击被拒绝（第3步下载前检查）."""
        # mock download_asset不被调用，因为路径包含..会直接返回None
        with patch("video_processing.oss_helpers.download_asset") as mock_dl:
            result = resolve_asset_path("../etc/passwd", tmp_path)
            assert result is None
            mock_dl.assert_not_called()

    def test_absolute_path_key_rejected(self, tmp_path):
        """以/开头的存储键在下载前检查被拒."""
        with patch("video_processing.oss_helpers.download_asset") as mock_dl:
            result = resolve_asset_path("/etc/passwd", tmp_path)
            assert result is None
            mock_dl.assert_not_called()

    def test_cache_hit_returns_cached_path(self, tmp_path):
        """缓存命中返回缓存路径."""
        asset_id = "test-asset-123"
        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
        cached_file = tmp_path / f"{cache_hash}.mp4"
        cached_file.write_bytes(b"fake video data")

        result = resolve_asset_path(asset_id, tmp_path)
        assert result == cached_file
        assert result.exists()

    def test_cache_empty_file_not_considered_hit(self, tmp_path):
        """空文件不算缓存命中."""
        asset_id = "empty-cache-file"
        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
        cached_file = tmp_path / f"{cache_hash}.mp4"
        cached_file.touch()  # 空文件

        with patch("video_processing.oss_helpers.download_asset", return_value=False):
            result = resolve_asset_path(asset_id, tmp_path)
            # 空文件不命中缓存，走下载，下载失败返回None
            assert result is None

    def test_download_success_returns_path(self, tmp_path):
        """下载成功返回本地路径."""
        asset_id = "remote-asset"
        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
        expected_path = tmp_path / f"{cache_hash}.mp4"

        def fake_download(storage_key, local_path):
            Path(local_path).write_bytes(b"downloaded data")
            return True

        with patch("video_processing.oss_helpers.download_asset", side_effect=fake_download):
            result = resolve_asset_path(asset_id, tmp_path)
            assert result == expected_path
            assert result.exists()
            assert result.stat().st_size > 0

    def test_download_failure_returns_none(self, tmp_path):
        """下载失败返回None."""
        with patch("video_processing.oss_helpers.download_asset", return_value=False):
            result = resolve_asset_path("nonexistent-asset", tmp_path)
            assert result is None

    def test_work_dir_not_exists_creates_on_demand(self, tmp_path):
        """work_dir不存在时也能处理."""
        asset_id = "new-dir-asset"
        new_dir = tmp_path / "subdir" / "nested"

        with patch("video_processing.oss_helpers.download_asset", return_value=False):
            # 不存在的work_dir，缓存检查也不会命中
            result = resolve_asset_path(asset_id, new_dir)
            assert result is None