kodi
262 lines
8.5 KiB
Python
262 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
|
|
|
|
from backend.app.dependencies import get_task_service
|
|
from backend.app.db.task_repository import TaskRepository
|
|
from backend.app.main import app
|
|
from backend.app.services.task_service import TaskService
|
|
|
|
|
|
class TasksApiGoldenTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.db_path = str(Path(self.temp_dir.name) / "tasks.db")
|
|
self.repo = TaskRepository(self.db_path)
|
|
self.service = TaskService(self.repo)
|
|
|
|
async def _override_task_service() -> TaskService:
|
|
return self.service
|
|
|
|
app.dependency_overrides[get_task_service] = _override_task_service
|
|
|
|
def tearDown(self) -> None:
|
|
app.dependency_overrides.clear()
|
|
self.temp_dir.cleanup()
|
|
|
|
def _get(self, url: str) -> httpx.Response:
|
|
async def _run() -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
return await client.get(url)
|
|
|
|
return asyncio.run(_run())
|
|
|
|
def _insert_task(
|
|
self,
|
|
*,
|
|
task_id: str,
|
|
operation: str,
|
|
status: str,
|
|
source: str,
|
|
destination: str,
|
|
created_at: str,
|
|
started_at: str | None = None,
|
|
finished_at: str | None = None,
|
|
done_bytes: int | None = None,
|
|
total_bytes: int | None = None,
|
|
done_items: int | None = None,
|
|
total_items: int | None = None,
|
|
current_item: str | None = None,
|
|
failed_item: str | None = None,
|
|
error_code: str | None = None,
|
|
error_message: str | None = None,
|
|
) -> None:
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": task_id,
|
|
"operation": operation,
|
|
"status": status,
|
|
"source": source,
|
|
"destination": destination,
|
|
"done_bytes": done_bytes,
|
|
"total_bytes": total_bytes,
|
|
"done_items": done_items,
|
|
"total_items": total_items,
|
|
"current_item": current_item,
|
|
"failed_item": failed_item,
|
|
"error_code": error_code,
|
|
"error_message": error_message,
|
|
"created_at": created_at,
|
|
"started_at": started_at,
|
|
"finished_at": finished_at,
|
|
}
|
|
)
|
|
|
|
def test_get_tasks_empty_list(self) -> None:
|
|
response = self._get("/api/tasks")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), {"items": []})
|
|
|
|
def test_get_tasks_list_shape(self) -> None:
|
|
self._insert_task(
|
|
task_id="task-old",
|
|
operation="copy",
|
|
status="completed",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
finished_at="2026-03-10T10:00:05Z",
|
|
)
|
|
self._insert_task(
|
|
task_id="task-new",
|
|
operation="move",
|
|
status="running",
|
|
source="storage1/b.txt",
|
|
destination="storage2/b.txt",
|
|
created_at="2026-03-10T10:01:00Z",
|
|
)
|
|
|
|
response = self._get("/api/tasks")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(
|
|
response.json(),
|
|
{
|
|
"items": [
|
|
{
|
|
"id": "task-new",
|
|
"operation": "move",
|
|
"status": "running",
|
|
"source": "storage1/b.txt",
|
|
"destination": "storage2/b.txt",
|
|
"created_at": "2026-03-10T10:01:00Z",
|
|
"finished_at": None,
|
|
},
|
|
{
|
|
"id": "task-old",
|
|
"operation": "copy",
|
|
"status": "completed",
|
|
"source": "storage1/a.txt",
|
|
"destination": "storage2/a.txt",
|
|
"created_at": "2026-03-10T10:00:00Z",
|
|
"finished_at": "2026-03-10T10:00:05Z",
|
|
},
|
|
]
|
|
},
|
|
)
|
|
|
|
def test_get_task_detail_queued(self) -> None:
|
|
self._insert_task(
|
|
task_id="task-queued",
|
|
operation="copy",
|
|
status="queued",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
)
|
|
|
|
response = self._get("/api/tasks/task-queued")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(
|
|
response.json(),
|
|
{
|
|
"id": "task-queued",
|
|
"operation": "copy",
|
|
"status": "queued",
|
|
"source": "storage1/a.txt",
|
|
"destination": "storage2/a.txt",
|
|
"done_bytes": None,
|
|
"total_bytes": None,
|
|
"done_items": None,
|
|
"total_items": None,
|
|
"current_item": None,
|
|
"failed_item": None,
|
|
"error_code": None,
|
|
"error_message": None,
|
|
"created_at": "2026-03-10T10:00:00Z",
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
},
|
|
)
|
|
|
|
def test_get_task_detail_running(self) -> None:
|
|
self._insert_task(
|
|
task_id="task-running",
|
|
operation="move",
|
|
status="running",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
started_at="2026-03-10T10:00:01Z",
|
|
done_bytes=1024,
|
|
total_bytes=2048,
|
|
done_items=1,
|
|
total_items=2,
|
|
current_item="storage1/a.txt",
|
|
)
|
|
|
|
response = self._get("/api/tasks/task-running")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["status"], "running")
|
|
self.assertEqual(body["done_bytes"], 1024)
|
|
self.assertEqual(body["total_bytes"], 2048)
|
|
self.assertEqual(body["current_item"], "storage1/a.txt")
|
|
|
|
def test_get_task_detail_completed(self) -> None:
|
|
self._insert_task(
|
|
task_id="task-completed",
|
|
operation="copy",
|
|
status="completed",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
started_at="2026-03-10T10:00:01Z",
|
|
finished_at="2026-03-10T10:00:03Z",
|
|
done_bytes=2048,
|
|
total_bytes=2048,
|
|
)
|
|
|
|
response = self._get("/api/tasks/task-completed")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["status"], "completed")
|
|
self.assertEqual(body["finished_at"], "2026-03-10T10:00:03Z")
|
|
self.assertEqual(body["error_code"], None)
|
|
|
|
def test_get_task_detail_failed(self) -> None:
|
|
self._insert_task(
|
|
task_id="task-failed",
|
|
operation="move",
|
|
status="failed",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
started_at="2026-03-10T10:00:01Z",
|
|
finished_at="2026-03-10T10:00:02Z",
|
|
failed_item="storage1/a.txt",
|
|
error_code="io_error",
|
|
error_message="write failed",
|
|
)
|
|
|
|
response = self._get("/api/tasks/task-failed")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["status"], "failed")
|
|
self.assertEqual(body["failed_item"], "storage1/a.txt")
|
|
self.assertEqual(body["error_code"], "io_error")
|
|
self.assertEqual(body["error_message"], "write failed")
|
|
|
|
def test_get_task_not_found(self) -> None:
|
|
response = self._get("/api/tasks/task-missing")
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(
|
|
response.json(),
|
|
{
|
|
"error": {
|
|
"code": "task_not_found",
|
|
"message": "Task was not found",
|
|
"details": {"task_id": "task-missing"},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|