144 lines
6.8 KiB
Python
144 lines
6.8 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from biliup_next.app.task_actions import bind_full_video_action, merge_session_action, rebind_session_full_video_action
|
|
from biliup_next.core.models import Task, TaskContext
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, task: Task, context: TaskContext | None = None, contexts: list[TaskContext] | None = None) -> None:
|
|
self.task = task
|
|
self.context = context
|
|
self.contexts = contexts or ([] if context is None else [context])
|
|
self.task_context_upserts: list[TaskContext] = []
|
|
self.session_binding_upserts = []
|
|
self.updated_session_bvid: tuple[str, str, str] | None = None
|
|
|
|
def get_task(self, task_id: str) -> Task | None:
|
|
return self.task if task_id == self.task.id else None
|
|
|
|
def get_task_context(self, task_id: str) -> TaskContext | None:
|
|
return self.context if task_id == self.task.id else None
|
|
|
|
def upsert_task_context(self, context: TaskContext) -> None:
|
|
self.context = context
|
|
self.task_context_upserts.append(context)
|
|
|
|
def upsert_session_binding(self, binding) -> None: # type: ignore[no-untyped-def]
|
|
self.session_binding_upserts.append(binding)
|
|
|
|
def add_action_record(self, record) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def list_task_contexts_by_session_key(self, session_key: str) -> list[TaskContext]:
|
|
return [context for context in self.contexts if context.session_key == session_key]
|
|
|
|
def update_session_full_video_bvid(self, session_key: str, full_video_bvid: str, updated_at: str) -> int:
|
|
self.updated_session_bvid = (session_key, full_video_bvid, updated_at)
|
|
return len(self.list_task_contexts_by_session_key(session_key))
|
|
|
|
def list_task_contexts_by_source_title(self, source_title: str) -> list[TaskContext]:
|
|
return [context for context in self.contexts if context.source_title == source_title]
|
|
|
|
|
|
class TaskActionsTests(unittest.TestCase):
|
|
def test_bind_full_video_action_persists_context_binding_and_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
task = Task("task-1", "local_file", "/tmp/source.mp4", "task-title", "created", "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00")
|
|
repo = FakeRepo(task)
|
|
state = {
|
|
"repo": repo,
|
|
"settings": {"paths": {"session_dir": str(Path(tmpdir) / "session")}},
|
|
}
|
|
|
|
with patch("biliup_next.app.task_actions.ensure_initialized", return_value=state), patch(
|
|
"biliup_next.app.task_actions.record_task_action"
|
|
):
|
|
result = bind_full_video_action("task-1", " BV1234567890 ")
|
|
|
|
self.assertEqual(result["full_video_bvid"], "BV1234567890")
|
|
self.assertEqual(repo.context.full_video_bvid, "BV1234567890")
|
|
self.assertEqual(len(repo.session_binding_upserts), 1)
|
|
self.assertTrue(Path(result["path"]).exists())
|
|
self.assertEqual(Path(result["path"]).read_text(encoding="utf-8"), "BV1234567890")
|
|
|
|
def test_rebind_session_full_video_action_updates_binding_and_all_task_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
task = Task("task-1", "local_file", "/tmp/source.mp4", "task-title", "published", "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00")
|
|
context = TaskContext(
|
|
id=None,
|
|
task_id="task-1",
|
|
session_key="session-1",
|
|
streamer="streamer",
|
|
room_id="room",
|
|
source_title="task-title",
|
|
segment_started_at=None,
|
|
segment_duration_seconds=None,
|
|
full_video_bvid="BVOLD",
|
|
created_at="2026-01-01T00:00:00+00:00",
|
|
updated_at="2026-01-01T00:00:00+00:00",
|
|
)
|
|
repo = FakeRepo(task, context=context, contexts=[context])
|
|
state = {
|
|
"repo": repo,
|
|
"settings": {"paths": {"session_dir": str(Path(tmpdir) / "session")}},
|
|
}
|
|
|
|
with patch("biliup_next.app.task_actions.ensure_initialized", return_value=state), patch(
|
|
"biliup_next.app.task_actions.record_task_action"
|
|
):
|
|
result = rebind_session_full_video_action("session-1", "BVNEW1234567")
|
|
|
|
self.assertEqual(result["updated_count"], 1)
|
|
self.assertEqual(repo.context.full_video_bvid, "BVNEW1234567")
|
|
self.assertIsNotNone(repo.updated_session_bvid)
|
|
self.assertEqual(len(repo.session_binding_upserts), 1)
|
|
self.assertEqual(repo.session_binding_upserts[-1].full_video_bvid, "BVNEW1234567")
|
|
persisted_path = Path(result["tasks"][0]["path"])
|
|
self.assertTrue(persisted_path.exists())
|
|
self.assertEqual(persisted_path.read_text(encoding="utf-8"), "BVNEW1234567")
|
|
|
|
def test_merge_session_action_reuses_persist_path_for_inherited_bvid(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
task = Task("task-1", "local_file", "/tmp/source.mp4", "task-title", "created", "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00")
|
|
existing_context = TaskContext(
|
|
id=None,
|
|
task_id="existing-task",
|
|
session_key="session-1",
|
|
streamer="streamer",
|
|
room_id="room",
|
|
source_title="existing-title",
|
|
segment_started_at=None,
|
|
segment_duration_seconds=None,
|
|
full_video_bvid="BVINHERITED123",
|
|
created_at="2026-01-01T00:00:00+00:00",
|
|
updated_at="2026-01-01T00:00:00+00:00",
|
|
)
|
|
repo = FakeRepo(task, contexts=[existing_context])
|
|
state = {
|
|
"repo": repo,
|
|
"settings": {"paths": {"session_dir": str(Path(tmpdir) / "session")}},
|
|
}
|
|
|
|
with patch("biliup_next.app.task_actions.ensure_initialized", return_value=state), patch(
|
|
"biliup_next.app.task_actions.record_task_action"
|
|
):
|
|
result = merge_session_action("session-1", ["task-1"])
|
|
|
|
self.assertEqual(result["merged_count"], 1)
|
|
self.assertEqual(repo.context.full_video_bvid, "BVINHERITED123")
|
|
self.assertEqual(len(repo.session_binding_upserts), 1)
|
|
self.assertEqual(repo.session_binding_upserts[0].full_video_bvid, "BVINHERITED123")
|
|
self.assertIn("path", result["tasks"][0])
|
|
persisted_path = Path(result["tasks"][0]["path"])
|
|
self.assertTrue(persisted_path.exists())
|
|
self.assertEqual(persisted_path.read_text(encoding="utf-8"), "BVINHERITED123")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|