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()