import importlib.util
|
import json
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from pathlib import Path
|
from unittest import mock
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "bili_video_download_bridge.py"
|
SPEC = importlib.util.spec_from_file_location("bili_video_download_bridge", MODULE_PATH)
|
bridge = importlib.util.module_from_spec(SPEC)
|
assert SPEC.loader is not None
|
sys.modules[SPEC.name] = bridge
|
SPEC.loader.exec_module(bridge)
|
|
|
def completed(stdout="", stderr="", returncode=0):
|
return subprocess.CompletedProcess([], returncode, stdout, stderr)
|
|
|
class BatchInputTests(unittest.TestCase):
|
def test_example_loads_two_exact_canonical_targets(self):
|
example = MODULE_PATH.with_name("bili_video_download_bridge.example.json")
|
batch = bridge.load_batch(example)
|
self.assertEqual("qingfengpushangq-20260805", batch.batch_id)
|
self.assertEqual(["BV1DVMX6XEPq", "BV1HA3o6oEJJ"], [item.bvid for item in batch.items])
|
self.assertTrue(all(not url.endswith(".webp") for url in (item.source_url for item in batch.items)))
|
|
def test_rejects_noncanonical_query_cover_and_duplicate(self):
|
base = {
|
"schema_version": "1.0",
|
"batch_id": "batch-1",
|
"items": [
|
{
|
"bvid": "BV1DVMX6XEPq",
|
"source_url": "https://www.bilibili.com/video/BV1DVMX6XEPq?token=secret",
|
"published_at": "2026-08-03T19:39:03+08:00",
|
}
|
],
|
}
|
with tempfile.TemporaryDirectory() as tmp:
|
path = Path(tmp) / "batch.json"
|
path.write_text(json.dumps(base), encoding="utf-8")
|
with self.assertRaises(bridge.InputError):
|
bridge.load_batch(path)
|
base["items"][0]["source_url"] = "https://example.test/cover.webp"
|
path.write_text(json.dumps(base), encoding="utf-8")
|
with self.assertRaises(bridge.InputError):
|
bridge.load_batch(path)
|
base["items"][0]["source_url"] = "https://www.bilibili.com/video/BV1DVMX6XEPq"
|
base["items"].append(dict(base["items"][0]))
|
path.write_text(json.dumps(base), encoding="utf-8")
|
with self.assertRaises(bridge.InputError):
|
bridge.load_batch(path)
|
|
|
class CommandTests(unittest.TestCase):
|
def setUp(self):
|
self.item = bridge.VideoItem(
|
bvid="BV1DVMX6XEPq",
|
source_url="https://www.bilibili.com/video/BV1DVMX6XEPq",
|
published_at="2026-08-03T19:39:03+08:00",
|
)
|
|
def test_probe_command_forbids_implicit_credentials_and_download(self):
|
command = bridge.build_probe_command("yt-dlp", self.item)
|
self.assertIn("--ignore-config", command)
|
self.assertIn("--no-cookies", command)
|
self.assertIn("--no-cookies-from-browser", command)
|
self.assertIn("--simulate", command)
|
self.assertNotIn("--netrc", command)
|
self.assertNotIn("--cookies", command)
|
self.assertNotIn("--cookies-from-browser", command)
|
self.assertEqual(self.item.source_url, command[-1])
|
|
def test_download_command_requests_complete_merged_media_in_staging(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
stage = Path(tmp) / "stage"
|
result = stage / "completed-path.txt"
|
command = bridge.build_download_command("yt-dlp", "ffmpeg", self.item, stage, result)
|
self.assertEqual("bv*+ba/b", command[command.index("--format") + 1])
|
self.assertEqual("mp4", command[command.index("--merge-output-format") + 1])
|
self.assertEqual("2", command[command.index("--retries") + 1])
|
self.assertIn("--no-overwrites", command)
|
self.assertIn("after_move:filepath", command)
|
self.assertNotIn("m4s", " ".join(command).casefold())
|
|
def test_probe_rejects_remote_identity_mismatch(self):
|
runner = mock.Mock(return_value=completed(json.dumps({"id": "BV0000000000", "duration": 12})))
|
with self.assertRaises(bridge.AccessError):
|
bridge.probe_item(self.item, "yt-dlp", runner=runner)
|
|
def test_successful_probe_is_labeled_metadata_only(self):
|
runner = mock.Mock(
|
return_value=completed(
|
json.dumps({"id": self.item.bvid, "duration": 1206.067, "is_live": False})
|
)
|
)
|
result = bridge.probe_item(self.item, "yt-dlp", runner=runner)
|
self.assertEqual("METADATA_PASS_ONLY", result["probe"])
|
|
def test_probe_rejects_playlist_and_auth_boundaries(self):
|
cases = [
|
{"id": self.item.bvid, "duration": 100, "_type": "playlist"},
|
{"id": self.item.bvid, "duration": 100, "entries": []},
|
{"id": self.item.bvid, "duration": 100, "availability": "needs_auth"},
|
{"id": self.item.bvid, "duration": 100, "availability": "premium_only"},
|
]
|
for metadata in cases:
|
with self.subTest(metadata=metadata):
|
runner = mock.Mock(return_value=completed(json.dumps(metadata)))
|
with self.assertRaises(bridge.AccessError):
|
bridge.probe_item(self.item, "yt-dlp", runner=runner)
|
|
def test_process_diagnostic_redacts_urls(self):
|
safe = bridge._safe_process_error("failed https://signed.example/x?token=secret")
|
self.assertNotIn("secret", safe)
|
self.assertIn("[URL_REDACTED]", safe)
|
|
def test_browser_file_command_has_no_cookie_or_download_switch(self):
|
args = bridge._parser().parse_args(
|
[
|
"--input",
|
"batch.json",
|
"accept-browser-file",
|
"--bvid",
|
self.item.bvid,
|
"--media-file",
|
"complete.mp4",
|
"--destination",
|
"destination",
|
]
|
)
|
self.assertEqual("accept-browser-file", args.command)
|
self.assertFalse(hasattr(args, "browser_profile"))
|
self.assertFalse(hasattr(args, "cookies"))
|
|
|
class ValidationAndPublishTests(unittest.TestCase):
|
def setUp(self):
|
self.item = bridge.VideoItem(
|
bvid="BV1DVMX6XEPq",
|
source_url="https://www.bilibili.com/video/BV1DVMX6XEPq",
|
published_at="2026-08-03T19:39:03+08:00",
|
)
|
|
def test_ffprobe_requires_video_and_audio_and_rejects_cover(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
video = Path(tmp) / "x.mp4"
|
video.write_bytes(b"media")
|
valid = {
|
"streams": [
|
{"codec_type": "video", "codec_name": "h264"},
|
{"codec_type": "audio", "codec_name": "aac"},
|
],
|
"format": {"duration": "10.5", "format_name": "mov,mp4"},
|
}
|
facts = bridge.probe_media_file(
|
video, "ffprobe", runner=mock.Mock(return_value=completed(json.dumps(valid)))
|
)
|
self.assertEqual("h264", facts.video_codec)
|
valid["streams"] = valid["streams"][:1]
|
with self.assertRaises(bridge.ValidationError):
|
bridge.probe_media_file(
|
video, "ffprobe", runner=mock.Mock(return_value=completed(json.dumps(valid)))
|
)
|
cover = Path(tmp) / "cover.webp"
|
cover.write_bytes(b"RIFF")
|
with self.assertRaises(bridge.ValidationError):
|
bridge.probe_media_file(cover, "ffprobe", runner=mock.Mock())
|
|
def test_publish_create_new_hash_mapping_and_collision(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
source = root / "staging.mp4"
|
source.write_bytes(b"complete-video-and-audio")
|
destination = root / "destination"
|
destination.mkdir()
|
facts = bridge.MediaFacts(12.5, "mov,mp4", "h264", "aac")
|
result = bridge.publish_validated_media(
|
source,
|
destination,
|
self.item,
|
{"title": "test title", "duration_seconds": 12.5},
|
facts,
|
)
|
media = destination / "BV1DVMX6XEPq.mp4"
|
mapping = destination / "BV1DVMX6XEPq.download.json"
|
self.assertTrue(media.is_file())
|
self.assertTrue(mapping.is_file())
|
self.assertEqual(bridge.sha256_file(source), result["sha256"])
|
self.assertEqual(result, json.loads(mapping.read_text(encoding="utf-8")))
|
with self.assertRaises(bridge.CollisionError):
|
bridge.publish_validated_media(
|
source,
|
destination,
|
self.item,
|
{"title": "test title", "duration_seconds": 12.5},
|
facts,
|
)
|
|
def test_publish_rolls_back_media_when_mapping_create_fails(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
source = root / "staging.mp4"
|
source.write_bytes(b"complete-video-and-audio")
|
destination = root / "destination"
|
destination.mkdir()
|
facts = bridge.MediaFacts(12.5, "mov,mp4", "h264", "aac")
|
with mock.patch.object(bridge, "_write_json_create_new", side_effect=KeyboardInterrupt):
|
with self.assertRaises(KeyboardInterrupt):
|
bridge.publish_validated_media(
|
source,
|
destination,
|
self.item,
|
{"title": "test title", "duration_seconds": 12.5},
|
facts,
|
)
|
|
def test_duration_consistency_allows_mux_drift_and_rejects_preview(self):
|
accepted = bridge.validate_duration_consistency(1206.067, 1206.067664)
|
self.assertLess(accepted["duration_delta_seconds"], accepted["duration_tolerance_seconds"])
|
tolerance = max(3.0, 4000.0 * 0.001)
|
bridge.validate_duration_consistency(4000.0, 4000.0 + tolerance)
|
with self.assertRaises(bridge.ValidationError):
|
bridge.validate_duration_consistency(4000.0, 4000.0 + tolerance + 0.000001)
|
with self.assertRaisesRegex(bridge.ValidationError, "possible preview or wrong part"):
|
bridge.validate_duration_consistency(3133.95, 600.133313)
|
self.assertEqual([], list(destination.iterdir()))
|
|
|
class BatchFlowTests(unittest.TestCase):
|
def test_download_item_stub_closes_complete_media_mapping_and_cleanup(self):
|
item = bridge.VideoItem(
|
"BV1DVMX6XEPq",
|
"https://www.bilibili.com/video/BV1DVMX6XEPq",
|
"2026-08-03T19:39:03+08:00",
|
)
|
|
def runner(command, **_kwargs):
|
if "--dump-single-json" in command:
|
return completed(
|
json.dumps(
|
{
|
"id": item.bvid,
|
"title": "A股缩量磨底阶段。",
|
"duration": 1206.067,
|
"is_live": False,
|
}
|
)
|
)
|
if "--format" in command:
|
stage = Path(command[command.index("--paths") + 1])
|
media = stage / f"{item.bvid}.mp4"
|
media.write_bytes(b"stub-complete-media")
|
path_file = Path(command[command.index("--print-to-file") + 2])
|
path_file.write_text(str(media), encoding="utf-8")
|
return completed()
|
if "-show_entries" in command:
|
return completed(
|
json.dumps(
|
{
|
"streams": [
|
{"codec_type": "video", "codec_name": "h264"},
|
{"codec_type": "audio", "codec_name": "aac"},
|
],
|
"format": {"duration": "1206.067", "format_name": "mov,mp4"},
|
}
|
)
|
)
|
raise AssertionError(command)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
destination = Path(tmp)
|
result = bridge.download_item(
|
item,
|
destination,
|
"batch-1",
|
"yt-dlp",
|
"ffmpeg",
|
"ffprobe",
|
runner=runner,
|
)
|
self.assertEqual("COMPLETE", result["status"])
|
self.assertTrue((destination / f"{item.bvid}.mp4").is_file())
|
self.assertTrue((destination / f"{item.bvid}.download.json").is_file())
|
self.assertFalse((destination / ".bili-download-staging").exists())
|
|
def test_download_batch_continues_after_one_item_failure(self):
|
items = (
|
bridge.VideoItem("BV1DVMX6XEPq", "https://www.bilibili.com/video/BV1DVMX6XEPq", "2026-08-03T19:39:03+08:00"),
|
bridge.VideoItem("BV1HA3o6oEJJ", "https://www.bilibili.com/video/BV1HA3o6oEJJ", "2026-08-02T21:11:11+08:00"),
|
)
|
batch = bridge.BatchSpec("batch-1", items)
|
with tempfile.TemporaryDirectory() as tmp:
|
with mock.patch.object(
|
bridge,
|
"download_item",
|
side_effect=[bridge.AccessError("public access denied"), {"status": "COMPLETE"}],
|
) as mocked:
|
results, failures = bridge.download_batch(
|
batch, Path(tmp), "yt-dlp", "ffmpeg", "ffprobe", runner=mock.Mock()
|
)
|
self.assertEqual(2, mocked.call_count)
|
self.assertEqual(1, failures)
|
self.assertEqual(["FAIL", "COMPLETE"], [item["status"] for item in results])
|
|
def test_preview_duration_mismatch_never_publishes_and_cleans_staging(self):
|
item = bridge.VideoItem(
|
"BV1HA3o6oEJJ",
|
"https://www.bilibili.com/video/BV1HA3o6oEJJ",
|
"2026-08-02T21:11:11+08:00",
|
)
|
|
def runner(command, **_kwargs):
|
if "--dump-single-json" in command:
|
return completed(
|
json.dumps(
|
{
|
"id": item.bvid,
|
"title": "full remote video",
|
"duration": 3133.95,
|
"is_live": False,
|
}
|
)
|
)
|
if "--format" in command:
|
stage = Path(command[command.index("--paths") + 1])
|
media = stage / f"{item.bvid}.mp4"
|
media.write_bytes(b"premium-preview-only")
|
path_file = Path(command[command.index("--print-to-file") + 2])
|
path_file.write_text(str(media), encoding="utf-8")
|
return completed()
|
if "-show_entries" in command:
|
return completed(
|
json.dumps(
|
{
|
"streams": [
|
{"codec_type": "video", "codec_name": "h264"},
|
{"codec_type": "audio", "codec_name": "aac"},
|
],
|
"format": {"duration": "600.133313", "format_name": "mov,mp4"},
|
}
|
)
|
)
|
raise AssertionError(command)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
destination = Path(tmp)
|
with self.assertRaisesRegex(bridge.ValidationError, "possible preview or wrong part"):
|
bridge.download_item(
|
item,
|
destination,
|
"batch-1",
|
"yt-dlp",
|
"ffmpeg",
|
"ffprobe",
|
runner=runner,
|
)
|
self.assertEqual([], list(destination.iterdir()))
|
|
|
class BrowserFileHandoffTests(unittest.TestCase):
|
def setUp(self):
|
self.item = bridge.VideoItem(
|
"BV1HA3o6oEJJ",
|
"https://www.bilibili.com/video/BV1HA3o6oEJJ",
|
"2026-08-02T21:11:11+08:00",
|
)
|
|
def runner_with_duration(self, local_duration):
|
def runner(command, **_kwargs):
|
if "--dump-single-json" in command:
|
self.assertIn("--no-cookies", command)
|
self.assertIn("--no-cookies-from-browser", command)
|
return completed(
|
json.dumps(
|
{
|
"id": self.item.bvid,
|
"title": "流水淘沙不暂停，前波未灭后波生。（周复盘）",
|
"duration": 3133.95,
|
"is_live": False,
|
}
|
)
|
)
|
if "-show_entries" in command:
|
return completed(
|
json.dumps(
|
{
|
"streams": [
|
{"codec_type": "video", "codec_name": "h264"},
|
{"codec_type": "audio", "codec_name": "aac"},
|
],
|
"format": {
|
"duration": str(local_duration),
|
"format_name": "mov,mp4",
|
},
|
}
|
)
|
)
|
raise AssertionError(command)
|
|
return mock.Mock(side_effect=runner)
|
|
def test_accept_browser_file_publishes_copy_and_keeps_source_unchanged(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
source = root / "browser-download.mp4"
|
source.write_bytes(b"complete-authorized-video-with-audio")
|
source_before = (source.read_bytes(), source.stat().st_mtime_ns)
|
destination = root / "destination"
|
destination.mkdir()
|
runner = self.runner_with_duration(3133.9504)
|
result = bridge.accept_browser_file(
|
self.item,
|
source,
|
destination,
|
"batch-1",
|
"yt-dlp",
|
"ffprobe",
|
runner=runner,
|
sleeper=lambda _seconds: None,
|
)
|
self.assertEqual("COMPLETE", result["status"])
|
self.assertEqual("authorized_browser_file_handoff", result["acquisition_mode"])
|
self.assertEqual(result["sha256"], result["handoff_source_sha256"])
|
self.assertEqual(source_before, (source.read_bytes(), source.stat().st_mtime_ns))
|
published = destination / f"{self.item.bvid}.mp4"
|
mapping = destination / f"{self.item.bvid}.download.json"
|
self.assertTrue(published.is_file())
|
self.assertTrue(mapping.is_file())
|
self.assertEqual(source.read_bytes(), published.read_bytes())
|
self.assertFalse((destination / ".bili-download-staging").exists())
|
commands = [call.args[0] for call in runner.call_args_list]
|
self.assertEqual(2, len(commands))
|
self.assertTrue(all("--cookies-from-browser" not in command for command in commands))
|
self.assertTrue(all("--cookies" not in command for command in commands))
|
|
def test_rejects_partial_and_growing_browser_files(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
partial = root / "video.mp4.crdownload"
|
partial.write_bytes(b"partial")
|
with self.assertRaisesRegex(bridge.ValidationError, "incomplete media suffix"):
|
bridge.validate_stable_local_media(partial, sleeper=lambda _seconds: None)
|
|
growing = root / "growing-video.mp4"
|
growing.write_bytes(b"first")
|
|
def grow(_seconds):
|
with growing.open("ab") as target:
|
target.write(b"more")
|
|
with self.assertRaisesRegex(bridge.ValidationError, "not stable"):
|
bridge.validate_stable_local_media(growing, sleeper=grow)
|
|
def test_preview_handoff_fails_before_publication_and_cleans_stage(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
source = root / "browser-download.mp4"
|
source.write_bytes(b"preview-only")
|
destination = root / "destination"
|
destination.mkdir()
|
with self.assertRaisesRegex(bridge.ValidationError, "possible preview or wrong part"):
|
bridge.accept_browser_file(
|
self.item,
|
source,
|
destination,
|
"batch-1",
|
"yt-dlp",
|
"ffprobe",
|
runner=self.runner_with_duration(600.133313),
|
sleeper=lambda _seconds: None,
|
)
|
self.assertEqual([], list(destination.iterdir()))
|
self.assertEqual(b"preview-only", source.read_bytes())
|
|
def test_base_exception_after_copy_cleans_stage_and_keeps_source(self):
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
source = root / "browser-download.mp4"
|
source.write_bytes(b"complete-media")
|
destination = root / "destination"
|
destination.mkdir()
|
runner = self.runner_with_duration(3133.95)
|
with mock.patch.object(bridge, "probe_media_file", side_effect=KeyboardInterrupt):
|
with self.assertRaises(KeyboardInterrupt):
|
bridge.accept_browser_file(
|
self.item,
|
source,
|
destination,
|
"batch-1",
|
"yt-dlp",
|
"ffprobe",
|
runner=runner,
|
sleeper=lambda _seconds: None,
|
)
|
self.assertEqual([], list(destination.iterdir()))
|
self.assertEqual(b"complete-media", source.read_bytes())
|
|
def test_browser_handoff_requires_exact_batch_bvid(self):
|
batch = bridge.BatchSpec("batch-1", (self.item,))
|
self.assertIs(self.item, bridge._find_batch_item(batch, self.item.bvid))
|
with self.assertRaises(bridge.InputError):
|
bridge._find_batch_item(batch, "BV1DVMX6XEPq")
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
