xiaoxia-saas/tests/unit/test_p02_worker_oss_fix.py

"""P0-2 深度修复：Worker 端 OSS 工具函数测试.

测试：
1. oss_bucket() endpoint 自动补 https:// 前缀
2. get_signed_download_url() 生成预签名 URL
3. upload_to_oss() 返回 HTTPS URL
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock, patch

# ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────


class TestOSSBucketEndpointScheme:
    """测试 oss_bucket() 自动为 endpoint 补 https:// 前缀."""

    def test_endpoint_without_scheme_adds_https(self):
        """endpoint 不带 scheme 时，自动补 https://."""
        from video_processing.oss_helpers import oss_bucket

        mock_bucket_instance = MagicMock()
        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
        ):
            bucket = oss_bucket()

            assert bucket is mock_bucket_instance
            # 验证 endpoint 传的是带 https:// 的
            call_args = mock_bucket_cls.call_args
            endpoint_arg = call_args[0][1]  # 第 2 个位置参数是 endpoint
            assert endpoint_arg.startswith("https://"), f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
            assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg

    def test_endpoint_with_https_keeps_as_is(self):
        """endpoint 已有 https:// 时，不重复添加."""
        from video_processing.oss_helpers import oss_bucket

        mock_bucket_instance = MagicMock()
        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
        ):
            oss_bucket()

            call_args = mock_bucket_cls.call_args
            endpoint_arg = call_args[0][1]
            # 不应该出现 https://https:// 这种双重前缀
            assert endpoint_arg.count("https://") == 1
            assert endpoint_arg == "https://oss-cn-hangzhou.aliyuncs.com"

    def test_endpoint_with_http_keeps_as_is(self):
        """endpoint 已有 http:// 时，不修改（保留用户选择）."""
        from video_processing.oss_helpers import oss_bucket

        mock_bucket_instance = MagicMock()
        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "http://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
        ):
            oss_bucket()

            call_args = mock_bucket_cls.call_args
            endpoint_arg = call_args[0][1]
            assert endpoint_arg == "http://oss-cn-hangzhou.aliyuncs.com"

    def test_missing_credentials_returns_none(self):
        """凭证缺失时返回 None."""
        from video_processing.oss_helpers import oss_bucket

        with patch("video_processing.oss_helpers.oss_settings", return_value=None):
            bucket = oss_bucket()
            assert bucket is None


# ── get_signed_download_url ──────────────────────────────────────────────────


class TestGetSignedDownloadUrl:
    """测试 get_signed_download_url() 预签名 URL 生成."""

    def test_returns_signed_url_with_storage_key(self):
        """传入 storage key 时，调用 sign_url 并返回结果."""
        from video_processing.oss_helpers import get_signed_download_url

        mock_bucket = MagicMock()
        mock_bucket.sign_url.return_value = (
            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
            "?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
        )

        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
        ):
            result = get_signed_download_url("generated/test.mp4", expires_seconds=3600)

            assert result is not None
            assert "Signature=" in result
            mock_bucket.sign_url.assert_called_once_with("GET", "generated/test.mp4", 3600)

    def test_normalizes_full_url_to_storage_key(self):
        """传入完整 URL 时，提取 storage key 再生成签名."""
        from video_processing.oss_helpers import get_signed_download_url

        mock_bucket = MagicMock()
        mock_bucket.sign_url.return_value = (
            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
        )

        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
        ):
            get_signed_download_url("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4")

            mock_bucket.sign_url.assert_called_once()
            # 验证传给 sign_url 的是纯 storage key，不是完整 URL
            call_key = mock_bucket.sign_url.call_args[0][1]
            assert not call_key.startswith("http")
            assert call_key == "generated/test.mp4"

    def test_returns_none_when_bucket_none(self):
        """bucket 为 None 时返回 None（不抛异常）."""
        from video_processing.oss_helpers import get_signed_download_url

        with patch("video_processing.oss_helpers.oss_settings", return_value=None):
            result = get_signed_download_url("generated/test.mp4")
            assert result is None

    def test_sign_url_exception_returns_none(self):
        """sign_url 抛异常时，返回 None（不向上抛出）."""
        from video_processing.oss_helpers import get_signed_download_url

        mock_bucket = MagicMock()
        mock_bucket.sign_url.side_effect = Exception("sign failed")

        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
        ):
            result = get_signed_download_url("generated/test.mp4")
            assert result is None


# ── upload_to_oss 返回 HTTPS URL ─────────────────────────────────────────────


class TestUploadToOSSReturnsHTTPS:
    """测试 upload_to_oss() 返回的 URL 始终是 HTTPS."""

    def test_endpoint_without_scheme_returns_https_url(self):
        """endpoint 不带 scheme 时，返回 HTTPS URL."""
        from video_processing.oss_helpers import upload_to_oss

        mock_bucket = MagicMock()
        mock_bucket.put_object_from_file = MagicMock()

        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
        ):
            result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")

            assert result is not None
            assert result.startswith("https://")
            assert "test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" in result

    def test_endpoint_with_https_returns_clean_url(self):
        """endpoint 带 https:// 时，URL 里不会有双重 https."""
        from video_processing.oss_helpers import upload_to_oss

        mock_bucket = MagicMock()
        mock_bucket.put_object_from_file = MagicMock()

        with (
            patch(
                "video_processing.oss_helpers.oss_settings",
                return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
            ),
            patch("video_processing.oss_helpers.oss2.Auth"),
            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
        ):
            result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")

            assert result is not None
            assert result.startswith("https://")
            # 不应该出现 https://https://
            assert result.count("https://") == 1