62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from biliup_next.core.models import Task, TaskContext
|
|
from biliup_next.modules.ingest.service import IngestService
|
|
|
|
|
|
class _FakeRepo:
|
|
def __init__(self, contexts: list[TaskContext]) -> None:
|
|
self.contexts = contexts
|
|
|
|
def find_recent_task_contexts(self, streamer: str) -> list[TaskContext]:
|
|
return [context for context in self.contexts if context.streamer == streamer]
|
|
|
|
|
|
class IngestSessionGroupingTests(unittest.TestCase):
|
|
def test_infer_session_key_groups_same_streamer_within_three_hours_to_earliest_title(self) -> None:
|
|
existing_context = TaskContext(
|
|
id=None,
|
|
task_id="task-1",
|
|
session_key="王海颖唱歌录播 04月04日 21时59分 p01 王海颖唱歌录播 04月04日 21时59分",
|
|
streamer="王海颖唱歌录播",
|
|
room_id=None,
|
|
source_title="王海颖唱歌录播 04月04日 21时59分 p01 王海颖唱歌录播 04月04日 21时59分",
|
|
segment_started_at="2026-04-04T21:59:00+08:00",
|
|
segment_duration_seconds=None,
|
|
full_video_bvid="BVFULL123",
|
|
created_at="2026-04-04T14:00:00+00:00",
|
|
updated_at="2026-04-04T14:00:00+00:00",
|
|
)
|
|
service = IngestService(registry=None, repo=_FakeRepo([existing_context])) # type: ignore[arg-type]
|
|
|
|
session_key, inherited_bvid = service._infer_session_key(
|
|
streamer="王海颖唱歌录播",
|
|
room_id=None,
|
|
segment_started_at="2026-04-05T00:30:00+08:00",
|
|
source_title="王海颖唱歌录播 04月05日 00时30分 p02 王海颖唱歌录播 04月05日 00时30分",
|
|
gap_minutes=60,
|
|
)
|
|
|
|
self.assertEqual(session_key, existing_context.session_key)
|
|
self.assertEqual(inherited_bvid, "BVFULL123")
|
|
|
|
def test_infer_session_key_uses_current_title_when_no_recent_context_matches(self) -> None:
|
|
service = IngestService(registry=None, repo=_FakeRepo([])) # type: ignore[arg-type]
|
|
|
|
session_key, inherited_bvid = service._infer_session_key(
|
|
streamer="王海颖唱歌录播",
|
|
room_id=None,
|
|
segment_started_at="2026-04-05T00:30:00+08:00",
|
|
source_title="王海颖唱歌录播 04月05日 00时30分 p02 王海颖唱歌录播 04月05日 00时30分",
|
|
gap_minutes=60,
|
|
)
|
|
|
|
self.assertEqual(session_key, "王海颖唱歌录播 04月05日 00时30分 p02 王海颖唱歌录播 04月05日 00时30分")
|
|
self.assertIsNone(inherited_bvid)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|