SMANAGE/gvsds/tests/test_storage.py

"""模块：storage.py 单元测试

覆盖 _NormalizeKey 路径穿越防护 + LocalBackend 的
Upload / UploadFile / Delete / Exists / SignedUrl 完整流程。
CosBackend / OssBackend 需真实云凭证，由集成测试覆盖。
"""
import io
import os
import shutil
import tempfile
from pathlib import Path
from unittest import mock

from django.test import SimpleTestCase, override_settings

from gvsds import storage as _ShimStorage  # noqa: F401  兼容转发层：为 SDK 注入 app.ini 配置提供者
from gvsdsdk import storage


class NormalizeKeyTestCase(SimpleTestCase):
    """_NormalizeKey 路径安全规范化。"""

    def testNormalizeKey_StripsLeadingSlash(self):
        self.assertEqual(storage._NormalizeKey('/assets/abc.jpg'), 'assets/abc.jpg')

    def testNormalizeKey_MultipleLeadingSlashes(self):
        self.assertEqual(storage._NormalizeKey('//assets/abc.jpg'), 'assets/abc.jpg')

    def testNormalizeKey_NormalizesBackslashes(self):
        self.assertEqual(storage._NormalizeKey('assets\\sub\\abc.jpg'), 'assets/sub/abc.jpg')

    def testNormalizeKey_RejectsEmpty(self):
        with self.assertRaises(ValueError):
            storage._NormalizeKey('')

    def testNormalizeKey_RejectsNone(self):
        with self.assertRaises(ValueError):
            storage._NormalizeKey(None)

    def testNormalizeKey_RejectsNonString(self):
        with self.assertRaises(ValueError):
            storage._NormalizeKey(123)

    def testNormalizeKey_RejectsParentTraversal(self):
        with self.assertRaises(ValueError):
            storage._NormalizeKey('../etc/passwd')

    def testNormalizeKey_RejectsEmbeddedParentTraversal(self):
        with self.assertRaises(ValueError):
            storage._NormalizeKey('assets/../../etc/passwd')

    def testNormalizeKey_AcceptsNestedPath(self):
        self.assertEqual(
            storage._NormalizeKey('assets/2024/06/abc.jpg'),
            'assets/2024/06/abc.jpg',
        )


