kodi
260 lines
8.7 KiB
Python
260 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
|
|
|
|
from backend.app.db.task_repository import TaskRepository
|
|
|
|
|
|
class TaskRepositoryTest(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)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_list_tasks_sorted_created_at_desc(self) -> None:
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": "task-old",
|
|
"operation": "copy",
|
|
"status": "queued",
|
|
"source": "storage1/a",
|
|
"destination": "storage2/a",
|
|
"created_at": "2026-03-10T09:00:00Z",
|
|
}
|
|
)
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": "task-new",
|
|
"operation": "move",
|
|
"status": "queued",
|
|
"source": "storage1/b",
|
|
"destination": "storage2/b",
|
|
"created_at": "2026-03-10T10:00:00Z",
|
|
}
|
|
)
|
|
|
|
tasks = self.repo.list_tasks()
|
|
|
|
self.assertEqual([task["id"] for task in tasks], ["task-new", "task-old"])
|
|
|
|
def test_insert_rejects_invalid_status(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": "task-x",
|
|
"operation": "copy",
|
|
"status": "unknown",
|
|
"source": "storage1/a",
|
|
"destination": "storage2/a",
|
|
"created_at": "2026-03-10T09:00:00Z",
|
|
}
|
|
)
|
|
|
|
def test_create_download_task_with_requested_status_and_artifact(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="download",
|
|
source="storage1/docs",
|
|
destination="docs.zip",
|
|
status="requested",
|
|
)
|
|
self.repo.upsert_artifact(
|
|
task_id=created["id"],
|
|
file_path="/tmp/archive.zip",
|
|
file_name="docs.zip",
|
|
expires_at="2026-03-10T10:30:00Z",
|
|
)
|
|
|
|
task = self.repo.get_task(created["id"])
|
|
artifact = self.repo.get_artifact(created["id"])
|
|
|
|
self.assertEqual(task["operation"], "download")
|
|
self.assertEqual(task["status"], "requested")
|
|
self.assertEqual(artifact["file_name"], "docs.zip")
|
|
|
|
def test_create_duplicate_task_is_allowed(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="duplicate",
|
|
source="storage1/report.txt",
|
|
destination="storage1/report copy.txt",
|
|
)
|
|
|
|
task = self.repo.get_task(created["id"])
|
|
|
|
self.assertEqual(task["operation"], "duplicate")
|
|
self.assertEqual(task["status"], "queued")
|
|
|
|
def test_mark_cancelled_transitions_requested_download_task(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="download",
|
|
source="storage1/docs",
|
|
destination="docs.zip",
|
|
status="requested",
|
|
)
|
|
|
|
changed = self.repo.mark_cancelled(created["id"])
|
|
task = self.repo.get_task(created["id"])
|
|
|
|
self.assertTrue(changed)
|
|
self.assertEqual(task["status"], "cancelled")
|
|
self.assertIsNotNone(task["finished_at"])
|
|
|
|
def test_request_cancellation_moves_running_file_task_to_cancelling(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="copy",
|
|
source="storage1/docs/a.txt",
|
|
destination="storage1/docs-copy/a.txt",
|
|
)
|
|
self.repo.mark_running(
|
|
created["id"],
|
|
done_items=0,
|
|
total_items=2,
|
|
current_item="storage1/docs/a.txt",
|
|
)
|
|
|
|
task = self.repo.request_cancellation(created["id"])
|
|
|
|
self.assertIsNotNone(task)
|
|
self.assertEqual(task["status"], "cancelling")
|
|
self.assertEqual(task["current_item"], "storage1/docs/a.txt")
|
|
self.assertIsNone(task["finished_at"])
|
|
|
|
def test_request_cancellation_moves_queued_file_task_to_cancelled(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="delete",
|
|
source="storage1/docs/a.txt",
|
|
destination="",
|
|
)
|
|
|
|
task = self.repo.request_cancellation(created["id"])
|
|
|
|
self.assertIsNotNone(task)
|
|
self.assertEqual(task["status"], "cancelled")
|
|
self.assertIsNone(task["current_item"])
|
|
self.assertIsNotNone(task["finished_at"])
|
|
|
|
def test_finalize_cancelled_transitions_cancelling_task(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="move",
|
|
source="storage1/docs/a.txt",
|
|
destination="storage1/archive/a.txt",
|
|
)
|
|
self.repo.mark_running(
|
|
created["id"],
|
|
done_items=0,
|
|
total_items=3,
|
|
current_item="storage1/docs/a.txt",
|
|
)
|
|
self.repo.request_cancellation(created["id"])
|
|
|
|
changed = self.repo.finalize_cancelled(
|
|
created["id"],
|
|
done_items=1,
|
|
total_items=3,
|
|
)
|
|
task = self.repo.get_task(created["id"])
|
|
|
|
self.assertTrue(changed)
|
|
self.assertEqual(task["status"], "cancelled")
|
|
self.assertEqual(task["done_items"], 1)
|
|
self.assertEqual(task["total_items"], 3)
|
|
self.assertIsNone(task["current_item"])
|
|
self.assertIsNotNone(task["finished_at"])
|
|
|
|
def test_reconcile_incomplete_tasks_marks_non_terminal_failed(self) -> None:
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": "task-running",
|
|
"operation": "copy",
|
|
"status": "running",
|
|
"source": "storage1/a",
|
|
"destination": "storage2/a",
|
|
"created_at": "2026-03-10T09:00:00Z",
|
|
"started_at": "2026-03-10T09:00:01Z",
|
|
"current_item": "storage1/a",
|
|
}
|
|
)
|
|
self.repo.insert_task_for_testing(
|
|
{
|
|
"id": "task-completed",
|
|
"operation": "copy",
|
|
"status": "completed",
|
|
"source": "storage1/b",
|
|
"destination": "storage2/b",
|
|
"created_at": "2026-03-10T09:05:00Z",
|
|
"finished_at": "2026-03-10T09:06:00Z",
|
|
}
|
|
)
|
|
|
|
changed = self.repo.reconcile_incomplete_tasks()
|
|
|
|
running = self.repo.get_task("task-running")
|
|
completed = self.repo.get_task("task-completed")
|
|
self.assertEqual(changed, ["task-running"])
|
|
self.assertEqual(running["status"], "failed")
|
|
self.assertEqual(running["error_code"], "task_interrupted")
|
|
self.assertEqual(running["error_message"], "Task was interrupted before completion")
|
|
self.assertIsNone(running["current_item"])
|
|
self.assertIsNotNone(running["finished_at"])
|
|
self.assertEqual(completed["status"], "completed")
|
|
|
|
def test_reconcile_incomplete_tasks_removes_stale_artifact(self) -> None:
|
|
created = self.repo.create_task(
|
|
operation="download",
|
|
source="storage1/docs",
|
|
destination="docs.zip",
|
|
status="preparing",
|
|
)
|
|
self.repo.upsert_artifact(
|
|
task_id=created["id"],
|
|
file_path="/tmp/docs.zip.partial",
|
|
file_name="docs.zip",
|
|
expires_at="2026-03-10T10:30:00Z",
|
|
)
|
|
|
|
changed = self.repo.reconcile_incomplete_tasks()
|
|
|
|
task = self.repo.get_task(created["id"])
|
|
self.assertEqual(changed, [created["id"]])
|
|
self.assertEqual(task["status"], "failed")
|
|
self.assertIsNone(self.repo.get_artifact(created["id"]))
|
|
|
|
def test_migrates_legacy_tasks_schema_missing_source_destination(self) -> None:
|
|
legacy_db_path = Path(self.temp_dir.name) / "legacy.db"
|
|
conn = sqlite3.connect(legacy_db_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE tasks (
|
|
id TEXT PRIMARY KEY,
|
|
operation TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
repo = TaskRepository(str(legacy_db_path))
|
|
created = repo.create_task(
|
|
operation="move",
|
|
source="storage1/a.txt",
|
|
destination="storage2/a.txt",
|
|
)
|
|
|
|
self.assertEqual(created["operation"], "move")
|
|
self.assertEqual(created["source"], "storage1/a.txt")
|
|
self.assertEqual(created["destination"], "storage2/a.txt")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|