更新股票估值台账至2026-08-14

Cai

2026-08-16 326f521cec4496e9cb8e7528a0d40c99c13805ed

 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)