123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
from fastapi.testclient import TestClient
|
|
from fastapi import BackgroundTasks
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
import asyncio
|
|
import os
|
|
from app.main import app
|
|
from app.services.file import FileService
|
|
from app.services.task import TaskService
|
|
|
|
client = TestClient(app)
|
|
|
|
# Constants for reused values
|
|
TEST_FILE_PATH = os.path.join(os.getcwd(), "tests/test_files", "manabox_test_file.csv")
|
|
DEFAULT_METADATA = {
|
|
"source": "manabox",
|
|
"type": "scan_export_rare"
|
|
}
|
|
|
|
def get_file_size_kb(file_path):
|
|
"""Helper to consistently calculate file size in KB"""
|
|
with open(file_path, "rb") as f:
|
|
return round(len(f.read()) / 1024, 2)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_manabox_file():
|
|
"""Test creating a new manabox file"""
|
|
# Open file within the test scope
|
|
with open(TEST_FILE_PATH, "rb") as test_file:
|
|
files = {"file": test_file}
|
|
|
|
# Make request
|
|
response = client.post("/api/files", data=DEFAULT_METADATA, files=files)
|
|
|
|
# Check response
|
|
assert response.status_code == 201
|
|
assert response.json()["success"] == True
|
|
|
|
file_data = response.json()["files"][0]
|
|
assert file_data["source"] == DEFAULT_METADATA["source"]
|
|
assert file_data["type"] == DEFAULT_METADATA["type"]
|
|
assert file_data["status"] == "pending"
|
|
assert file_data["service"] == None
|
|
assert file_data["filename"] == "manabox_test_file.csv"
|
|
assert file_data["filesize_kb"] == get_file_size_kb(TEST_FILE_PATH)
|
|
assert file_data["id"] is not None
|
|
|
|
# Execute background tasks if they were added
|
|
background_tasks = BackgroundTasks()
|
|
for task in background_tasks.tasks:
|
|
await task()
|
|
|
|
def test_get_file():
|
|
"""Test retrieving a specific file"""
|
|
# Create a file first
|
|
with open(TEST_FILE_PATH, "rb") as test_file:
|
|
files = {"file": test_file}
|
|
create_response = client.post("/api/files", data=DEFAULT_METADATA, files=files)
|
|
file_id = create_response.json()["files"][0]["id"]
|
|
|
|
# Get the file
|
|
response = client.get(f"/api/files/{file_id}")
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] == True
|
|
|
|
file_data = response.json()["files"][0]
|
|
assert file_data["source"] == DEFAULT_METADATA["source"]
|
|
assert file_data["type"] == DEFAULT_METADATA["type"]
|
|
assert file_data["status"] == "completed"
|
|
assert file_data["service"] == None
|
|
assert file_data["filename"] == "manabox_test_file.csv"
|
|
assert file_data["filesize_kb"] == get_file_size_kb(TEST_FILE_PATH)
|
|
assert file_data["id"] == file_id
|
|
|
|
def test_delete_file():
|
|
"""Test file deletion"""
|
|
# Create a file first
|
|
with open(TEST_FILE_PATH, "rb") as test_file:
|
|
files = {"file": test_file}
|
|
create_response = client.post("/api/files", data=DEFAULT_METADATA, files=files)
|
|
file_id = create_response.json()["files"][0]["id"]
|
|
|
|
# Delete the file
|
|
response = client.delete(f"/api/files/{file_id}")
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] == True
|
|
|
|
file_data = response.json()["files"][0]
|
|
assert file_data["source"] == DEFAULT_METADATA["source"]
|
|
assert file_data["type"] == DEFAULT_METADATA["type"]
|
|
assert file_data["status"] == "deleted"
|
|
assert file_data["service"] == None
|
|
assert file_data["filename"] == "manabox_test_file.csv"
|
|
assert file_data["filesize_kb"] == get_file_size_kb(TEST_FILE_PATH)
|
|
assert file_data["id"] == file_id
|
|
|
|
def test_get_prepared_files():
|
|
"""Test retrieving files filtered by status"""
|
|
# Create a test file first
|
|
with open(TEST_FILE_PATH, "rb") as test_file:
|
|
files = {"file": test_file}
|
|
create_response = client.post("/api/files", data=DEFAULT_METADATA, files=files)
|
|
file_id = create_response.json()["files"][0]["id"]
|
|
|
|
# Get prepared files
|
|
response = client.get("/api/files?status=completed")
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] == True
|
|
|
|
# get file from id
|
|
file_data = [file for file in response.json()["files"] if file["id"] == file_id][0]
|
|
assert file_data["source"] == DEFAULT_METADATA["source"]
|
|
assert file_data["type"] == DEFAULT_METADATA["type"]
|
|
assert file_data["status"] == "completed"
|
|
assert file_data["service"] == None
|
|
assert file_data["filename"] == "manabox_test_file.csv"
|
|
assert file_data["filesize_kb"] == get_file_size_kb(TEST_FILE_PATH) |