331 lines
14 KiB
Python
331 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from biliup_next.core.models import Artifact, Task, utc_now_iso
|
|
from biliup_next.infra.adapters.biliup_cli import BiliupCliAdapter
|
|
from biliup_next.modules.publish.providers.biliup_cli import BiliupCliPublishProvider
|
|
|
|
|
|
class _FakeBiliupAdapter:
|
|
def __init__(self) -> None:
|
|
self.optional_calls: list[dict] = []
|
|
self.run_calls: list[dict] = []
|
|
|
|
def run_optional(self, cmd: list[str], *, label: str, timeout_seconds: int | None = None, log_path: Path | None = None) -> None:
|
|
self.optional_calls.append(
|
|
{"cmd": cmd, "label": label, "timeout_seconds": timeout_seconds, "log_path": log_path}
|
|
)
|
|
|
|
def run(self, cmd: list[str], *, label: str, timeout_seconds: int | None = None, log_path: Path | None = None) -> subprocess.CompletedProcess[str]:
|
|
self.run_calls.append(
|
|
{"cmd": cmd, "label": label, "timeout_seconds": timeout_seconds, "log_path": log_path}
|
|
)
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='{"bvid":"BV1TEST12345"}', stderr="")
|
|
|
|
|
|
class BiliupCliAdapterTests(unittest.TestCase):
|
|
def test_run_writes_publish_log(self) -> None:
|
|
adapter = BiliupCliAdapter()
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_path = Path(tmpdir) / "publish.log"
|
|
result = adapter.run(
|
|
[sys.executable, "-c", "print('hello from biliup adapter')"],
|
|
label="adapter smoke",
|
|
timeout_seconds=5,
|
|
log_path=log_path,
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
content = log_path.read_text(encoding="utf-8")
|
|
self.assertIn("adapter smoke", content)
|
|
self.assertIn("timeout_seconds: 5", content)
|
|
self.assertIn("exit: 0", content)
|
|
self.assertIn("hello from biliup adapter", content)
|
|
|
|
|
|
class BiliupCliPublishProviderTests(unittest.TestCase):
|
|
def test_publish_uses_source_path_parent_when_task_title_differs(self) -> None:
|
|
adapter = _FakeBiliupAdapter()
|
|
provider = BiliupCliPublishProvider(adapter=adapter)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-id-dir"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-id",
|
|
source_type="bilibili_url",
|
|
source_path=str(work_dir / "task-id.mp4"),
|
|
title="display-title",
|
|
status="split_done",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song"}]}), encoding="utf-8")
|
|
upload_config = root / "upload_config.json"
|
|
upload_config.write_text("{}", encoding="utf-8")
|
|
clip_path = work_dir / "clip-1.mp4"
|
|
clip_path.write_text("fake", encoding="utf-8")
|
|
clip = Artifact(
|
|
id=None,
|
|
task_id=task.id,
|
|
artifact_type="clip_video",
|
|
path=str(clip_path),
|
|
metadata_json="{}",
|
|
created_at=utc_now_iso(),
|
|
)
|
|
|
|
record = provider.publish(
|
|
task,
|
|
[clip],
|
|
{
|
|
"session_dir": str(root),
|
|
"upload_config_file": str(upload_config),
|
|
"biliup_path": "runtime/biliup",
|
|
"cookie_file": "runtime/cookies.json",
|
|
"retry_count": 1,
|
|
"command_timeout_seconds": 123,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(record.bvid, "BV1TEST12345")
|
|
self.assertEqual(adapter.optional_calls[0]["log_path"], work_dir / "publish.log")
|
|
self.assertTrue((work_dir / "bvid.txt").exists())
|
|
self.assertTrue((work_dir / "upload_done.flag").exists())
|
|
|
|
def test_publish_passes_timeout_and_log_path(self) -> None:
|
|
adapter = _FakeBiliupAdapter()
|
|
provider = BiliupCliPublishProvider(adapter=adapter)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-1"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-1",
|
|
source_type="local_file",
|
|
source_path=str(work_dir / "source.mp4"),
|
|
title="task-1",
|
|
status="split_done",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song"}]}), encoding="utf-8")
|
|
upload_config = root / "upload_config.json"
|
|
upload_config.write_text("{}", encoding="utf-8")
|
|
clip_path = work_dir / "clip-1.mp4"
|
|
clip_path.write_text("fake", encoding="utf-8")
|
|
clip = Artifact(
|
|
id=None,
|
|
task_id=task.id,
|
|
artifact_type="clip_video",
|
|
path=str(clip_path),
|
|
metadata_json="{}",
|
|
created_at=utc_now_iso(),
|
|
)
|
|
|
|
record = provider.publish(
|
|
task,
|
|
[clip],
|
|
{
|
|
"session_dir": str(root),
|
|
"upload_config_file": str(upload_config),
|
|
"biliup_path": "runtime/biliup",
|
|
"cookie_file": "runtime/cookies.json",
|
|
"retry_count": 2,
|
|
"command_timeout_seconds": 123,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(record.bvid, "BV1TEST12345")
|
|
self.assertEqual(adapter.optional_calls[0]["timeout_seconds"], 123)
|
|
self.assertEqual(adapter.optional_calls[0]["log_path"], work_dir / "publish.log")
|
|
self.assertEqual(adapter.run_calls[0]["timeout_seconds"], 123)
|
|
self.assertEqual(adapter.run_calls[0]["log_path"], work_dir / "publish.log")
|
|
self.assertTrue((work_dir / "bvid.txt").exists())
|
|
self.assertTrue((work_dir / "upload_done.flag").exists())
|
|
|
|
def test_extract_bvid_supports_rust_debug_string_format(self) -> None:
|
|
provider = BiliupCliPublishProvider()
|
|
|
|
output = 'ResponseData { code: 0, data: Some(Object {"bvid": String("BV1N5DrBQEBg")}), message: "0" }'
|
|
|
|
self.assertEqual(provider._extract_bvid(output), "BV1N5DrBQEBg")
|
|
|
|
def test_publish_does_not_reuse_stale_bvid_without_upload_done_flag(self) -> None:
|
|
adapter = _FakeBiliupAdapter()
|
|
adapter.run = lambda cmd, *, label, timeout_seconds=None, log_path=None: subprocess.CompletedProcess( # type: ignore[method-assign]
|
|
cmd, 0, stdout='ResponseData { code: 0, data: Some(Object {"bvid": String("BV1NEW1234567")}) }', stderr=""
|
|
)
|
|
provider = BiliupCliPublishProvider(adapter=adapter)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-1"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-1",
|
|
source_type="local_file",
|
|
source_path=str(work_dir / "source.mp4"),
|
|
title="task-1",
|
|
status="split_done",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song"}]}), encoding="utf-8")
|
|
(work_dir / "bvid.txt").write_text("BVOLD1234567", encoding="utf-8")
|
|
upload_config = root / "upload_config.json"
|
|
upload_config.write_text("{}", encoding="utf-8")
|
|
clip_path = work_dir / "clip-1.mp4"
|
|
clip_path.write_text("fake", encoding="utf-8")
|
|
clip = Artifact(
|
|
id=None,
|
|
task_id=task.id,
|
|
artifact_type="clip_video",
|
|
path=str(clip_path),
|
|
metadata_json="{}",
|
|
created_at=utc_now_iso(),
|
|
)
|
|
|
|
record = provider.publish(
|
|
task,
|
|
[clip],
|
|
{
|
|
"session_dir": str(root),
|
|
"upload_config_file": str(upload_config),
|
|
"biliup_path": "runtime/biliup",
|
|
"cookie_file": "runtime/cookies.json",
|
|
"retry_count": 2,
|
|
"command_timeout_seconds": 123,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(record.bvid, "BV1NEW1234567")
|
|
self.assertEqual((work_dir / "bvid.txt").read_text(encoding="utf-8"), "BV1NEW1234567")
|
|
|
|
def test_publish_resumes_append_when_bvid_exists_without_upload_done(self) -> None:
|
|
adapter = _FakeBiliupAdapter()
|
|
provider = BiliupCliPublishProvider(adapter=adapter)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-1"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-1",
|
|
source_type="local_file",
|
|
source_path=str(work_dir / "source.mp4"),
|
|
title="task-1",
|
|
status="split_done",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song"}]}), encoding="utf-8")
|
|
(work_dir / "bvid.txt").write_text("BV1RESUME1234", encoding="utf-8")
|
|
(work_dir / "publish_progress.json").write_text(
|
|
json.dumps({"bvid": "BV1RESUME1234", "completed_append_batches": []}),
|
|
encoding="utf-8",
|
|
)
|
|
upload_config = root / "upload_config.json"
|
|
upload_config.write_text("{}", encoding="utf-8")
|
|
clips = []
|
|
for index in range(1, 11):
|
|
clip_path = work_dir / f"clip-{index}.mp4"
|
|
clip_path.write_text("fake", encoding="utf-8")
|
|
clips.append(
|
|
Artifact(
|
|
id=None,
|
|
task_id=task.id,
|
|
artifact_type="clip_video",
|
|
path=str(clip_path),
|
|
metadata_json="{}",
|
|
created_at=utc_now_iso(),
|
|
)
|
|
)
|
|
|
|
with patch("biliup_next.modules.publish.providers.biliup_cli.time.sleep", return_value=None):
|
|
record = provider.publish(
|
|
task,
|
|
clips,
|
|
{
|
|
"session_dir": str(root),
|
|
"upload_config_file": str(upload_config),
|
|
"biliup_path": "runtime/biliup",
|
|
"cookie_file": "runtime/cookies.json",
|
|
"retry_count": 2,
|
|
"command_timeout_seconds": 123,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(record.bvid, "BV1RESUME1234")
|
|
self.assertEqual(len(adapter.run_calls), 1)
|
|
self.assertIn("append", adapter.run_calls[0]["cmd"])
|
|
self.assertIn("BV1RESUME1234", adapter.run_calls[0]["cmd"])
|
|
self.assertTrue((work_dir / "upload_done.flag").exists())
|
|
|
|
def test_publish_creates_progress_from_existing_bvid_for_append_resume(self) -> None:
|
|
adapter = _FakeBiliupAdapter()
|
|
provider = BiliupCliPublishProvider(adapter=adapter)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
work_dir = root / "task-1"
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
task = Task(
|
|
id="task-1",
|
|
source_type="local_file",
|
|
source_path=str(work_dir / "source.mp4"),
|
|
title="task-1",
|
|
status="split_done",
|
|
created_at=utc_now_iso(),
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
(work_dir / "songs.txt").write_text("00:00:00 Test Song - Tester\n", encoding="utf-8")
|
|
(work_dir / "songs.json").write_text(json.dumps({"songs": [{"title": "Test Song"}]}), encoding="utf-8")
|
|
(work_dir / "bvid.txt").write_text("BV1RESUME1234", encoding="utf-8")
|
|
upload_config = root / "upload_config.json"
|
|
upload_config.write_text("{}", encoding="utf-8")
|
|
clips = []
|
|
for index in range(1, 11):
|
|
clip_path = work_dir / f"clip-{index}.mp4"
|
|
clip_path.write_text("fake", encoding="utf-8")
|
|
clips.append(
|
|
Artifact(
|
|
id=None,
|
|
task_id=task.id,
|
|
artifact_type="clip_video",
|
|
path=str(clip_path),
|
|
metadata_json="{}",
|
|
created_at=utc_now_iso(),
|
|
)
|
|
)
|
|
|
|
with patch("biliup_next.modules.publish.providers.biliup_cli.time.sleep", return_value=None):
|
|
record = provider.publish(
|
|
task,
|
|
clips,
|
|
{
|
|
"session_dir": str(root),
|
|
"upload_config_file": str(upload_config),
|
|
"biliup_path": "runtime/biliup",
|
|
"cookie_file": "runtime/cookies.json",
|
|
"retry_count": 2,
|
|
"command_timeout_seconds": 123,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(record.bvid, "BV1RESUME1234")
|
|
self.assertEqual(len(adapter.run_calls), 1)
|
|
self.assertIn("append", adapter.run_calls[0]["cmd"])
|
|
self.assertFalse((work_dir / "publish_progress.json").exists())
|
|
self.assertTrue((work_dir / "upload_done.flag").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|