488 lines
18 KiB
Python
488 lines
18 KiB
Python
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import HTTPException, status
|
|
from fastapi.testclient import TestClient
|
|
from minio.error import S3Error
|
|
from starlette.datastructures import UploadFile
|
|
|
|
from app.api.files import get_audit_log_service, get_file_service, require_current_user
|
|
from app.main import app
|
|
|
|
|
|
class StubAuditLogService:
|
|
def __init__(self):
|
|
self.events = []
|
|
|
|
def record_event(self, **kwargs):
|
|
self.events.append(kwargs)
|
|
return kwargs
|
|
|
|
|
|
class StubFileService:
|
|
def list_files(self, current_user, page=1, page_size=20, file_type=None):
|
|
items = [
|
|
{
|
|
"id": "file-123",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"uploaded_by": current_user["id"],
|
|
"bucket": "default",
|
|
"object_key": "tenant-demo-001/document/abc/demo.txt",
|
|
"file_name": "demo.txt",
|
|
"mime_type": "text/plain",
|
|
"file_size": 11,
|
|
"file_hash": "hash-demo",
|
|
"file_type": "document",
|
|
"status": "active",
|
|
"created_at": "2026-04-17T00:00:00+00:00",
|
|
}
|
|
]
|
|
if file_type:
|
|
items = [item for item in items if item["file_type"] == file_type]
|
|
return {
|
|
"items": items,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total": len(items),
|
|
}
|
|
|
|
def upload_file(self, current_user, file, file_type, bucket="default"):
|
|
return {
|
|
"id": "file-123",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"uploaded_by": current_user["id"],
|
|
"bucket": bucket,
|
|
"object_key": f"{current_user['tenant_id']}/{file_type}/abc/{file.filename}",
|
|
"file_name": file.filename,
|
|
"mime_type": file.content_type or "application/octet-stream",
|
|
"file_size": 11,
|
|
"file_hash": "hash-demo",
|
|
"file_type": file_type,
|
|
"status": "active",
|
|
"created_at": "2026-04-17T00:00:00+00:00",
|
|
}
|
|
|
|
def get_file(self, current_user, file_id):
|
|
return self.list_files(current_user)["items"][0]
|
|
|
|
def delete_file(self, current_user, file_id):
|
|
return True
|
|
|
|
def get_download_url(self, current_user, file_id):
|
|
return {
|
|
"id": file_id,
|
|
"file_name": "demo.txt",
|
|
"download_url": f"/api/v1/files/{file_id}/download",
|
|
"expires_in": 3600,
|
|
}
|
|
|
|
def get_download_payload(self, current_user, file_id, allow_owner_access=False):
|
|
return {
|
|
"id": file_id,
|
|
"file_name": "demo.txt",
|
|
"mime_type": "text/plain",
|
|
"content": b"hello world",
|
|
}
|
|
|
|
|
|
class FilesApiTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.audit_log_service = StubAuditLogService()
|
|
app.dependency_overrides[get_file_service] = lambda: StubFileService()
|
|
app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self):
|
|
app.dependency_overrides.clear()
|
|
|
|
def test_list_files_returns_filtered_total(self):
|
|
response = self.client.get("/api/v1/files", params={"file_type": "document"})
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["code"], 0)
|
|
self.assertEqual(body["data"]["total"], 1)
|
|
self.assertEqual(body["data"]["items"][0]["file_type"], "document")
|
|
|
|
def test_upload_file_records_audit_event(self):
|
|
response = self.client.post(
|
|
"/api/v1/files",
|
|
data={"file_type": "document", "bucket": "default"},
|
|
files={"file": ("demo.txt", b"hello world", "text/plain")},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["file_name"], "demo.txt")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "upload_file")
|
|
self.assertEqual(
|
|
self.audit_log_service.events[0]["detail_json"],
|
|
{
|
|
"file_name": "demo.txt",
|
|
"file_type": "document",
|
|
"file_size": 11,
|
|
"mime_type": "text/plain",
|
|
"file_hash": "hash-demo",
|
|
"bucket": "default",
|
|
},
|
|
)
|
|
|
|
def test_get_file_records_audit_event(self):
|
|
response = self.client.get("/api/v1/files/file-123")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["id"], "file-123")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "get_file")
|
|
self.assertEqual(
|
|
self.audit_log_service.events[0]["detail_json"],
|
|
{
|
|
"file_name": "demo.txt",
|
|
"file_type": "document",
|
|
"mime_type": "text/plain",
|
|
"file_size": 11,
|
|
"file_hash": "hash-demo",
|
|
},
|
|
)
|
|
|
|
def test_delete_file_records_audit_event(self):
|
|
response = self.client.delete("/api/v1/files/file-123")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertTrue(body["data"]["success"])
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "delete_file")
|
|
self.assertEqual(
|
|
self.audit_log_service.events[0]["detail_json"],
|
|
{
|
|
"file_name": "demo.txt",
|
|
"file_hash": "hash-demo",
|
|
"mime_type": "text/plain",
|
|
"file_size": 11,
|
|
"file_type": "document",
|
|
},
|
|
)
|
|
|
|
def test_get_download_url_records_audit_event(self):
|
|
response = self.client.get("/api/v1/files/file-123/download-url")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["id"], "file-123")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "get_download_url")
|
|
self.assertEqual(
|
|
self.audit_log_service.events[0]["detail_json"],
|
|
{
|
|
"file_name": "demo.txt",
|
|
"download_url": "/api/v1/files/file-123/download",
|
|
"expires_in": 3600,
|
|
},
|
|
)
|
|
|
|
def test_download_file_returns_attachment_response(self):
|
|
response = self.client.get("/api/v1/files/file-123/download")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.content, b"hello world")
|
|
self.assertEqual(response.headers["content-type"], "text/plain; charset=utf-8")
|
|
self.assertEqual(response.headers["content-disposition"], "attachment; filename*=UTF-8''demo.txt")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "download_file")
|
|
|
|
|
|
class FileServiceStorageTestCase(unittest.TestCase):
|
|
def test_upload_download_delete_manage_minio_storage(self):
|
|
from app.services.file_service import FileService
|
|
|
|
class FakeObjectResponse:
|
|
def __init__(self, content: bytes):
|
|
self._content = content
|
|
|
|
def read(self):
|
|
return self._content
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
def release_conn(self):
|
|
return None
|
|
|
|
class FakeMinioClient:
|
|
def __init__(self):
|
|
self.buckets = set()
|
|
self.objects = {}
|
|
|
|
def bucket_exists(self, bucket):
|
|
return bucket in self.buckets
|
|
|
|
def make_bucket(self, bucket):
|
|
self.buckets.add(bucket)
|
|
|
|
def put_object(self, bucket_name, object_name, data, length, content_type):
|
|
if bucket_name not in self.buckets:
|
|
raise RuntimeError("bucket missing")
|
|
self.objects[(bucket_name, object_name)] = data.read(length)
|
|
|
|
def get_object(self, bucket_name, object_name):
|
|
key = (bucket_name, object_name)
|
|
if key not in self.objects:
|
|
raise S3Error(
|
|
code="NoSuchKey",
|
|
message="Not Found",
|
|
resource=object_name,
|
|
request_id="req",
|
|
host_id="host",
|
|
response=None,
|
|
bucket_name=bucket_name,
|
|
object_name=object_name,
|
|
)
|
|
return FakeObjectResponse(self.objects[key])
|
|
|
|
def remove_object(self, bucket_name, object_name):
|
|
key = (bucket_name, object_name)
|
|
if key not in self.objects:
|
|
raise S3Error(
|
|
code="NoSuchKey",
|
|
message="Not Found",
|
|
resource=object_name,
|
|
request_id="req",
|
|
host_id="host",
|
|
response=None,
|
|
bucket_name=bucket_name,
|
|
object_name=object_name,
|
|
)
|
|
del self.objects[key]
|
|
|
|
class InMemoryRepository:
|
|
def __init__(self):
|
|
self.created = None
|
|
self.deleted = None
|
|
|
|
def create(self, file_record):
|
|
if file_record.created_at is None:
|
|
file_record.created_at = datetime.now(timezone.utc)
|
|
self.created = file_record
|
|
return file_record
|
|
|
|
def get_by_id(self, file_id, tenant_id):
|
|
if self.created and self.created.id == file_id and self.created.tenant_id == tenant_id and self.created.status == "active":
|
|
return self.created
|
|
return None
|
|
|
|
def soft_delete(self, file_record):
|
|
file_record.status = "deleted"
|
|
self.deleted = file_record
|
|
return file_record
|
|
|
|
def list_by_type(self, tenant_id, file_type, page, page_size):
|
|
return []
|
|
|
|
def list_paginated(self, tenant_id, page, page_size):
|
|
return []
|
|
|
|
def count(self, tenant_id, file_type=None):
|
|
return 0
|
|
|
|
service = FileService(db=None)
|
|
repository = InMemoryRepository()
|
|
fake_minio = FakeMinioClient()
|
|
service.repository = repository
|
|
service.minio_client = fake_minio
|
|
current_user = {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
upload = UploadFile(filename="nested/contract.txt", file=io.BytesIO(b"payload"), headers={"content-type": "text/plain"})
|
|
|
|
created = service.upload_file(current_user, upload, "document", "contracts")
|
|
object_key = repository.created.object_key
|
|
|
|
self.assertEqual(created["file_name"], "contract.txt")
|
|
self.assertIn("contracts", fake_minio.buckets)
|
|
self.assertEqual(fake_minio.objects[("contracts", object_key)], b"payload")
|
|
|
|
download = service.get_download_payload(current_user, repository.created.id)
|
|
self.assertEqual(download["content"], b"payload")
|
|
|
|
service.delete_file(current_user, repository.created.id)
|
|
self.assertNotIn(("contracts", object_key), fake_minio.objects)
|
|
self.assertEqual(repository.deleted.status, "deleted")
|
|
|
|
def test_download_missing_object_returns_404(self):
|
|
from app.services.file_service import FileService
|
|
|
|
class FakeMinioClient:
|
|
def get_object(self, bucket_name, object_name):
|
|
raise S3Error(
|
|
code="NoSuchKey",
|
|
message="Not Found",
|
|
resource=object_name,
|
|
request_id="req",
|
|
host_id="host",
|
|
response=None,
|
|
bucket_name=bucket_name,
|
|
object_name=object_name,
|
|
)
|
|
|
|
class InMemoryRepository:
|
|
def __init__(self):
|
|
self.created = type(
|
|
"FileRecord",
|
|
(),
|
|
{
|
|
"id": "file-123",
|
|
"tenant_id": "tenant-demo-001",
|
|
"status": "active",
|
|
"bucket": "contracts",
|
|
"object_key": "tenant-demo-001/document/object.txt",
|
|
"file_name": "object.txt",
|
|
"mime_type": "text/plain",
|
|
},
|
|
)()
|
|
|
|
def get_by_id(self, file_id, tenant_id):
|
|
if self.created.id == file_id and self.created.tenant_id == tenant_id and self.created.status == "active":
|
|
return self.created
|
|
return None
|
|
|
|
def list_by_type(self, tenant_id, file_type, page, page_size):
|
|
return []
|
|
|
|
def list_paginated(self, tenant_id, page, page_size):
|
|
return []
|
|
|
|
def count(self, tenant_id, file_type=None):
|
|
return 0
|
|
|
|
service = FileService(db=None)
|
|
service.repository = InMemoryRepository()
|
|
service.minio_client = FakeMinioClient()
|
|
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
service.get_download_payload({"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"}, "file-123")
|
|
|
|
self.assertEqual(ctx.exception.status_code, status.HTTP_404_NOT_FOUND)
|
|
self.assertEqual(ctx.exception.detail, "file content not found")
|
|
|
|
def test_owner_can_download_without_admin_role_when_enabled(self):
|
|
from app.services.file_service import FileService
|
|
|
|
class FakeMinioClient:
|
|
def get_object(self, bucket_name, object_name):
|
|
class FakeObjectResponse:
|
|
def read(self_inner):
|
|
return b"owner-payload"
|
|
|
|
def close(self_inner):
|
|
return None
|
|
|
|
def release_conn(self_inner):
|
|
return None
|
|
|
|
return FakeObjectResponse()
|
|
|
|
class InMemoryRepository:
|
|
def __init__(self):
|
|
self.created = type(
|
|
"FileRecord",
|
|
(),
|
|
{
|
|
"id": "file-123",
|
|
"tenant_id": "tenant-demo-001",
|
|
"uploaded_by": "user-owner",
|
|
"status": "active",
|
|
"bucket": "contracts",
|
|
"object_key": "tenant-demo-001/document/object.txt",
|
|
"file_name": "object.txt",
|
|
"mime_type": "text/plain",
|
|
},
|
|
)()
|
|
|
|
def get_by_id(self, file_id, tenant_id):
|
|
if self.created.id == file_id and self.created.tenant_id == tenant_id and self.created.status == "active":
|
|
return self.created
|
|
return None
|
|
|
|
service = FileService(db=None)
|
|
service.repository = InMemoryRepository()
|
|
service.minio_client = FakeMinioClient()
|
|
|
|
payload = service.get_download_payload(
|
|
{"id": "user-owner", "tenant_id": "tenant-demo-001", "role": "member"},
|
|
"file-123",
|
|
allow_owner_access=True,
|
|
)
|
|
|
|
self.assertEqual(payload["content"], b"owner-payload")
|
|
|
|
def test_local_storage_backend_persists_downloads_and_deletes_files(self):
|
|
from app.services.file_service import FileService
|
|
|
|
class InMemoryRepository:
|
|
def __init__(self):
|
|
self.created = None
|
|
self.deleted = None
|
|
|
|
def create(self, file_record):
|
|
if file_record.created_at is None:
|
|
file_record.created_at = datetime.now(timezone.utc)
|
|
self.created = file_record
|
|
return file_record
|
|
|
|
def get_by_id(self, file_id, tenant_id):
|
|
if self.created and self.created.id == file_id and self.created.tenant_id == tenant_id and self.created.status == "active":
|
|
return self.created
|
|
return None
|
|
|
|
def soft_delete(self, file_record):
|
|
file_record.status = "deleted"
|
|
self.deleted = file_record
|
|
return file_record
|
|
|
|
def list_by_type(self, tenant_id, file_type, page, page_size):
|
|
return []
|
|
|
|
def list_paginated(self, tenant_id, page, page_size):
|
|
return []
|
|
|
|
def count(self, tenant_id, file_type=None):
|
|
return 0
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
service = FileService(db=None)
|
|
service.repository = InMemoryRepository()
|
|
service.storage_backend = "local"
|
|
service.local_storage_dir = Path(temp_dir)
|
|
current_user = {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
upload = UploadFile(filename="nested/contract.txt", file=io.BytesIO(b"local-payload"), headers={"content-type": "text/plain"})
|
|
|
|
created = service.upload_file(current_user, upload, "document", "contracts")
|
|
object_path = service._resolve_local_object_path("contracts", service.repository.created.object_key)
|
|
|
|
self.assertEqual(created["file_name"], "contract.txt")
|
|
self.assertTrue(object_path.exists())
|
|
self.assertEqual(object_path.read_bytes(), b"local-payload")
|
|
|
|
download = service.get_download_payload(current_user, service.repository.created.id)
|
|
self.assertEqual(download["content"], b"local-payload")
|
|
|
|
service.delete_file(current_user, service.repository.created.id)
|
|
self.assertFalse(object_path.exists())
|
|
self.assertEqual(service.repository.deleted.status, "deleted")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|