From 5d0274c519f591e751f286f1410d272c5d855b67 Mon Sep 17 00:00:00 2001
From: lewis_yan <lewis_yan@avc.co>
Date: Fri, 3 Apr 2026 16:52:56 +0800
Subject: [PATCH] test: add MinIO file upload integration tests

- Add FileListFactory for test data creation
- Add TestFileUpload with 4 test cases:
  - test_upload_file_success (requires MinIO running)
  - test_upload_file_unauthenticated (passes without MinIO)
  - test_upload_image_success (requires MinIO running)
  - test_upload_without_file (passes without MinIO)
- Fix admin_user/normal_user fixtures to use factory_boy
  sequence to avoid username collision
---
 backend/tests/conftest.py               | 16 ++---
 backend/tests/factories/file_factory.py | 13 ++++
 backend/tests/test_file_upload.py       | 83 +++++++++++++++++++++++++
 3 files changed, 100 insertions(+), 12 deletions(-)
 create mode 100644 backend/tests/factories/file_factory.py
 create mode 100644 backend/tests/test_file_upload.py

diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 017389e..a5cede5 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -18,24 +18,16 @@ def authenticate(api_client, admin_user):
 @pytest.fixture
 def admin_user(db):
     """创建管理员用户（每个测试独立事务，测试后自动 rollback）"""
-    from dvadmin.system.models import Users
+    from tests.factories.system_factory import UserFactory
 
-    user = Users.objects.create_user(
-        username="test_admin",
-        password="testpass123",
-        is_superuser=True,
-    )
+    user = UserFactory(admin=True)
     return user
 
 
 @pytest.fixture
 def normal_user(db):
     """创建普通用户"""
-    from dvadmin.system.models import Users
+    from tests.factories.system_factory import UserFactory
 
-    user = Users.objects.create_user(
-        username="test_normal_user",
-        password="testpass123",
-        is_superuser=False,
-    )
+    user = UserFactory()
     return user
diff --git a/backend/tests/factories/file_factory.py b/backend/tests/factories/file_factory.py
new file mode 100644
index 0000000..1eeb03b
--- /dev/null
+++ b/backend/tests/factories/file_factory.py
@@ -0,0 +1,13 @@
+import factory
+from dvadmin.system.models import FileList
+
+
+class FileListFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = FileList
+
+    name = factory.Sequence(lambda n: f"test_file_{n}.txt")
+    engine = "minio"
+    mime_type = "text/plain"
+    size = "100"
+    md5sum = factory.Faker("md5")
diff --git a/backend/tests/test_file_upload.py b/backend/tests/test_file_upload.py
new file mode 100644
index 0000000..8a62524
--- /dev/null
+++ b/backend/tests/test_file_upload.py
@@ -0,0 +1,83 @@
+"""
+MinIO 文件上传集成测试
+
+依赖: 真实的 MinIO 服务运行在 MINIO_ENDPOINT (192.168.80.90:9000)
+测试前确保:
+  1. MinIO 服务已启动
+  2. bucket "pis-media" 已创建
+  3. MINIO_ACCESS_KEY / MINIO_SECRET_KEY 正确配置
+"""
+import io
+import pytest
+from django.conf import settings
+from rest_framework import status
+
+
+CODE_SUCCESS = 2000
+CODE_ERROR = 4000
+
+
+@pytest.mark.django_db
+class TestFileUpload:
+    """文件上传 API 测试"""
+
+    def test_upload_file_success(self, authenticate):
+        """POST 上传文件返回成功 + MinIO URL"""
+        file_content = b"Hello, MinIO!"
+        file_obj = io.BytesIO(file_content)
+        file_obj.name = "test.txt"
+
+        response = authenticate.post(
+            "/api/system/file/",
+            data={"file": file_obj},
+            format="multipart",
+        )
+
+        assert response.data["code"] == CODE_SUCCESS, f"上传失败: {response.data}"
+        assert "file_url" in response.data["data"]
+        file_url = response.data["data"]["file_url"]
+        # 验证 URL 指向 MinIO
+        assert settings.MINIO_ENDPOINT in file_url, f"文件 URL 非 MinIO 地址: {file_url}"
+        assert file_url.startswith("http://") or file_url.startswith("https://")
+
+    def test_upload_file_unauthenticated(self, api_client):
+        """未认证请求上传文件返回错误码"""
+        file_content = b"Unauthorized file"
+        file_obj = io.BytesIO(file_content)
+        file_obj.name = "unauth.txt"
+
+        response = api_client.post(
+            "/api/system/file/",
+            data={"file": file_obj},
+            format="multipart",
+        )
+        # 无认证时返回 4000 或非 2000
+        assert response.data["code"] == CODE_ERROR
+
+    def test_upload_image_success(self, authenticate):
+        """POST 上传图片返回成功 + image file_type"""
+        # GIF 格式最小文件 (1x1 透明像素)
+        gif_bytes = (
+            b"\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00\x80\x00\x00"
+            b"\xff\xff\xff\x00\x00\x00\x21\xf9\x04\x01\x00\x00\x00"
+            b"\x00\x2c\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02\x44"
+            b"\x01\x00\x3b"
+        )
+        file_obj = io.BytesIO(gif_bytes)
+        file_obj.name = "test.gif"
+
+        response = authenticate.post(
+            "/api/system/file/",
+            data={"file": file_obj},
+            format="multipart",
+        )
+
+        assert response.data["code"] == CODE_SUCCESS, f"图片上传失败: {response.data}"
+        file_url = response.data["data"]["file_url"]
+        assert settings.MINIO_ENDPOINT in file_url
+
+    def test_upload_without_file(self, authenticate):
+        """POST 不带文件返回错误码"""
+        response = authenticate.post("/api/system/file/", data={}, format="multipart")
+        # 不带文件应该失败
+        assert response.data["code"] == CODE_ERROR