Daniil
294 lines
9.7 KiB
Python
294 lines
9.7 KiB
Python
"""
|
|
Tests for file management endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import uuid
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.modules.files.models import File
|
|
from cpv3.modules.users.models import User
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_file(test_db_session: AsyncSession, test_user: User) -> File:
|
|
"""Create a test file entry owned by test_user."""
|
|
file = File(
|
|
id=uuid.uuid4(),
|
|
owner_id=test_user.id,
|
|
original_filename="test-document.pdf",
|
|
path="uploads/test-document.pdf",
|
|
storage_backend="LOCAL",
|
|
mime_type="application/pdf",
|
|
size_bytes=1024,
|
|
is_uploaded=True,
|
|
is_active=True,
|
|
)
|
|
test_db_session.add(file)
|
|
await test_db_session.commit()
|
|
await test_db_session.refresh(file)
|
|
return file
|
|
|
|
|
|
@pytest.fixture
|
|
async def other_file(test_db_session: AsyncSession, other_user: User) -> File:
|
|
"""Create a file entry owned by another user."""
|
|
file = File(
|
|
id=uuid.uuid4(),
|
|
owner_id=other_user.id,
|
|
original_filename="other-document.pdf",
|
|
path="uploads/other-document.pdf",
|
|
storage_backend="LOCAL",
|
|
mime_type="application/pdf",
|
|
size_bytes=2048,
|
|
is_uploaded=True,
|
|
is_active=True,
|
|
)
|
|
test_db_session.add(file)
|
|
await test_db_session.commit()
|
|
await test_db_session.refresh(file)
|
|
return file
|
|
|
|
|
|
class TestUploadFileEndpoint:
|
|
"""Tests for POST /api/files/upload/."""
|
|
|
|
async def test_upload_file_success(self, auth_client: AsyncClient):
|
|
"""Test successful file upload."""
|
|
file_content = b"test file content"
|
|
files = {"file": ("testfile.txt", io.BytesIO(file_content), "text/plain")}
|
|
data = {"folder": "uploads"}
|
|
|
|
response = await auth_client.post("/api/files/upload/", files=files, data=data)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "file_path" in data
|
|
assert "file_url" in data
|
|
|
|
async def test_upload_file_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test uploading file without auth returns 401."""
|
|
file_content = b"test file content"
|
|
files = {"file": ("testfile.txt", io.BytesIO(file_content), "text/plain")}
|
|
|
|
response = await async_client.post("/api/files/upload/", files=files)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestGetFileInfoEndpoint:
|
|
"""Tests for GET /api/files/get_file/."""
|
|
|
|
async def test_get_file_info_success(self, auth_client: AsyncClient):
|
|
"""Test getting file info by path."""
|
|
response = await auth_client.get(
|
|
"/api/files/get_file/", params={"file_path": "uploads/test-file.txt"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "file_path" in data
|
|
assert "file_url" in data
|
|
|
|
async def test_get_file_info_not_found(
|
|
self, auth_client: AsyncClient, mock_storage
|
|
):
|
|
"""Test getting info for nonexistent file returns 404."""
|
|
mock_storage.exists.return_value = False
|
|
|
|
response = await auth_client.get(
|
|
"/api/files/get_file/", params={"file_path": "nonexistent/file.txt"}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_get_file_info_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test getting file info without auth returns 401."""
|
|
response = await async_client.get(
|
|
"/api/files/get_file/", params={"file_path": "uploads/test.txt"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestListFileEntriesEndpoint:
|
|
"""Tests for GET /api/files/files/."""
|
|
|
|
async def test_list_file_entries(self, auth_client: AsyncClient, test_file: File):
|
|
"""Test listing file entries."""
|
|
response = await auth_client.get("/api/files/files/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
async def test_list_file_entries_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test listing file entries without auth returns 401."""
|
|
response = await async_client.get("/api/files/files/")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestCreateFileEntryEndpoint:
|
|
"""Tests for POST /api/files/files/."""
|
|
|
|
async def test_create_file_entry_success(self, auth_client: AsyncClient):
|
|
"""Test creating a file entry."""
|
|
response = await auth_client.post(
|
|
"/api/files/files/",
|
|
json={
|
|
"original_filename": "new-file.pdf",
|
|
"path": "uploads/new-file.pdf",
|
|
"storage_backend": "LOCAL",
|
|
"mime_type": "application/pdf",
|
|
"size_bytes": 4096,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["original_filename"] == "new-file.pdf"
|
|
assert data["path"] == "uploads/new-file.pdf"
|
|
|
|
async def test_create_file_entry_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test creating file entry without auth returns 401."""
|
|
response = await async_client.post(
|
|
"/api/files/files/",
|
|
json={
|
|
"original_filename": "test.pdf",
|
|
"path": "test.pdf",
|
|
"storage_backend": "LOCAL",
|
|
"mime_type": "application/pdf",
|
|
"size_bytes": 1024,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestRetrieveFileEntryEndpoint:
|
|
"""Tests for GET /api/files/files/{file_id}/."""
|
|
|
|
async def test_retrieve_own_file_entry(
|
|
self, auth_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test retrieving own file entry."""
|
|
response = await auth_client.get(f"/api/files/files/{test_file.id}/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == str(test_file.id)
|
|
assert data["original_filename"] == test_file.original_filename
|
|
|
|
async def test_retrieve_other_file_as_staff(
|
|
self, staff_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test staff can retrieve any file entry."""
|
|
response = await staff_client.get(f"/api/files/files/{test_file.id}/")
|
|
|
|
assert response.status_code == 200
|
|
|
|
async def test_retrieve_nonexistent_file_entry(self, auth_client: AsyncClient):
|
|
"""Test retrieving nonexistent file entry returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.get(f"/api/files/files/{fake_id}/")
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_retrieve_other_file_forbidden(
|
|
self, auth_client: AsyncClient, other_file: File
|
|
):
|
|
"""Test regular user cannot retrieve other user's file entry."""
|
|
response = await auth_client.get(f"/api/files/files/{other_file.id}/")
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestPatchFileEntryEndpoint:
|
|
"""Tests for PATCH /api/files/files/{file_id}/."""
|
|
|
|
async def test_patch_own_file_entry(
|
|
self, auth_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test updating own file entry."""
|
|
response = await auth_client.patch(
|
|
f"/api/files/files/{test_file.id}/",
|
|
json={"original_filename": "renamed-file.pdf"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["original_filename"] == "renamed-file.pdf"
|
|
|
|
async def test_patch_other_file_as_staff(
|
|
self, staff_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test staff can update any file entry."""
|
|
response = await staff_client.patch(
|
|
f"/api/files/files/{test_file.id}/",
|
|
json={"original_filename": "staff-renamed.pdf"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
async def test_patch_nonexistent_file_entry(self, auth_client: AsyncClient):
|
|
"""Test patching nonexistent file entry returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.patch(
|
|
f"/api/files/files/{fake_id}/",
|
|
json={"original_filename": "test.pdf"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_patch_other_file_forbidden(
|
|
self, auth_client: AsyncClient, other_file: File
|
|
):
|
|
"""Test regular user cannot update other user's file entry."""
|
|
response = await auth_client.patch(
|
|
f"/api/files/files/{other_file.id}/",
|
|
json={"original_filename": "hacked.pdf"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestDeleteFileEntryEndpoint:
|
|
"""Tests for DELETE /api/files/files/{file_id}/."""
|
|
|
|
async def test_delete_own_file_entry(
|
|
self, auth_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test deleting own file entry."""
|
|
response = await auth_client.delete(f"/api/files/files/{test_file.id}/")
|
|
|
|
assert response.status_code == 204
|
|
|
|
async def test_delete_other_file_as_staff(
|
|
self, staff_client: AsyncClient, test_file: File
|
|
):
|
|
"""Test staff can delete any file entry."""
|
|
response = await staff_client.delete(f"/api/files/files/{test_file.id}/")
|
|
|
|
assert response.status_code == 204
|
|
|
|
async def test_delete_nonexistent_file_entry(self, auth_client: AsyncClient):
|
|
"""Test deleting nonexistent file entry returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.delete(f"/api/files/files/{fake_id}/")
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_delete_other_file_forbidden(
|
|
self, auth_client: AsyncClient, other_file: File
|
|
):
|
|
"""Test regular user cannot delete other user's file entry."""
|
|
response = await auth_client.delete(f"/api/files/files/{other_file.id}/")
|
|
|
|
assert response.status_code == 403
|