xiaoxia-saas/tests/unit/test_p02_oss_credential_validation.py

"""P0-2 修复：OSS 凭证验证 + 启动诊断。

验证：
1. 非开发环境 OSS_ACCESS_KEY_ID/SECRET 为空时启动失败
2. 开发环境允许空凭证
3. diagnose() 方法正确输出配置状态
"""

import logging
import os
import sys
from unittest.mock import MagicMock, patch

import pytest


def _fresh_settings(env: str):
    """清除 config 模块缓存，以指定 APP_ENV 重新导入 Settings。

    为非开发环境预设 OSS 环境变量，确保模块级 get_settings() 能成功完成导入。
    测试方法内可根据需要清除这些变量来测试验证器。
    """
    for mod_name in [m for m in list(sys.modules) if "app.config" in m]:
        del sys.modules[mod_name]
    os.environ["APP_ENV"] = env
    # 非开发环境下，为模块级导入提供有效凭证（避免导入时验证失败）
    if env != "development":
        os.environ.setdefault("OSS_ACCESS_KEY_ID", "test-key-for-import")
        os.environ.setdefault("OSS_ACCESS_KEY_SECRET", "test-secret-for-import")
    # 重置单例，让测试方法自行控制实例化
    from apps.api.app import config as _cfg
    from apps.api.app.config import Settings

    _cfg._settings = None
    return Settings


class TestOSSCredentialValidation:
    """测试 OSS 凭证验证器（直接调用验证器类方法）。"""

    def test_empty_oss_key_id_rejected_in_staging(self):
        """非开发环境 OSS_ACCESS_KEY_ID 为空应报错。"""
        Settings = _fresh_settings("staging")
        with pytest.raises(Exception, match="OSS_ACCESS_KEY_ID"):
            Settings.validate_oss_access_key_id("")

    def test_empty_oss_key_secret_rejected_in_staging(self):
        """非开发环境 OSS_ACCESS_KEY_SECRET 为空应报错。"""
        Settings = _fresh_settings("staging")
        with pytest.raises(Exception, match="OSS_ACCESS_KEY_SECRET"):
            Settings.validate_oss_access_key_secret("")

    def test_empty_oss_credentials_allowed_in_development(self):
        """开发环境允许空 OSS 凭证。"""
        Settings = _fresh_settings("development")
        assert Settings.validate_oss_access_key_id("") == ""
        assert Settings.validate_oss_access_key_secret("") == ""

    def test_valid_credentials_pass_validation(self):
        """有效凭证应通过验证。"""
        Settings = _fresh_settings("staging")
        assert Settings.validate_oss_access_key_id("test-key-id") == "test-key-id"
        assert Settings.validate_oss_access_key_secret("test-key-secret") == "test-key-secret"

    def test_valid_credentials_instantiation_succeeds(self):
        """有效凭证应能成功创建 Settings 实例。"""
        os.environ.pop("OSS_ACCESS_KEY_ID", None)
        os.environ.pop("OSS_ACCESS_KEY_SECRET", None)
        Settings = _fresh_settings("staging")
        os.environ["OSS_ACCESS_KEY_ID"] = "test-key-id"
        os.environ["OSS_ACCESS_KEY_SECRET"] = "test-key-secret"
        s = Settings(_env_file=None)
        assert s.OSS_ACCESS_KEY_ID == "test-key-id"
        assert s.OSS_ACCESS_KEY_SECRET == "test-key-secret"


class TestOSSDiagnose:
    """测试 SharedStorageService.diagnose() 方法。"""

    @patch("packages.shared.storage.oss2", None)
    @patch("packages.shared.storage.get_shared_settings")
    def test_diagnose_logs_error_when_bucket_none(self, mock_settings, caplog):
        """bucket=None 时 diagnose 应输出 ERROR 日志。"""
        from packages.shared.storage import SharedStorageService

        mock_settings.return_value.oss_bucket_name = "test-bucket"
        mock_settings.return_value.oss_endpoint = "oss-cn-test.com"
        mock_settings.return_value.oss_access_key_id = ""
        mock_settings.return_value.oss_access_key_secret = ""

        service = SharedStorageService()
        assert service.bucket is None

        with caplog.at_level(logging.ERROR, logger="packages.shared.storage"):
            service.diagnose()

        assert any("❌" in record.message for record in caplog.records)

    @patch("packages.shared.storage.oss2")
    @patch("packages.shared.storage.get_shared_settings")
    def test_diagnose_logs_success_when_bucket_configured(self, mock_settings, mock_oss2, caplog):
        """bucket 已配置时 diagnose 应输出成功日志。"""
        from packages.shared.storage import SharedStorageService

        mock_settings.return_value.oss_bucket_name = "test-bucket"
        mock_settings.return_value.oss_endpoint = "oss-cn-test.com"
        mock_settings.return_value.oss_access_key_id = "test-key-id"
        mock_settings.return_value.oss_access_key_secret = "test-key-secret"
        mock_oss2.Bucket.return_value = MagicMock()

        service = SharedStorageService()
        assert service.bucket is not None

        with caplog.at_level(logging.INFO, logger="packages.shared.storage"):
            service.diagnose()

        assert any("OSS诊断" in record.message for record in caplog.records)


class TestOSSHTTPSEndpoint:
    """测试 P0-2 真正根因：sign_url 必须返回 HTTPS URL。"""

    @patch("packages.shared.storage.oss2")
    @patch("packages.shared.storage.get_shared_settings")
    def test_endpoint_without_scheme_gets_https_prefix(self, mock_settings, mock_oss2):
        """endpoint 无 scheme 时应自动加 https://，确保 sign_url 生成 HTTPS URL。"""
        from packages.shared.storage import SharedStorageService

        mock_settings.return_value.oss_bucket_name = "test-bucket"
        mock_settings.return_value.oss_endpoint = "oss-cn-hangzhou.aliyuncs.com"
        mock_settings.return_value.oss_access_key_id = "test-key-id"
        mock_settings.return_value.oss_access_key_secret = "test-key-secret"
        mock_oss2.Bucket.return_value = MagicMock()

        SharedStorageService()

        # 验证传给 oss2.Bucket 的 endpoint 带了 https://
        call_args = mock_oss2.Bucket.call_args
        endpoint_passed = call_args[0][1]  # 第二个位置参数
        assert endpoint_passed == "https://oss-cn-hangzhou.aliyuncs.com"

    @patch("packages.shared.storage.oss2")
    @patch("packages.shared.storage.get_shared_settings")
    def test_endpoint_with_existing_https_not_doubled(self, mock_settings, mock_oss2):
        """endpoint 已有 https:// 时不应重复添加。"""
        from packages.shared.storage import SharedStorageService

        mock_settings.return_value.oss_bucket_name = "test-bucket"
        mock_settings.return_value.oss_endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
        mock_settings.return_value.oss_access_key_id = "test-key-id"
        mock_settings.return_value.oss_access_key_secret = "test-key-secret"
        mock_oss2.Bucket.return_value = MagicMock()

        SharedStorageService()

        call_args = mock_oss2.Bucket.call_args
        endpoint_passed = call_args[0][1]
        assert endpoint_passed == "https://oss-cn-hangzhou.aliyuncs.com"