109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from minio.error import S3Error
|
|
|
|
from app.minio_client import MinioClient
|
|
|
|
|
|
class _FakeObject:
|
|
def __init__(self, object_name, size, last_modified):
|
|
self.object_name = object_name
|
|
self.size = size
|
|
self.last_modified = last_modified
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, data: bytes):
|
|
self._data = data
|
|
self.closed = False
|
|
self.released = False
|
|
|
|
def read(self):
|
|
return self._data
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def release_conn(self):
|
|
self.released = True
|
|
|
|
|
|
class FakeMinioSDK:
|
|
def __init__(self):
|
|
self.objects = {}
|
|
self.put_calls = []
|
|
self.removed_keys = []
|
|
|
|
def list_objects(self, bucket, prefix="", recursive=False):
|
|
for key, (data, content_type, last_modified) in self.objects.items():
|
|
if key.startswith(prefix):
|
|
yield _FakeObject(key, len(data), last_modified)
|
|
|
|
def get_object(self, bucket, key):
|
|
if key not in self.objects:
|
|
raise S3Error("NoSuchKey", "Object does not exist", key, "req-1", "host-1", None)
|
|
data, _, _ = self.objects[key]
|
|
return _FakeResponse(data)
|
|
|
|
def put_object(self, bucket, key, stream, length, content_type):
|
|
data = stream.read()
|
|
self.objects[key] = (data, content_type, datetime.now(timezone.utc))
|
|
self.put_calls.append((bucket, key, content_type))
|
|
|
|
def remove_object(self, bucket, key):
|
|
del self.objects[key]
|
|
self.removed_keys.append(key)
|
|
|
|
|
|
def test_list_objects_returns_keys_matching_prefix():
|
|
sdk = FakeMinioSDK()
|
|
sdk.objects["产品文档/架构设计.md"] = (b"# hi", "text/markdown", datetime(2026, 8, 1, tzinfo=timezone.utc))
|
|
sdk.objects["团队规范/编码规范.md"] = (b"# rules", "text/markdown", datetime(2026, 8, 1, tzinfo=timezone.utc))
|
|
client = MinioClient(sdk, bucket="dochub")
|
|
|
|
result = client.list_objects("产品文档/")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["key"] == "产品文档/架构设计.md"
|
|
assert result[0]["size"] == 4
|
|
|
|
|
|
def test_get_object_returns_bytes():
|
|
sdk = FakeMinioSDK()
|
|
sdk.objects["产品文档/架构设计.md"] = (b"# architecture", "text/markdown", datetime.now(timezone.utc))
|
|
client = MinioClient(sdk, bucket="dochub")
|
|
|
|
data = client.get_object("产品文档/架构设计.md")
|
|
|
|
assert data == b"# architecture"
|
|
|
|
|
|
def test_get_object_raises_file_not_found_for_missing_key():
|
|
sdk = FakeMinioSDK()
|
|
client = MinioClient(sdk, bucket="dochub")
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
client.get_object("不存在/文档.md")
|
|
|
|
|
|
def test_put_object_stores_data_with_content_type():
|
|
sdk = FakeMinioSDK()
|
|
client = MinioClient(sdk, bucket="dochub")
|
|
|
|
client.put_object("产品文档/新文档.md", b"# new doc", "text/markdown")
|
|
|
|
assert sdk.objects["产品文档/新文档.md"][0] == b"# new doc"
|
|
assert sdk.put_calls == [("dochub", "产品文档/新文档.md", "text/markdown")]
|
|
|
|
|
|
def test_delete_object_removes_key():
|
|
sdk = FakeMinioSDK()
|
|
sdk.objects["产品文档/旧文档.md"] = (b"old", "text/markdown", datetime.now(timezone.utc))
|
|
client = MinioClient(sdk, bucket="dochub")
|
|
|
|
client.delete_object("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in sdk.objects
|
|
assert sdk.removed_keys == ["产品文档/旧文档.md"]
|