From cb8756d4cad0474d3475cfe1f8989869bf2d792c Mon Sep 17 00:00:00 2001
From: MB-X Bilibili Pipeline <mbx-bili-pipeline@localhost>
Date: Sat, 05 Sep 2026 22:30:52 +0800
Subject: [PATCH] chore(project-info): archive Bilibili dynamic a1574f7e7ab4
---
dev/project-dev/test/test_transcribe_media.py | 111 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 111 insertions(+), 0 deletions(-)
diff --git a/dev/project-dev/test/test_transcribe_media.py b/dev/project-dev/test/test_transcribe_media.py
index ba72522..d6304a4 100644
--- a/dev/project-dev/test/test_transcribe_media.py
+++ b/dev/project-dev/test/test_transcribe_media.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
+import os
import subprocess
import sys
import tempfile
@@ -138,11 +139,121 @@
if Path(command[0]).name.lower().startswith("ffmpeg")
)
self.assertIn("-n", ffmpeg_command)
+ self.assertEqual(
+ ffmpeg_command[ffmpeg_command.index("-threads") + 1],
+ str(transcribe_media.MEDIA_DECODE_THREADS),
+ )
+ self.assertLess(ffmpeg_command.index("-threads"), ffmpeg_command.index("-i"))
self.assertEqual(ffmpeg_command[ffmpeg_command.index("-map") + 1], "0:a:0")
self.assertEqual(ffmpeg_command[ffmpeg_command.index("-ac") + 1], "1")
self.assertEqual(ffmpeg_command[ffmpeg_command.index("-ar") + 1], "16000")
self.assertEqual(ffmpeg_command[ffmpeg_command.index("-c:a") + 1], "flac")
+ def test_default_process_runner_uses_below_normal_creation_flags(self) -> None:
+ completed = subprocess.CompletedProcess(["tool"], 0, "", "")
+ with patch.object(transcribe_media, "_is_windows", return_value=True), patch.object(
+ transcribe_media.subprocess, "run", return_value=completed
+ ) as run:
+ self.assertIs(transcribe_media._run_process(["tool"]), completed)
+ self.assertEqual(
+ run.call_args.kwargs["creationflags"],
+ getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0x00004000),
+ )
+
+ @unittest.skipUnless(os.name == "nt", "Windows named mutex and priority are required")
+ def test_named_mutex_blocks_second_process_and_restores_priority(self) -> None:
+ import ctypes
+
+ kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+ kernel32.GetCurrentProcess.restype = ctypes.c_void_p
+ kernel32.GetPriorityClass.argtypes = (ctypes.c_void_p,)
+ kernel32.GetPriorityClass.restype = ctypes.c_uint32
+ process_handle = kernel32.GetCurrentProcess()
+ original_priority = int(kernel32.GetPriorityClass(process_handle))
+ child_code = (
+ "import sys\n"
+ f"sys.path.insert(0, {str(PROJECT_DEV)!r})\n"
+ "import transcribe_media as module\n"
+ "try:\n"
+ " with module.media_host_guard():\n"
+ " pass\n"
+ "except module.MediaHostGuardError:\n"
+ " raise SystemExit(23)\n"
+ )
+ with transcribe_media.media_host_guard():
+ self.assertEqual(int(kernel32.GetPriorityClass(process_handle)), 0x00004000)
+ busy = subprocess.run([sys.executable, "-c", child_code], check=False)
+ self.assertEqual(busy.returncode, 23)
+ self.assertEqual(int(kernel32.GetPriorityClass(process_handle)), original_priority)
+ free = subprocess.run([sys.executable, "-c", child_code], check=False)
+ self.assertEqual(free.returncode, 0)
+
+ @unittest.skipUnless(os.name == "nt", "Windows named mutex is required")
+ def test_both_real_clis_reject_busy_before_path_or_external_actions(self) -> None:
+ scripts = (
+ (PROJECT_DEV / "transcribe_media.py", 1),
+ (PROJECT_DEV / "extract_ppt_slides.py", 2),
+ )
+ with transcribe_media.media_host_guard():
+ for script, expected_code in scripts:
+ with self.subTest(script=script.name):
+ result = subprocess.run(
+ [sys.executable, str(script), "definitely-missing-media.mp4"],
+ capture_output=True,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ env={**os.environ, "PYTHONIOENCODING": "utf-8"},
+ check=False,
+ )
+ self.assertEqual(result.returncode, expected_code)
+ self.assertIn("已有重型媒体任务运行", result.stderr)
+ self.assertNotIn("视频文件不存在", result.stderr)
+ self.assertNotIn("输入视频不存在", result.stderr)
+
+ def test_win32_guard_failures_close_every_created_handle(self) -> None:
+ def kernel(**values):
+ defaults = {
+ "CreateMutexW": unittest.mock.Mock(return_value=101),
+ "WaitForSingleObject": unittest.mock.Mock(return_value=0),
+ "ReleaseMutex": unittest.mock.Mock(return_value=True),
+ "CloseHandle": unittest.mock.Mock(return_value=True),
+ "GetCurrentProcess": unittest.mock.Mock(return_value=202),
+ "GetPriorityClass": unittest.mock.Mock(return_value=0x20),
+ "SetPriorityClass": unittest.mock.Mock(return_value=True),
+ }
+ defaults.update(values)
+ return types.SimpleNamespace(**defaults)
+
+ scenarios = (
+ ("wait", kernel(WaitForSingleObject=unittest.mock.Mock(return_value=0x102))),
+ ("get", kernel(GetPriorityClass=unittest.mock.Mock(return_value=0))),
+ ("set", kernel(SetPriorityClass=unittest.mock.Mock(return_value=False))),
+ )
+ for name, fake in scenarios:
+ with self.subTest(name=name), patch.object(
+ transcribe_media, "_is_windows", return_value=True
+ ), patch.object(transcribe_media, "_windows_kernel32", return_value=fake), patch.object(
+ transcribe_media, "_win32_last_error", return_value=5
+ ):
+ with self.assertRaises(transcribe_media.MediaHostGuardError):
+ with transcribe_media.media_host_guard():
+ self.fail("guard must not yield")
+ fake.CloseHandle.assert_called_once_with(101)
+ if name == "wait":
+ fake.ReleaseMutex.assert_not_called()
+ else:
+ fake.ReleaseMutex.assert_called_once_with(101)
+
+ create_failed = kernel(CreateMutexW=unittest.mock.Mock(return_value=0))
+ with patch.object(transcribe_media, "_is_windows", return_value=True), patch.object(
+ transcribe_media, "_windows_kernel32", return_value=create_failed
+ ), patch.object(transcribe_media, "_win32_last_error", return_value=5):
+ with self.assertRaises(transcribe_media.MediaHostGuardError):
+ with transcribe_media.media_host_guard():
+ self.fail("guard must not yield")
+ create_failed.CloseHandle.assert_not_called()
+
def test_existing_output_is_not_overwritten(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
--
Gitblit v1.9.3