63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
|
import pytest
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
from botocore.exceptions import ClientError
|
||
|
|
|
||
|
|
from app.clients.storage.rustfs_client import RustFSClient
|
||
|
|
from app.core.exceptions import StorageError
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def client():
|
||
|
|
with patch("app.clients.storage.rustfs_client.boto3") as mock_boto3:
|
||
|
|
c = RustFSClient(
|
||
|
|
endpoint="http://rustfs:9000",
|
||
|
|
access_key="key",
|
||
|
|
secret_key="secret",
|
||
|
|
)
|
||
|
|
c._s3 = MagicMock()
|
||
|
|
return c
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_download_bytes_returns_bytes(client):
|
||
|
|
client._s3.get_object.return_value = {"Body": MagicMock(read=lambda: b"hello")}
|
||
|
|
result = await client.download_bytes("source-data", "text/test.txt")
|
||
|
|
assert result == b"hello"
|
||
|
|
client._s3.get_object.assert_called_once_with(Bucket="source-data", Key="text/test.txt")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_download_bytes_raises_storage_error(client):
|
||
|
|
client._s3.get_object.side_effect = ClientError(
|
||
|
|
{"Error": {"Code": "NoSuchKey", "Message": "Not Found"}}, "GetObject"
|
||
|
|
)
|
||
|
|
with pytest.raises(StorageError, match="存储下载失败"):
|
||
|
|
await client.download_bytes("source-data", "missing.txt")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_object_size_returns_content_length(client):
|
||
|
|
client._s3.head_object.return_value = {"ContentLength": 1024}
|
||
|
|
size = await client.get_object_size("source-data", "video/test.mp4")
|
||
|
|
assert size == 1024
|
||
|
|
client._s3.head_object.assert_called_once_with(Bucket="source-data", Key="video/test.mp4")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_object_size_raises_storage_error(client):
|
||
|
|
client._s3.head_object.side_effect = ClientError(
|
||
|
|
{"Error": {"Code": "NoSuchKey", "Message": "Not Found"}}, "HeadObject"
|
||
|
|
)
|
||
|
|
with pytest.raises(StorageError, match="获取文件大小失败"):
|
||
|
|
await client.get_object_size("source-data", "video/missing.mp4")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_upload_bytes_calls_put_object(client):
|
||
|
|
client._s3.put_object.return_value = {}
|
||
|
|
await client.upload_bytes("source-data", "frames/1/0.jpg", b"jpeg-data", "image/jpeg")
|
||
|
|
client._s3.put_object.assert_called_once()
|
||
|
|
call_kwargs = client._s3.put_object.call_args
|
||
|
|
assert call_kwargs.kwargs["Bucket"] == "source-data"
|
||
|
|
assert call_kwargs.kwargs["Key"] == "frames/1/0.jpg"
|