class LocalBackendTestCase(SimpleTestCase):
    """LocalBackend 本地磁盘存储。"""

    def setUp(self):
        # 创建临时 MEDIA_ROOT
        self._tmpdir = tempfile.mkdtemp(prefix='gvsds_test_storage_')
        self._patcher = override_settings(MEDIA_ROOT=self._tmpdir)
        self._patcher.enable()
        # 重置 Storage 单例缓存
        storage.Storage._Backend = None

    def tearDown(self):
        self._patcher.disable()
        shutil.rmtree(self._tmpdir, ignore_errors=True)
        storage.Storage._Backend = None

    def _MakeBackend(self):
        """构造 LocalBackend，绕过 app.ini 读取。"""
        backend = storage.LocalBackend.__new__(storage.LocalBackend)
        backend.BaseDir = Path(self._tmpdir)
        backend.BaseUrl = '/media/'
        backend.BaseDir.mkdir(parents=True, exist_ok=True)
        return backend

    def testUpload_Bytes_WritesFile(self):
        backend = self._MakeBackend()
        url = backend.Upload('assets/abc.txt', b'hello world')
        self.assertEqual(url, '/media/assets/abc.txt')
        FilePath = Path(self._tmpdir) / 'assets' / 'abc.txt'
        self.assertTrue(FilePath.exists())
        self.assertEqual(FilePath.read_bytes(), b'hello world')

    def testUpload_FileLike_ReadsAndWrites(self):
        backend = self._MakeBackend()
        FileObj = io.BytesIO(b'file content')
        url = backend.Upload('assets/file.txt', FileObj)
        self.assertEqual(url, '/media/assets/file.txt')
        FilePath = Path(self._tmpdir) / 'assets' / 'file.txt'
        self.assertEqual(FilePath.read_bytes(), b'file content')

    def testUpload_RejectsNonBytes(self):
        backend = self._MakeBackend()
        with self.assertRaises(RuntimeError):
            backend.Upload('assets/abc.txt', 'not bytes')

    def testUpload_NestedPath_CreatesDirs(self):
        backend = self._MakeBackend()
        backend.Upload('a/b/c/d.txt', b'deep')
        self.assertTrue((Path(self._tmpdir) / 'a' / 'b' / 'c' / 'd.txt').exists())

    def testUpload_RejectsPathTraversal(self):
        backend = self._MakeBackend()
        with self.assertRaises(RuntimeError):
            backend.Upload('../escape.txt', b'evil')

    def testExists_ExistingFile_True(self):
        backend = self._MakeBackend()
        backend.Upload('assets/abc.txt', b'data')
        self.assertTrue(backend.Exists('assets/abc.txt'))

    def testExists_MissingFile_False(self):
        backend = self._MakeBackend()
        self.assertFalse(backend.Exists('assets/nope.txt'))

    def testExists_RejectsPathTraversal(self):
        backend = self._MakeBackend()
        # 路径穿越在 _NormalizeKey 阶段被拒，Exists 包装为 RuntimeError
        with self.assertRaises(RuntimeError):
            backend.Exists('../escape.txt')

    def testDelete_ExistingFile_Removes(self):
        backend = self._MakeBackend()
        backend.Upload('assets/abc.txt', b'data')
        FilePath = Path(self._tmpdir) / 'assets' / 'abc.txt'
        self.assertTrue(FilePath.exists())
        backend.Delete('assets/abc.txt')
        self.assertFalse(FilePath.exists())

    def testDelete_MissingFile_NoError(self):
        backend = self._MakeBackend()
        # 删除不存在的文件不应抛异常
        backend.Delete('assets/never.txt')

    def testDelete_RejectsPathTraversal(self):
        backend = self._MakeBackend()
        with self.assertRaises(RuntimeError):
            backend.Delete('../escape.txt')

    def testSignedUrl_ReturnsBaseUrl(self):
        backend = self._MakeBackend()
        url = backend.SignedUrl('assets/abc.txt', Expires=3600)
        # local 后端忽略 Expires，直接返回公开 URL
        self.assertEqual(url, '/media/assets/abc.txt')


class StorageFacadeTestCase(SimpleTestCase):
    """Storage 单例门面。"""

    def setUp(self):
        self._tmpdir = tempfile.mkdtemp(prefix='gvsds_test_facade_')
        self._patcher = override_settings(MEDIA_ROOT=self._tmpdir)
        self._patcher.enable()
        storage.Storage._Backend = None

    def tearDown(self):
        self._patcher.disable()
        shutil.rmtree(self._tmpdir, ignore_errors=True)
        storage.Storage._Backend = None

    def _MockConfigGet(self, backend='local'):
        """mock _config_get：仅 BACKEND 返回指定值，其他返回 None 以触发 fallback。"""
        def _SideEffect(section, key, fallback=None):
            if key == 'BACKEND':
                return backend
            return None
        return mock.patch('gvsdsdk.storage._config_get', side_effect=_SideEffect)

    def testGetBackend_Local_CachesInstance(self):
        with self._MockConfigGet():
            b1 = storage.Storage._GetBackend()
            b2 = storage.Storage._GetBackend()
            self.assertIs(b1, b2)

    def testUpload_FacadeDelegatesToBackend(self):
        with self._MockConfigGet():
            url = storage.Storage.Upload('assets/facade.txt', b'data')
            self.assertEqual(url, '/media/assets/facade.txt')
            self.assertTrue((Path(self._tmpdir) / 'assets' / 'facade.txt').exists())

    def testUploadFile_FacadeDelegatesToBackend(self):
        with self._MockConfigGet():
            FileObj = io.BytesIO(b'file data')
            url = storage.Storage.UploadFile('assets/file.txt', FileObj)
            self.assertEqual(url, '/media/assets/file.txt')

    def testDelete_FacadeDelegatesToBackend(self):
        with self._MockConfigGet():
            storage.Storage.Upload('assets/del.txt', b'data')
            storage.Storage.Delete('assets/del.txt')
            self.assertFalse(storage.Storage.Exists('assets/del.txt'))

    def testGetBackend_UnsupportedBackend_Raises(self):
        with self._MockConfigGet(backend='ftp'):
            with self.assertRaises(Exception):
                storage.Storage._GetBackend()